fix(personas): close review findings across the envelope, resume, and RBAC lanes

Provider search gating (replace-only): native web search now stands in for
a client web_search def that survived the persona visibility filter — on
both OpenAI surfaces and both injection lanes (web_search_options, the
server_side_tools loop, and _convert_tools' capability lane). A scribe or
any envelope hiding web_search stays search-free on search-capable models;
coordinators and tool-less utility calls stop receiving search too.

Resume stamp discipline: resume() loads config and parses the target's
stamp BEFORE touching session identity/history, so a corrupt stamp raises
with the session intact instead of half-adopting and then 'repairing' the
target's stamp on the next config save. The MCP lever now follows the
stamp on mid-session adoption: an MCP-off stamp drops the live surface in
place (listeners deregistered, toolsets reset); adopting an MCP-on stamp
into a session whose persona gated the client off is refused loudly (the
surface cannot be rebuilt post-construction). The REPL /resume handler
reports these errors instead of crashing the CLI.

Fail-closed default lane: a FAILED default-persona lookup at create is a
503 (routes) / clear exit (CLI) instead of silently degrading to the
unstamped stock envelope; a clean 'no default configured' still creates
legacy. resolve_persona_for_kind reports storage-unavailable distinctly
from unknown-persona.

Soft-set governance: tool_search expansion under a persona visibility set
recomposes the system prompt so tool-gated policy segments land with the
tool they gate. MCP resource/prompt catalogs gate on read_resource /
use_prompt visibility. Spawn judge/audit projections carry persona (the
human approval header already did). Active-list rows carry persona like
their project_id twin.

RBAC catalogs: persona.{create,read,write} join _VALID_PERMISSIONS and
the roles-editor sections, making the documented grant-outward path real.

Storage hardening: default-persona invariants move to a shared _utils
helper (validate + demote) with a pg advisory xact lock serializing
promotions and a post-promote single-default assertion; create maps the
unique-name race to the same ValueError as the pre-check; reads validate
JSON shape loudly (naming the persona); serialize enforces size caps;
field validation runs before invariant checks so malformed input is a 400,
never a TypeError-500. org_id guards explicit null and caps at 64.

Also: base_override='' means 'no override' at the compose boundary;
persona tag flattened/capped before the spawn approval header; /creative
redirect resolves the writer persona before advertising it; memory-nudge
gating unified through _nudges_enabled.

Provider/row-shape tests updated to the new contracts (the old ones
pinned the injection hole and the pre-persona row shape).

tests/test_coordinator_endpoints.py

@@ -521,11 +521,16 @@ def test_active_list_row_shape_includes_unified_fields(storage):
         "parent_ws_id",
         "user_id",
         "project_id",
+        "persona",
     }
     assert row["name"] == "lifted-coord"
     assert row["kind"] == "coordinator"
     assert row["parent_ws_id"] is None
     assert row["user_id"] == "u1"
+    # mgr.create without a persona kwarg stamps nothing at this layer
+    # (default resolution lives in the HTTP create handler), so the
+    # row carries the null slug — not a fabricated default.
+    assert row["persona"] is None
 
 
 def test_create_returns_ws_id_and_records_audit(storage):

tests/test_governance_endpoints.py

@@ -225,6 +225,22 @@ class TestRoles:
         assert resp.status_code == 200, resp.json()
         assert "model.skills.write" in resp.json()["permissions"]
 
+    def test_create_role_with_persona_permissions(self, client):
+        """``persona.{create,read,write}`` (migration 063) are enumerated in
+        ``_VALID_PERMISSIONS`` and pass role-create validation.  Before the fix
+        they 400'd — a custom role could never carry a persona grant."""
+        resp = client.post(
+            "/v1/api/admin/roles",
+            json=_role_payload(
+                name="personaeditor",
+                permissions="read,persona.create,persona.read,persona.write",
+            ),
+        )
+        assert resp.status_code == 200, resp.json()
+        perms = resp.json()["permissions"]
+        for p in ("persona.create", "persona.read", "persona.write"):
+            assert p in perms
+
     def test_permission_sections_js_covers_valid_permissions(self):
         """F-5: ``_PERMISSION_SECTIONS`` in governance.js mirrors
         ``_VALID_PERMISSIONS`` in console/server.py.  A new perm added
@@ -355,6 +371,19 @@ class TestRoles:
         assert role["display_name"] == "Senior Analyst"
         assert role["permissions"] == "read,write,approve"
 
+    def test_update_role_accepts_persona_permissions(self, client):
+        """Editing a custom role to carry ``persona.*`` must validate (they were
+        rejected before 063 added them to ``_VALID_PERMISSIONS``)."""
+        create_resp = client.post("/v1/api/admin/roles", json=_role_payload())
+        role_id = create_resp.json()["role_id"]
+        resp = client.put(
+            f"/v1/api/admin/roles/{role_id}",
+            json={"permissions": "read,persona.read,persona.write"},
+        )
+        assert resp.status_code == 200, resp.json()
+        perms = resp.json()["permissions"]
+        assert "persona.read" in perms and "persona.write" in perms
+
     def test_update_nonexistent_role(self, client):
         resp = client.put(
             "/v1/api/admin/roles/nonexistent",
@@ -451,6 +480,20 @@ class TestRoleOverrides:
         assert "model.skills.write" in body["effective"]
         assert body["grants"] == ["model.skills.write"]
 
+    def test_overrides_grant_persona_write(self, client, storage):
+        # persona.write is admin-default (063) but grantable to any builtin
+        # role via the overrides layer — the endpoint must accept it, not 400
+        # it as an unknown permission.
+        _seed_builtin_admin(storage, "read,write,admin.roles")
+        resp = client.put(
+            "/v1/api/admin/roles/builtin-admin/overrides",
+            json={"grant": ["persona.write"], "revoke": []},
+        )
+        assert resp.status_code == 200, resp.json()
+        body = resp.json()
+        assert "persona.write" in body["effective"]
+        assert body["grants"] == ["persona.write"]
+
     def test_overrides_replace_semantics(self, client, storage):
         _seed_builtin_admin(storage, "read,write,admin.roles")
         client.put(

tests/test_provider_xai.py

@@ -208,7 +208,26 @@ class TestBuildKwargs:
     def test_web_search_tool_injected_into_tools_list(self, provider: XAIProvider) -> None:
         # The inherited generalised injection in
         # OpenAIResponsesProvider._build_kwargs walks server_side_tools;
-        # grok-4.3 declares `("web_search",)`.
+        # grok-4.3 declares `("web_search",)`.  Injection is replace-only:
+        # it stands in for a client web_search def that survived the
+        # session's visibility filter, so the def must be present.
+        kwargs = provider._build_kwargs(
+            model="grok-4.3",
+            messages=[{"role": "user", "content": "hi"}],
+            tools=[{"type": "function", "function": {"name": "web_search"}}],
+            max_tokens=512,
+            temperature=0.5,
+            reasoning_effort="low",
+            deferred_names=None,
+            capabilities=None,
+            replay_reasoning_to_model=False,
+        )
+        tools = kwargs.get("tools") or []
+        assert {"type": "web_search"} in tools
+
+    def test_web_search_not_injected_without_client_def(self, provider: XAIProvider) -> None:
+        # A request whose envelope hides web_search (persona visibility
+        # set, tool-less utility call) gains no native search.
         kwargs = provider._build_kwargs(
             model="grok-4.3",
             messages=[{"role": "user", "content": "hi"}],
@@ -221,7 +240,7 @@ class TestBuildKwargs:
             replay_reasoning_to_model=False,
         )
         tools = kwargs.get("tools") or []
-        assert {"type": "web_search"} in tools
+        assert {"type": "web_search"} not in tools
 
 
 # ---------------------------------------------------------------------------

tests/test_providers.py

@@ -2752,6 +2752,23 @@ class TestOpenAIWebSearch:
         result = self.provider._apply_web_search(kwargs, caps, tools)
         assert result is None
 
+    def test_apply_web_search_no_op_when_client_def_absent(self) -> None:
+        """Replace-only: a search model with a NON-EMPTY toolset that never
+        advertised web_search (a persona visibility set or coordinator
+        toolset) must NOT gain native search — the option stays off and the
+        tools pass through untouched.  Contrast test_apply_web_search_with_
+        no_tools, which covers the tool-less utility-call case."""
+        caps = self.provider.get_capabilities("gpt-5-search-api")
+        assert caps.supports_web_search is True
+        kwargs: dict[str, Any] = {"model": "gpt-5-search-api"}
+        tools: list[dict[str, Any]] = [
+            {"type": "function", "function": {"name": "bash", "description": "Run bash"}},
+            {"type": "function", "function": {"name": "read_file", "description": "Read"}},
+        ]
+        result = self.provider._apply_web_search(kwargs, caps, tools)
+        assert "web_search_options" not in kwargs
+        assert result is tools  # unchanged, not filtered or replaced
+
     def test_format_citations_appends_sources(self) -> None:
         """url_citation annotations should be formatted as footnote sources."""
         ann = MagicMock()
@@ -2810,13 +2827,27 @@ class TestOpenAIWebSearch:
         assert "Sources:" not in result
 
     def test_apply_web_search_with_no_tools(self) -> None:
-        """Search model with tools=None should still inject web_search_options."""
+        """No client web_search def ⇒ no injection (replace-only semantics).
+
+        A request that never advertised the web_search tool — persona
+        visibility set, coordinator toolset, or a tool-less utility call —
+        must not gain native search at the provider layer.
+        """
         caps = self.provider.get_capabilities("gpt-5-search-api")
         kwargs: dict[str, Any] = {}
         result = self.provider._apply_web_search(kwargs, caps, None)
-        assert "web_search_options" in kwargs
+        assert "web_search_options" not in kwargs
         assert result is None
 
+    def test_apply_web_search_replaces_client_def(self) -> None:
+        """With the client def present, it is filtered and the option set."""
+        caps = self.provider.get_capabilities("gpt-5-search-api")
+        kwargs: dict[str, Any] = {}
+        tools = [{"type": "function", "function": {"name": "web_search"}}]
+        result = self.provider._apply_web_search(kwargs, caps, tools)
+        assert "web_search_options" in kwargs
+        assert result is None  # the lone def was filtered away
+
     def test_streaming_creates_with_web_search_options(self) -> None:
         """Streaming with a search model should pass web_search_options."""
         client = MagicMock()
@@ -4193,8 +4224,13 @@ class TestResponsesParamBuilding:
         )
         assert kwargs["instructions"] == "Be helpful"
 
-    def test_web_search_injected_with_no_tools(self) -> None:
-        """Search-capable models get web_search tool even when tools=None."""
+    def test_web_search_not_injected_with_no_tools(self) -> None:
+        """No client web_search def ⇒ no server-side web_search entry.
+
+        Replace-only semantics: a request whose envelope hides web_search
+        (persona visibility set, coordinator toolset, tool-less utility
+        call) must not gain native search at the provider layer.
+        """
         kwargs = self.provider._build_kwargs(
             model="gpt-5-search-api",
             messages=[{"role": "user", "content": "Hi"}],
@@ -4204,10 +4240,43 @@ class TestResponsesParamBuilding:
             reasoning_effort="none",
             deferred_names=None,
         )
+        tool_types = [t.get("type") for t in kwargs.get("tools") or []]
+        assert "web_search" not in tool_types
+
+    def test_web_search_injected_with_client_def(self) -> None:
+        """The server-side entry stands in for a surviving client def."""
+        kwargs = self.provider._build_kwargs(
+            model="gpt-5-search-api",
+            messages=[{"role": "user", "content": "Hi"}],
+            tools=[{"type": "function", "function": {"name": "web_search"}}],
+            max_tokens=4096,
+            temperature=0.5,
+            reasoning_effort="none",
+            deferred_names=None,
+        )
         assert "tools" in kwargs
         tool_types = [t.get("type") for t in kwargs["tools"]]
         assert "web_search" in tool_types
 
+    def test_web_search_not_injected_for_nonempty_toolset_without_def(self) -> None:
+        """A non-empty toolset lacking web_search gains no native search.
+
+        Guards the _convert_tools lane: capability alone must not inject —
+        a persona visibility set or the coordinator toolset that hides
+        web_search stays search-free on search-capable models.
+        """
+        kwargs = self.provider._build_kwargs(
+            model="gpt-5-search-api",
+            messages=[{"role": "user", "content": "Hi"}],
+            tools=[{"type": "function", "function": {"name": "read_file"}}],
+            max_tokens=4096,
+            temperature=0.5,
+            reasoning_effort="none",
+            deferred_names=None,
+        )
+        tool_types = [t.get("type") for t in kwargs.get("tools") or []]
+        assert "web_search" not in tool_types
+
 
 class TestResponsesCitationFormat:
     """Test format_citations handles Responses API flat annotation format."""

tests/test_server_authz.py

@@ -647,6 +647,7 @@ class TestListWorkstreamsTrustedTeamVisibility:
             "parent_ws_id",
             "user_id",
             "project_id",
+            "persona",
         }
         assert row["kind"] == "interactive"
         assert row["user_id"] == "user-shape"

turnstone/cli.py

@@ -888,15 +888,20 @@ def resolve_cli_persona_kwargs(
         return {}
     if persona_arg:
         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}"))
+        if row is None:
+            print(red(err))
             sys.exit(1)
         return {"persona": row["name"], "persona_snapshot": snapshot_from_persona(row)}
     if storage is not None:
         try:
             default_row = storage.get_default_persona("interactive")
-        except Exception:
-            default_row = None
+        except Exception as exc:
+            # A failed lookup must not silently start an unrestricted
+            # session — the operator may have promoted a restricted
+            # persona to default.  (A clean None — no default configured —
+            # still yields the unstamped legacy session below.)
+            print(red(f"Default persona lookup failed: {exc}"))
+            sys.exit(1)
         if default_row:
             return {
                 "persona": default_row["name"],

turnstone/console/server.py

@@ -6441,6 +6441,13 @@ _VALID_PERMISSIONS = frozenset(
         # Project deletion — destroys the container and its scoped memory, so
         # it is a distinct capability from project.write (admin-default).
         "project.delete",
+        # Personas — workstream capability-envelope templates.  Granted to
+        # builtin-admin via migration 063; grantable outward via custom
+        # roles / the overrides editor (selection at workstream creation is
+        # deliberately ungated — these gate authoring and management only).
+        "persona.create",
+        "persona.read",
+        "persona.write",
     }
 )
 
@@ -11941,7 +11948,9 @@ async def admin_create_persona(request: Request) -> JSONResponse:
         {
             "persona_id": persona_id,
             "name": name,
-            "org_id": str(body.get("org_id", "")).strip(),
+            # ``or ""`` guards explicit JSON null (str(None) would persist
+            # the literal "None"); [:64] matches the org-handler caps.
+            "org_id": str(body.get("org_id") or "").strip()[:64],
             "created_by": audit_uid,
         }
     )

turnstone/console/static/admin.js

@@ -2931,52 +2931,13 @@ function confirmDeleteProject(pid, name) {
 let _adminPersonas = [];
 let _personaShelfWired = false;
 
-// 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.
+// Builtin tool inventories per kind for the visibility checklist ride the
+// GET /v1/api/admin/personas response (tool_inventory, derived server-side
+// from core/tools.py) — deliberately NO hand-mirrored fallback constant
+// here, which would silently drift every time a tool ships.  Until the
+// first list response lands the checklist renders empty; the free-text
+// "extra tools" input still accepts any name in that window.
 let _personaToolInventory = null; // {interactive: [...], coordinator: [...]}
-const _PERSONA_TOOLS = {
-  interactive: [
-    "bash",
-    "diff_file",
-    "edit_file",
-    "memory",
-    "notify",
-    "read_file",
-    "read_resource",
-    "recall",
-    "search",
-    "skills",
-    "task_agent",
-    "tool_search",
-    "use_prompt",
-    "watch",
-    "web_fetch",
-    "web_search",
-    "write_file",
-  ],
-  coordinator: [
-    "cancel_workstream",
-    "close_all_children",
-    "close_workstream",
-    "delete_workstream",
-    "inspect_workstream",
-    "list_nodes",
-    "list_workstreams",
-    "memory",
-    "notify",
-    "send_to_workstream",
-    "skills",
-    "spawn_batch",
-    "spawn_workstream",
-    "tasks",
-    "tool_search",
-    "wait_for_workstream",
-  ],
-};
 
 function loadAdminPersonas() {
   authFetch("/v1/api/admin/personas")
@@ -3059,8 +3020,7 @@ function _renderPersonas(personas) {
               : [
                   {
                     label: "make default",
-                    title:
-                      "Make this the kind default (demotes the incumbent)",
+                    title: "Make this the kind default (demotes the incumbent)",
                     attrs: { "data-default-persona": p.persona_id },
                   },
                 ],
@@ -3158,8 +3118,31 @@ function _personaShelfWire() {
     .getElementById("pr-tools-mode")
     .addEventListener("change", _personaToolsModeChanged);
   document.getElementById("pr-kinds").addEventListener("change", function () {
-    // The checklist shows the ACTIVE kind's inventory.
-    _renderPersonaToolChecklist([]);
+    // Re-render the ACTIVE kind's inventory carrying the checked names
+    // over, so dual-kind tools (memory, notify, skills, tool_search)
+    // survive the flip; checked names outside the new kind's inventory
+    // migrate to the extra field instead of silently dropping from the
+    // allowlist the operator is editing.
+    const kept = [];
+    document
+      .querySelectorAll("#pr-tools-checklist [data-persona-tool]")
+      .forEach(function (input) {
+        if (input.checked) kept.push(input.value);
+      });
+    _renderPersonaToolChecklist(kept);
+    const kind = document.getElementById("pr-kinds").value || "interactive";
+    const known = (_personaToolInventory || {})[kind] || [];
+    const extra = document.getElementById("pr-tools-extra");
+    const extras = (extra.value || "")
+      .split(",")
+      .map(function (s) {
+        return s.trim();
+      })
+      .filter(Boolean);
+    kept.forEach(function (n) {
+      if (known.indexOf(n) < 0 && extras.indexOf(n) < 0) extras.push(n);
+    });
+    extra.value = extras.join(", ");
   });
 }
 
@@ -3171,7 +3154,7 @@ function _personaToolsModeChanged() {
 function _renderPersonaToolChecklist(checked) {
   const kind = document.getElementById("pr-kinds").value || "interactive";
   const host = document.getElementById("pr-tools-checklist");
-  const inventory = _personaToolInventory || _PERSONA_TOOLS;
+  const inventory = _personaToolInventory || {};
   const names = inventory[kind] || [];
   host.replaceChildren();
   names.forEach(function (name) {
@@ -3229,7 +3212,7 @@ function _personaFillToolsForm(allowlist) {
   } else {
     modeSel.value = "list";
     const kind = document.getElementById("pr-kinds").value || "interactive";
-    const known = (_personaToolInventory || _PERSONA_TOOLS)[kind] || [];
+    const known = (_personaToolInventory || {})[kind] || [];
     _renderPersonaToolChecklist(allowlist);
     extra.value = allowlist
       .filter(function (n) {
@@ -3303,9 +3286,7 @@ function submitPersonaShelf() {
     display_name: (
       document.getElementById("pr-display-name").value || ""
     ).trim(),
-    description: (
-      document.getElementById("pr-description").value || ""
-    ).trim(),
+    description: (document.getElementById("pr-description").value || "").trim(),
     base_prompt: prompt.trim() ? prompt : null,
     tool_allowlist: _personaToolsFromForm(),
     mcp_enabled: document.getElementById("pr-mcp").checked,

turnstone/console/static/app.js

@@ -2086,13 +2086,18 @@ function _initSavedCoordTable() {
       },
     },
   });
-  // The PROJECT column resolves names from the shared projects cache,
-  // which fills asynchronously — re-render once names arrive.
+  // The PROJECT and PERSONA columns resolve names from the shared caches,
+  // which fill asynchronously — re-render once names arrive.
   if (window.TurnstoneProjects) {
     window.TurnstoneProjects.onProjectsChange(function () {
       if (_coordTable) _coordTable.render();
     });
   }
+  if (window.TurnstonePersonas) {
+    window.TurnstonePersonas.onPersonasChange(function () {
+      if (_coordTable) _coordTable.render();
+    });
+  }
 }
 
 // HTML inline-onclick wrappers — keep the global names the markup binds

turnstone/console/static/governance.js

@@ -363,6 +363,10 @@ const _PERMISSION_SECTIONS = [
       "project.delete",
     ],
   },
+  {
+    label: "Personas",
+    permissions: ["persona.create", "persona.read", "persona.write"],
+  },
   {
     label: "Coordinator",
     permissions: ["coordinator.trust.send"],

turnstone/core/personas.py

@@ -59,9 +59,7 @@ class PersonaSnapshot:
         return {
             "persona": self.name,
             "persona_prompt": self.prompt,
-            "persona_tools": (
-                json.dumps(sorted(self.tools)) if self.tools is not None else "null"
-            ),
+            "persona_tools": (json.dumps(sorted(self.tools)) if self.tools is not None else "null"),
             "persona_mcp": "1" if self.mcp else "0",
             "persona_memory": "1" if self.memory else "0",
         }
@@ -77,8 +75,12 @@ def resolve_persona_for_kind(
     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.
+    ``storage is None`` reports a distinct storage-unavailable error — a
+    storage outage must never masquerade as "unknown persona".
     """
-    row = storage.get_persona_by_name(name) if storage else None
+    if storage is None:
+        return None, "persona storage unavailable"
+    row = storage.get_persona_by_name(name)
     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 []):

turnstone/core/providers/_openai_chat.py

@@ -103,10 +103,15 @@ class OpenAIChatCompletionsProvider:
         """
         if not caps.supports_web_search:
             return tools
-        if tools:
-            tools = [t for t in tools if t.get("function", {}).get("name") != "web_search"]
-            if not tools:
-                tools = None
+        # Replace-only: native search stands in for the client ``web_search``
+        # def. When the request never advertised one (persona visibility set,
+        # coordinator toolset), injecting the option would hand the model a
+        # capability its envelope hides.
+        if not tools or not any(t.get("function", {}).get("name") == "web_search" for t in tools):
+            return tools
+        tools = [t for t in tools if t.get("function", {}).get("name") != "web_search"]
+        if not tools:
+            tools = None
         kwargs["web_search_options"] = {}
         return tools
 

turnstone/core/providers/_openai_responses.py

@@ -313,8 +313,14 @@ class OpenAIResponsesProvider:
                 item["defer_loading"] = True
             converted.append(item)
 
-        # Inject native web search tool
-        if has_web_search_func or caps.supports_web_search:
+        # Inject native web search — replace-only: it stands in for a client
+        # ``web_search`` def that survived the session's visibility filter.
+        # ``caps.supports_web_search`` alone must NOT inject, or a toolset
+        # whose envelope hides web_search (persona visibility set,
+        # coordinator toolset) gains native search on capable models; the
+        # capability-only lane for def-less requests is handled (and gated
+        # the same way) by the server_side_tools loop in _build_kwargs.
+        if has_web_search_func:
             converted.append({"type": "web_search"})
 
         # Responses API requires a tool_search tool when defer_loading is used
@@ -369,7 +375,16 @@ class OpenAIResponsesProvider:
         # ``{"type": "web_search"}`` appended.  Subclasses (e.g.
         # ``XAIProvider``) opt their own provider-specific server tools
         # into ``caps.server_side_tools`` and inherit this injection.
+        has_client_web_search = any(
+            t.get("function", {}).get("name") == "web_search" for t in tools or []
+        )
         for tool_type in resolve_server_side_tools(caps):
+            # Replace-only: the native web_search entry stands in for the
+            # client def. A request whose envelope hides ``web_search``
+            # (persona visibility set, coordinator toolset) gets no native
+            # search either.
+            if tool_type == "web_search" and not has_client_web_search:
+                continue
             converted_tools = converted_tools or []
             if not any(t.get("type") == tool_type for t in converted_tools):
                 converted_tools.append({"type": tool_type})

turnstone/core/session.py

@@ -115,7 +115,11 @@ from turnstone.core.metacognition import (
     should_nudge,
 )
 from turnstone.core.nudge_queue import TOOL_DRAIN, USER_DRAIN, NudgeQueue
-from turnstone.core.personas import PersonaSnapshot, snapshot_from_config
+from turnstone.core.personas import (
+    PersonaSnapshot,
+    resolve_persona_for_kind,
+    snapshot_from_config,
+)
 from turnstone.core.providers import create_provider
 from turnstone.core.ratelimit import TokenBucket
 from turnstone.core.safety import is_command_blocked, sanitize_command
@@ -1454,6 +1458,11 @@ class ChatSession:
         # refresh callbacks stay inert (they all guard on _mcp_client).
         # Task agents keep their native tools; only the MCP surface closes.
         self._mcp_client = mcp_client if self._persona_mcp else None
+        # True when a real client was withheld by the persona gate (as
+        # opposed to no MCP in the deployment at all).  Mid-session
+        # ``resume()`` refuses to adopt an MCP-on stamp in that case —
+        # the dropped surface cannot be rebuilt post-construction.
+        self._mcp_gated_off = mcp_client is not None and not self._persona_mcp
         self._mcp_refresh_cb: Any = None  # Callable | None (avoid import)
         self._mcp_resource_cb: Any = None
         self._mcp_prompt_cb: Any = None
@@ -2636,6 +2645,39 @@ class ChatSession:
                 log.debug("chat_session.coord_client_close_failed", exc_info=True)
         self._cleanup_skill_resources()
 
+    def _drop_mcp_surface(self) -> None:
+        """Drop the live MCP surface to match an adopted MCP-off stamp.
+
+        Mirror of the constructor's persona MCP gate for the one path that
+        changes the lever after construction: a mid-session ``resume()``
+        adopting an MCP-off stamp.  Deregisters the three listeners (same
+        ``_mcp_listener_user_id`` identity rule as ``close()``), drops the
+        client reference, and resets both toolsets to the builtin lists.
+        Only reachable on interactive sessions — coordinators never hold a
+        client.  The caller rebuilds tool search and recomposes the prompt.
+        """
+        if self._mcp_client is None:
+            return
+        if self._mcp_refresh_cb:
+            self._mcp_client.remove_listener(
+                self._mcp_refresh_cb, user_id=self._mcp_listener_user_id
+            )
+            self._mcp_refresh_cb = None
+        if self._mcp_resource_cb:
+            self._mcp_client.remove_resource_listener(
+                self._mcp_resource_cb, user_id=self._mcp_listener_user_id
+            )
+            self._mcp_resource_cb = None
+        if self._mcp_prompt_cb:
+            self._mcp_client.remove_prompt_listener(
+                self._mcp_prompt_cb, user_id=self._mcp_listener_user_id
+            )
+            self._mcp_prompt_cb = None
+        self._mcp_client = None
+        self._tools = merge_mcp_tools(INTERACTIVE_TOOLS, [])
+        self._task_tools = merge_mcp_tools(TASK_AGENT_TOOLS, [])
+        self._render_agent_tool_descriptions()
+
     def _handle_mcp_refresh(self, arg: str) -> None:
         """Handle ``/mcp refresh [server]``."""
         assert self._mcp_client is not None
@@ -3031,7 +3073,27 @@ class ChatSession:
         turns = load_message_turns(ws_id)
         if not turns:
             return False
+        # Load persisted config and parse the persona stamp BEFORE touching
+        # session identity/history: a corrupt stamp must raise while this
+        # session is still intact — the web /command surface reports the
+        # error and continues on the CURRENT workstream, and a half-adopted
+        # resume would run the corrupt target under this session's envelope,
+        # after which the next _save_config would "repair" the target's
+        # stamp with a persona the operator never chose for it.
+        config = load_workstream_config(ws_id)
+        snap: PersonaSnapshot | None = None
         if not fork:
+            snap = snapshot_from_config(config or {})
+            if (snap.mcp if snap else True) and self._mcp_gated_off:
+                # The MCP lever is construction-time: narrowing is applied
+                # in place during adoption below, but a session whose
+                # persona dropped the client at construction cannot rebuild
+                # it — refuse rather than run out of step with the stamp.
+                raise ValueError(
+                    f"cannot resume {ws_id} in place: its persona enables "
+                    "MCP, which this session's persona dropped at "
+                    "construction — open the workstream fresh instead"
+                )
             self._ws_id = ws_id
             # A non-fork resume repoints this session at a DIFFERENT existing
             # workstream's identity (fork keeps self._ws_id, so its nonces stay
@@ -3062,26 +3124,25 @@ class ChatSession:
             type(self._provider).__name__,
             self.model,
         )
-        # Restore persisted config
-        config = load_workstream_config(ws_id)
+        # Restore persisted config (loaded and stamp-parsed above, before
+        # any session state was touched).
         # Adopt the persona stamp of the workstream being resumed — a
         # non-fork resume adopts the target's identity, so keeping this
         # session's creation-time stamp would clobber the target's on the
         # next _save_config.  The prompt/visibility/memory levers reapply
         # via the _init_system_messages() recompose below and the per-call
-        # visibility filter; the MCP lever is construction-time only (the
-        # startup resume paths — SessionManager.open, CLI --resume —
-        # construct with the stamp, so only a mid-session REPL /resume
-        # keeps the constructing persona's MCP surface).  A fork keeps its
-        # own creation-time stamp.  Corrupt stamps raise — never silently
-        # rewritten.
+        # visibility filter; the MCP lever narrows in place when the
+        # adopted stamp is MCP-off (_drop_mcp_surface below — widening was
+        # refused before adoption).  A fork keeps its own creation-time
+        # stamp.  Corrupt stamps raise — never silently rewritten.
         if not fork:
-            snap = snapshot_from_config(config or {})
             self._persona_name = snap.name if snap else ""
             self._persona_prompt = snap.prompt if snap else ""
             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
+            if not self._persona_mcp and self._mcp_client is not None:
+                self._drop_mcp_surface()
             # 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
@@ -3240,9 +3301,10 @@ class ChatSession:
 
         Memory-directed nudge types (``MEMORY_NUDGE_TYPES``) are suppressed
         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.
+        being off OR an allowlist that hides ``memory`` (a tool_search
+        expansion that re-adds it re-enables them) — because their copy
+        directs the model at that tool.  Every other nudge type passes
+        straight through to the config gate.
         """
         if not self._mem_cfg.nudges:
             return False
@@ -3309,8 +3371,7 @@ class ChatSession:
             kind=self._kind,
             # Persona lever 1: replaces ONLY the BASE module.  ENV /
             # CONTEXT / TOOLS / POLICIES keep composing, so mandatory
-            # prompt policies ride on top of every persona — unlike the
-            # removed /creative fork, which bypassed composition entirely.
+            # prompt policies ride on top of every persona.
             base_override=self._persona_prompt or None,
         )
         dev_parts = [composed]
@@ -3354,8 +3415,12 @@ class ChatSession:
                 "\n\nAdditional tools are available via tool_search. "
                 "Use it when you need a capability not in your current tool set."
             )
-        # MCP resource catalog (lets the model know what's available for read_resource)
-        if self._mcp_client:
+        # MCP resource catalog (lets the model know what's available for
+        # read_resource).  Gated on the tool's visibility, not just the
+        # client: a persona allowlist that hides read_resource must drop
+        # the catalog AND its call instruction — same rule as the memory
+        # advisory — or the prompt instructs calls the wire hides.
+        if self._mcp_client and self._persona_tool_visible("read_resource"):
             # Per-user merge: pool entries for the effective user (acting
             # user on shared workstreams, owner otherwise) are included;
             # other users' pool resources are not.
@@ -3382,8 +3447,9 @@ class ChatSession:
                 lines.append("</mcp-resources>")
                 lines.append("Use read_resource(uri='...') to access the resources listed above.")
                 dev_parts.append("\n".join(lines))
-        # MCP prompt catalog (lets the model know what's available for use_prompt)
-        if self._mcp_client:
+        # MCP prompt catalog (lets the model know what's available for
+        # use_prompt) — visibility-gated like the resource catalog above.
+        if self._mcp_client and self._persona_tool_visible("use_prompt"):
             # Per-user merge: pool entries for the effective user (acting
             # user on shared workstreams, owner otherwise) are included;
             # other users' pool prompts are not.
@@ -4467,7 +4533,7 @@ class ChatSession:
         Persona gating (levers 2+4) runs LAST so the wire never advertises
         a tool the visibility set hides — whatever branch produced the list.
         With a persona visibility set, tool search always runs in
-        client-side mode (see ``_get_capabilities`` note below): native
+        client-side mode (see ``_get_deferred_names``): native
         defer_loading would strip deferred tools here before the provider
         could offer them for discovery.
         """
@@ -7480,8 +7546,8 @@ class ChatSession:
                 # skill_data dict (or ``None``), not the raw string the
                 # LLM passed.  Mirror the ``spawn_workstream`` projection
                 # so heuristic ``arg_pattern`` rules can match on skill
-                # name and the judge / audit row sees which persona and
-                # model override were selected.
+                # name and the judge / audit row sees which skill was
+                # selected.
                 skill_dict = it.get("skill") or {}
                 it["func_args"] = {
                     "prompt": honest_truncate(it.get("prompt") or "", arg_budget),
@@ -7495,6 +7561,7 @@ class ChatSession:
             elif name == "spawn_workstream":
                 it["func_args"] = {
                     "skill": it.get("skill", ""),
+                    "persona": it.get("persona", ""),
                     "initial_message": honest_truncate(it.get("initial_message") or "", arg_budget),
                     "target_node": it.get("target_node", ""),
                     "name": it.get("name", ""),
@@ -7510,7 +7577,9 @@ class ChatSession:
                 # ``initial_message`` is honestly truncated to its share.
                 # ``name`` (cosmetic) and ``model`` (registry alias) skipped
                 # to keep the payload lean; risk-relevant fields are skill,
-                # initial_message, target_node.
+                # persona, initial_message, target_node — a persona swaps the
+                # child's entire base prompt and tool envelope, so the judge
+                # must see it like the human approval header does.
                 children = it.get("children") or []
                 per_child = max(arg_budget // max(len(children), 1), 0)
                 it["func_args"] = {
@@ -7518,6 +7587,7 @@ class ChatSession:
                     "children": [
                         {
                             "skill": c.get("skill", "") if isinstance(c, dict) else "",
+                            "persona": c.get("persona", "") if isinstance(c, dict) else "",
                             "initial_message": (
                                 honest_truncate(c.get("initial_message") or "", per_child)
                                 if isinstance(c, dict)
@@ -9302,7 +9372,15 @@ class ChatSession:
         results = self._tool_search.search(query)
         # Expand discovered tools into the visible set
         names = [t.get("function", {}).get("name", "") for t in results]
-        self._tool_search.expand_visible(names)
+        newly_added = self._tool_search.expand_visible(names)
+        if newly_added and self._persona_tools is not None:
+            # Under a soft persona set the prompt was composed against the
+            # pre-expansion visible names, so tool-gated policy segments
+            # for the just-expanded tools were dropped — recompose so the
+            # operator's guidance lands with the tool.  The wire tools
+            # array changes on expansion anyway (client-side mode), so the
+            # prefix-cache invalidation is already being paid.
+            self._init_system_messages()
         output = self._tool_search.format_search_results(results)
         return item["call_id"], output
 
@@ -10045,7 +10123,7 @@ class ChatSession:
             # Reset so the model gets a clean slate after the warning.
             # If it repeats again, a new warning fires.
             self._repeat_detector.clear()
-            if self._mem_cfg.nudges and should_nudge(
+            if self._nudges_enabled("repeat") and should_nudge(
                 "repeat",
                 self._metacog_state,
                 message_count=len(self.messages),
@@ -10159,7 +10237,12 @@ class ChatSession:
         model = (args.get("model") or "").strip()
         target_node = (args.get("target_node") or "").strip()
         project = (args.get("project") or "").strip()
-        persona = (args.get("persona") or "").strip()
+        # Flatten whitespace and cap before the tag reaches the trusted
+        # approval-header chrome — the value is model-authored and, when
+        # storage is down, reaches the header unvalidated.  Real slugs are
+        # ≤64 chars with no whitespace, so this only reshapes invalid names
+        # (which then fail resolution showing the sanitized string).
+        persona = " ".join((args.get("persona") or "").split())[:64]
         if persona:
             persona_err = self._validate_child_persona(persona)
             if persona_err:
@@ -10207,8 +10290,6 @@ class ChatSession:
         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:
@@ -14069,18 +14150,18 @@ class ChatSession:
                 "ws_id": self._ws_id,
                 "node_id": self._node_id or "",
             }
-            persona = _render_template(skill_data["content"], context)
-            if len(persona) > _MAX_SKILL_CONTENT:
+            skill_identity = _render_template(skill_data["content"], context)
+            if len(skill_identity) > _MAX_SKILL_CONTENT:
                 log.warning(
                     "skill_content.truncated",
-                    length=len(persona),
+                    length=len(skill_identity),
                     agent="task",
                     skill=skill_data.get("name", ""),
                 )
-                persona = persona[:_MAX_SKILL_CONTENT]
+                skill_identity = skill_identity[:_MAX_SKILL_CONTENT]
         else:
-            persona = self._TASK_DEFAULT_IDENTITY
-        identity = persona + "\n\n" + self._TASK_OPERATING_GUIDANCE
+            skill_identity = self._TASK_DEFAULT_IDENTITY
+        identity = skill_identity + "\n\n" + self._TASK_OPERATING_GUIDANCE
         # Task agent gets the base system prompt (tool patterns) merged
         # with its own identity in a single system message. No conversation
         # history — it's an autonomous sub-agent. Merged to avoid
@@ -15342,15 +15423,24 @@ class ChatSession:
                     self.ui.on_info(f"Workstream not found: {arg.strip()}")
                 elif target_id == self._ws_id:
                     self.ui.on_info("Already in that workstream.")
-                elif self.resume(target_id):
-                    self.ui.on_info(
-                        f"Resumed {bold(target_id)} ({len(self.messages)} messages loaded)"
-                    )
-                    name = get_workstream_display_name(target_id)
-                    if name:
-                        self.ui.on_rename(name)
                 else:
-                    self.ui.on_info(f"Workstream {arg.strip()} has no messages.")
+                    try:
+                        resumed: bool | None = self.resume(target_id)
+                    except ValueError as exc:
+                        # Corrupt stamp or MCP-lever refusal: resume parses
+                        # before mutating, so this session is untouched —
+                        # report and stay on the current workstream.
+                        self.ui.on_error(f"Cannot resume {target_id}: {exc}")
+                        resumed = None
+                    if resumed:
+                        self.ui.on_info(
+                            f"Resumed {bold(target_id)} ({len(self.messages)} messages loaded)"
+                        )
+                        name = get_workstream_display_name(target_id)
+                        if name:
+                            self.ui.on_rename(name)
+                    elif resumed is False:
+                        self.ui.on_info(f"Workstream {arg.strip()} has no messages.")
 
         elif cmd == "/name":
             if not arg:
@@ -15487,11 +15577,17 @@ class ChatSession:
                 self.ui.on_info("Compaction cancelled.")
 
         elif cmd == "/creative":
-            # Removed in 1.7 — the writer persona replaces it (issue #683).
-            self.ui.on_info(
-                "/creative was removed. Start a session with the 'writer' "
-                "persona instead: turnstone --persona writer"
-            )
+            # Recognized but decommissioned: print a live migration pointer
+            # instead of an "unknown command" dead end.  The 'writer' seed
+            # is archivable, so resolve it before advertising it.
+            replacement = "a writing persona (see the personas list)"
+            try:
+                row, _ = resolve_persona_for_kind(get_storage(), "writer", "interactive")
+                if row is not None:
+                    replacement = "the 'writer' persona: turnstone --persona writer"
+            except Exception:
+                log.debug("creative_redirect.writer_resolve_failed", exc_info=True)
+            self.ui.on_info(f"/creative was removed. Start a session with {replacement}.")
 
         elif cmd == "/debug":
             self.debug = not self.debug

turnstone/core/session_routes.py

@@ -2490,9 +2490,7 @@ def make_create_handler(
             # engineer/orchestrator defaults.  Resume skips resolution:
             # the resumed session restores its own stamp from config.
             body_persona_raw = body.get("persona") or ""
-            body_persona = (
-                body_persona_raw.strip() if isinstance(body_persona_raw, str) else ""
-            )
+            body_persona = body_persona_raw.strip() if isinstance(body_persona_raw, str) else ""
             persona_snapshot = None
             if isinstance(resume_ws_id_raw, str) and resume_ws_id_raw:
                 # Fork-resume adopts the SOURCE workstream's stamp, resolved
@@ -2511,10 +2509,7 @@ def make_create_handler(
                 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 {}
+                            await asyncio.to_thread(_st.load_workstream_config, resume_target) or {}
                         )
                     except ValueError as exc:
                         return JSONResponse(
@@ -2539,19 +2534,24 @@ def make_create_handler(
                     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
-                    # is behavior-identical to the shipped defaults —
-                    # never a reason to fail the create.
-                    try:
-                        _st = _get_storage()
-                        if _st is not None:
+                    # No explicit persona: stamp the kind's default. A clean
+                    # ``None`` (no default configured — pre-seed DB) creates
+                    # unstamped legacy, but a FAILED lookup must not: the
+                    # operator may have promoted a restricted persona to
+                    # default, and degrading to the stock envelope on a
+                    # storage blip would silently widen it.
+                    _st = _get_storage()
+                    if _st is not None:
+                        try:
                             persona_row = await asyncio.to_thread(
                                 _st.get_default_persona, mgr.kind.value
                             )
-                    except Exception:
-                        log.debug("ws.create.default_persona_lookup_failed", exc_info=True)
-                        persona_row = None
+                        except Exception:
+                            log.warning("ws.create.default_persona_lookup_failed", exc_info=True)
+                            return JSONResponse(
+                                {"error": "persona resolution unavailable"},
+                                status_code=503,
+                            )
                 if persona_row is not None:
                     persona_snapshot = snapshot_from_persona(persona_row)
 
@@ -2749,6 +2749,10 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler:
             for ws in wss:
                 raw_pid = getattr(ws, "project_id", "")
                 project_id = raw_pid if isinstance(raw_pid, str) else ""
+                # Same guarded read as project_id — test doubles and older
+                # node payloads may lack the attribute.
+                raw_persona = getattr(ws, "persona", "")
+                persona = raw_persona if isinstance(raw_persona, str) else ""
                 # Private-project tenancy — drop rows the requester may
                 # not see (same predicate as the saved list).
                 if not visibility.ws_visible(project_id, ws_owner=ws.user_id or ""):
@@ -2763,6 +2767,7 @@ def make_list_handler(cfg: SessionEndpointConfig) -> Handler:
                         "parent_ws_id": ws.parent_ws_id,
                         "user_id": ws.user_id,
                         "project_id": project_id or None,
+                        "persona": persona or None,
                     }
                 )
             return rows

turnstone/core/storage/_postgresql.py

@@ -123,6 +123,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     VERDICT_MUTABLE as _VERDICT_MUTABLE,
 )
+from turnstone.core.storage._utils import (
+    assert_single_default_persona as _assert_single_default_persona,
+)
 from turnstone.core.storage._utils import (
     build_attachments_by_msg as _build_attachments_by_msg,
 )
@@ -168,6 +171,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     split_perms as _split_perms,
 )
+from turnstone.core.storage._utils import (
+    validate_and_clear_default_persona as _validate_and_clear_default_persona,
+)
 from turnstone.core.workstream import BULK_CLOSE_STATE_VALUES, WorkstreamKind
 
 log = get_logger(__name__)
@@ -5603,40 +5609,62 @@ class PostgreSQLBackend:
             ).fetchone()
             if existing is not None:
                 raise ValueError(f"persona name already exists: {values['name']}")
+            default_kinds = persona.get("applies_to_kinds") or ["interactive"]
             if values.get("is_default"):
-                self._validate_and_clear_default(
+                # Serialize default promotions cluster-wide: under READ
+                # COMMITTED two concurrent promotions can each miss the
+                # other's uncommitted flag and commit two defaults.  The
+                # xact-scoped advisory lock releases on commit/rollback.
+                conn.execute(
+                    sa.text("SELECT pg_advisory_xact_lock(hashtext('turnstone_personas_default'))")
+                )
+                _validate_and_clear_default_persona(
                     conn,
+                    personas,
                     persona_id=values["persona_id"],
-                    kinds=persona.get("applies_to_kinds") or ["interactive"],
+                    kinds=default_kinds,
                     enabled=persona.get("enabled", True),
                     now=now,
                 )
-            conn.execute(
-                sa.insert(personas),
-                {
-                    "persona_id": values["persona_id"],
-                    "name": values["name"],
-                    "display_name": values.get("display_name", ""),
-                    "description": values.get("description", ""),
-                    "base_prompt": values.get("base_prompt"),
-                    "tool_allowlist": values.get("tool_allowlist"),
-                    "mcp_enabled": values.get("mcp_enabled", 1),
-                    "memory_enabled": values.get("memory_enabled", 1),
-                    "applies_to_kinds": values.get("applies_to_kinds", '["interactive"]'),
-                    "is_default": values.get("is_default", 0),
-                    "enabled": values.get("enabled", 1),
-                    "org_id": values.get("org_id", ""),
-                    "created_by": values.get("created_by", ""),
-                    "created": now,
-                    "updated": now,
-                },
-            )
+            try:
+                conn.execute(
+                    sa.insert(personas),
+                    {
+                        "persona_id": values["persona_id"],
+                        "name": values["name"],
+                        "display_name": values.get("display_name", ""),
+                        "description": values.get("description", ""),
+                        "base_prompt": values.get("base_prompt"),
+                        "tool_allowlist": values.get("tool_allowlist"),
+                        "mcp_enabled": values.get("mcp_enabled", 1),
+                        "memory_enabled": values.get("memory_enabled", 1),
+                        "applies_to_kinds": values.get("applies_to_kinds", '["interactive"]'),
+                        "is_default": values.get("is_default", 0),
+                        "enabled": values.get("enabled", 1),
+                        "org_id": values.get("org_id", ""),
+                        "created_by": values.get("created_by", ""),
+                        "created": now,
+                        "updated": now,
+                    },
+                )
+            except sa.exc.IntegrityError as exc:
+                # SELECT-then-INSERT loser on unique(name): surface the
+                # same ValueError the pre-check raises so callers map one
+                # error shape (400), not an opaque 500.
+                raise ValueError(f"persona name already exists: {values['name']}") from exc
+            if values.get("is_default"):
+                _assert_single_default_persona(conn, personas, default_kinds[0])
             conn.commit()
 
     def update_persona(self, persona_id: str, **fields: Any) -> bool:
         fields = {k: v for k, v in fields.items() if k in _PERSONA_MUTABLE}
         if not fields:
             return False
+        # Validate/serialize BEFORE the invariant checks so malformed input
+        # (explicit-None kinds, wrong types) surfaces as the serializer's
+        # precise ValueError instead of a TypeError escaping the routes'
+        # 400 mapping as a 500.
+        values = _serialize_persona_fields(fields)
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         with self._conn() as conn:
             row = conn.execute(
@@ -5650,64 +5678,37 @@ class PostgreSQLBackend:
                     raise ValueError("the default persona cannot be archived")
                 if "is_default" in fields and not fields["is_default"]:
                     raise ValueError(
-                        "cannot unset is_default directly; set it on the "
-                        "successor persona instead"
+                        "cannot unset is_default directly; set it on the successor persona instead"
                     )
                 if "applies_to_kinds" in fields and sorted(
                     fields["applies_to_kinds"] or []
                 ) != sorted(current["applies_to_kinds"]):
                     raise ValueError("cannot change applies_to_kinds of the default persona")
-            if fields.get("is_default") and not current["is_default"]:
-                self._validate_and_clear_default(
+            promote = bool(fields.get("is_default")) and not current["is_default"]
+            promote_kinds = fields.get("applies_to_kinds", current["applies_to_kinds"])
+            if promote:
+                # Serialize default promotions cluster-wide (see
+                # create_persona for the READ COMMITTED rationale).
+                conn.execute(
+                    sa.text("SELECT pg_advisory_xact_lock(hashtext('turnstone_personas_default'))")
+                )
+                _validate_and_clear_default_persona(
                     conn,
+                    personas,
                     persona_id=persona_id,
-                    kinds=fields.get("applies_to_kinds", current["applies_to_kinds"]),
+                    kinds=promote_kinds,
                     enabled=fields.get("enabled", current["enabled"]),
                     now=now,
                 )
-            values = _serialize_persona_fields(fields)
             values["updated"] = now
             conn.execute(
                 sa.update(personas).where(personas.c.persona_id == persona_id).values(**values)
             )
+            if promote:
+                _assert_single_default_persona(conn, personas, promote_kinds[0])
             conn.commit()
             return True
 
-    @staticmethod
-    def _validate_and_clear_default(
-        conn: sa.engine.Connection,
-        *,
-        persona_id: str,
-        kinds: list[str],
-        enabled: Any,
-        now: str,
-    ) -> None:
-        """Enforce the default-persona invariants and demote the previous
-        holder, inside the caller's transaction.
-
-        Exactly one default per kind: a default must be single-kind (a
-        two-kind default would get orphan-cleared when either kind's crown
-        moves) and enabled; the incumbent default for that kind loses the
-        flag in the same transaction.
-        """
-        if len(kinds) != 1:
-            raise ValueError("a default persona must apply to exactly one kind")
-        if not enabled:
-            raise ValueError("a disabled persona cannot be the default")
-        kind = kinds[0]
-        others = conn.execute(
-            sa.select(personas.c.persona_id, personas.c.applies_to_kinds).where(
-                sa.and_(personas.c.is_default == 1, personas.c.persona_id != persona_id)
-            )
-        ).fetchall()
-        for oid, okinds_raw in others:
-            if kind in json.loads(okinds_raw or "[]"):
-                conn.execute(
-                    sa.update(personas)
-                    .where(personas.c.persona_id == oid)
-                    .values(is_default=0, updated=now)
-                )
-
     # -- Prompt policies -------------------------------------------------------
 
     def list_prompt_policies(self, org_id: str = "") -> list[dict[str, Any]]:

turnstone/core/storage/_sqlite.py

@@ -123,6 +123,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     VERDICT_MUTABLE as _VERDICT_MUTABLE,
 )
+from turnstone.core.storage._utils import (
+    assert_single_default_persona as _assert_single_default_persona,
+)
 from turnstone.core.storage._utils import (
     build_attachments_by_msg as _build_attachments_by_msg,
 )
@@ -168,6 +171,9 @@ from turnstone.core.storage._utils import (
 from turnstone.core.storage._utils import (
     split_perms as _split_perms,
 )
+from turnstone.core.storage._utils import (
+    validate_and_clear_default_persona as _validate_and_clear_default_persona,
+)
 from turnstone.core.workstream import BULK_CLOSE_STATE_VALUES, WorkstreamKind
 
 log = get_logger(__name__)
@@ -5766,40 +5772,55 @@ class SQLiteBackend:
             ).fetchone()
             if existing is not None:
                 raise ValueError(f"persona name already exists: {values['name']}")
+            default_kinds = persona.get("applies_to_kinds") or ["interactive"]
             if values.get("is_default"):
-                self._validate_and_clear_default(
+                _validate_and_clear_default_persona(
                     conn,
+                    personas,
                     persona_id=values["persona_id"],
-                    kinds=persona.get("applies_to_kinds") or ["interactive"],
+                    kinds=default_kinds,
                     enabled=persona.get("enabled", True),
                     now=now,
                 )
-            conn.execute(
-                sa.insert(personas),
-                {
-                    "persona_id": values["persona_id"],
-                    "name": values["name"],
-                    "display_name": values.get("display_name", ""),
-                    "description": values.get("description", ""),
-                    "base_prompt": values.get("base_prompt"),
-                    "tool_allowlist": values.get("tool_allowlist"),
-                    "mcp_enabled": values.get("mcp_enabled", 1),
-                    "memory_enabled": values.get("memory_enabled", 1),
-                    "applies_to_kinds": values.get("applies_to_kinds", '["interactive"]'),
-                    "is_default": values.get("is_default", 0),
-                    "enabled": values.get("enabled", 1),
-                    "org_id": values.get("org_id", ""),
-                    "created_by": values.get("created_by", ""),
-                    "created": now,
-                    "updated": now,
-                },
-            )
+            try:
+                conn.execute(
+                    sa.insert(personas),
+                    {
+                        "persona_id": values["persona_id"],
+                        "name": values["name"],
+                        "display_name": values.get("display_name", ""),
+                        "description": values.get("description", ""),
+                        "base_prompt": values.get("base_prompt"),
+                        "tool_allowlist": values.get("tool_allowlist"),
+                        "mcp_enabled": values.get("mcp_enabled", 1),
+                        "memory_enabled": values.get("memory_enabled", 1),
+                        "applies_to_kinds": values.get("applies_to_kinds", '["interactive"]'),
+                        "is_default": values.get("is_default", 0),
+                        "enabled": values.get("enabled", 1),
+                        "org_id": values.get("org_id", ""),
+                        "created_by": values.get("created_by", ""),
+                        "created": now,
+                        "updated": now,
+                    },
+                )
+            except sa.exc.IntegrityError as exc:
+                # SELECT-then-INSERT loser on unique(name): surface the
+                # same ValueError the pre-check raises so callers map one
+                # error shape (400), not an opaque 500.
+                raise ValueError(f"persona name already exists: {values['name']}") from exc
+            if values.get("is_default"):
+                _assert_single_default_persona(conn, personas, default_kinds[0])
             conn.commit()
 
     def update_persona(self, persona_id: str, **fields: Any) -> bool:
         fields = {k: v for k, v in fields.items() if k in _PERSONA_MUTABLE}
         if not fields:
             return False
+        # Validate/serialize BEFORE the invariant checks so malformed input
+        # (explicit-None kinds, wrong types) surfaces as the serializer's
+        # precise ValueError instead of a TypeError escaping the routes'
+        # 400 mapping as a 500.
+        values = _serialize_persona_fields(fields)
         now = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
         with self._conn() as conn:
             row = conn.execute(
@@ -5813,64 +5834,32 @@ class SQLiteBackend:
                     raise ValueError("the default persona cannot be archived")
                 if "is_default" in fields and not fields["is_default"]:
                     raise ValueError(
-                        "cannot unset is_default directly; set it on the "
-                        "successor persona instead"
+                        "cannot unset is_default directly; set it on the successor persona instead"
                     )
                 if "applies_to_kinds" in fields and sorted(
                     fields["applies_to_kinds"] or []
                 ) != sorted(current["applies_to_kinds"]):
                     raise ValueError("cannot change applies_to_kinds of the default persona")
-            if fields.get("is_default") and not current["is_default"]:
-                self._validate_and_clear_default(
+            promote = bool(fields.get("is_default")) and not current["is_default"]
+            promote_kinds = fields.get("applies_to_kinds", current["applies_to_kinds"])
+            if promote:
+                _validate_and_clear_default_persona(
                     conn,
+                    personas,
                     persona_id=persona_id,
-                    kinds=fields.get("applies_to_kinds", current["applies_to_kinds"]),
+                    kinds=promote_kinds,
                     enabled=fields.get("enabled", current["enabled"]),
                     now=now,
                 )
-            values = _serialize_persona_fields(fields)
             values["updated"] = now
             conn.execute(
                 sa.update(personas).where(personas.c.persona_id == persona_id).values(**values)
             )
+            if promote:
+                _assert_single_default_persona(conn, personas, promote_kinds[0])
             conn.commit()
             return True
 
-    @staticmethod
-    def _validate_and_clear_default(
-        conn: sa.engine.Connection,
-        *,
-        persona_id: str,
-        kinds: list[str],
-        enabled: Any,
-        now: str,
-    ) -> None:
-        """Enforce the default-persona invariants and demote the previous
-        holder, inside the caller's transaction.
-
-        Exactly one default per kind: a default must be single-kind (a
-        two-kind default would get orphan-cleared when either kind's crown
-        moves) and enabled; the incumbent default for that kind loses the
-        flag in the same transaction.
-        """
-        if len(kinds) != 1:
-            raise ValueError("a default persona must apply to exactly one kind")
-        if not enabled:
-            raise ValueError("a disabled persona cannot be the default")
-        kind = kinds[0]
-        others = conn.execute(
-            sa.select(personas.c.persona_id, personas.c.applies_to_kinds).where(
-                sa.and_(personas.c.is_default == 1, personas.c.persona_id != persona_id)
-            )
-        ).fetchall()
-        for oid, okinds_raw in others:
-            if kind in json.loads(okinds_raw or "[]"):
-                conn.execute(
-                    sa.update(personas)
-                    .where(personas.c.persona_id == oid)
-                    .values(is_default=0, updated=now)
-                )
-
     # -- Prompt policies -------------------------------------------------------
 
     def list_prompt_policies(self, org_id: str = "") -> list[dict[str, Any]]:

turnstone/core/storage/_utils.py

@@ -664,23 +664,64 @@ PERSONA_KINDS = frozenset({"interactive", "coordinator"})
 def persona_row_to_dict(row: Any) -> dict[str, Any]:
     """Convert a personas row to the Python-typed dict shape the Protocol
     documents: JSON columns parsed, 0/1 columns as bool.  ``tool_allowlist``
-    keeps its tri-state — None (unrestricted) vs [] (hard empty) vs [names]."""
+    keeps its tri-state — None (unrestricted) vs [] (hard empty) vs [names].
+
+    Raises ValueError on a corrupt row: a malformed allowlist or kinds
+    column must fail loudly (mirroring ``snapshot_from_config``), never
+    decode into a garbage envelope or mask a broken invariant.
+    """
     d = row_to_dict(row, "mcp_enabled", "memory_enabled", "is_default", "enabled")
+
+    def _load_json(column: str, raw_value: Any) -> Any:
+        # Re-raise parser errors with the persona named — a bare
+        # JSONDecodeError message doesn't say WHICH row is corrupt.
+        try:
+            return json.loads(raw_value)
+        except ValueError as exc:
+            raise ValueError(f"corrupt {column} on persona {d.get('persona_id')!r}: {exc}") from exc
+
     raw = d.get("tool_allowlist")
-    d["tool_allowlist"] = None if raw is None else json.loads(raw)
-    d["applies_to_kinds"] = json.loads(d.get("applies_to_kinds") or '["interactive"]')
+    if raw is None:
+        d["tool_allowlist"] = None
+    else:
+        tools = _load_json("tool_allowlist", raw)
+        if not isinstance(tools, list) or not all(isinstance(t, str) for t in tools):
+            raise ValueError(f"corrupt tool_allowlist on persona {d.get('persona_id')!r}")
+        d["tool_allowlist"] = tools
+    kinds_raw = d.get("applies_to_kinds")
+    kinds = _load_json("applies_to_kinds", kinds_raw) if kinds_raw else None
+    if not isinstance(kinds, list) or not kinds or not all(isinstance(k, str) for k in kinds):
+        raise ValueError(f"corrupt applies_to_kinds on persona {d.get('persona_id')!r}")
+    d["applies_to_kinds"] = kinds
     return d
 
 
+# Storage-layer size bounds for operator-authored persona fields.  The
+# console route truncates its inputs to the same shape, but the storage
+# edge is the layer every future ingress (SDK-direct, admin CLI) inherits —
+# reject rather than silently truncate here.
+PERSONA_FIELD_CAPS: dict[str, int] = {
+    "display_name": 128,
+    "description": 1024,
+    "base_prompt": 32768,
+}
+PERSONA_ALLOWLIST_MAX_ENTRIES = 512
+PERSONA_ALLOWLIST_MAX_NAME_LEN = 256
+
+
 def serialize_persona_fields(fields: dict[str, Any]) -> dict[str, Any]:
     """Validate + serialize Python-typed persona fields to column values.
 
-    Shared by both backends so the tri-state allowlist encoding and the
-    kinds validation can't drift between them.  Raises ValueError on
-    malformed input; unknown keys pass through (callers filter to the
-    mutable set first where that matters).
+    Shared by both backends so the tri-state allowlist encoding, the kinds
+    validation, and the size bounds can't drift between them.  Raises
+    ValueError on malformed or oversized input; unknown keys pass through
+    (callers filter to the mutable set first where that matters).
     """
     out = dict(fields)
+    for key, cap in PERSONA_FIELD_CAPS.items():
+        val = out.get(key)
+        if val is not None and key in out and len(str(val)) > cap:
+            raise ValueError(f"{key} exceeds {cap} characters")
     if "applies_to_kinds" in out:
         kinds = out["applies_to_kinds"]
         if not isinstance(kinds, list) or not kinds or not set(kinds) <= PERSONA_KINDS:
@@ -692,11 +733,79 @@ def serialize_persona_fields(fields: dict[str, Any]) -> dict[str, Any]:
         tools = out["tool_allowlist"]
         if not isinstance(tools, list) or not all(isinstance(t, str) for t in tools):
             raise ValueError("tool_allowlist must be None or a list of tool names")
+        if len(tools) > PERSONA_ALLOWLIST_MAX_ENTRIES:
+            raise ValueError(f"tool_allowlist exceeds {PERSONA_ALLOWLIST_MAX_ENTRIES} entries")
+        if any(len(t) > PERSONA_ALLOWLIST_MAX_NAME_LEN for t in tools):
+            raise ValueError(
+                f"tool_allowlist entries are capped at {PERSONA_ALLOWLIST_MAX_NAME_LEN} chars"
+            )
         out["tool_allowlist"] = json.dumps(tools)
     for key in ("mcp_enabled", "memory_enabled", "is_default", "enabled"):
         if key in out:
             out[key] = 1 if out[key] else 0
     return out
+
+
+def validate_and_clear_default_persona(
+    conn: Any,
+    personas_table: Any,
+    *,
+    persona_id: str,
+    kinds: list[str],
+    enabled: Any,
+    now: str,
+) -> None:
+    """Enforce the default-persona invariants and demote the incumbent,
+    inside the caller's transaction.
+
+    Shared by both backends (fully dialect-neutral) so the invariants —
+    exactly one default per kind, single-kind, enabled — cannot drift.
+    A corrupt incumbent row raises rather than being skipped: silently
+    not-demoting it would commit two defaults, the exact state this
+    helper exists to prevent.  Concurrency: the PostgreSQL backend
+    serializes promotions with an advisory xact lock before calling this;
+    ``assert_single_default_persona`` runs post-promote as the backstop.
+    """
+    if not isinstance(kinds, list) or len(kinds) != 1:
+        raise ValueError("a default persona must apply to exactly one kind")
+    if not enabled:
+        raise ValueError("a disabled persona cannot be the default")
+    kind = kinds[0]
+    others = conn.execute(
+        sa.select(personas_table.c.persona_id, personas_table.c.applies_to_kinds).where(
+            sa.and_(
+                personas_table.c.is_default == 1,
+                personas_table.c.persona_id != persona_id,
+            )
+        )
+    ).fetchall()
+    for oid, okinds_raw in others:
+        okinds = json.loads(okinds_raw) if okinds_raw else None
+        if not isinstance(okinds, list):
+            raise ValueError(f"corrupt applies_to_kinds on persona {oid!r}")
+        if kind in okinds:
+            conn.execute(
+                sa.update(personas_table)
+                .where(personas_table.c.persona_id == oid)
+                .values(is_default=0, updated=now)
+            )
+
+
+def assert_single_default_persona(conn: Any, personas_table: Any, kind: str) -> None:
+    """Post-promote backstop: raise (rolling back the enclosing
+    transaction) if more than one enabled default applies to *kind* — a
+    concurrent promotion that slipped past serialization must fail loudly,
+    never commit a nondeterministic default."""
+    rows = conn.execute(
+        sa.select(personas_table.c.persona_id, personas_table.c.applies_to_kinds).where(
+            personas_table.c.is_default == 1
+        )
+    ).fetchall()
+    holders = [oid for oid, kr in rows if kind in (json.loads(kr) if kr else [])]
+    if len(holders) > 1:
+        raise ValueError(f"concurrent default-persona change detected for kind {kind!r}; retry")
+
+
 HEURISTIC_RULE_MUTABLE = frozenset(
     {
         "name",

turnstone/prompts/__init__.py

@@ -273,8 +273,11 @@ 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").
-    #    A persona's base_override replaces exactly this module.
-    if base_override is not None:
+    #    A persona's base_override replaces exactly this module.  Truthy
+    #    check, not ``is not None``: the stamp codec documents ``""`` as
+    #    "use the kind's stock BASE", so the empty string must never
+    #    compose an empty BASE regardless of which caller forwards it.
+    if base_override:
         parts.append(base_override)
     else:
         base_module = "base_coordinator.md" if kind == WorkstreamKind.COORDINATOR else "base.md"

turnstone/ui/static/app.js

@@ -700,7 +700,12 @@ function loadDashboard() {
   const projP = window.TurnstoneProjects
     ? window.TurnstoneProjects.refreshProjects()
     : Promise.resolve();
-  Promise.all([dashP, sessP, projP])
+  // Same for the personas cache — the PERSONA column falls back to raw
+  // slugs until display names arrive.
+  const persP = window.TurnstonePersonas
+    ? window.TurnstonePersonas.refreshPersonas()
+    : Promise.resolve();
+  Promise.all([dashP, sessP, projP, persP])
     .then(function (res) {
       const dashData = res[0];
       const wsList = dashData.workstreams || [];
@@ -931,13 +936,18 @@ function _initSavedWsTable() {
       },
     },
   });
-  // The PROJECT column resolves names from the shared projects cache,
-  // which fills asynchronously — re-render once names arrive.
+  // The PROJECT and PERSONA columns resolve names from the shared caches,
+  // which fill asynchronously — re-render once names arrive.
   if (window.TurnstoneProjects) {
     window.TurnstoneProjects.onProjectsChange(function () {
       if (_wsTable) _wsTable.render();
     });
   }
+  if (window.TurnstonePersonas) {
+    window.TurnstonePersonas.onPersonasChange(function () {
+      if (_wsTable) _wsTable.render();
+    });
+  }
 }
 
 // HTML inline-onclick wrappers — keep the global names the existing markup
@@ -2183,6 +2193,10 @@ function applyRosterSnapshot(list, opts) {
     cur.state = ws.state || cur.state || "idle";
     cur.parent_ws_id = ws.parent_ws_id || null;
     cur.project_id = ws.project_id || null;
+    // Preserve-on-missing (unlike project_id's hard overwrite): roster
+    // snapshots from pre-persona nodes omit the field during a rolling
+    // upgrade, and persona is immutable post-create — keeping the known
+    // value beats flapping labels to the slug fallback.
     cur.persona = ws.persona || cur.persona || "";
     workstreams[ws.id] = cur;
   });