fix(server): close review cleanup items from PRs #374 / #375 review (#376)

Third and final PR of the retrospective-review series.  Addresses the
remaining bug / perf / doc findings from the original multi-stage review
plus the three inline comments left on #374 and #375.

From the original review:

- bug-3: delete_workstream now nulls out parent_ws_id on every child
  row before dropping the target — previously, deleting a coordinator
  left orphaned parent_ws_id pointers and list_workstreams(parent_ws_id=
  <deleted>) kept returning ghost-parented rows.  Fix lives at the
  storage edge so both SQLite and PostgreSQL benefit without a schema
  migration.
- perf-1 / perf-2 / perf-3: new migration 041 drops the low-cardinality
  idx_workstreams_kind outright, rebuilds idx_workstreams_parent as a
  partial index (WHERE parent_ws_id IS NOT NULL) to halve its btree,
  and uses CREATE INDEX CONCURRENTLY on postgres so the rebuild
  doesn't take ACCESS EXCLUSIVE on populated tables.  Dialect-guarded;
  sqlite path is a straight partial CREATE INDEX.
- perf-5: _rebuild_children_from_storage bumps its limit sentinel to
  10_000 and logs a warning when the cap is hit instead of silently
  truncating the tail on every console cold-start.
- q-2: turnstone.core.memory.list_workstreams wrapper deleted (zero
  live callers; PR #374 kept it forward-compatible with the new
  kwargs as a stepping stone).
- q-5: migration 039's docstring now warns operators that downgrade
  drops parent_ws_id irreversibly and notes the 041 dependency.
- q-7: GET /v1/api/workstreams row shape now includes kind +
  parent_ws_id to match /v1/api/dashboard; the Pydantic
  WorkstreamInfo schema follows so SDK consumers see the same fields.

Inline review comments:

- #374 (copilot): console/server.py::coordinator_children now pushes
  user_id into the SQL filter for non-admin callers, so forged /
  migration-era rows with matching parent_ws_id but a different
  owner can't leak through.  Admins bypass the filter — they're
  expected to see the full subtree.
- #375 (copilot, delete handler): storage.get_workstream(ws_id) for
  the audit snapshot moved inside the try: block so a transient DB
  error surfaces through the endpoint's redacted 500 handler instead
  of an unhandled exception.
- #375 (copilot, _require_ws_access): added optional mgr= kwarg —
  when the workstream is live in the in-memory manager, trust its
  cached user_id instead of round-tripping storage.  In-memory-only
  handlers (approve / plan / cancel / command / close / events_sse /
  refresh-title / set-title) pass mgr= so they stay functional
  during transient DB outages and skip one query on the hot path.
  Storage-backed handlers (/delete, /open) omit mgr= and keep the
  storage path for persisted-but-not-loaded rows.

Tests:

- tests/test_workstream_kind.py adds regression tests for the cascade
  null-out on delete and the new user_id SQL filter.
- tests/test_workstream_endpoints.py updated so the title-handler
  tests exercise the in-memory fast path (MagicMock manager returning
  None falls through to storage; explicit ws.user_id set where the
  mock ws is used).

Lint (ruff), typecheck (strict mypy), pytest -m 'not live' all green
(4209 passing).

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

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=<deleted>) 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."""

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

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:

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)

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:

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

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=<deleted>)`` 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

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

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 = <coord_ws>``,
+  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"])

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