fix(personas): apply review findings — stamp adoption on fork/restore, PATCH semantics, gating

Review pass over the branch surfaced real defects, all fixed here with regression guards: - Fork-resume (resume_ws) adopts the SOURCE workstream's stamp, resolved pre-construction so all four levers (including the construction-time MCP gate) bind the fork; a corrupt source stamp is a loud 400, an unstamped legacy source forks unstamped — never the kind default. Watch-restore and CLI --resume thread the stamp the same way, closing an MCP leak where a restored MCP-off workstream re-merged the catalog. - SessionManager.open parses the stamp inside the install guard so a corrupt stamp releases the reserved slot; a retry reproduces the loud error instead of 'already tracked'. - Mid-session resume() adopting a stamp rebuilds the tool_search pathway to match (hard set drops it, soft set force-constructs it); soft persona sets survive the global tool-search setting being off. - Memory nudges gate on actual memory-tool VISIBILITY, not just the memory lever, so an allowlist that hides the tool also silences the nudges that point at it; post-compaction resume gets a no-recall nudge variant when the pointer would dangle. - Console PATCH: explicit null flags from UpdatePersonaRequest no longer archive the persona or flip levers on a rename; multi-kind personas survive a shelf edit; admin list ships the per-kind tool_inventory so the shelf checklist tracks the server inventory instead of a hardcoded JS list; admin CRUD moved off the event loop. - Migration 063 converts legacy creative_mode workstreams to the full writer stamp (downgrade removes all persona keys). - REPL: /new passes the persona; /workstreams unpacks the widened row. - Shared resolve_persona_for_kind is the single eligibility rule for the HTTP handler, CLI, and spawn precheck; spawn_batch memoizes the persona lookup; ToolSearchManager.is_expanded gives the visibility tail an O(1) probe.
2026-08-12 23:12:23 -06:00 · 2026-07-02 13:18:08 +00:00
parent 0d6d7ebae1
commit e2dcd2bd6b
17 changed files with 647 additions and 104 deletions
@@ -145,6 +145,64 @@ class TestMigration063:
        finally:
            engine.dispose()

+    def test_converts_legacy_creative_workstreams_to_writer(self, tmp_path: Path) -> None:
+        db_path = tmp_path / "063-creative.db"
+        cfg = _alembic_cfg(db_path)
+        command.upgrade(cfg, "062")
+
+        engine = sa.create_engine(f"sqlite:///{db_path}")
+        try:
+            with engine.begin() as conn:
+                for ws_id, mode in (("ws-creative", "True"), ("ws-plain", "False")):
+                    conn.execute(
+                        sa.text(
+                            "INSERT INTO workstreams (ws_id, name, state, created, updated) "
+                            "VALUES (:ws, :ws, 'closed', '2026-01-01T00:00:00', "
+                            "'2026-01-01T00:00:00')"
+                        ),
+                        {"ws": ws_id},
+                    )
+                    conn.execute(
+                        sa.text(
+                            "INSERT INTO workstream_config (ws_id, key, value) "
+                            "VALUES (:ws, 'creative_mode', :mode)"
+                        ),
+                        {"ws": ws_id, "mode": mode},
+                    )
+
+            command.upgrade(cfg, "063")
+
+            with engine.connect() as conn:
+                stamped = {
+                    str(r[0]): str(r[1])
+                    for r in conn.execute(
+                        sa.text(
+                            "SELECT ws_id, value FROM workstream_config WHERE key='persona'"
+                        )
+                    ).fetchall()
+                }
+                cols = conn.execute(
+                    sa.text(
+                        "SELECT key, value FROM workstream_config "
+                        "WHERE ws_id='ws-creative' AND key LIKE 'persona%'"
+                    )
+                ).fetchall()
+                row_persona = conn.execute(
+                    sa.text("SELECT persona FROM workstreams WHERE ws_id='ws-creative'")
+                ).fetchone()
+            # creative_mode='True' → the full writer stamp (all five keys)…
+            assert stamped == {"ws-creative": "writer"}
+            keys = {str(k): str(v) for k, v in cols}
+            assert keys["persona_tools"] == "[]"
+            assert keys["persona_mcp"] == "0"
+            assert keys["persona_memory"] == "1"
+            assert "creative writing partner" in keys["persona_prompt"]
+            assert row_persona is not None and row_persona[0] == "writer"
+            # …while creative_mode='False' workstreams stay legacy-unstamped.
+            assert "ws-plain" not in stamped
+        finally:
+            engine.dispose()
+
    def test_downgrade_reverses_everything(self, tmp_path: Path) -> None:
        db_path = tmp_path / "063-down.db"
        cfg = _alembic_cfg(db_path)
@@ -174,6 +174,39 @@ class TestRouteContracts:
        assert resp.status_code == 400
        assert "archived" in resp.json()["error"]

+    def test_patch_null_flags_leave_persona_unchanged(self, tmp_db: Any, seeded: str) -> None:
+        # Clients built from UpdatePersonaRequest (every flag boolean|null)
+        # serialize unset fields as explicit null — a rename must not archive
+        # the persona or flip its levers as a side effect.
+        c = _client(tmp_db, _ALL)
+        resp = c.patch(
+            "/v1/api/admin/personas/" + seeded,
+            json={
+                "display_name": "Renamed",
+                "enabled": None,
+                "mcp_enabled": None,
+                "memory_enabled": None,
+                "is_default": None,
+                "applies_to_kinds": None,
+            },
+        )
+        assert resp.status_code == 200
+        row = resp.json()
+        assert row["display_name"] == "Renamed"
+        assert row["enabled"] is True  # NOT archived by the null
+        assert row["mcp_enabled"] is False  # seeded value preserved
+        assert row["applies_to_kinds"] == ["interactive"]
+
+    def test_list_carries_tool_inventory(self, tmp_db: Any, seeded: str) -> None:
+        c = _client(tmp_db, _ALL)
+        inv = c.get("/v1/api/admin/personas").json()["tool_inventory"]
+        assert "read_file" in inv["interactive"]
+        assert "spawn_workstream" in inv["coordinator"]
+        # tool_search is synthetic but listed — its membership decides
+        # whether an authored set is soft or hard.
+        assert "tool_search" in inv["interactive"]
+        assert "tool_search" in inv["coordinator"]
+
    def test_missing_persona_is_404(self, tmp_db: Any) -> None:
        c = _client(tmp_db, _ALL)
        assert c.get("/v1/api/admin/personas/nope").status_code == 404
@@ -157,6 +157,41 @@ class TestToolSearchEscape:
        names = _wire_names(session)
        assert "mcp_widget" in names

+    def test_soft_set_survives_global_setting_off(self, tmp_db, mock_openai_client) -> None:
+        # The authored escape hatch must not silently degrade to a hard set
+        # on deployments where the global setting/threshold wouldn't have
+        # constructed a ToolSearchManager.
+        session = _session(
+            mock_openai_client,
+            mcp_client=self._mcp_client(),
+            tool_search="off",
+            persona_snapshot=_snap(tools=frozenset({"read_file", "tool_search"})),
+        )
+        assert session._tool_search is not None
+        assert "tool_search" in _wire_names(session)
+
+    def test_mid_session_adopt_of_hard_set_drops_tool_search(
+        self, tmp_db, mock_openai_client
+    ) -> None:
+        from turnstone.core.memory import register_workstream, save_workstream_config
+
+        # Unstamped session with a live ToolSearchManager and a discovered tool.
+        session = _session(mock_openai_client, mcp_client=self._mcp_client(), tool_search="on")
+        assert session._tool_search is not None
+        session._tool_search.expand_visible(["mcp_widget"])
+        # Adopt a hard-set (scribe-shaped) stamp via non-fork resume.
+        register_workstream("t" * 32)
+        from turnstone.core.memory import save_message
+
+        save_message("t" * 32, "user", "hi")
+        save_workstream_config(
+            "t" * 32, _snap(name="scribe", tools=frozenset(), memory=False).to_config()
+        )
+        assert session.resume("t" * 32)
+        # The pathway is re-gated: no manager, no hint target, no escape hatch.
+        assert session._tool_search is None
+        assert _wire_names(session) == []
+
    def test_omitted_is_hard(self, tmp_db, mock_openai_client) -> None:
        session = _session(
            mock_openai_client,
@@ -194,6 +229,19 @@ class TestMemoryOff:
        assert session._nudges_enabled("repeat")
        assert session._nudges_enabled("compaction_pending")

+    def test_allowlist_hiding_memory_tool_also_gates_nudges(
+        self, tmp_db, mock_openai_client
+    ) -> None:
+        # memory lever ON but the visibility set omits the memory tool —
+        # nudges directing the model at memory(...) must not fire either.
+        session = _session(
+            mock_openai_client,
+            persona_snapshot=_snap(tools=frozenset({"read_file"}), memory=True),
+        )
+        session._memory_config.nudges = True
+        assert not session._nudges_enabled("start")
+        assert session._nudges_enabled("repeat")
+
    def test_recall_pointer_gates_on_visibility(self, tmp_db, mock_openai_client) -> None:
        # scribe-shaped: empty toolset hides recall — the compaction pointer
        # must not direct the model at a tool it can't call.
@@ -402,6 +450,23 @@ class TestRehydrateThreading:
        with pytest.raises(ValueError, match="corrupt persona snapshot"):
            mgr.open(ws_id)

+    def test_corrupt_stamp_releases_the_slot(self) -> None:
+        # The parse raise must unwind exactly like a build_session failure:
+        # the placeholder slot is released (no max_active pin) and a retry
+        # raises the SAME loud error instead of "already tracked".
+        from tests.test_session_manager import _make_manager
+
+        mgr, _, storage = _make_manager()
+        ws = mgr.create(user_id="u1")
+        ws_id = ws.id
+        storage.ws_config[ws_id] = {"persona": "scribe"}
+        mgr.close(ws_id)
+        with pytest.raises(ValueError, match="corrupt persona snapshot"):
+            mgr.open(ws_id)
+        assert mgr.get(ws_id) is None  # slot released, not a stuck placeholder
+        with pytest.raises(ValueError, match="corrupt persona snapshot"):
+            mgr.open(ws_id)  # retry reproduces the loud error, not RuntimeError
+

 # ---------------------------------------------------------------------------
 # Guard 10 — mandatory prompt policies compose under EVERY persona, including
@@ -559,3 +624,166 @@ def test_snapshot_roundtrip_via_json() -> None:
    # The stamp's config form is plain strings — JSON-safe end to end.
    snap = _snap(name="s", prompt="p", tools=frozenset({"a"}), mcp=False, memory=False)
    assert json.loads(json.dumps(snap.to_config())) == snap.to_config()
+
+
+# ---------------------------------------------------------------------------
+# Guard 9, fork lane — creating with ``resume_ws`` adopts the SOURCE
+# workstream's stamp at construction time (all four levers, including the
+# construction-time MCP gate), never the kind default; a corrupt source
+# stamp is a loud 400; an unstamped legacy source forks unstamped.
+# ---------------------------------------------------------------------------
+
+
+class TestForkAdoptsStamp:
+    @pytest.fixture()
+    def _fork_app(self, tmp_db):
+        """The production ``make_create_handler`` over a real SessionManager,
+        with a session factory that forwards ``persona_snapshot`` (the same
+        contract the server factory honors)."""
+        import queue
+        import threading
+
+        from starlette.applications import Starlette
+        from starlette.middleware import Middleware
+        from starlette.middleware.base import BaseHTTPMiddleware
+        from starlette.routing import Mount, Route
+        from starlette.testclient import TestClient
+
+        from turnstone.core.adapters.interactive_adapter import InteractiveAdapter
+        from turnstone.core.auth import AuthResult
+        from turnstone.core.session_manager import SessionManager
+        from turnstone.core.session_routes import SessionEndpointConfig, make_create_handler
+        from turnstone.server import (
+            WebUI,
+            _interactive_create_build_kwargs,
+            _interactive_create_post_install,
+            _interactive_create_validate_request,
+            _interactive_manager_lookup,
+            _interactive_tenant_check,
+        )
+
+        class _Auth(BaseHTTPMiddleware):
+            async def dispatch(self, request: Any, call_next: Any) -> Any:
+                request.state.auth_result = AuthResult(
+                    user_id="test-user",
+                    scopes=frozenset({"approve"}),
+                    token_source="config",
+                    permissions=frozenset({"read", "write", "approve"}),
+                )
+                return await call_next(request)
+
+        def _session_factory(ui: Any, model_alias: Any = None, ws_id: Any = None, **kw: Any):
+            return ChatSession(
+                client=MagicMock(),
+                model=model_alias or "test-model",
+                ui=ui,
+                instructions=None,
+                temperature=0.5,
+                max_tokens=1000,
+                tool_timeout=10,
+                ws_id=ws_id,
+                persona_snapshot=kw.get("persona_snapshot"),
+            )
+
+        gq: queue.Queue[dict[str, Any]] = queue.Queue()
+        WebUI._global_queue = gq
+        adapter = InteractiveAdapter(
+            global_queue=gq,
+            ui_factory=lambda ws: WebUI(
+                ws_id=ws.id,
+                user_id=ws.user_id,
+                kind=ws.kind,
+                parent_ws_id=ws.parent_ws_id,
+            ),
+            session_factory=_session_factory,
+        )
+        mgr = SessionManager(
+            adapter, storage=get_storage(), max_active=10, event_emitter=adapter
+        )
+        handler = make_create_handler(
+            SessionEndpointConfig(
+                permission_gate=None,
+                manager_lookup=_interactive_manager_lookup,
+                tenant_check=_interactive_tenant_check,
+                not_found_label="Workstream not found",
+                audit_action_prefix="workstream",
+                create_supports_attachments=True,
+                create_supports_user_id_override=True,
+                create_validate_request=_interactive_create_validate_request,
+                create_build_kwargs=_interactive_create_build_kwargs,
+                create_post_install=_interactive_create_post_install,
+            )
+        )
+        app = Starlette(
+            routes=[
+                Mount(
+                    "/v1",
+                    routes=[Route("/api/workstreams/new", handler, methods=["POST"])],
+                )
+            ],
+            middleware=[Middleware(_Auth)],
+        )
+        app.state.workstreams = mgr
+        app.state.skip_permissions = True
+        app.state.global_queue = gq
+        app.state.global_listeners = []
+        app.state.global_listeners_lock = threading.Lock()
+
+        yield TestClient(app, raise_server_exceptions=False), mgr
+
+    def _seed_default(self) -> None:
+        get_storage().create_persona(
+            {
+                "persona_id": "pd",
+                "name": "engineer",
+                "applies_to_kinds": ["interactive"],
+                "is_default": True,
+            }
+        )
+
+    def test_fork_adopts_source_stamp_not_default(self, _fork_app) -> None:
+        from turnstone.core.memory import register_workstream, save_workstream_config
+
+        client, mgr = _fork_app
+        self._seed_default()  # present, and must LOSE to the source stamp
+        src = "s" * 32
+        register_workstream(src)
+        snap = _snap(name="scribe", tools=frozenset(), mcp=False, memory=False)
+        save_workstream_config(src, snap.to_config())
+
+        resp = client.post("/v1/api/workstreams/new", json={"resume_ws": src})
+        assert resp.status_code == 200, resp.text
+        ws = mgr.get(resp.json()["ws_id"])
+        assert ws is not None and ws.session is not None
+        assert ws.persona == "scribe"
+        assert ws.session._persona_name == "scribe"
+        assert ws.session._persona_tools == frozenset()
+        assert ws.session._persona_mcp is False
+        assert ws.session._persona_memory is False
+
+    def test_corrupt_source_stamp_is_400(self, _fork_app) -> None:
+        from turnstone.core.memory import register_workstream, save_workstream_config
+
+        client, mgr = _fork_app
+        src = "c" * 32
+        register_workstream(src)
+        save_workstream_config(src, {"persona": "scribe"})  # partial = corrupt
+
+        resp = client.post("/v1/api/workstreams/new", json={"resume_ws": src})
+        assert resp.status_code == 400
+        assert "cannot fork" in resp.json()["error"]
+
+    def test_unstamped_legacy_source_forks_unstamped(self, _fork_app) -> None:
+        from turnstone.core.memory import register_workstream
+
+        client, mgr = _fork_app
+        self._seed_default()  # the default must NOT leak onto the fork
+        src = "l" * 32
+        register_workstream(src)
+
+        resp = client.post("/v1/api/workstreams/new", json={"resume_ws": src})
+        assert resp.status_code == 200, resp.text
+        ws = mgr.get(resp.json()["ws_id"])
+        assert ws is not None and ws.session is not None
+        assert ws.session._persona_name == ""  # unstamped, not the default
+        assert not ws.persona
@@ -493,6 +493,7 @@ class TestSavedListPagination:
            None,
            project_id,
            "alice",
+            None,  # persona
        )

    def _cfg(self):
@@ -182,28 +182,28 @@ def test_rail_seam_exposed_and_bottom_bar_retired() -> None:
    assert "cluster-status-bar" not in index, "the #cluster-status-bar markup must be deleted"


-def test_rail_conveys_state_and_persona() -> None:
+def test_rail_conveys_state_and_kind() -> None:
    """rail.js conveys state by shape+colour via the shared ui-base .ui-glyph-*
    vocabulary (not a private glyph class), nests children via the shared bucket
-    helper, and tags sessions by persona (COORD/INT)."""
+    helper, and tags sessions by KIND (COORD/INT)."""
    body = _RAIL_JS.read_text(encoding="utf-8")
    assert "ui-glyph-" in body, "rail must use ui-base .ui-glyph-* for state (shape+colour)"
    assert "bucketByParent" in body, "rail must nest children via the shared bucket helper"
-    assert "COORD" in body and "INT" in body, "rail must tag sessions by persona"
+    assert "COORD" in body and "INT" in body, "rail must tag sessions by kind"


-def test_console_launcher_routes_by_persona() -> None:
-    """Step 2b: the dashboard launcher carries a persona kind, scope-gates the
-    interactive option, branches submit + create by kind (coordinator =
+def test_console_launcher_routes_by_kind() -> None:
+    """Step 2b: the dashboard launcher carries a workstream kind, scope-gates
+    the interactive option, branches submit + create by kind (coordinator =
    console-local, interactive = node-proxy), routes saved activation to the
    node for interactive rows, and the active-coordinators home table is gone
    (the rail covers it).  Pins the console-JS convention for the new logic."""
    app = _CONSOLE_APP.read_text(encoding="utf-8")
    assert "function _setLauncherKind" in app and "_launcherKind" in app
    assert "function _hasInteractivePermission" in app, (
-        "launcher must scope-gate the interactive persona"
+        "launcher must scope-gate the interactive kind"
    )
-    assert 'kind === "interactive"' in app, "submitHomeCoord must branch by persona kind"
+    assert 'kind === "interactive"' in app, "submitHomeCoord must branch by kind"
    assert "function _createInteractive" in app, "the interactive create path must exist"
    assert '"/v1/api/cluster/workstreams/new"' in app, (
        "interactive create must use the node-proxy endpoint"
@@ -217,11 +217,16 @@ def test_console_launcher_routes_by_persona() -> None:
    assert 'id="active-coordinators"' not in index, (
        "the active-coordinators table must be removed (the rail covers it)"
    )
-    assert 'id="launcher-personas"' in index, "the persona toggle must be in the launcher panel"
+    # "personas" now means the capability-bundle feature; the kind toggle
+    # ids were reclaimed to kind-* (launcher-kinds / kind-coordinator / ...).
+    assert 'id="launcher-kinds"' in index, "the kind toggle must be in the launcher panel"
+    assert 'id="launcher-personas"' not in index, (
+        "the old persona-squatting toggle id must stay gone"
+    )


 def test_console_launcher_creates_open_panes() -> None:
-    """Workstream-lifecycle bugfix: BOTH launcher personas open the new session as
+    """Workstream-lifecycle bugfix: BOTH launcher kinds open the new session as
    an L-shell PANE (openPane), not a full-page nav — coordinator and interactive
    alike.  Full-page nav survives only as the shell-absent fallback."""
    app = _CONSOLE_APP.read_text(encoding="utf-8")
@@ -258,13 +263,13 @@ def test_console_resolve_interactive_node_seam() -> None:
 def test_console_launcher_node_strategy() -> None:
    """Workstream-lifecycle bugfix: the interactive launcher gains a node-selection
    strategy (Least loaded | Specific node) with a live node picker, and the shared
-    composer's task hint + node fields track the active persona."""
+    composer's task hint + node fields track the active kind."""
    app = _CONSOLE_APP.read_text(encoding="utf-8")
    assert 'id: "node_strategy"' in app and 'id: "node_id"' in app, (
        "launcher must expose the node-strategy + node-picker option fields"
    )
    assert "function _applyLauncherFields" in app, (
-        "persona switch must update the hint + node-field visibility"
+        "kind switch must update the hint + node-field visibility"
    )
    assert "function _populateLauncherNodes" in app, (
        "the specific-node picker must populate from the live cluster snapshot"
@@ -873,7 +873,11 @@ def resolve_cli_persona_kwargs(
    an unrestricted session.  A corrupt stamp on the resume target raises
    (``snapshot_from_config``) for the same reason.
    """
-    from turnstone.core.personas import snapshot_from_config, snapshot_from_persona
+    from turnstone.core.personas import (
+        resolve_persona_for_kind,
+        snapshot_from_config,
+        snapshot_from_persona,
+    )

    if resume_target and storage is not None:
        if persona_arg:
@@ -883,12 +887,9 @@ def resolve_cli_persona_kwargs(
            return {"persona": snap.name, "persona_snapshot": snap}
        return {}
    if persona_arg:
-        row = storage.get_persona_by_name(persona_arg) if storage else None
-        if not row or not row.get("enabled", False):
-            print(red(f"Persona not found or disabled: {persona_arg}"))
-            sys.exit(1)
-        if "interactive" not in (row.get("applies_to_kinds") or []):
-            print(red(f"Persona '{persona_arg}' does not apply to interactive sessions"))
+        row, err = resolve_persona_for_kind(storage, persona_arg, "interactive")
+        if err or row is None:
+            print(red(err or f"Persona not found or disabled: {persona_arg}"))
            sys.exit(1)
        return {"persona": row["name"], "persona_snapshot": snapshot_from_persona(row)}
    if storage is not None:
@@ -11850,22 +11850,35 @@ def _parse_persona_body(body: dict[str, Any]) -> tuple[dict[str, Any] | None, JS
                status_code=400,
            )
        fields["tool_allowlist"] = tools
-    if "applies_to_kinds" in body:
+    if "applies_to_kinds" in body and body.get("applies_to_kinds") is not None:
        kinds = body.get("applies_to_kinds")
        if not isinstance(kinds, list) or not all(isinstance(k, str) for k in kinds):
            return None, JSONResponse(
                {"error": "applies_to_kinds must be a list of kinds"}, status_code=400
            )
        fields["applies_to_kinds"] = kinds
+    # Explicit JSON null on a flag means "leave unchanged" (the
+    # UpdatePersonaRequest schema types every flag as boolean|null) —
+    # coercing null with bool() would silently archive a persona as a side
+    # effect of an unrelated PATCH.
    for flag in ("mcp_enabled", "memory_enabled", "is_default", "enabled"):
-        if flag in body:
+        if flag in body and body.get(flag) is not None:
            fields[flag] = bool(body.get(flag))
    return fields, None


 async def admin_list_personas(request: Request) -> JSONResponse:
-    """GET /v1/api/admin/personas — all personas, archived included."""
+    """GET /v1/api/admin/personas — all personas, archived included.
+
+    Also carries ``tool_inventory``: the per-kind builtin tool names (plus
+    the synthetic ``tool_search``) so the shelf's visibility checklist is
+    derived from the server's authoritative sets instead of a hand-mirrored
+    JS constant that drifts every time a tool ships.
+    """
+    import asyncio
+
    from turnstone.core.auth import require_permission
+    from turnstone.core.tools import COORDINATOR_TOOLS, INTERACTIVE_TOOLS
    from turnstone.core.web_helpers import require_storage_or_503

    storage, err = require_storage_or_503(request)
@@ -11875,7 +11888,21 @@ async def admin_list_personas(request: Request) -> JSONResponse:
    if err:
        return err

-    return JSONResponse({"personas": storage.list_personas(include_disabled=True)})
+    def _names(tools: list[dict[str, Any]]) -> list[str]:
+        # ``tool_search`` is synthetic (not a builtin) but listed because its
+        # membership decides whether a visibility set is soft or hard.
+        return sorted({t["function"]["name"] for t in tools} | {"tool_search"})
+
+    personas = await asyncio.to_thread(storage.list_personas, include_disabled=True)
+    return JSONResponse(
+        {
+            "personas": personas,
+            "tool_inventory": {
+                "interactive": _names(INTERACTIVE_TOOLS),
+                "coordinator": _names(COORDINATOR_TOOLS),
+            },
+        }
+    )


 async def admin_create_persona(request: Request) -> JSONResponse:
@@ -11918,8 +11945,10 @@ async def admin_create_persona(request: Request) -> JSONResponse:
            "created_by": audit_uid,
        }
    )
+    import asyncio
+
    try:
-        storage.create_persona(fields)
+        await asyncio.to_thread(storage.create_persona, fields)
    except ValueError as exc:
        return JSONResponse({"error": str(exc)}, status_code=400)

@@ -11933,7 +11962,7 @@ async def admin_create_persona(request: Request) -> JSONResponse:
        ip,
    )

-    return JSONResponse(storage.get_persona(persona_id) or {})
+    return JSONResponse(await asyncio.to_thread(storage.get_persona, persona_id) or {})


 async def admin_get_persona(request: Request) -> JSONResponse:
@@ -11948,7 +11977,9 @@ async def admin_get_persona(request: Request) -> JSONResponse:
    if err:
        return err

-    persona = storage.get_persona(request.path_params["persona_id"])
+    import asyncio
+
+    persona = await asyncio.to_thread(storage.get_persona, request.path_params["persona_id"])
    if persona is None:
        return JSONResponse({"error": "Persona not found"}, status_code=404)
    return JSONResponse(persona)
@@ -11971,8 +12002,11 @@ async def admin_update_persona(request: Request) -> JSONResponse:
    if err:
        return err

+    import asyncio
+    import functools
+
    persona_id = request.path_params["persona_id"]
-    existing = storage.get_persona(persona_id)
+    existing = await asyncio.to_thread(storage.get_persona, persona_id)
    if existing is None:
        return JSONResponse({"error": "Persona not found"}, status_code=404)

@@ -11987,7 +12021,7 @@ async def admin_update_persona(request: Request) -> JSONResponse:
        return JSONResponse({"error": "no editable fields in body"}, status_code=400)

    try:
-        storage.update_persona(persona_id, **fields)
+        await asyncio.to_thread(functools.partial(storage.update_persona, persona_id, **fields))
    except ValueError as exc:
        # Storage-enforced invariants: default not archivable / must stay
        # single-kind / can't unset is_default directly / kinds validation.
@@ -12004,7 +12038,7 @@ async def admin_update_persona(request: Request) -> JSONResponse:
        ip,
    )

-    return JSONResponse(storage.get_persona(persona_id) or {})
+    return JSONResponse(await asyncio.to_thread(storage.get_persona, persona_id) or {})


 # ---------------------------------------------------------------------------
@@ -2931,11 +2931,13 @@ function confirmDeleteProject(pid, name) {
 let _adminPersonas = [];
 let _personaShelfWired = false;

-// Builtin tool inventories per kind, for the visibility checklist.  Mirrored
-// from turnstone/core/tools.py (INTERACTIVE_TOOLS / COORDINATOR_TOOLS) — keep
-// in sync when a tool is added; unknown/dynamic names ride the free-text row.
-// "tool_search" is synthetic (not a builtin) but listed because its presence
-// decides whether a visibility set is soft (expandable) or hard.
+// FALLBACK builtin tool inventories per kind, for the visibility checklist.
+// The authoritative sets ride the GET /v1/api/admin/personas response
+// (tool_inventory, derived server-side from core/tools.py) and are cached in
+// _personaToolInventory; this constant only covers the render-before-load
+// window.  "tool_search" is synthetic (not a builtin) but listed because its
+// presence decides whether a visibility set is soft (expandable) or hard.
+let _personaToolInventory = null; // {interactive: [...], coordinator: [...]}
 const _PERSONA_TOOLS = {
  interactive: [
    "bash",
@@ -2984,6 +2986,7 @@ function loadAdminPersonas() {
    })
    .then(function (data) {
      _adminPersonas = data.personas || [];
+      if (data.tool_inventory) _personaToolInventory = data.tool_inventory;
      _renderPersonas(_adminPersonas);
    })
    .catch(function () {
@@ -3168,7 +3171,8 @@ function _personaToolsModeChanged() {
 function _renderPersonaToolChecklist(checked) {
  const kind = document.getElementById("pr-kinds").value || "interactive";
  const host = document.getElementById("pr-tools-checklist");
-  const names = _PERSONA_TOOLS[kind] || [];
+  const inventory = _personaToolInventory || _PERSONA_TOOLS;
+  const names = inventory[kind] || [];
  host.replaceChildren();
  names.forEach(function (name) {
    const label = document.createElement("label");
@@ -3225,7 +3229,7 @@ function _personaFillToolsForm(allowlist) {
  } else {
    modeSel.value = "list";
    const kind = document.getElementById("pr-kinds").value || "interactive";
-    const known = _PERSONA_TOOLS[kind] || [];
+    const known = (_personaToolInventory || _PERSONA_TOOLS)[kind] || [];
    _renderPersonaToolChecklist(allowlist);
    extra.value = allowlist
      .filter(function (n) {
@@ -3293,7 +3297,8 @@ function submitPersonaShelf() {
  if (!editing && !name) return _showModalError(errEl, "Name is required");

  const prompt = document.getElementById("pr-base-prompt").value;
-  const wasDefault = editing && !!(_personaById(pid) || {}).is_default;
+  const original = editing ? _personaById(pid) || {} : {};
+  const wasDefault = editing && !!original.is_default;
  const body = {
    display_name: (
      document.getElementById("pr-display-name").value || ""
@@ -3313,9 +3318,20 @@ function submitPersonaShelf() {
  // when it's actually turning ON.
  if (wantDefault && !wasDefault) body.is_default = true;
  if (!editing) body.name = name;
-  if (editing && wasDefault) {
-    // Storage forbids changing a default persona's kinds — don't send it.
-    delete body.applies_to_kinds;
+  if (editing) {
+    const originalKinds = original.applies_to_kinds || [];
+    if (wasDefault) {
+      // Storage forbids changing a default persona's kinds — don't send it.
+      delete body.applies_to_kinds;
+    } else if (
+      originalKinds.length !== 1 &&
+      body.applies_to_kinds[0] === originalKinds[0]
+    ) {
+      // The single-value select can only show one kind; on a multi-kind
+      // persona an unrelated edit must not silently strip the others.
+      // Send kinds only when the operator actually changed the selection.
+      delete body.applies_to_kinds;
+    }
  }

  errEl.classList.remove("is-visible");
@@ -269,6 +269,7 @@ def register_workstream(
    kind: WorkstreamKind | str = WorkstreamKind.INTERACTIVE,
    parent_ws_id: str | None = None,
    project_id: str | None = None,
+    persona: str | None = None,
 ) -> None:
    """Persist a new workstream (no-op if already exists)."""
    try:
@@ -283,6 +284,7 @@ def register_workstream(
            kind=kind,
            parent_ws_id=parent_ws_id,
            project_id=project_id,
+            persona=persona,
        )
    except Exception:
        log.warning("Failed to register workstream ws=%s", ws_id, exc_info=True)
@@ -128,6 +128,18 @@ NUDGE_COMPACTION_RESUME = (
    "conversation. If the task is already complete, give your final answer."
 )

+# Variant for sessions whose persona hides the recall tool (empty/hard
+# visibility sets): same resume instruction, no pointer at a tool that isn't
+# on the wire — mirrors the compaction summary's own recall-pointer gating.
+NUDGE_COMPACTION_RESUME_NO_RECALL = (
+    "The conversation was just compacted to free context. If there is remaining "
+    "work, continue from the summary above — pick up the open tasks and next "
+    "steps you recorded and keep going without waiting for further instructions. "
+    "The summary is a digest, not the record: the full transcript remains in "
+    "stored conversation history. If the task is already complete, give your "
+    "final answer."
+)
+
 _NUDGE_MAP: dict[str, str] = {
    "correction": NUDGE_CORRECTION,
    "denial": NUDGE_DENIAL,
@@ -67,6 +67,25 @@ class PersonaSnapshot:
        }


+def resolve_persona_for_kind(
+    storage: Any, name: str, kind: str
+) -> tuple[dict[str, Any] | None, str]:
+    """Resolve a persona slug for attaching to a ``kind`` workstream.
+
+    Returns ``(row, "")`` on success or ``(None, error)`` when the name is
+    unknown, disabled, or does not apply to the kind.  ONE shared eligibility
+    rule — the HTTP create handler, the CLI ``--persona`` path, and the
+    coordinator spawn precheck all consume this, so a future rule change
+    (per-org personas, a new kind) cannot leave the surfaces disagreeing.
+    """
+    row = storage.get_persona_by_name(name) if storage else None
+    if not row or not row.get("enabled", False):
+        return None, f"Persona not found or disabled: {name}"
+    if kind not in (row.get("applies_to_kinds") or []):
+        return None, f"Persona {name!r} does not apply to kind {kind!r}"
+    return row, ""
+
+
 def snapshot_from_persona(persona: Mapping[str, Any]) -> PersonaSnapshot:
    """Build the stamp from a storage persona row — the resolve-once moment."""
    tools = persona.get("tool_allowlist")
@@ -106,6 +106,7 @@ from turnstone.core.memory_relevance import (
 from turnstone.core.metacognition import (
    MEMORY_NUDGE_TYPES,
    NUDGE_COMPACTION_RESUME,
+    NUDGE_COMPACTION_RESUME_NO_RECALL,
    RepeatDetector,
    detect_completion,
    detect_correction,
@@ -1527,8 +1528,16 @@ class ChatSession:
            # the provider-native defer_loading mode where there is no
            # synthetic tool name to filter out.
            pass
-        elif tool_search == "on" or (
-            tool_search == "auto" and len(self._tools) > tool_search_threshold
+        elif (
+            # A persona visibility set that DOES include ``tool_search`` is
+            # a soft set: the author explicitly granted discovery, so the
+            # manager is constructed regardless of the global setting or the
+            # catalog-size threshold — otherwise the authored escape hatch
+            # silently degrades to a hard set on small/tool_search=off
+            # deployments.
+            self._persona_tools is not None
+            or tool_search == "on"
+            or (tool_search == "auto" and len(self._tools) > tool_search_threshold)
        ):
            # always_on_names is the set of builtin tools present in
            # *this* session — kind-aware, so coordinator sessions never
@@ -2496,8 +2505,15 @@ class ChatSession:
            # Hard persona set — the pathway stays disabled across MCP
            # catalog refreshes too (mirrors the constructor gate).
            self._tool_search = None
-        elif self._tool_search_setting == "on" or (
-            self._tool_search_setting == "auto" and len(self._tools) > self._tool_search_threshold
+        elif (
+            # Soft persona set — the authored escape hatch stays live
+            # regardless of the global setting (mirrors the constructor).
+            self._persona_tools is not None
+            or self._tool_search_setting == "on"
+            or (
+                self._tool_search_setting == "auto"
+                and len(self._tools) > self._tool_search_threshold
+            )
        ):
            self._tool_search = ToolSearchManager(
                self._tools,
@@ -3066,6 +3082,11 @@ class ChatSession:
            self._persona_tools = snap.tools if snap else None
            self._persona_mcp = snap.mcp if snap else True
            self._persona_memory = snap.memory if snap else True
+            # Re-gate tool search under the adopted stamp: a hard set must
+            # drop the ToolSearchManager (else the recomposed prompt keeps
+            # the tool_search hint and the expanded-names escape hatch keeps
+            # since-discovered tools visible), and a soft set must gain one.
+            self._rebuild_tool_search()
        if config:
            # Restore model via registry (same path as /model command)
            saved_alias = config.get("model_alias", "")
@@ -3218,13 +3239,14 @@ class ChatSession:
        """Config gate + persona lever 4 for metacognitive nudges.

        Memory-directed nudge types (``MEMORY_NUDGE_TYPES``) are suppressed
-        when the persona's memory is off — their copy directs the model at
-        the memory tool the persona hides.  Behavioural nudges (repeat,
+        whenever the persona's envelope hides the memory tool — the lever
+        being off OR a visibility set that omits ``memory`` — because their
+        copy directs the model at that tool.  Behavioural nudges (repeat,
        compaction, idle children, watches) stay on the config gate only.
        """
        if not self._mem_cfg.nudges:
            return False
-        return self._persona_memory or nudge_type not in MEMORY_NUDGE_TYPES
+        return nudge_type not in MEMORY_NUDGE_TYPES or self._persona_tool_visible("memory")

    def _init_system_messages(self) -> None:
        """Build the system/developer prefix messages.
@@ -4415,7 +4437,7 @@ class ChatSession:
        if name in self._persona_tools:
            return True
        ts = self._tool_search
-        return ts is not None and name in ts.get_expanded_names()
+        return ts is not None and ts.is_expanded(name)

    def _apply_persona_visibility(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """Filter a tool-definition list to the persona's visible set."""
@@ -5426,7 +5448,9 @@ class ChatSession:
                            # from).  The prompt lets a genuinely-finished model
                            # give its final answer and stop.
                            self._append_user_turn(
-                                NUDGE_COMPACTION_RESUME,
+                                NUDGE_COMPACTION_RESUME
+                                if self._persona_tool_visible("recall")
+                                else NUDGE_COMPACTION_RESUME_NO_RECALL,
                                (),
                                source="compaction_resume",
                            )
@@ -10177,27 +10201,23 @@ class ChatSession:

        Returns an error string (empty = valid).  Children are always
        ``kind=interactive`` — see ``CoordinatorClient.spawn``.  The
-        receiving node's create handler re-resolves and stamps; this gate
+        receiving node's create handler re-resolves through the SAME
+        shared rule (``resolve_persona_for_kind``) and stamps; this gate
        just turns an inevitable HTTP 400 into a clean tool error the
        model can react to.  Best-effort: a storage blip defers the
        verdict to the create handler rather than blocking the spawn.
        """
+        from turnstone.core.personas import resolve_persona_for_kind
+
        try:
            storage = get_storage()
            if storage is None:
                return ""
-            row = storage.get_persona_by_name(persona)
+            _row, err = resolve_persona_for_kind(storage, persona, "interactive")
        except Exception:
            log.debug("spawn.persona_precheck_failed persona=%s", persona, exc_info=True)
            return ""
-        if not row or not row.get("enabled", False):
-            return f"unknown or disabled persona: {persona!r}"
-        if "interactive" not in (row.get("applies_to_kinds") or []):
-            return (
-                f"persona {persona!r} does not apply to interactive workstreams "
-                "(spawned children are always interactive)"
-            )
-        return ""
+        return err

    def _exec_spawn_workstream(self, item: dict[str, Any]) -> tuple[str, str]:
        call_id = item["call_id"]
@@ -10289,6 +10309,9 @@ class ChatSession:
        # poison the other approved spawns.
        normalised: list[dict[str, Any]] = []
        preview_rows: list[str] = []
+        # Persona prechecks hit storage; a fan-out batch usually repeats one
+        # persona across all children, so memoize per prepare call.
+        persona_verdicts: dict[str, str] = {}
        for idx, raw in enumerate(raw_children):
            if not isinstance(raw, dict):
                normalised.append({"idx": idx, "_error": "child spec must be an object"})
@@ -10313,9 +10336,10 @@ class ChatSession:
                # Same prep-time gate as spawn_workstream, but surfaced as
                # a per-row denial (partial-success semantics) rather than
                # failing the whole batch.
-                persona_err = self._validate_child_persona(persona)
-                if persona_err:
-                    spec["_error"] = persona_err
+                if persona not in persona_verdicts:
+                    persona_verdicts[persona] = self._validate_child_persona(persona)
+                if persona_verdicts[persona]:
+                    spec["_error"] = persona_verdicts[persona]
            normalised.append(spec)
            if spec.get("_error"):
                preview_rows.append(f"  {idx}. [invalid — {spec['_error']}]")
@@ -15278,7 +15302,14 @@ class ChatSession:
            self._envelope_nonce = fence.mint_nonce()
            self._sender_label_nonce = fence.mint_nonce()
            self._title_generated = False
-            register_workstream(self._ws_id, node_id=self._node_id)
+            # The session keeps its persona across /new (the stamp is
+            # re-written by _save_config below) — carry the display slug
+            # onto the fresh row so projections agree with the config.
+            register_workstream(
+                self._ws_id,
+                node_id=self._node_id,
+                persona=self._persona_name or None,
+            )
            self._save_config()
            self.ui.on_info("New workstream started.")

@@ -15288,7 +15319,9 @@ class ChatSession:
                self.ui.on_info("No saved workstreams.")
            else:
                lines = ["Workstreams:\n"]
-                for wid, alias, title, _created, updated, count, *_extra in rows:
+                # Column order from the storage SELECT: ws_id, alias, title,
+                # name, created, updated, message_count, … (tail ignored).
+                for wid, alias, title, _name, _created, updated, count, *_extra in rows:
                    display_name = alias or wid
                    display_title = f"  {title}" if title else ""
                    marker = " *" if wid == self._ws_id else "  "
@@ -665,20 +665,24 @@ class SessionManager:
                    )
                    saved_alias = None

-                # Persona snapshot rides the same pre-construction lane as
-                # the saved alias: the constructor applies the four levers
-                # (tool merge, MCP gate, composition) inside __init__, so
-                # the stamp must land as a kwarg — resume() is too late.
-                # A corrupt/partial stamp raises here (loud construction
-                # error), never silently reverting to a default envelope.
-                # No stamp = legacy pre-persona workstream: the kwarg is
-                # omitted entirely so factories that predate it keep working.
-                persona_snapshot = snapshot_from_config(saved_cfg or {})
-                extra_build_kwargs: dict[str, Any] = {}
-                if persona_snapshot is not None:
-                    extra_build_kwargs["persona_snapshot"] = persona_snapshot
-
                try:
+                    # Persona snapshot rides the same pre-construction lane as
+                    # the saved alias: the constructor applies the four levers
+                    # (tool merge, MCP gate, composition) inside __init__, so
+                    # the stamp must land as a kwarg — resume() is too late.
+                    # A corrupt/partial stamp raises here (loud construction
+                    # error), never silently reverting to a default envelope.
+                    # No stamp = legacy pre-persona workstream: the kwarg is
+                    # omitted entirely so factories that predate it keep
+                    # working.  Inside the unwind bracket: a parse raise must
+                    # release the placeholder slot exactly like a
+                    # build_session failure, or the ws_id stays tracked
+                    # forever (pinning a max_active slot and turning every
+                    # later open() into the already-tracked RuntimeError).
+                    persona_snapshot = snapshot_from_config(saved_cfg or {})
+                    extra_build_kwargs: dict[str, Any] = {}
+                    if persona_snapshot is not None:
+                        extra_build_kwargs["persona_snapshot"] = persona_snapshot
                    ws.session = self._adapter.build_session(
                        ws, model=saved_alias, **extra_build_kwargs
                    )
@@ -2493,32 +2493,51 @@ def make_create_handler(
            body_persona = (
                body_persona_raw.strip() if isinstance(body_persona_raw, str) else ""
            )
-            persona_row: dict[str, Any] | None = None
-            if not (isinstance(resume_ws_id_raw, str) and resume_ws_id_raw):
+            persona_snapshot = None
+            if isinstance(resume_ws_id_raw, str) and resume_ws_id_raw:
+                # Fork-resume adopts the SOURCE workstream's stamp, resolved
+                # pre-construction so all four levers (including the
+                # construction-time MCP gate) apply to the fork.  The four
+                # levers a fork runs under must be the ones its conversation
+                # was authored under — never a fresh default.  A corrupt
+                # stamp is a loud 400, mirroring the rehydrate contract; an
+                # unstamped (legacy) source forks unstamped.
+                from turnstone.core.memory import resolve_workstream
+                from turnstone.core.personas import snapshot_from_config
                from turnstone.core.storage._registry import get_storage as _get_storage

+                _st = _get_storage()
+                resume_target = await asyncio.to_thread(resolve_workstream, resume_ws_id_raw)
+                if _st is not None and resume_target:
+                    try:
+                        persona_snapshot = snapshot_from_config(
+                            await asyncio.to_thread(
+                                _st.load_workstream_config, resume_target
+                            )
+                            or {}
+                        )
+                    except ValueError as exc:
+                        return JSONResponse(
+                            {"error": f"cannot fork {resume_ws_id_raw}: {exc}"},
+                            status_code=400,
+                        )
+            else:
+                from turnstone.core.personas import (
+                    resolve_persona_for_kind,
+                    snapshot_from_persona,
+                )
+                from turnstone.core.storage._registry import get_storage as _get_storage
+
+                persona_row: dict[str, Any] | None = None
                if body_persona:
                    _st = _get_storage()
                    if _st is None:
                        return JSONResponse({"error": "storage unavailable"}, status_code=503)
-                    persona_row = await asyncio.to_thread(
-                        _st.get_persona_by_name, body_persona
+                    persona_row, persona_err = await asyncio.to_thread(
+                        resolve_persona_for_kind, _st, body_persona, mgr.kind.value
                    )
-                    if not persona_row or not persona_row.get("enabled", False):
-                        return JSONResponse(
-                            {"error": f"Persona not found or disabled: {body_persona}"},
-                            status_code=400,
-                        )
-                    if mgr.kind.value not in (persona_row.get("applies_to_kinds") or []):
-                        return JSONResponse(
-                            {
-                                "error": (
-                                    f"Persona {body_persona!r} does not apply to "
-                                    f"kind {mgr.kind.value!r}"
-                                )
-                            },
-                            status_code=400,
-                        )
+                    if persona_err:
+                        return JSONResponse({"error": persona_err}, status_code=400)
                else:
                    # Default-persona lookup is best-effort: a storage blip
                    # here degrades to an unstamped (legacy) create, which
@@ -2533,18 +2552,18 @@ def make_create_handler(
                    except Exception:
                        log.debug("ws.create.default_persona_lookup_failed", exc_info=True)
                        persona_row = None
+                if persona_row is not None:
+                    persona_snapshot = snapshot_from_persona(persona_row)

            kwargs = cfg.create_build_kwargs(
                request, body, uid, skill_data, skill_id_resolved, applied_skill_version
            )
-            if persona_row is not None:
-                from turnstone.core.personas import snapshot_from_persona
-
+            if persona_snapshot is not None:
                # ``persona`` is SessionManager.create's explicit param
                # (Workstream attr + workstreams row); the snapshot rides
                # **extra_session_kwargs into the session factory.
-                kwargs["persona"] = persona_row["name"]
-                kwargs["persona_snapshot"] = snapshot_from_persona(persona_row)
+                kwargs["persona"] = persona_snapshot.name
+                kwargs["persona_snapshot"] = persona_snapshot
            # Deferred emit — committed below post-attachment-
            # validation. See handler docstring's Ordering invariants.
            ws = await asyncio.to_thread(mgr.create, defer_emit_created=True, **kwargs)
@@ -287,12 +287,59 @@ def upgrade() -> None:
    for perm in _PERSONA_PERMS:
        _append_permission(conn, perm)

+    # Convert legacy creative-mode workstreams to the writer stamp so "a
+    # creative workstream resumes as a creative workstream" survives the
+    # /creative removal: pre-063 code persisted creative_mode='True' in
+    # workstream_config; post-063 code reads only the persona keys.  The
+    # writer seed is /creative's designated successor (same prompt lineage,
+    # tools off, MCP off, memory on).  The stale creative_mode key is left
+    # in place — nothing reads it, and downgrade needs it intact.
+    creative_rows = conn.execute(
+        sa.text(
+            "SELECT ws_id FROM workstream_config "
+            "WHERE key = 'creative_mode' AND value = 'True' "
+            "AND ws_id NOT IN "
+            "  (SELECT ws_id FROM workstream_config WHERE key = 'persona')"
+        )
+    ).fetchall()
+    writer_stamp = {
+        "persona": "writer",
+        "persona_prompt": _WRITER_PROMPT,
+        "persona_tools": "[]",
+        "persona_mcp": "0",
+        "persona_memory": "1",
+    }
+    for (ws_id,) in creative_rows:
+        for key, value in writer_stamp.items():
+            conn.execute(
+                sa.text(
+                    "INSERT INTO workstream_config (ws_id, key, value) "
+                    "VALUES (:ws, :k, :v)"
+                ),
+                {"ws": ws_id, "k": key, "v": value},
+            )
+        conn.execute(
+            sa.text("UPDATE workstreams SET persona = 'writer' WHERE ws_id = :ws"),
+            {"ws": ws_id},
+        )
+

 def downgrade() -> None:
    conn = op.get_bind()
    for perm in reversed(_PERSONA_PERMS):
        _remove_permission(conn, perm)

+    # Remove every persona stamp (including the writer stamps the upgrade
+    # synthesized from creative_mode rows — creative_mode itself was left in
+    # place, so pre-063 code resumes those workstreams as creative again).
+    conn.execute(
+        sa.text(
+            "DELETE FROM workstream_config WHERE key IN "
+            "('persona', 'persona_prompt', 'persona_tools', "
+            "'persona_mcp', 'persona_memory')"
+        )
+    )
+
    op.drop_column("workstreams", "persona")

    op.drop_index("idx_personas_enabled", table_name="personas")
@@ -123,6 +123,11 @@ class ToolSearchManager:
        """Return names of currently expanded (discovered) tools."""
        return list(self._expanded.keys())

+    def is_expanded(self, name: str) -> bool:
+        """O(1) membership check for the discovered set — the per-tool
+        visibility filter runs per LLM turn, so no list construction."""
+        return name in self._expanded
+
    def expand_visible(self, tool_names: list[str]) -> list[dict[str, Any]]:
        """Promote discovered tools to the visible set.

@@ -2982,7 +2982,9 @@ async def list_personas_endpoint(request: Request) -> JSONResponse:
    if uerr:
        return uerr
    storage = get_storage()
-    rows = storage.list_personas() if storage else []
+    # Off the event loop: the picker feed runs on every page load, and a
+    # slow storage round-trip here would stall all in-flight SSE streams.
+    rows = await asyncio.to_thread(storage.list_personas) if storage else []
    personas = [
        {
            "name": r["name"],
@@ -4980,6 +4982,23 @@ def main() -> None:
    interactive_adapter.attach(manager)
    WebUI._workstream_mgr = manager

+    def _resume_persona_kwargs(target_ws_id: str) -> dict[str, Any]:
+        """Pre-read a resume target's persona stamp for ``manager.create``.
+
+        The create-then-resume paths below construct the session BEFORE
+        ``resume()`` runs, and the persona MCP gate is construction-time
+        only — without this, restoring an MCP-off workstream would merge
+        the live MCP catalog back in.  A corrupt stamp raises (ValueError),
+        matching the rehydrate contract; no stamp = legacy, no kwargs.
+        """
+        from turnstone.core.memory import load_workstream_config
+        from turnstone.core.personas import snapshot_from_config
+
+        snap = snapshot_from_config(load_workstream_config(target_ws_id) or {})
+        if snap is None:
+            return {}
+        return {"persona": snap.name, "persona_snapshot": snap}
+
    def _watch_restore_fn(ws_id: str) -> Any:
        """Restore an evicted workstream so a watch can deliver results.

@@ -4989,7 +5008,9 @@ def main() -> None:
        :class:`NudgeQueue` without a second pass through ``restore_fn``.
        """
        try:
-            ws = manager.create(user_id="", name="watch-restore")
+            ws = manager.create(
+                user_id="", name="watch-restore", **_resume_persona_kwargs(ws_id)
+            )
            # Restored workstreams run unattended — auto-approve tool calls
            # to avoid blocking forever on approval with no connected user.
            if isinstance(ws.ui, WebUI):
@@ -5000,6 +5021,11 @@ def main() -> None:
                return _watch_runner.get_dispatch_fn(ws.session._ws_id)
        except RuntimeError:
            log.warning("watch_restore: cannot restore ws %s (all slots active)", ws_id)
+        except ValueError:
+            # Corrupt persona stamp — refuse to run the watch under an
+            # envelope the operator didn't choose (it would be unattended
+            # AND auto-approved); the watch stays queued for a manual open.
+            log.warning("watch_restore: corrupt persona stamp on ws %s", ws_id, exc_info=True)
        return None

    _watch_runner = WatchRunner(
@@ -5020,7 +5046,7 @@ def main() -> None:
        if not target_id:
            log.error("Workstream not found: %s", args.resume)
            sys.exit(1)
-        ws = manager.create(user_id="", name="resumed")
+        ws = manager.create(user_id="", name="resumed", **_resume_persona_kwargs(target_id))
        if not isinstance(ws.ui, WebUI):
            raise TypeError(f"Expected WebUI, got {type(ws.ui).__name__}")
        if args.skip_permissions or config_store.get("tools.skip_permissions"):