diff --git a/docs/architecture.md b/docs/architecture.md index bdd75414..dcb3c26d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -884,7 +884,7 @@ and are the single source of truth for both backends and Alembic migrations. | `update_workstream_title(ws_id, title)` | Set/update LLM-generated title | | `update_workstream_state(ws_id, state)` | Update workstream state and bump timestamp | | `update_workstream_name(ws_id, name)` | Update workstream display name | -| `list_workstreams(node_id, limit)` | List workstreams, optionally by node | +| `list_workstreams(node_id, limit, *, parent_ws_id, kind, user_id)` | List workstreams, optionally filtered by node, parent, kind, or owning user | | `kv_get(key)` / `kv_set(key, value)` / `kv_delete(key)` | Generic key-value store (backs memories table) | | `kv_list()` / `kv_search(query)` | List or search key-value pairs | | `search_history(query, limit)` | Full-text search (FTS5 on SQLite, tsvector on PostgreSQL) | diff --git a/docs/diagrams/14-storage-architecture.puml b/docs/diagrams/14-storage-architecture.puml index 5d11f24b..9a9b1a37 100644 --- a/docs/diagrams/14-storage-architecture.puml +++ b/docs/diagrams/14-storage-architecture.puml @@ -23,7 +23,7 @@ interface "StorageBackend" as SB <> { +resolve_workstream(alias_or_id) → str | None +delete_workstream(ws_id) → bool +prune_workstreams(retention_days) → (int, int) - +list_workstreams(node_id, limit) → list + +list_workstreams(node_id, limit, *, parent_ws_id, kind, user_id) → list +save_workstream_config(ws_id, config) +load_workstream_config(ws_id) → dict +kv_get(key) → str | None diff --git a/tests/test_coordinator_client.py b/tests/test_coordinator_client.py index 396c363e..7a2ce408 100644 --- a/tests/test_coordinator_client.py +++ b/tests/test_coordinator_client.py @@ -272,7 +272,8 @@ def test_non_2xx_response_populates_error(): def populated_storage(tmp_path): st = SQLiteBackend(str(tmp_path / "coord.db")) # Coord + 2 interactive children + 1 child coordinator (excluded) + - # 1 unrelated ws. + # 1 unrelated ws + 1 cross-tenant child (excluded by the user_id SQL + # filter: belongs to user-2 but forged parent_ws_id=coord-1). st.register_workstream("coord-1", kind="coordinator", user_id="user-1") st.register_workstream( "child-a", @@ -280,6 +281,7 @@ def populated_storage(tmp_path): parent_ws_id="coord-1", state="idle", skill_id="skill-x", + user_id="user-1", ) st.register_workstream( "child-b", @@ -287,13 +289,21 @@ def populated_storage(tmp_path): parent_ws_id="coord-1", state="running", skill_id="skill-y", + user_id="user-1", ) st.register_workstream( "child-coord", kind="coordinator", parent_ws_id="coord-1", + user_id="user-1", + ) + st.register_workstream("unrelated", kind="interactive", user_id="user-1") + st.register_workstream( + "cross-tenant-child", + kind="interactive", + parent_ws_id="coord-1", + user_id="user-2", ) - st.register_workstream("unrelated", kind="interactive") return st @@ -316,7 +326,9 @@ def test_list_children_returns_only_interactive_children(populated_storage): assert set(result.keys()) == {"children", "truncated"} rows = result["children"] names = {r["ws_id"] for r in rows} - assert names == {"child-a", "child-b"} # excludes child-coord + unrelated + # Excludes child-coord (kind filter), unrelated (parent filter), + # cross-tenant-child (user_id filter). + assert names == {"child-a", "child-b"} for r in rows: assert r["kind"] == "interactive" assert r["parent_ws_id"] == "coord-1" @@ -324,6 +336,14 @@ def test_list_children_returns_only_interactive_children(populated_storage): assert result["truncated"] is False +def test_list_children_excludes_cross_tenant_child(populated_storage): + """SQL-level user_id filter drops forged parent_ws_id rows owned by another user.""" + client = _make_read_client(populated_storage) + result = client.list_children("coord-1") + names = {r["ws_id"] for r in result["children"]} + assert "cross-tenant-child" not in names + + def test_list_children_filters_by_state(populated_storage): client = _make_read_client(populated_storage) result = client.list_children("coord-1", state="running") @@ -385,12 +405,26 @@ def test_list_children_excludes_closed_by_default(tmp_path): post-hoc filter them. An explicit state filter still wins.""" st = SQLiteBackend(str(tmp_path / "closed.db")) st.register_workstream("coord-1", kind="coordinator", user_id="user-1") - st.register_workstream("child-active", kind="interactive", parent_ws_id="coord-1", state="idle") st.register_workstream( - "child-closed", kind="interactive", parent_ws_id="coord-1", state="closed" + "child-active", + kind="interactive", + parent_ws_id="coord-1", + state="idle", + user_id="user-1", ) st.register_workstream( - "child-deleted", kind="interactive", parent_ws_id="coord-1", state="deleted" + "child-closed", + kind="interactive", + parent_ws_id="coord-1", + state="closed", + user_id="user-1", + ) + st.register_workstream( + "child-deleted", + kind="interactive", + parent_ws_id="coord-1", + state="deleted", + user_id="user-1", ) client = _make_read_client(st) result = client.list_children("coord-1") diff --git a/tests/test_coordinator_end_to_end.py b/tests/test_coordinator_end_to_end.py index 59a0e193..57e51484 100644 --- a/tests/test_coordinator_end_to_end.py +++ b/tests/test_coordinator_end_to_end.py @@ -289,13 +289,16 @@ def seeded_storage(tmp_path): st = SQLiteBackend(str(tmp_path / "seed.db")) # Parent coordinator. st.register_workstream("coord-root", kind="coordinator", user_id="user-1") - # Two interactive children — one idle, one running. + # Two interactive children — one idle, one running. Children inherit + # the coord's user_id by construction (server-side create gate), which + # the list_children SQL filter now enforces. st.register_workstream( "child-idle", kind="interactive", parent_ws_id="coord-root", state="idle", skill_id="skill-alpha", + user_id="user-1", ) st.register_workstream( "child-running", @@ -303,15 +306,17 @@ def seeded_storage(tmp_path): parent_ws_id="coord-root", state="running", skill_id="skill-beta", + user_id="user-1", ) # Coordinator child — MUST be excluded from list_children results. st.register_workstream( "child-coord", kind="coordinator", parent_ws_id="coord-root", + user_id="user-1", ) # Unrelated workstream with no parent — MUST be excluded. - st.register_workstream("unrelated-ws", kind="interactive") + st.register_workstream("unrelated-ws", kind="interactive", user_id="user-1") return st diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index eacd36eb..1f2ab2f4 100644 --- a/turnstone/api/server_schemas.py +++ b/turnstone/api/server_schemas.py @@ -6,6 +6,8 @@ from typing import Literal from pydantic import BaseModel, Field, model_validator +from turnstone.core.workstream import WorkstreamKind + # --------------------------------------------------------------------------- # Workstream management # --------------------------------------------------------------------------- @@ -140,8 +142,8 @@ class CreateWorkstreamRequest(BaseModel): "lands. Auto-generated when omitted." ), ) - kind: str = Field( - default="interactive", + kind: WorkstreamKind = Field( + default=WorkstreamKind.INTERACTIVE, description=( "Workstream kind — 'interactive' (default) or 'coordinator'. " "Coordinator workstreams are created by the console's own " diff --git a/turnstone/cli.py b/turnstone/cli.py index 75c9690d..98393ebe 100644 --- a/turnstone/cli.py +++ b/turnstone/cli.py @@ -17,7 +17,12 @@ from typing import TYPE_CHECKING, Any from turnstone.core.judge import JudgeConfig from turnstone.core.session import ChatSession, SessionUI -from turnstone.core.workstream import Workstream, WorkstreamManager, WorkstreamState +from turnstone.core.workstream import ( + Workstream, + WorkstreamKind, + WorkstreamManager, + WorkstreamState, +) from turnstone.ui.colors import ( BOLD, DIM, @@ -1130,7 +1135,7 @@ def main() -> None: *, skill: str | None = None, client_type: str = "", - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> ChatSession: assert ui is not None, "session_factory requires a non-None UI" diff --git a/turnstone/console/collector.py b/turnstone/console/collector.py index db7550c4..1562dfd3 100644 --- a/turnstone/console/collector.py +++ b/turnstone/console/collector.py @@ -22,6 +22,8 @@ from typing import TYPE_CHECKING, Any import httpx import httpx_sse +from turnstone.core.workstream import WorkstreamKind + if TYPE_CHECKING: from turnstone.console.metrics import ConsoleMetrics from turnstone.console.router import ConsoleRouter @@ -397,7 +399,7 @@ class ClusterCollector: "name": ws.get("title", "") or ws.get("name", ""), "title": ws.get("title", ""), "node_id": node_id, - "kind": ws.get("kind", "interactive"), + "kind": WorkstreamKind.from_raw(ws.get("kind")), "parent_ws_id": ws.get("parent_ws_id"), } ) @@ -419,7 +421,7 @@ class ClusterCollector: "node_id": node_id, "tokens": new_w.get("tokens", 0), "content": new_w.get("content", ""), - "kind": new_w.get("kind", "interactive"), + "kind": WorkstreamKind.from_raw(new_w.get("kind")), "parent_ws_id": new_w.get("parent_ws_id"), } ) @@ -485,7 +487,7 @@ class ClusterCollector: "node_id": node_id, "tokens": data.get("tokens", 0), "content": data.get("content", ""), - "kind": ws.get("kind", "interactive"), + "kind": WorkstreamKind.from_raw(ws.get("kind")), "parent_ws_id": ws.get("parent_ws_id"), } ) @@ -500,7 +502,7 @@ class ClusterCollector: elif etype == "ws_created": ws_id = data.get("ws_id", "") - ws_kind = data.get("kind", "interactive") + ws_kind = WorkstreamKind.from_raw(data.get("kind")) ws_parent = data.get("parent_ws_id") # user_id travels on the event so console-side fan-out # can enforce tenant isolation — a coordinator must diff --git a/turnstone/console/coordinator.py b/turnstone/console/coordinator.py index eda36673..8bb646ea 100644 --- a/turnstone/console/coordinator.py +++ b/turnstone/console/coordinator.py @@ -36,6 +36,7 @@ from typing import TYPE_CHECKING, Any from turnstone.core.log import get_logger from turnstone.core.workstream import ( Workstream, + WorkstreamKind, WorkstreamManager, WorkstreamState, ) @@ -177,7 +178,7 @@ class CoordinatorManager: node_id=self.NODE_ID, user_id=user_id, name=ws.name, - kind="coordinator", + kind=WorkstreamKind.COORDINATOR, parent_ws_id=None, ) except Exception: @@ -196,7 +197,7 @@ class CoordinatorManager: None, # model_alias — factory reads coordinator.model_alias ws_id, skill=skill, - kind="coordinator", + kind=WorkstreamKind.COORDINATOR, parent_ws_id=None, ) except Exception: @@ -278,7 +279,7 @@ class CoordinatorManager: return existing row = self._storage.get_workstream(ws_id) - if row is None or row.get("kind") != "coordinator": + if row is None or row.get("kind") != WorkstreamKind.COORDINATOR: return None # close()/delete() only soft-mark the row — refuse to # resurrect those sessions on any subsequent GET, or the @@ -325,7 +326,7 @@ class CoordinatorManager: None, ws_id, skill=None, - kind="coordinator", + kind=WorkstreamKind.COORDINATOR, parent_ws_id=None, ) except Exception: @@ -577,7 +578,7 @@ class CoordinatorManager: self._order.remove(oldest.id) evicted = oldest ws = Workstream(id=ws_id, name=name or f"coord-{ws_id[:4]}") - ws.kind = "coordinator" + ws.kind = WorkstreamKind.COORDINATOR ws.user_id = user_id ws.parent_ws_id = None # ui_factory is fast (just allocates a ConsoleCoordinatorUI) @@ -690,18 +691,29 @@ class CoordinatorManager: would drop that child. """ # Tenant filter: only accept persisted children whose owning - # user_id matches the coordinator's. A forged parent_ws_id - # that slipped past the server-side create gate (migration-era - # data, downgrade path) would otherwise keep re-leaking into - # the fan-out set on every console restart. + # user_id matches the coordinator's. Pushed into SQL via the + # ``user_id`` kwarg so cross-tenant rows never leave the DB — + # previously we fetched everything and filtered in Python, which + # depended on the in-loop user_id check catching forged + # parent_ws_id rows (migration-era data, downgrade path). An + # empty coord_user_id is fail-closed: skip the rebuild entirely + # rather than matching rows with blank owners (system-owned or + # legacy rows would otherwise leak into the fan-out set). with self._lock: coord_ws = self._workstreams.get(coord_ws_id) coord_user_id = coord_ws.user_id if coord_ws is not None else "" + if not coord_user_id: + log.debug( + "coord_mgr.rebuild_skipped_empty_owner coord=%s", + coord_ws_id[:8], + ) + return try: rows = self._storage.list_workstreams( limit=1000, parent_ws_id=coord_ws_id, kind=None, + user_id=coord_user_id, ) except Exception: log.debug( @@ -715,23 +727,10 @@ class CoordinatorManager: try: m = r._mapping child_id = m["ws_id"] - row_user = m.get("user_id", "") or "" - except (AttributeError, KeyError): + except AttributeError: child_id = r[0] if r else "" - row_user = "" if not child_id: continue - # Fail-closed: empty coord owner or mismatch → skip. If - # coord_user_id was unavailable (coordinator evicted - # mid-rebuild) we'd rather lose the registry seed than - # leak cross-tenant. - if not coord_user_id or row_user != coord_user_id: - log.debug( - "coord_mgr.rebuild_skipped_cross_tenant coord=%s child=%s", - coord_ws_id[:8], - str(child_id)[:8], - ) - continue child_ids.append(child_id) with self._children_lock: self._merge_child_ids_locked(coord_ws_id, child_ids) diff --git a/turnstone/console/coordinator_client.py b/turnstone/console/coordinator_client.py index bba1c56f..ee595ff4 100644 --- a/turnstone/console/coordinator_client.py +++ b/turnstone/console/coordinator_client.py @@ -35,6 +35,7 @@ import httpx from turnstone.core.auth import JWT_AUD_CONSOLE, create_jwt from turnstone.core.log import get_logger +from turnstone.core.workstream import WorkstreamKind _TASK_STATUSES = frozenset({"pending", "in_progress", "done", "blocked"}) # Hard cap on tasks per coordinator — the full list is read and re-serialized @@ -284,7 +285,7 @@ class CoordinatorClient: ) -> dict[str, Any]: """Create a child workstream via the routing proxy.""" body: dict[str, Any] = { - "kind": "interactive", + "kind": WorkstreamKind.INTERACTIVE.value, "parent_ws_id": parent_ws_id, "user_id": user_id, "initial_message": initial_message, @@ -372,10 +373,16 @@ class CoordinatorClient: """ if parent_ws_id != self._coord_ws_id: return {"children": [], "truncated": False} + # Tenant filter: push the coord's owner into SQL. Children of a + # coord share its owner by construction (the create-path + # parent_ws_id gate at server.py enforces this), but the filter + # is defense-in-depth for migration-era rows where that gate + # didn't exist yet. raw = self._storage.list_workstreams( limit=limit, parent_ws_id=parent_ws_id, - kind="interactive", + kind=WorkstreamKind.INTERACTIVE, + user_id=self._user_id or None, ) _terminal_states = {"closed", "deleted"} children: list[dict[str, Any]] = [] @@ -395,7 +402,7 @@ class CoordinatorClient: "state": row[3], "created": row[4], "updated": row[5], - "kind": row[6] if len(row) > 6 else "interactive", + "kind": WorkstreamKind.from_raw(row[6] if len(row) > 6 else None), "parent_ws_id": row[7] if len(row) > 7 else None, "skill_id": row[8] if len(row) > 8 else None, "skill_version": row[9] if len(row) > 9 else None, diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 1580ff30..4381e09f 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -51,6 +51,7 @@ from turnstone.core.auth import ( jwt_version_slot, ) from turnstone.core.hash_ring import NoAvailableNodeError +from turnstone.core.workstream import WorkstreamKind if TYPE_CHECKING: from collections.abc import AsyncGenerator @@ -414,7 +415,7 @@ def _coordinator_rows(request: Request) -> list[dict[str, Any]]: "activity": "", "activity_state": "", "tool_calls": 0, - "kind": "coordinator", + "kind": WorkstreamKind.COORDINATOR.value, "parent_ws_id": None, } ) @@ -586,14 +587,14 @@ async def _fetch_live_block( handler; internal degradations stay silent. """ row_node_id = row.get("node_id") or "" - row_kind = row.get("kind") or "interactive" + row_kind = WorkstreamKind.from_raw(row.get("kind")) # Kind is the authoritative discriminator; the `"console"` node_id # sentinel is paired with coordinator rows only (see # ``CoordinatorManager.NODE_ID``). Branching purely on kind avoids # a subtle collision if a real node ever registers with # ``node_id="console"``. - if row_kind == "coordinator": + if row_kind == WorkstreamKind.COORDINATOR: coord_mgr = getattr(request.app.state, "coord_mgr", None) if coord_mgr is None: return None @@ -2069,7 +2070,7 @@ def _resolve_coordinator_or_404( except Exception: log.debug("resolve_coordinator.storage_failed ws=%s", ws_id[:8], exc_info=True) return None, miss - if row is None or row.get("kind") != "coordinator": + if row is None or row.get("kind") != WorkstreamKind.COORDINATOR: return None, miss err = _check_row_owner_or_404( request, @@ -2586,7 +2587,7 @@ def _coord_children_row(row: Any) -> dict[str, Any]: "state": row[3], "created": row[4], "updated": row[5], - "kind": row[6] if len(row) > 6 else "interactive", + "kind": WorkstreamKind.from_raw(row[6] if len(row) > 6 else None), "parent_ws_id": row[7] if len(row) > 7 else None, "skill_id": row[8] if len(row) > 8 else None, "skill_version": row[9] if len(row) > 9 else None, @@ -2667,7 +2668,7 @@ async def coordinator_children(request: Request) -> JSONResponse: # produce them (WorkstreamManager rejects kind!=interactive), but # the filter keeps the tree UI contract stable across schema # changes. - if serialized.get("kind") == "coordinator": + if serialized.get("kind") == WorkstreamKind.COORDINATOR: continue items.append(serialized) if len(items) >= _CHILDREN_PAGE_LIMIT: diff --git a/turnstone/console/session_factory.py b/turnstone/console/session_factory.py index 7fa26bff..19fee167 100644 --- a/turnstone/console/session_factory.py +++ b/turnstone/console/session_factory.py @@ -24,6 +24,7 @@ from typing import TYPE_CHECKING from turnstone.core.log import get_logger from turnstone.core.session import ChatSession +from turnstone.core.workstream import WorkstreamKind from turnstone.prompts import ClientType if TYPE_CHECKING: @@ -86,13 +87,13 @@ def build_console_session_factory( *, skill: str | None = None, client_type: str = "web", - kind: str = "coordinator", + kind: WorkstreamKind = WorkstreamKind.COORDINATOR, parent_ws_id: str | None = None, ) -> ChatSession: assert ui is not None, "console session_factory requires a non-None UI" - if kind != "coordinator": + if kind != WorkstreamKind.COORDINATOR: raise ValueError( - f"console session factory only supports kind='coordinator', got {kind!r}" + f"console session factory only supports kind=COORDINATOR, got {kind!r}" ) # Resolve coordinator.model_alias from settings if caller didn't @@ -196,7 +197,7 @@ def build_console_session_factory( if client_type in {ct.value for ct in ClientType} else ClientType.WEB, username=_username, - kind="coordinator", + kind=WorkstreamKind.COORDINATOR, parent_ws_id=parent_ws_id, coord_client=coord_client, ) diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index 93a9b427..1bbc4876 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -18,6 +18,7 @@ import sqlalchemy as sa from turnstone.core.hash_ring import bucket_of as _bucket_of from turnstone.core.log import get_logger from turnstone.core.storage import get_storage +from turnstone.core.workstream import WorkstreamKind if TYPE_CHECKING: from collections.abc import Callable @@ -272,7 +273,7 @@ def register_workstream( skill_id: str = "", skill_version: int = 0, user_id: str | None = None, - kind: str = "interactive", + kind: WorkstreamKind | str = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> None: """Persist a new workstream (no-op if already exists).""" @@ -343,10 +344,23 @@ 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) -> list[Any]: - """List workstreams, optionally filtered by node_id.""" +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) + 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 [] diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 9f3f7e01..11cbcac3 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -103,6 +103,7 @@ from turnstone.core.tools import ( merge_mcp_tools, ) from turnstone.core.web import check_ssrf, strip_html +from turnstone.core.workstream import WorkstreamKind from turnstone.prompts import ClientType, SessionContext, compose_system_message from turnstone.ui.colors import DIM, GRAY, GREEN, RED, RESET, YELLOW, bold, cyan, dim @@ -330,14 +331,14 @@ class ChatSession: web_search_backend: str = "", client_type: ClientType = ClientType.CLI, username: str = "", - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, coord_client: Any = None, ): self.client = client self.model = model # Coordinator plumbing: populated by the console's session factory - # only — ``kind == "coordinator"`` sessions run COORDINATOR_TOOLS + # only — ``kind == COORDINATOR`` sessions run COORDINATOR_TOOLS # and dispatch tool execs through ``coord_client``. self._kind = kind self._parent_ws_id = parent_ws_id if parent_ws_id else None @@ -460,7 +461,7 @@ class ChatSession: # listeners register so tool/resource/prompt refreshes flow # through to this session. # * interactive (no mcp) — INTERACTIVE_TOOLS. - if kind == "coordinator": + if kind == WorkstreamKind.COORDINATOR: self._tools = list(COORDINATOR_TOOLS) self._task_tools = [] self._agent_tools = [] @@ -817,7 +818,7 @@ class ChatSession: return # Coordinator sessions don't consume MCP tools — the tool set # is fixed at COORDINATOR_TOOLS. Ignore MCP server changes. - if self._kind == "coordinator": + if self._kind == WorkstreamKind.COORDINATOR: return mcp_tools = self._mcp_client.get_tools() self._tools = merge_mcp_tools(INTERACTIVE_TOOLS, mcp_tools) diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index b2fddb8d..b3d8dc2f 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -99,6 +99,7 @@ from turnstone.core.storage._utils import sanitize_text from turnstone.core.storage._utils import ( scan_skill_content as _scan_skill_content, ) +from turnstone.core.workstream import WorkstreamKind log = get_logger(__name__) @@ -538,12 +539,14 @@ class PostgreSQLBackend: title: str | None = None, skill_id: str = "", skill_version: int = 0, - kind: str = "interactive", + kind: WorkstreamKind | str = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> None: from sqlalchemy.dialects.postgresql import insert as pg_insert now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + # Validate kind at the storage edge — see the sqlite sibling for rationale. + norm_kind = WorkstreamKind(kind).value # Normalize empty-string parent to NULL so WHERE parent_ws_id IS NULL # filters remain correct. norm_parent = parent_ws_id if parent_ws_id else None @@ -560,7 +563,7 @@ class PostgreSQLBackend: title=title, skill_id=skill_id, skill_version=skill_version, - kind=kind, + kind=norm_kind, parent_ws_id=norm_parent, created=now, updated=now, @@ -855,7 +858,8 @@ class PostgreSQLBackend: limit: int = 100, *, parent_ws_id: str | None = None, - kind: str | None = None, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, ) -> list[Any]: with self._conn() as conn: q = ( @@ -880,7 +884,9 @@ class PostgreSQLBackend: if parent_ws_id is not None: q = q.where(workstreams.c.parent_ws_id == parent_ws_id) if kind is not None: - q = q.where(workstreams.c.kind == kind) + q = q.where(workstreams.c.kind == WorkstreamKind(kind).value) + if user_id is not None: + q = q.where(workstreams.c.user_id == user_id) return list(conn.execute(q).fetchall()) # -- Conversation search --------------------------------------------------- diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 6727e63e..645ef34c 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -2,7 +2,10 @@ from __future__ import annotations -from typing import Any, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +if TYPE_CHECKING: + from turnstone.core.workstream import WorkstreamKind @runtime_checkable @@ -351,12 +354,14 @@ class StorageBackend(Protocol): title: str | None = None, skill_id: str = "", skill_version: int = 0, - kind: str = "interactive", + kind: WorkstreamKind | str = "interactive", parent_ws_id: str | None = None, ) -> None: """Create a workstreams row (no-op if already exists). - ``kind`` is ``"interactive"`` (default) or ``"coordinator"``. + ``kind`` accepts a ``WorkstreamKind`` member or its raw string value + (``"interactive"`` / ``"coordinator"``); the storage edge validates + the value and rejects unknown kinds with ``ValueError``. ``parent_ws_id`` is non-NULL for children spawned by a coordinator; the storage edge normalizes the empty string to ``None``. """ @@ -380,19 +385,24 @@ class StorageBackend(Protocol): limit: int = 100, *, parent_ws_id: str | None = None, - kind: str | None = None, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, ) -> list[Any]: """List workstreams, optionally filtered. - Filters are additive. When ``parent_ws_id`` / ``kind`` are ``None`` - (default) they are not applied — behavior is identical to the - pre-1.5 two-arg call shape. + Filters are additive. When ``parent_ws_id`` / ``kind`` / ``user_id`` + are ``None`` (default) they are not applied — behavior is identical + to the pre-1.5 two-arg call shape. + + ``user_id`` pushes ``WHERE user_id = :user_id`` into SQL so tenant + scoping is enforced server-side rather than relying on every + handler to remember a client-side filter. Pass the authenticated + caller's uid unless the caller holds a service scope. Returns a list of SQLAlchemy ``Row`` objects. **Prefer dict access via ``row._mapping[]``**; positional indexing is brittle against - future SELECT reorders. The current column order is ``ws_id, - node_id, name, state, created, updated, kind, parent_ws_id, - skill_id, skill_version`` but new code should not rely on that. + future SELECT reorders and against new columns appearing in the + tail (the select currently ends with ``user_id``). """ ... diff --git a/turnstone/core/storage/_schema.py b/turnstone/core/storage/_schema.py index 7ee103e1..b858c948 100644 --- a/turnstone/core/storage/_schema.py +++ b/turnstone/core/storage/_schema.py @@ -54,8 +54,9 @@ workstreams = sa.Table( sa.Column("state", sa.Text, nullable=False, server_default="idle"), sa.Column("skill_id", sa.Text, nullable=False, server_default=""), sa.Column("skill_version", sa.Integer, nullable=False, server_default="0"), - # kind: "interactive" (default) | "coordinator" | future kinds. - # Added in migration 039. Coordinator entries live only on the console. + # kind: "interactive" (default) | "coordinator". See WorkstreamKind in + # turnstone.core.workstream. Added in migration 039; coordinator entries + # live only on the console. sa.Column("kind", sa.Text, nullable=False, server_default="interactive"), # parent_ws_id: non-NULL for children spawned by a coordinator. NULL for # top-level workstreams (including the coordinators themselves). See diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index 96f9c8f1..35b67d69 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -99,6 +99,7 @@ from turnstone.core.storage._utils import sanitize_text from turnstone.core.storage._utils import ( scan_skill_content as _scan_skill_content, ) +from turnstone.core.workstream import WorkstreamKind log = get_logger(__name__) @@ -638,10 +639,17 @@ class SQLiteBackend: title: str | None = None, skill_id: str = "", skill_version: int = 0, - kind: str = "interactive", + kind: WorkstreamKind | str = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> None: now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S") + # Kind validation at the storage edge — third of three layers + # (HTTP handler in server.py returns 400, WorkstreamManager.create + # raises for in-process callers, here we reject direct storage + # callers: SDK inserts, restore paths, test doubles). Each layer + # targets a different audience; trimming any one of them opens a + # corresponding path to silently corrupt the NOT NULL column. + norm_kind = WorkstreamKind(kind).value # Normalize empty-string parent to NULL so WHERE parent_ws_id IS NULL # filters remain correct. norm_parent = parent_ws_id if parent_ws_id else None @@ -658,7 +666,7 @@ class SQLiteBackend: "state": state, "skill_id": skill_id, "skill_version": skill_version, - "kind": kind, + "kind": norm_kind, "parent_ws_id": norm_parent, "created": now, "updated": now, @@ -957,7 +965,8 @@ class SQLiteBackend: limit: int = 100, *, parent_ws_id: str | None = None, - kind: str | None = None, + kind: WorkstreamKind | str | None = None, + user_id: str | None = None, ) -> list[Any]: with self._conn() as conn: q = ( @@ -982,7 +991,9 @@ class SQLiteBackend: if parent_ws_id is not None: q = q.where(workstreams.c.parent_ws_id == parent_ws_id) if kind is not None: - q = q.where(workstreams.c.kind == kind) + q = q.where(workstreams.c.kind == WorkstreamKind(kind).value) + if user_id is not None: + q = q.where(workstreams.c.user_id == user_id) return list(conn.execute(q).fetchall()) # -- Conversation search --------------------------------------------------- diff --git a/turnstone/core/workstream.py b/turnstone/core/workstream.py index 36b1174e..3902ac77 100644 --- a/turnstone/core/workstream.py +++ b/turnstone/core/workstream.py @@ -30,11 +30,56 @@ if TYPE_CHECKING: *, skill: str | None = ..., client_type: str = ..., - kind: str = ..., + kind: WorkstreamKind = ..., parent_ws_id: str | None = ..., ) -> ChatSession: ... +# --------------------------------------------------------------------------- +# Kind enum — single source of truth for the workstream dispatch classifier +# --------------------------------------------------------------------------- + + +class WorkstreamKind(enum.StrEnum): + """Classifier for which manager hosts a workstream. + + StrEnum so members are drop-in ``str`` replacements for the DB column, + JSON payloads, and existing ``==`` comparisons against raw strings. + Narrow internal annotations to this type; wide boundaries (HTTP body, + DB row) stay ``str`` and parse via ``WorkstreamKind(raw)`` / ``from_raw`` + at the edge. + """ + + INTERACTIVE = "interactive" # hosted by turnstone-server.WorkstreamManager + COORDINATOR = "coordinator" # hosted by turnstone-console.CoordinatorManager + + @classmethod + def from_raw( + cls, + value: WorkstreamKind | str | None, + *, + default: WorkstreamKind | None = None, + ) -> WorkstreamKind: + """Parse an externally-supplied kind value with a fallback for missing data. + + Handles the three shapes that arrive from storage rows and wire + payloads — already-an-enum, non-empty string, None/empty — so the + ``WorkstreamKind(x or WorkstreamKind.INTERACTIVE.value)`` dance + (``or`` short-circuits on a truthy enum member and skips the + default, forcing every caller to reach for ``.value``) collapses + into a single predictable call. + + ``default`` defaults to ``INTERACTIVE`` when omitted. Raises + ``ValueError`` for a non-empty string that doesn't match any + known kind — callers that want to coerce unknowns to the default + should catch and fall back explicitly. + """ + effective_default = default if default is not None else cls.INTERACTIVE + if value is None or value == "": + return effective_default + return cls(value) + + # --------------------------------------------------------------------------- # State enum # --------------------------------------------------------------------------- @@ -70,9 +115,9 @@ class Workstream: # Owning user_id. Populated by the console's CoordinatorManager so # attribution survives across restarts / lazy rehydration. user_id: str = "" - # "interactive" (default) or "coordinator". Reused by both - # WorkstreamManager and CoordinatorManager — no parallel type hierarchy. - kind: str = "interactive" + # Classifier reused by both WorkstreamManager and CoordinatorManager — + # no parallel type hierarchy. + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE # Non-None for children spawned by a coordinator. parent_ws_id: str | None = None _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) @@ -152,7 +197,7 @@ class WorkstreamManager: client_type: str = "", judge_model: str | None = None, user_id: str = "", - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> Workstream: """Create a new workstream. Returns the new ws. @@ -170,16 +215,26 @@ class WorkstreamManager: ws_id: Optional workstream ID. If non-empty, used as-is instead of generating a new UUID. user_id: Owning user id, persisted on the dataclass + storage row. - kind: Workstream kind — must be ``"interactive"``. Coordinator - workstreams live on the console process and are created - via ``CoordinatorManager``; routing one through + kind: Workstream kind — must be ``WorkstreamKind.INTERACTIVE``. + Coordinator workstreams live on the console process and are + created via ``CoordinatorManager``; routing one through ``WorkstreamManager`` silently builds a coordinator-kind session with ``coord_client=None``, so this guard refuses at the boundary rather than failing deep inside tool execution. parent_ws_id: Optional parent workstream id (coordinator children). """ - if kind != "interactive": + # Kind validation lives in three layers. This one is the + # in-process guard: direct callers (tests, CLI factory, any + # future non-HTTP entry point) land here without passing + # through the server's body_kind parser, so a defensive raise + # stops a bug upstream from silently constructing a coord-kind + # session with coord_client=None. The HTTP layer + # (server.py POST /v1/api/workstreams/new) returns 400 for + # user-facing feedback; the storage layer (register_workstream + # in both backends) re-checks via WorkstreamKind(kind).value so + # restore paths and SDK inserts can't corrupt the column. + if kind != WorkstreamKind.INTERACTIVE: raise ValueError( f"WorkstreamManager only hosts interactive workstreams; refusing kind={kind!r}", ) diff --git a/turnstone/prompts/__init__.py b/turnstone/prompts/__init__.py index c3a0f76c..d3ed28df 100644 --- a/turnstone/prompts/__init__.py +++ b/turnstone/prompts/__init__.py @@ -15,6 +15,8 @@ from datetime import datetime from pathlib import Path from typing import Any +from turnstone.core.workstream import WorkstreamKind + log = logging.getLogger(__name__) _PROMPTS_DIR = Path(__file__).resolve().parent # turnstone/prompts/ @@ -88,7 +90,7 @@ def compose_system_message( available_tools: frozenset[str], policies: list[str] | None = None, db_policies: list[dict[str, Any]] | None = None, - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, ) -> str: """Compose a system message from modular components. @@ -125,7 +127,7 @@ def compose_system_message( # model as an IC engineer ("you read before you edit, commits # you make..."); coordinators need an orchestrator framing # instead ("you decompose, delegate, monitor, synthesise"). - base_module = "base_coordinator.md" if kind == "coordinator" else "base.md" + base_module = "base_coordinator.md" if kind == WorkstreamKind.COORDINATOR else "base.md" parts.append(_load(base_module)) # 2. ENV — exactly one, selected by client type @@ -140,7 +142,7 @@ def compose_system_message( # 4. TOOLS — kind-specific patterns. Coordinators get the # orchestrator block; interactive sessions get the IC block. if available_tools: - tools_module = "tools_coordinator.md" if kind == "coordinator" else "tools.md" + tools_module = "tools_coordinator.md" if kind == WorkstreamKind.COORDINATOR else "tools.md" parts.append(_load(tools_module)) # 5. POLICIES — resolve from DB first, fall back to files diff --git a/turnstone/server.py b/turnstone/server.py index 1ad1120f..d81027a7 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -49,7 +49,12 @@ from turnstone.core.ratelimit import resolve_client_ip from turnstone.core.session import ChatSession, GenerationCancelled, SessionUI # noqa: F401 from turnstone.core.tools import TOOLS # noqa: F401 — available for introspection from turnstone.core.web_helpers import version_html as _version_html -from turnstone.core.workstream import Workstream, WorkstreamManager, WorkstreamState +from turnstone.core.workstream import ( + Workstream, + WorkstreamKind, + WorkstreamManager, + WorkstreamState, +) from turnstone.prompts import ClientType if TYPE_CHECKING: @@ -102,7 +107,7 @@ class WebUI: ws_id: str = "", user_id: str = "", *, - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> None: self.ws_id = ws_id @@ -111,7 +116,11 @@ class WebUI: # the lifetime of the workstream, so locking the manager on # every state/activity tick to re-read them burns lock budget. self._kind = kind - self._parent_ws_id = parent_ws_id + # Normalize empty string to None at the UI boundary so the + # invariant "parent_ws_id is either a non-empty string or None" + # holds in every ws_state/ws_activity event payload — mirrors + # the storage-layer normalization at register_workstream. + self._parent_ws_id = parent_ws_id if parent_ws_id else None self._listeners: list[queue.Queue[dict[str, Any]]] = [] self._listeners_lock = threading.Lock() self._approval_event = threading.Event() @@ -170,7 +179,7 @@ class WebUI: with self._listeners_lock, contextlib.suppress(ValueError): self._listeners.remove(client_queue) - def _ws_kind_and_parent(self) -> tuple[str, str | None]: + def _ws_kind_and_parent(self) -> tuple[WorkstreamKind, str | None]: """Return cached (kind, parent_ws_id) for broadcast event payloads. Stored on the UI at construction time — both fields are @@ -2370,12 +2379,21 @@ async def create_workstream(request: Request) -> JSONResponse: ) # Kind + parent relationship: coordinator-spawned children forward # these from the console's CoordinatorClient. Default kind is - # "interactive"; coordinators themselves are created by the console's + # ``INTERACTIVE``; coordinators themselves are created by the console's # own CoordinatorManager and never land in this handler. Only - # ``"interactive"`` is accepted here — requests that try to create a - # coordinator via the generic workstream endpoint are rejected. - body_kind = body.get("kind", "interactive") or "interactive" - if body_kind != "interactive": + # ``INTERACTIVE`` is accepted here — requests that try to create a + # coordinator via the generic workstream endpoint are rejected, and + # unknown kinds (typos, future-only values) are 400 rather than being + # silently coerced. User-facing edge: the storage_edge and manager + # layers below re-validate, see comments at those sites for why. + try: + body_kind = WorkstreamKind.from_raw(body.get("kind")) + except ValueError: + return JSONResponse( + {"error": f"unknown workstream kind {body.get('kind')!r}"}, + status_code=400, + ) + if body_kind != WorkstreamKind.INTERACTIVE: return JSONResponse( {"error": "coordinator workstreams must be created via /v1/api/coordinator/new"}, status_code=400, @@ -2401,7 +2419,10 @@ async def create_workstream(request: Request) -> JSONResponse: {"error": "parent_ws_id does not reference a known workstream"}, status_code=400, ) - if parent_row.get("kind") != "coordinator" or (parent_row.get("user_id") or "") != uid: + if ( + parent_row.get("kind") != WorkstreamKind.COORDINATOR + or (parent_row.get("user_id") or "") != uid + ): return JSONResponse( {"error": "parent_ws_id must reference a coordinator you own"}, status_code=403, @@ -3144,7 +3165,7 @@ async def open_workstream(request: Request) -> JSONResponse: # Coordinator workstreams live on the console process, not on server # nodes. Refuse to rehydrate one here so we don't silently build a # coordinator-kind ChatSession with coord_client=None (A/S1 guard). - if ws_row.get("kind") != "interactive": + if ws_row.get("kind") != WorkstreamKind.INTERACTIVE: return JSONResponse( {"error": "Workstream is not an interactive kind"}, status_code=400, @@ -3164,7 +3185,7 @@ async def open_workstream(request: Request) -> JSONResponse: ui_factory=lambda wid, **kw: WebUI(ws_id=wid, user_id=owner_uid, **kw), ws_id=resolved_id, user_id=owner_uid, - kind=ws_row.get("kind") or "interactive", + kind=WorkstreamKind.from_raw(ws_row.get("kind")), parent_ws_id=ws_row.get("parent_ws_id"), ) except Exception as e: @@ -4676,7 +4697,7 @@ def main() -> None: skill: str | None = None, client_type: str = "", judge_model: str | None = None, - kind: str = "interactive", + kind: WorkstreamKind = WorkstreamKind.INTERACTIVE, parent_ws_id: str | None = None, ) -> ChatSession: assert ui is not None