diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 2dc950f4..005e4380 100644 --- a/tests/test_workstream_endpoints.py +++ b/tests/test_workstream_endpoints.py @@ -197,9 +197,12 @@ class TestDeleteWorkstream: class TestSetWorkstreamTitle: def test_set_title_success(self, title_client, storage): client, mock_mgr = title_client - storage.register_workstream("ws-abc", "node-1", name="test") - mock_ws = MagicMock() - mock_mgr.get.return_value = mock_ws + storage.register_workstream("ws-abc", "node-1", name="test", user_id="test-user") + # mgr.get returning None makes _require_ws_access fall through to + # the storage-backed ownership check (caller == "test-user" matches + # the registered owner). Tests that need a ws returned from the + # manager set up mock_ws.user_id explicitly. + mock_mgr.get.return_value = None r = client.post( "/v1/api/workstreams/ws-abc/title", json={"title": "New Title"}, @@ -208,8 +211,9 @@ class TestSetWorkstreamTitle: assert r.json()["title"] == "New Title" def test_set_title_empty(self, title_client, storage): - client, _ = title_client + client, mock_mgr = title_client storage.register_workstream("ws-abc", "node-1", name="test", user_id="test-user") + mock_mgr.get.return_value = None r = client.post( "/v1/api/workstreams/ws-abc/title", json={"title": ""}, @@ -218,8 +222,9 @@ class TestSetWorkstreamTitle: assert "required" in r.json()["error"].lower() def test_set_title_missing_body(self, title_client, storage): - client, _ = title_client + client, mock_mgr = title_client storage.register_workstream("ws-abc", "node-1", name="test", user_id="test-user") + mock_mgr.get.return_value = None r = client.post( "/v1/api/workstreams/ws-abc/title", json={}, @@ -228,8 +233,8 @@ class TestSetWorkstreamTitle: def test_set_title_truncation(self, title_client, storage): client, mock_mgr = title_client - storage.register_workstream("ws-abc", "node-1", name="test") - mock_mgr.get.return_value = MagicMock() + storage.register_workstream("ws-abc", "node-1", name="test", user_id="test-user") + mock_mgr.get.return_value = None long_title = "x" * 200 r = client.post( "/v1/api/workstreams/ws-abc/title", @@ -239,10 +244,11 @@ class TestSetWorkstreamTitle: assert len(r.json()["title"]) <= 80 def test_set_title_alias_conflict(self, title_client, storage): - client, _ = title_client - storage.register_workstream("ws-1", "node-1", name="first") - storage.register_workstream("ws-2", "node-1", name="second") + client, mock_mgr = title_client + storage.register_workstream("ws-1", "node-1", name="first", user_id="test-user") + storage.register_workstream("ws-2", "node-1", name="second", user_id="test-user") storage.set_workstream_alias("ws-1", "taken-name") + mock_mgr.get.return_value = None r = client.post( "/v1/api/workstreams/ws-2/title", json={"title": "taken-name"}, @@ -259,7 +265,11 @@ class TestRefreshWorkstreamTitle: def test_refresh_success(self, title_client, storage): client, mock_mgr = title_client storage.register_workstream("ws-abc", "node-1", name="test", user_id="test-user") + # The in-memory fast path on _require_ws_access checks ws.user_id + # before falling back to storage, so the mock returned by + # mgr.get must carry the expected owner. mock_ws = MagicMock() + mock_ws.user_id = "test-user" mock_ws.session = MagicMock() mock_mgr.get.return_value = mock_ws with patch("turnstone.core.memory.get_workstream_display_name", return_value="Old Title"): diff --git a/tests/test_workstream_kind.py b/tests/test_workstream_kind.py index 045ab261..47e50e71 100644 --- a/tests/test_workstream_kind.py +++ b/tests/test_workstream_kind.py @@ -78,6 +78,29 @@ def test_register_rejects_unknown_kind(storage): assert storage.get_workstream("ws-bogus") is None +def test_delete_workstream_nulls_child_parent_ws_id(storage): + """Deleting a coordinator must null-out its children's parent_ws_id + so list_workstreams(parent_ws_id=) doesn't keep returning + ghost-parented rows.""" + storage.register_workstream("coord", kind="coordinator", user_id="user-1") + storage.register_workstream( + "child-a", kind="interactive", parent_ws_id="coord", user_id="user-1" + ) + storage.register_workstream( + "child-b", kind="interactive", parent_ws_id="coord", user_id="user-1" + ) + + assert storage.delete_workstream("coord") is True + + # Children still exist but with NULL parent_ws_id. + for cid in ("child-a", "child-b"): + row = storage.get_workstream(cid) + assert row is not None, f"{cid} should survive parent deletion" + assert row["parent_ws_id"] is None, f"{cid} still points at ghost coord" + # No rows match the deleted coord's parent filter. + assert storage.list_workstreams(parent_ws_id="coord") == [] + + def test_list_workstreams_filter_by_user_id(storage): """The user_id kwarg pushes tenant scoping into SQL so callers can't forget to filter client-side.""" diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index 1f2ab2f4..e2aa2712 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -191,6 +191,8 @@ class WorkstreamInfo(BaseModel): id: str name: str state: str + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE + parent_ws_id: str | None = None class ListWorkstreamsResponse(BaseModel): @@ -210,6 +212,9 @@ class DashboardWorkstream(BaseModel): node: str = "" model: str = "" model_alias: str = "" + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE + parent_ws_id: str | None = None + user_id: str = "" class DashboardAggregate(BaseModel): diff --git a/turnstone/console/coordinator.py b/turnstone/console/coordinator.py index 8bb646ea..fbab89ec 100644 --- a/turnstone/console/coordinator.py +++ b/turnstone/console/coordinator.py @@ -708,9 +708,18 @@ class CoordinatorManager: coord_ws_id[:8], ) return + # Cap is a sentinel, not a hard limit. The rebuild runs at most + # once per console cold-start per coordinator, so the fetch cost + # is irrelevant; what matters is visibility when a coord has + # more children than the cap — previously the tail was dropped + # silently on every restart. Fetch ``_rebuild_limit + 1`` so a + # coord with exactly ``_rebuild_limit`` children (no truncation + # yet) doesn't trigger a false-positive warning; the +1 is the + # sentinel that proves there's at least one more row in storage. + _rebuild_limit = 10_000 try: rows = self._storage.list_workstreams( - limit=1000, + limit=_rebuild_limit + 1, parent_ws_id=coord_ws_id, kind=None, user_id=coord_user_id, @@ -722,6 +731,13 @@ class CoordinatorManager: exc_info=True, ) rows = [] + if len(rows) > _rebuild_limit: + log.warning( + "coord_mgr.rebuild_children_truncated ws=%s limit=%d", + coord_ws_id[:8], + _rebuild_limit, + ) + rows = rows[:_rebuild_limit] child_ids: list[str] = [] for r in rows: try: diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 4381e09f..9f6dae0a 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -2637,11 +2637,20 @@ async def coordinator_children(request: Request) -> JSONResponse: if err404 is not None: return err404 + # Tenant filter: push the caller's user_id into SQL so forged / + # migration-era rows with the same parent_ws_id but a different + # owner can't leak through. Admins bypass the filter — they're + # expected to see the full subtree. _resolve_coordinator_or_404 + # already validated that non-admin callers own the coord itself, + # so the caller's uid is the correct scope for children by the + # "children share the coord's owner by construction" invariant. + filter_user_id = None if _is_admin(request) else (user_id or None) try: raw = storage.list_workstreams( limit=_CHILDREN_PAGE_LIMIT + 1, parent_ws_id=ws_id, kind=None, + user_id=filter_user_id, ) except Exception: correlation_id = secrets.token_hex(4) diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 1bbc4876..b3b81ffb 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -344,28 +344,6 @@ 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( - node_id: str | None = None, - limit: int = 100, - *, - parent_ws_id: str | None = None, - kind: WorkstreamKind | str | None = None, - user_id: str | None = None, -) -> list[Any]: - """List workstreams, optionally filtered by node/parent/kind/user.""" - try: - return get_storage().list_workstreams( - node_id, - limit, - parent_ws_id=parent_ws_id, - kind=kind, - user_id=user_id, - ) - except Exception: - log.warning("Failed to list workstreams", exc_info=True) - return [] - - def list_workstreams_with_history(limit: int = 20) -> list[Any]: """List workstreams that have conversation messages.""" try: diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index b3d8dc2f..ab9b8108 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -603,6 +603,12 @@ class PostgreSQLBackend: conn.execute( sa.delete(workstream_overrides).where(workstream_overrides.c.ws_id == ws_id) ) + # Null-out parent_ws_id on children — see sqlite sibling for rationale. + conn.execute( + sa.update(workstreams) + .where(workstreams.c.parent_ws_id == ws_id) + .values(parent_ws_id=None) + ) result = conn.execute(sa.delete(workstreams).where(workstreams.c.ws_id == ws_id)) conn.commit() return result.rowcount > 0 diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 35b67d69..99c63095 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -704,6 +704,17 @@ class SQLiteBackend: conn.execute( sa.delete(workstream_overrides).where(workstream_overrides.c.ws_id == ws_id) ) + # Null-out parent_ws_id on children before dropping the row — + # otherwise a deleted coordinator leaves orphaned pointers and + # ``list_workstreams(parent_ws_id=)`` keeps returning + # ghost-parented rows. Cheaper than a schema-level FK with + # ON DELETE SET NULL and avoids rewriting the workstreams + # table on SQLite. + conn.execute( + sa.update(workstreams) + .where(workstreams.c.parent_ws_id == ws_id) + .values(parent_ws_id=None) + ) result = conn.execute(sa.delete(workstreams).where(workstreams.c.ws_id == ws_id)) conn.commit() return result.rowcount > 0 diff --git a/turnstone/core/storage/migrations/versions/039_workstream_kind.py b/turnstone/core/storage/migrations/versions/039_workstream_kind.py index bf96fcf7..bfa70a87 100644 --- a/turnstone/core/storage/migrations/versions/039_workstream_kind.py +++ b/turnstone/core/storage/migrations/versions/039_workstream_kind.py @@ -12,6 +12,20 @@ filters stay cheap on both SQLite and PostgreSQL backends. ``NOT NULL`` default ``'interactive'`` on ``kind`` keeps existing rows valid; ``parent_ws_id`` is nullable because most workstreams have no parent. +.. warning:: + + ``downgrade()`` drops both columns, which **irreversibly destroys** + every ``parent_ws_id`` value. Coordinator → child lineage cannot be + reconstructed after a downgrade on a populated DB. If you need to + roll back on a production cluster, export the workstreams table + first (``SELECT ws_id, kind, parent_ws_id FROM workstreams``) so + the relationships can be re-applied after re-upgrading. + + The index-tuning follow-up (migration 041) assumes this migration + is applied — downgrading past 039 also invalidates 041's downgrade + path, so revert migrations in order (041 → 040 → 039) rather than + skipping. + Revision ID: 039 Revises: 038 Create Date: 2026-04-16 diff --git a/turnstone/core/storage/migrations/versions/041_workstream_index_tuning.py b/turnstone/core/storage/migrations/versions/041_workstream_index_tuning.py new file mode 100644 index 00000000..f4afed8a --- /dev/null +++ b/turnstone/core/storage/migrations/versions/041_workstream_index_tuning.py @@ -0,0 +1,117 @@ +"""Tune migration 039's workstream indexes for the real query mix. + +Follow-up to the retrospective review of the workstream-kind feature. +Migration 039 added two single-column btree indexes on the workstreams +table — ``idx_workstreams_kind`` (on ``kind``) and +``idx_workstreams_parent`` (on ``parent_ws_id``). Profiling every +``list_workstreams(...)`` call site revealed: + +- **kind has 2 values** (``"interactive"`` / ``"coordinator"``). Every + query that filters by ``kind`` also supplies a more selective predicate + (``parent_ws_id``, ``node_id``, or ``user_id``), so the planner never + chooses the low-cardinality ``kind`` index — it's write-amplification + overhead on every INSERT/UPDATE to workstreams for zero read benefit. + Drop it. + +- **parent_ws_id is mostly NULL** (the migration's own docstring notes + "most workstreams have no parent"). The existing full btree indexes + all those NULL rows. Every real query is ``parent_ws_id = ``, + never ``IS NULL``. Replace with a partial index ``WHERE parent_ws_id + IS NOT NULL`` — halves the btree size and write cost on interactive + workstreams without changing any query plan. + +On PostgreSQL the rebuild follows a gap-free pattern so the ``workstreams`` +table is never unindexed on ``parent_ws_id`` during the migration: + +1. ``CREATE INDEX CONCURRENTLY`` a new partial index under a temporary + name (``idx_workstreams_parent_new``). Builds without blocking writes. +2. ``DROP INDEX CONCURRENTLY`` the old full index. Also non-blocking. +3. ``ALTER INDEX ... RENAME`` the new index into the canonical slot. + +A naïve "drop then create" in-transaction would leave a minutes-long window +with no parent_ws_id index at all — severe query slowdowns for any reader +filtering on that column. The concurrent-build/drop/rename dance keeps +at least one index serving ``parent_ws_id`` queries for the full duration. + +Both concurrent ops require stepping out of Alembic's managed transaction +via an autocommit block. + +SQLite has no concurrent-index concept and its table-level write lock +already serializes readers and writers, so the partial rebuild there is +a straight drop-then-create (no gap-free concern). Partial-index support +has been stable since SQLite 3.8 (predates any modern Python stdlib). + +Revision ID: 041 +Revises: 040 +Create Date: 2026-04-18 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "041" +down_revision = "040" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + bind = op.get_bind() + dialect = bind.dialect.name + + if dialect == "postgresql": + # Gap-free rebuild — see module docstring for the rationale. + # Every op runs inside the autocommit block; ``DROP INDEX + # CONCURRENTLY`` and ``CREATE INDEX CONCURRENTLY`` both require + # running outside any transaction. + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS " + "idx_workstreams_parent_new ON workstreams (parent_ws_id) " + "WHERE parent_ws_id IS NOT NULL" + ) + op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_workstreams_parent") + op.execute("ALTER INDEX idx_workstreams_parent_new RENAME TO idx_workstreams_parent") + # Kind index has no replacement — drop concurrently to avoid + # the brief ACCESS EXCLUSIVE lock a plain DROP INDEX would take. + op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_workstreams_kind") + else: + # SQLite: partial indexes stable since 3.8; no concurrent concept. + # The table-level write lock means there's no useful distinction + # between gap-free and drop-then-create here. + op.drop_index("idx_workstreams_kind", table_name="workstreams") + op.drop_index("idx_workstreams_parent", table_name="workstreams") + op.create_index( + "idx_workstreams_parent", + "workstreams", + ["parent_ws_id"], + sqlite_where=sa.text("parent_ws_id IS NOT NULL"), + ) + + +def downgrade() -> None: + """Restore migration 039's non-partial full indexes. + + Symmetric rollback that also preserves the gap-free invariant on + PostgreSQL: build the full replacement under a temp name, drop the + partial index concurrently, then rename. Data is untouched — indexes + are pure read-path structures. + """ + bind = op.get_bind() + dialect = bind.dialect.name + + if dialect == "postgresql": + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS " + "idx_workstreams_parent_old ON workstreams (parent_ws_id)" + ) + op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_workstreams_parent") + op.execute("ALTER INDEX idx_workstreams_parent_old RENAME TO idx_workstreams_parent") + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_workstreams_kind ON workstreams (kind)" + ) + else: + op.drop_index("idx_workstreams_parent", table_name="workstreams") + op.create_index("idx_workstreams_parent", "workstreams", ["parent_ws_id"]) + op.create_index("idx_workstreams_kind", "workstreams", ["kind"]) diff --git a/turnstone/server.py b/turnstone/server.py index 168229e4..ee55f269 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -978,7 +978,8 @@ async def events_sse(request: Request) -> Response: # Subscribing to another tenant's stream would leak their messages, # tool calls, and pending approvals in real time. Gate before # _register_listener so non-owners get a flat 404 (no enumeration). - _owner, err = _require_ws_access(request, ws_id or "") + # In-memory fast path via mgr= keeps SSE resilient to DB blips. + _owner, err = _require_ws_access(request, ws_id or "", mgr=mgr) if err: return err ws, ui = _get_ws(mgr, ws_id) @@ -1224,11 +1225,17 @@ async def list_workstreams(request: Request) -> JSONResponse: result = [] for ws in _visible_workstreams(request, mgr.list_all()): title = get_workstream_display_name(ws.id) or ws.name + # kind + parent_ws_id mirror the shape /dashboard returns below so + # client consumers (SDK, frontend, integrators) see one consistent + # row schema across adjacent endpoints instead of a subset here + # and a superset there. result.append( { "id": ws.id, "name": title, "state": ws.state.value, + "kind": ws.kind, + "parent_ws_id": ws.parent_ws_id, } ) return JSONResponse({"workstreams": result}) @@ -1803,13 +1810,15 @@ async def approve(request: Request) -> JSONResponse: feedback = body.get("feedback") always = body.get("always", False) ws_id = body.get("ws_id") + mgr = request.app.state.workstreams # Cross-tenant guard: resolving a pending tool approval on another # tenant's workstream is RCE-adjacent (the victim queued a command # expecting to decide themselves). Gate before touching the UI. - _owner, err = _require_ws_access(request, str(ws_id or "")) + # Pass mgr= so the check uses the in-memory ws.user_id and survives + # transient storage outages. + _owner, err = _require_ws_access(request, str(ws_id or ""), mgr=mgr) if err: return err - mgr = request.app.state.workstreams ws, ui = _get_ws(mgr, ws_id) if not ws or not ui: return JSONResponse({"error": "Unknown workstream"}, status_code=404) @@ -1836,10 +1845,10 @@ async def plan_feedback(request: Request) -> JSONResponse: return body feedback = body.get("feedback", "") ws_id = body.get("ws_id") - _owner, err = _require_ws_access(request, str(ws_id or "")) + mgr = request.app.state.workstreams + _owner, err = _require_ws_access(request, str(ws_id or ""), mgr=mgr) if err: return err - mgr = request.app.state.workstreams ws, ui = _get_ws(mgr, ws_id) if not ws or not ui: return JSONResponse({"error": "Unknown workstream"}, status_code=404) @@ -1855,10 +1864,10 @@ async def cancel_generation(request: Request) -> JSONResponse: if isinstance(body, JSONResponse): return body ws_id = body.get("ws_id") - _owner, err = _require_ws_access(request, str(ws_id or "")) + mgr = request.app.state.workstreams + _owner, err = _require_ws_access(request, str(ws_id or ""), mgr=mgr) if err: return err - mgr = request.app.state.workstreams ws, ui = _get_ws(mgr, ws_id) if not ws or not ui: return JSONResponse({"error": "Unknown workstream"}, status_code=404) @@ -1900,11 +1909,11 @@ async def command(request: Request) -> JSONResponse: ws_id = body.get("ws_id") if not cmd: return JSONResponse({"error": "Empty command"}, status_code=400) - _owner, err = _require_ws_access(request, str(ws_id or "")) + mgr = request.app.state.workstreams + _owner, err = _require_ws_access(request, str(ws_id or ""), mgr=mgr) if err: return err - mgr = request.app.state.workstreams ws, ui = _get_ws(mgr, ws_id) if not ws or not ui: return JSONResponse({"error": "Unknown workstream"}, status_code=404) @@ -2732,13 +2741,13 @@ async def close_workstream(request: Request) -> JSONResponse: if isinstance(body, JSONResponse): return body ws_id = str(body.get("ws_id", "")) + mgr = request.app.state.workstreams # Cross-tenant close would abort another tenant's running generation. # _require_ws_access returns 404 on non-owner — same shape as the # "last workstream" (400) / "not found" (404) branches below. - owner_uid, err = _require_ws_access(request, ws_id) + owner_uid, err = _require_ws_access(request, ws_id, mgr=mgr) if err: return err - mgr = request.app.state.workstreams # Distinguish "last workstream" (400) from "not found" (404). # Note: get() and close() acquire the manager lock independently, so a # concurrent close between the two could produce a wrong error code. @@ -2783,15 +2792,19 @@ async def delete_workstream_endpoint(request: Request) -> JSONResponse: owner_uid, err = _require_ws_access(request, ws_id) if err: return err - # Snapshot kind/parent for the audit record before deletion wipes the row. storage = getattr(request.app.state, "auth_storage", None) - row: dict[str, Any] = {} - if storage is not None: - row = storage.get_workstream(ws_id) or {} - kind = row.get("kind", "") - parent_ws_id = row.get("parent_ws_id") + kind: str = "" + parent_ws_id: str | None = None _, ip = _audit_context(request) try: + # Snapshot kind/parent for the audit record before the delete + # wipes the row. Inside the try so a transient storage error + # surfaces through the endpoint's redacted 500 handler below + # rather than as an unhandled exception. + if storage is not None: + row = storage.get_workstream(ws_id) or {} + kind = row.get("kind", "") + parent_ws_id = row.get("parent_ws_id") if delete_workstream(ws_id): log.info("ws.deleted", ws_id=ws_id[:8]) if storage is not None: @@ -2820,13 +2833,13 @@ async def refresh_workstream_title(request: Request, ws_id: str = "") -> JSONRes log = get_logger(__name__) ws_id = request.path_params.get("ws_id", "") log.info("ws.title.refresh_requested", ws_id=ws_id[:8] if ws_id else "empty") + mgr = request.app.state.workstreams # Cross-tenant rename is a phishing / denial-of-use vector — a # malicious caller could push the victim's title to a misleading # string visible in list/dashboard responses. - _owner, err = _require_ws_access(request, ws_id) + _owner, err = _require_ws_access(request, ws_id, mgr=mgr) if err: return err - mgr = request.app.state.workstreams ws = mgr.get(ws_id) if not ws or not ws.session: log.warning( @@ -2858,8 +2871,9 @@ async def set_workstream_title(request: Request, ws_id: str = "") -> JSONRespons log.info("ws.title.set_requested", ws_id=ws_id[:8] if ws_id else "empty") if not ws_id: return JSONResponse({"error": "ws_id is required"}, status_code=400) + mgr = request.app.state.workstreams # Cross-tenant rename gate — same rationale as refresh-title above. - _owner, err = _require_ws_access(request, ws_id) + _owner, err = _require_ws_access(request, ws_id, mgr=mgr) if err: return err body = await read_json_or_400(request) @@ -2876,7 +2890,6 @@ async def set_workstream_title(request: Request, ws_id: str = "") -> JSONRespons status_code=409, ) log.info("ws.title.set_alias_updated", ws_id=ws_id[:8]) - mgr = request.app.state.workstreams ws = mgr.get(ws_id) if ws and ws.session and ws.session.ui: ws.session.ui.on_rename(title) @@ -3034,21 +3047,48 @@ def _auth_scopes(request: Request) -> set[str]: return set(getattr(auth, "scopes", []) or []) -def _require_ws_access(request: Request, ws_id: str) -> tuple[str, JSONResponse | None]: +def _require_ws_access( + request: Request, + ws_id: str, + *, + mgr: WorkstreamManager | None = None, +) -> tuple[str, JSONResponse | None]: """Resolve ``ws_id`` to its owner after verifying the caller has access. Service-scoped tokens (internal callers) bypass ownership checks. Returns ``(owner_user_id, None)`` on success. The owner id is what attachments should be filed under. + + When ``mgr`` is provided and the workstream is live in the manager, + trust its cached ``user_id`` instead of round-tripping storage — + keeps in-memory-only handlers (approve / plan / cancel / command / + close / SSE / title) functional during transient DB outages and + trims the hot-path by one query. Handlers that act on + persisted-but-not-loaded workstreams (``/delete``, ``/open``) omit + ``mgr`` and fall through to the storage path. """ + caller = _auth_user_id(request) + scopes = _auth_scopes(request) + is_service = "service" in scopes + + if mgr is not None: + ws_mem = mgr.get(ws_id) + if ws_mem is not None: + owner_mem = ws_mem.user_id + if is_service: + return owner_mem or caller, None + if owner_mem and owner_mem != caller: + return "", JSONResponse({"error": "Workstream not found"}, status_code=404) + return caller, None + # Not in memory — fall through to storage so /delete etc. + # still resolve persisted-but-not-loaded rows. + from turnstone.core.memory import get_workstream_owner owner = get_workstream_owner(ws_id) if owner is None: return "", JSONResponse({"error": "Workstream not found"}, status_code=404) - caller = _auth_user_id(request) - scopes = _auth_scopes(request) - if "service" in scopes: + if is_service: # Trust the service caller; file under its own user_id if no owner # is set, otherwise under the existing owner. return owner or caller, None