feat(core): WorkstreamKind enum + list_workstreams user_id filter (#374)

Foundation PR for the multi-stage-review follow-up.  Introduces a
single source of truth for workstream kind values and pushes tenant
scoping into the storage protocol so list callers can't forget to
filter client-side.

- WorkstreamKind(StrEnum) replaces bare "interactive" / "coordinator"
  literals across 17 production modules.  Strict mypy narrows every
  internal call site; raw strings still work at wide boundaries
  (HTTP body, DB row) via WorkstreamKind(raw) parse at the edge.
- StorageBackend.list_workstreams(..., user_id=None) adds a SQL-level
  WHERE user_id = :user_id gate on both sqlite and postgres impls.
  Memory wrapper forwards the new filters.
- register_workstream now validates kind at the storage edge so SDK /
  restore / internal callers can't silently corrupt the NOT NULL
  column with empty / mis-cased / unknown values.
- WebUI.__init__ normalizes empty-string parent_ws_id to None, matching
  the storage-edge and WorkstreamManager invariants.
- POST /v1/api/workstreams/new parses body["kind"] through the enum
  and returns 400 on unknown kinds instead of silent coercion.

Absorbs bug-1, bug-2, bug-4/q-6, q-1, q-8, and partial q-2 (wrapper
signature forwards the new filters; full deletion of the unused
wrapper stays in the cleanup PR).

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) |

docs/diagrams/14-storage-architecture.puml

@@ -23,7 +23,7 @@ interface "StorageBackend" as SB <<protocol>> {
     +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

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")

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
 
 

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 "

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"

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

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)

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,

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:

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,
         )

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 []

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)

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 ---------------------------------------------------

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[<col>]``**; 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``).
         """
         ...
 

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

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 ---------------------------------------------------

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}",
             )

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

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