mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
fix(server,console): kind filter on saved-workstreams + closed coords on landing (#380)
* 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.
This commit is contained in:
Patrick Buckley
@@ -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) == []
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
+83
-11
@@ -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
|
||||
|
||||
@@ -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 []
|
||||
|
||||
@@ -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()
|
||||
)
|
||||
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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()
|
||||
)
|
||||
|
||||
|
||||
+33
-3
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user