From 334edbd5808e09604e76e58bee88b16213b113d1 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Sat, 18 Apr 2026 03:10:19 -0700 Subject: [PATCH] fix(server,console): kind filter on saved-workstreams + closed coords on landing (#380) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(server,console): kind filter on saved-workstreams + closed coords on landing Two independent bugs folded into one hotfix: 1. Coordinators leaking into the interactive UI's "saved workstreams" sidebar. ``list_workstreams_with_history`` (SQLite + postgres) was kind-agnostic — every coordinator row with conversation history came back alongside interactive rows, and ``list_saved_workstreams`` serialized them uniformly with no kind field so the interactive UI rendered coordinators as regular interactive entries. Fix: add optional ``kind: WorkstreamKind | str | None = None`` kwarg on ``list_workstreams_with_history`` (storage protocol + both backends + the ``turnstone.core.memory`` helper). Pass ``kind=WorkstreamKind.INTERACTIVE`` from the /v1/api/workstreams/saved handler so the interactive surface only sees interactive rows. Default ``None`` preserves legacy all-kinds behaviour for any other caller that wants both. 2. Closed coordinators vanish from the console landing page. ``_coordinator_rows`` in console/server.py built dashboard rows exclusively from the in-memory ``CoordinatorManager`` registry, which pops rows on ``close()``. The persisted storage row stays (state='closed') but never reached the landing-page poller at /v1/api/cluster/workstreams?node=console. Fix: two-lane merge in ``_coordinator_rows``. The in-memory lane (manager) stays authoritative for live session state (model / model_alias / current state / tokens). A new persisted lane queries ``storage.list_workstreams(kind=COORDINATOR, user_id=uid, limit=200)`` and appends rows NOT already in the in-memory set — surfacing closed / error / deleted coordinators so the operator can still see them on the landing page. Ownership semantics unchanged — non-admin callers only see their own tenant, admin-bypass via admin.users/admin.roles honored on both lanes, empty-string defense-in-depth matches _check_row_owner_or_404. Tests: - tests/test_storage_sqlite.py — two new tests: kind filter excludes coordinators from the history list; string form of kind accepted (matches the memory.py forwarding shape). - tests/test_coordinator_endpoints.py — four new tests: - closed coordinators from storage surface alongside active ones. - in-memory row wins on ws_id dedup (live state authoritative). - persisted rows respect tenant filter (non-admin, admin bypass). - orphan rows (empty user_id) never leak to empty-sub callers. Gate: ruff + mypy + pytest -m "not live" (4315 passed) all clean. * fix(server,console): address Copilot review on PR #380 Three review comments folded in: 1. Tenancy leak in /v1/api/workstreams/saved — the handler called list_workstreams_with_history without a user_id filter, so any authenticated user could see every other user's saved workstream aliases / titles / names. Fix: - Add ``user_id: str | None = None`` kwarg to list_workstreams_with_history on the protocol + both backends (SQLite + postgres). Pushes the filter into SQL. - memory.py helper forwards the kwarg. - /v1/api/workstreams/saved reads ``_auth_scopes(request)``: a service-scoped caller gets cluster-wide visibility (None), a non-service caller with a blank ``sub`` returns an empty list, otherwise the SQL filter is scoped to the caller's uid. Matches the _visible_workstreams pattern used on /workstreams and /dashboard. 2. Loose type annotation on the memory.py helper — ``kind: Any`` tightened to ``WorkstreamKind | str | None`` so mypy catches invalid callers. WorkstreamKind was already imported in the module. 3. Brittle positional indexing in _coordinator_rows persisted-rows lane — ``row[10]`` for user_id encoded a column offset that would silently corrupt the projection on any future SELECT reorder. Drop the test-double fallback entirely; the storage-protocol contract already requires SQLAlchemy Row with _mapping, and every real caller (SQLite + postgres) provides it. Tests: - test_server_authz.py TestSavedWorkstreamsTenantScoping — four new regression tests covering: non-service caller sees only own rows, service scope sees cluster-wide, blank-sub non-service returns empty, and coordinator rows excluded even for service callers. Gate: ruff + mypy + pytest -m "not live" (4319 passed) all clean. --- tests/test_coordinator_endpoints.py | 159 ++++++++++++++++++++++++++ tests/test_server_authz.py | 95 +++++++++++++++ tests/test_storage_sqlite.py | 37 ++++++ turnstone/console/server.py | 94 +++++++++++++-- turnstone/core/memory.py | 27 ++++- turnstone/core/storage/_postgresql.py | 22 +++- turnstone/core/storage/_protocol.py | 23 +++- turnstone/core/storage/_sqlite.py | 30 ++++- turnstone/server.py | 36 +++++- 9 files changed, 500 insertions(+), 23 deletions(-) diff --git a/tests/test_coordinator_endpoints.py b/tests/test_coordinator_endpoints.py index 075c5ce4..5460744a 100644 --- a/tests/test_coordinator_endpoints.py +++ b/tests/test_coordinator_endpoints.py @@ -1056,3 +1056,162 @@ def test_coordinator_rows_empty_user_id_returns_empty(storage): permissions=frozenset({"read"}), ) assert _coordinator_rows(request) == [] + + +def _persisted_rows_request(storage, mgr, user_id: str, perms: frozenset[str]): + """Build a _coordinator_rows-shaped request with auth_storage wired + up so the persisted-rows merge path fires.""" + from unittest.mock import MagicMock + + from turnstone.core.auth import AuthResult + + request = MagicMock() + request.app.state.coord_mgr = mgr + request.app.state.auth_storage = storage + request.state.auth_result = AuthResult( + user_id=user_id, + scopes=frozenset({"read"}), + token_source="test", + permissions=perms, + ) + return request + + +def test_coordinator_rows_surfaces_closed_coordinators_from_storage(storage): + """Closed coordinators get popped from ``self._workstreams`` but + their persisted row stays in storage with ``state='closed'``. The + landing page polls _coordinator_rows via + /v1/api/cluster/workstreams?node=console — the persisted-rows + merge path surfaces closed rows so the operator can still see + them alongside active ones.""" + from turnstone.console.server import _coordinator_rows + from turnstone.core.workstream import WorkstreamKind + + mgr = _build_mgr(storage) + # Seed a persisted-but-not-loaded closed coordinator directly — + # register + soft-close via storage primitives. + storage.register_workstream( + "a" * 32, + node_id="console", + user_id="alice", + name="historical-coord", + state="closed", + kind=WorkstreamKind.COORDINATOR, + parent_ws_id=None, + ) + # Also seed a live coordinator via the manager to prove merge. + mgr.create(user_id="alice", name="live-coord") + + request = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"})) + rows = _coordinator_rows(request) + names = {r["name"] for r in rows} + assert names == {"live-coord", "historical-coord"} + # Closed coord carries its persisted state so the UI can render + # it with the correct state glyph. + closed = next(r for r in rows if r["name"] == "historical-coord") + assert closed["state"] == "closed" + assert closed["kind"] == "coordinator" + + +def test_coordinator_rows_dedupes_by_ws_id_in_memory_wins(storage): + """When a coordinator is both in-memory (manager) AND in storage, + _coordinator_rows must prefer the in-memory row so live session + state (model / model_alias / current state) stays authoritative. + The storage row has stale fields after every restart / refresh, + so merging it twice is strictly worse.""" + from turnstone.console.server import _coordinator_rows + + mgr = _build_mgr(storage) + live = mgr.create(user_id="alice", name="alice-live") + # Persist an explicit storage-only shape for the SAME ws_id — + # mgr.create already did this, but we deliberately corrupt the + # stored row to prove the in-memory row wins. Update the state + # to something the manager would never produce so the dedup check + # is unambiguous. + storage.update_workstream_state(live.id, "error") + + request = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"})) + rows = _coordinator_rows(request) + assert len(rows) == 1 + assert rows[0]["id"] == live.id + # In-memory WorkstreamState wins over the persisted "error" tweak + # — the manager reports "idle" for a freshly-created coordinator. + assert rows[0]["state"] == "idle" + + +def test_coordinator_rows_persisted_respects_tenant_filter(storage): + """Non-admin callers must not see OTHER tenants' persisted + (closed) coordinators either. Regression lock — the user_id + kwarg on list_workstreams is pushed through; the defense-in-depth + client-side empty-string check catches the tail.""" + from turnstone.console.server import _coordinator_rows + from turnstone.core.workstream import WorkstreamKind + + mgr = _build_mgr(storage) + storage.register_workstream( + "a" * 32, + node_id="console", + user_id="alice", + name="alice-closed", + state="closed", + kind=WorkstreamKind.COORDINATOR, + parent_ws_id=None, + ) + storage.register_workstream( + "b" * 32, + node_id="console", + user_id="bob", + name="bob-closed", + state="closed", + kind=WorkstreamKind.COORDINATOR, + parent_ws_id=None, + ) + + alice_req = _persisted_rows_request(storage, mgr, "alice", frozenset({"read"})) + alice_rows = _coordinator_rows(alice_req) + assert {r["name"] for r in alice_rows} == {"alice-closed"} + + bob_req = _persisted_rows_request(storage, mgr, "bob", frozenset({"read"})) + bob_rows = _coordinator_rows(bob_req) + assert {r["name"] for r in bob_rows} == {"bob-closed"} + + admin_req = _persisted_rows_request(storage, mgr, "admin-1", frozenset({"read", "admin.users"})) + admin_rows = _coordinator_rows(admin_req) + assert {r["name"] for r in admin_rows} == {"alice-closed", "bob-closed"} + + +def test_coordinator_rows_persisted_skips_orphan_rows_for_non_admin(storage): + """Defense-in-depth empty-string tenancy check — a persisted row + with empty user_id (migration-artifact / system-owned) must NOT + be visible to a non-admin caller with empty caller_uid either. + The SQL user_id filter above already enforces this, but duplicate + the check client-side so orphan rows never leak to any + hypothetical empty-sub JWT.""" + from unittest.mock import MagicMock + + from turnstone.console.server import _coordinator_rows + from turnstone.core.auth import AuthResult + from turnstone.core.workstream import WorkstreamKind + + mgr = _build_mgr(storage) + storage.register_workstream( + "c" * 32, + node_id="console", + user_id="", # orphan / system row + name="orphan-closed", + state="closed", + kind=WorkstreamKind.COORDINATOR, + parent_ws_id=None, + ) + + # Non-admin with empty caller_uid must not see the orphan. + request = MagicMock() + request.app.state.coord_mgr = mgr + request.app.state.auth_storage = storage + request.state.auth_result = AuthResult( + user_id="", + scopes=frozenset({"read"}), + token_source="test", + permissions=frozenset({"read"}), + ) + assert _coordinator_rows(request) == [] diff --git a/tests/test_server_authz.py b/tests/test_server_authz.py index be491830..41a682f2 100644 --- a/tests/test_server_authz.py +++ b/tests/test_server_authz.py @@ -410,6 +410,101 @@ class TestDashboardFiltered: assert owners == {"user-b"} +class TestSavedWorkstreamsTenantScoping: + """Regression for Copilot review on #380: /v1/api/workstreams/saved + used to call list_workstreams_with_history with no tenant filter, + so every authenticated user could see every other user's saved + workstream aliases / titles / names. Fix tightens to + ``list_workstreams_with_history(user_id=caller)`` with the + service-scope bypass matching _visible_workstreams.""" + + def _seed(self, client): + """Create two workstreams per user, each with a message so they + land in list_workstreams_with_history (the SQL gates on an + EXISTS conversation).""" + from turnstone.core.storage import get_storage + + storage = get_storage() + assert storage is not None + _register_ws(storage, "alice-saved", "alice") + storage.save_message("alice-saved", "user", "alice's plan") + _register_ws(storage, "bob-saved", "bob") + storage.save_message("bob-saved", "user", "bob's plan") + return storage + + def test_non_service_caller_sees_only_own_rows(self, app_client): + client, _mgr = app_client + self._seed(client) + resp = client.get("/v1/api/workstreams/saved", headers=_auth("alice")) + assert resp.status_code == 200 + rows = resp.json()["workstreams"] + ids = {r["ws_id"] for r in rows} + assert ids == {"alice-saved"}, f"alice must not see bob's saved rows: {ids}" + + def test_service_scope_sees_all_rows(self, app_client): + """Cluster-wide visibility is preserved for service callers + (console collector, cluster tooling) so they can still hydrate + cross-tenant state when needed.""" + client, _mgr = app_client + self._seed(client) + resp = client.get( + "/v1/api/workstreams/saved", + headers=_auth("cluster-collector", scopes=frozenset({"read", "service"})), + ) + assert resp.status_code == 200 + rows = resp.json()["workstreams"] + ids = {r["ws_id"] for r in rows} + assert {"alice-saved", "bob-saved"}.issubset(ids) + + def test_blank_sub_non_service_returns_empty(self, app_client): + """Defense-in-depth — a non-service token with an empty ``sub`` + claim (orphan / migration-artifact auth path) must not match + every workstream with empty ``user_id``. Fail closed.""" + client, _mgr = app_client + storage = self._seed(client) + # Also seed an orphan row so the test would fail loudly if the + # handler leaked it. + _register_ws(storage, "orphan-saved", "") + storage.save_message("orphan-saved", "user", "orphan content") + resp = client.get( + "/v1/api/workstreams/saved", + headers=_auth("", scopes=frozenset({"read"})), + ) + assert resp.status_code == 200 + assert resp.json()["workstreams"] == [] + + def test_coordinator_rows_excluded_even_for_service(self, app_client): + """kind filter is orthogonal to the user_id filter — even a + service caller (cluster-wide) must not see coordinator rows on + the interactive 'saved workstreams' endpoint.""" + from turnstone.core.storage import get_storage + from turnstone.core.workstream import WorkstreamKind + + client, _mgr = app_client + storage = get_storage() + assert storage is not None + storage.register_workstream( + "coord-row", + node_id="console", + user_id="alice", + name="alice-coord", + kind=WorkstreamKind.COORDINATOR, + parent_ws_id=None, + ) + storage.save_message("coord-row", "user", "planning") + _register_ws(storage, "alice-interactive", "alice") + storage.save_message("alice-interactive", "user", "interactive") + + resp = client.get( + "/v1/api/workstreams/saved", + headers=_auth("alice", scopes=frozenset({"read", "service"})), + ) + assert resp.status_code == 200 + ids = {r["ws_id"] for r in resp.json()["workstreams"]} + assert "alice-interactive" in ids + assert "coord-row" not in ids + + class TestGlobalEventsServiceGate: def test_non_service_rejected(self, app_client): client, _mgr = app_client diff --git a/tests/test_storage_sqlite.py b/tests/test_storage_sqlite.py index 2b057e1a..77f70f54 100644 --- a/tests/test_storage_sqlite.py +++ b/tests/test_storage_sqlite.py @@ -309,6 +309,43 @@ class TestListWorkstreamsWithHistory: rows = backend.list_workstreams_with_history(limit=3) assert len(rows) == 3 + def test_kind_filter_excludes_coordinators(self, backend): + """The interactive 'saved workstreams' sidebar calls this with + kind=INTERACTIVE so coordinator rows (which also persist + conversation history) don't leak into the interactive UI.""" + from turnstone.core.workstream import WorkstreamKind + + backend.register_workstream("interactive-1", kind=WorkstreamKind.INTERACTIVE) + backend.save_message("interactive-1", "user", "hi") + backend.register_workstream("coord-1", kind=WorkstreamKind.COORDINATOR) + backend.save_message("coord-1", "user", "plan something") + + # Default (no filter) returns both — preserves legacy behaviour. + rows_all = backend.list_workstreams_with_history() + assert {r[0] for r in rows_all} == {"interactive-1", "coord-1"} + + # kind=INTERACTIVE drops the coordinator row at the SQL layer. + rows_i = backend.list_workstreams_with_history(kind=WorkstreamKind.INTERACTIVE) + assert {r[0] for r in rows_i} == {"interactive-1"} + + # kind=COORDINATOR symmetric — for admin tooling that wants + # the opposite view. + rows_c = backend.list_workstreams_with_history(kind=WorkstreamKind.COORDINATOR) + assert {r[0] for r in rows_c} == {"coord-1"} + + def test_kind_filter_accepts_string(self, backend): + """String form (``"interactive"``) works too — matches how the + memory.py helper forwards caller-supplied values.""" + from turnstone.core.workstream import WorkstreamKind + + backend.register_workstream("interactive-1", kind=WorkstreamKind.INTERACTIVE) + backend.save_message("interactive-1", "user", "hi") + backend.register_workstream("coord-1", kind=WorkstreamKind.COORDINATOR) + backend.save_message("coord-1", "user", "plan") + + rows = backend.list_workstreams_with_history(kind="interactive") + assert {r[0] for r in rows} == {"interactive-1"} + class TestDeleteWorkstream: def test_deletes_all_data(self, backend): diff --git a/turnstone/console/server.py b/turnstone/console/server.py index eb794795..6371f035 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -371,24 +371,35 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]: them into the cluster view so the dashboard tree grouping can nest spawned children under their coordinator parent. - Ownership filter mirrors :meth:`CoordinatorManager.list_for_user`'s - invariant: non-admin callers must not see other tenants' coordinator - rows (``ws_id`` + name + state would otherwise leak cross-tenant via - the unified dashboard). Admins (``admin.users`` / ``admin.roles``) - get the full set. Unauthenticated callers shouldn't reach this path - — the endpoint sits behind the global auth middleware — but the - filter defaults to empty on a missing ``user_id`` rather than - leaking via ``list_all``. + Sources two lanes and merges by ws_id: + + - **In-memory** via :meth:`CoordinatorManager.list_for_user` / + ``list_all`` — carries live session state (model / model_alias / + current workstream state) for currently-loaded coordinators. + - **Persisted** via ``storage.list_workstreams(kind=COORDINATOR)`` + — includes closed / error / soft-deleted rows the manager has + evicted from memory. Without this, closed coordinators + disappeared from the landing page the moment ``close`` fired. + + In-memory wins on ws_id conflict so live state stays authoritative + for active sessions. + + Ownership filter: non-admin callers must not see other tenants' + coordinator rows. Admins (``admin.users`` / ``admin.roles``) get + the full set. Unauthenticated callers shouldn't reach this path + — the endpoint sits behind global auth — but the filter defaults + to empty on a missing ``user_id`` rather than leaking. """ coord_mgr = getattr(request.app.state, "coord_mgr", None) if coord_mgr is None: return [] + is_admin = _is_admin(request) + caller_uid = _auth_user_id(request) try: - if _is_admin(request): + if is_admin: wss = coord_mgr.list_all() else: - user_id = _auth_user_id(request) - wss = coord_mgr.list_for_user(user_id) if user_id else [] + wss = coord_mgr.list_for_user(caller_uid) if caller_uid else [] except Exception: log.debug("cluster_workstreams.coord_list_failed", exc_info=True) return [] @@ -398,6 +409,7 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]: return val if isinstance(val, str) else "" rows: list[dict[str, Any]] = [] + seen: set[str] = set() for ws in wss: sess = getattr(ws, "session", None) rows.append( @@ -417,6 +429,66 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]: "tool_calls": 0, "kind": WorkstreamKind.COORDINATOR.value, "parent_ws_id": None, + "user_id": ws.user_id or "", + } + ) + seen.add(ws.id) + + # Second lane — persisted coordinator rows, used to surface + # closed / error / deleted coordinators the manager has already + # evicted from ``self._workstreams``. Same ownership semantics as + # the in-memory lane (admin bypass, empty user_id fails closed). + storage = getattr(request.app.state, "auth_storage", None) + if storage is None: + return rows + if not is_admin and not caller_uid: + return rows + try: + persisted = storage.list_workstreams( + kind=WorkstreamKind.COORDINATOR, + user_id=None if is_admin else caller_uid, + limit=200, + ) + except Exception: + log.debug("cluster_workstreams.coord_persisted_failed", exc_info=True) + return rows + + for row in persisted: + # SQLAlchemy Row — access via _mapping so future SELECT reorders + # / new columns don't silently corrupt the projection (per the + # storage-protocol guidance on list_workstreams). Test doubles + # must expose the same ._mapping attribute; positional indexing + # was removed because it hard-coded column offsets that drift + # with migrations. + m = row._mapping + row_id = m.get("ws_id") or "" + if not row_id or row_id in seen: + continue + row_owner = m.get("user_id") or "" + # Defense-in-depth empty-string tenancy check — same pattern as + # _check_row_owner_or_404. The SQL user_id filter above should + # already enforce this, but duplicate the check client-side for + # migration-artifact rows with blank owners. + if not is_admin and (not caller_uid or not row_owner or row_owner != caller_uid): + continue + rows.append( + { + "id": row_id, + "name": m.get("name") or f"coord-{row_id[:4]}", + "state": str(m.get("state") or "idle"), + "title": "", + "node": "console", + "server_url": "", + "model": "", + "model_alias": "", + "tokens": 0, + "context_ratio": 0.0, + "activity": "", + "activity_state": "", + "tool_calls": 0, + "kind": WorkstreamKind.COORDINATOR.value, + "parent_ws_id": None, + "user_id": row_owner, } ) return rows diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index b3b81ffb..fa5ed736 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -344,10 +344,31 @@ def update_workstream_name(ws_id: str, name: str) -> None: log.warning("Failed to update workstream name ws=%s", ws_id, exc_info=True) -def list_workstreams_with_history(limit: int = 20) -> list[Any]: - """List workstreams that have conversation messages.""" +def list_workstreams_with_history( + limit: int = 20, + *, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, +) -> list[Any]: + """List workstreams that have conversation messages. + + ``kind`` forwards to the storage layer's SQL-side filter — pass + ``WorkstreamKind.INTERACTIVE`` from the interactive "saved + workstreams" endpoint so coordinator rows (which persist + conversation history too) don't leak into that sidebar. Default + ``None`` preserves legacy all-kinds behaviour. + + ``user_id`` enforces tenant scoping at the SQL layer. Pass the + authenticated caller's uid from any tenant-visible endpoint; + leaving it as ``None`` means cluster-wide (service-scoped + callers only). + """ try: - return get_storage().list_workstreams_with_history(limit) + return get_storage().list_workstreams_with_history( + limit, + kind=kind, + user_id=user_id, + ) except Exception: log.warning("Failed to list workstreams with history", exc_info=True) return [] diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index bd9f63f0..780d6b9d 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -300,7 +300,23 @@ class PostgreSQLBackend: # -- Workstream management ------------------------------------------------- - def list_workstreams_with_history(self, limit: int = 20) -> list[Any]: + def list_workstreams_with_history( + self, + limit: int = 20, + *, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, + ) -> list[Any]: + # See SQLite sibling for the rationale on the kind + user_id filters. + params: dict[str, Any] = {"limit": limit} + kind_clause = "" + user_clause = "" + if kind is not None: + params["kind"] = WorkstreamKind(kind).value + kind_clause = "AND w.kind = :kind " + if user_id is not None: + params["user_id"] = user_id + user_clause = "AND w.user_id = :user_id " with self._conn() as conn: return list( conn.execute( @@ -312,9 +328,11 @@ class PostgreSQLBackend: "FROM workstreams w " "WHERE EXISTS " " (SELECT 1 FROM conversations c WHERE c.ws_id = w.ws_id) " + f"{kind_clause}" + f"{user_clause}" "ORDER BY w.updated DESC LIMIT :limit" ), - {"limit": limit}, + params, ).fetchall() ) diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index db87dd46..333e8d97 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -207,8 +207,27 @@ class StorageBackend(Protocol): # -- Workstream management ------------------------------------------------- - def list_workstreams_with_history(self, limit: int = 20) -> list[Any]: - """List workstreams that have messages, ordered by updated DESC.""" + def list_workstreams_with_history( + self, + limit: int = 20, + *, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, + ) -> list[Any]: + """List workstreams that have messages, ordered by updated DESC. + + ``kind`` filters at the SQL layer — pass ``WorkstreamKind.INTERACTIVE`` + from the interactive "saved workstreams" sidebar so coordinator rows + (which also persist conversation history) don't leak into that + surface. Default ``None`` preserves the legacy all-kinds behaviour. + + ``user_id`` pushes ``WHERE user_id = :user_id`` into SQL so tenant + scoping is enforced server-side rather than relying on handlers to + remember a client-side filter. Pass the authenticated caller's + uid from any tenant-visible endpoint; pass ``None`` for + service-scoped callers that legitimately need cluster-wide + visibility. Mirrors the same contract on ``list_workstreams``. + """ ... def prune_workstreams(self, retention_days: int = 90) -> tuple[int, int]: diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index c9761d27..2b59ed1c 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -383,7 +383,31 @@ class SQLiteBackend: # -- Workstream management ------------------------------------------------- - def list_workstreams_with_history(self, limit: int = 20) -> list[Any]: + def list_workstreams_with_history( + self, + limit: int = 20, + *, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, + ) -> list[Any]: + # ``kind`` filter applied at the SQL layer so coordinator rows + # (which persist conversation history the same way interactive + # workstreams do) don't leak into the interactive UI's "saved + # workstreams" sidebar. Default None preserves legacy + # all-kinds behaviour for callers that want both. + # ``user_id`` pushes tenancy into SQL so the "saved" endpoint + # can't accidentally leak another tenant's workstreams. None + # = cluster-wide (service callers); empty string is a separate + # filter value the caller chose deliberately. + params: dict[str, Any] = {"limit": limit} + kind_clause = "" + user_clause = "" + if kind is not None: + params["kind"] = WorkstreamKind(kind).value + kind_clause = "AND w.kind = :kind " + if user_id is not None: + params["user_id"] = user_id + user_clause = "AND w.user_id = :user_id " with self._conn() as conn: return list( conn.execute( @@ -395,9 +419,11 @@ class SQLiteBackend: "FROM workstreams w " "WHERE EXISTS " " (SELECT 1 FROM conversations c WHERE c.ws_id = w.ws_id) " + f"{kind_clause}" + f"{user_clause}" "ORDER BY w.updated DESC LIMIT :limit" ), - {"limit": limit}, + params, ).fetchall() ) diff --git a/turnstone/server.py b/turnstone/server.py index 0319aa8a..9269bf22 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -1302,10 +1302,40 @@ async def dashboard(request: Request) -> JSONResponse: async def list_saved_workstreams(request: Request) -> JSONResponse: - """GET /v1/api/workstreams/saved — list saved workstreams with conversation history.""" - from turnstone.core.memory import list_workstreams_with_history + """GET /v1/api/workstreams/saved — list saved workstreams with conversation history. - rows = list_workstreams_with_history(limit=50) + Tenant-scoped — service-scoped callers (console collector, cluster + tooling) see cluster-wide rows; end-user callers see only their + own workstreams (matching ``_visible_workstreams``). A non-service + call with a blank ``user_id`` returns an empty list rather than + leaking orphan rows. + + Restricted to ``kind="interactive"`` — the interactive UI's "saved + workstreams" sidebar is not a coordinator surface, and coordinator + rows (which persist conversation history too) would otherwise leak + into it. + """ + from turnstone.core.memory import list_workstreams_with_history + from turnstone.core.workstream import WorkstreamKind + + scopes = _auth_scopes(request) + if "service" in scopes: + # Cluster-wide visibility for service-scoped callers. + user_filter: str | None = None + else: + caller_uid = _auth_user_id(request) + if not caller_uid: + # Blank sub on a non-service token — fail closed instead of + # matching every orphan / migration-artifact row with empty + # user_id. Mirrors _visible_workstreams. + return JSONResponse({"workstreams": []}) + user_filter = caller_uid + + rows = list_workstreams_with_history( + limit=50, + kind=WorkstreamKind.INTERACTIVE, + user_id=user_filter, + ) result = [ { "ws_id": wid,