test(personas): guard suite — rank guard, levers, spawn, RBAC, immutability

The 15 guards from the design brief: the approval path is untouched
under any persona (rank guard); empty-toolset personas compose no tools
block and put zero definitions on the wire; the tool_search escape hatch
is soft when included (discovered tools union with the allowlist) and
hard when omitted (pathway disabled, including native defer_loading);
memory-off suppresses recall injection, the memory tool, and
memory-directed nudges while behavioural nudges and task-agent tools
survive; MCP-off is session-wide and refresh-proof; spawn validates
persona at prep time and never inherits the parent's; task_agent's
schema stays persona-free; the stamp is immutable, survives
SessionManager.open threading, and corrupt stamps fail construction
loudly; mandatory prompt policies compose under every persona; CLI
resolution (extracted to resolve_cli_persona_kwargs for testability)
loads seeds, exits clearly on unknown names, and adopts the resume
target's stamp; persona edits/archives never touch stamped workstreams;
and the row-shape contract twins carry the persona field.

RBAC endpoint coverage: admin CRUD 403s without persona.* and succeeds
with it; the picker feed needs no persona permission and hides archived
personas; invariant violations surface as 400s; DELETE is 405.

tests/test_persona_endpoints.py

@@ -0,0 +1,187 @@
+"""Endpoint tests for the personas surface (guard 12 + route contracts).
+
+RBAC: the console admin CRUD is gated per-verb on ``persona.{create,read,
+write}``; the picker feed (``GET /v1/api/personas``) is authenticated but
+deliberately gated by NO persona permission — selecting a persona at
+creation is a user action, authoring is the admin surface.  No DELETE
+route exists (archive-only).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import pytest
+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
+
+if TYPE_CHECKING:
+    from starlette.requests import Request
+    from starlette.responses import Response
+
+from turnstone.console.server import (
+    admin_create_persona,
+    admin_get_persona,
+    admin_list_personas,
+    admin_update_persona,
+)
+from turnstone.core.auth import AuthResult
+from turnstone.server import list_personas_endpoint
+
+
+class _InjectAuthMiddleware(BaseHTTPMiddleware):
+    """Injects an AuthResult whose permissions the test controls via
+    ``app.state.test_permissions`` (empty set = authenticated, no grants)."""
+
+    async def dispatch(self, request: Request, call_next: Any) -> Response:
+        request.state.auth_result = AuthResult(
+            user_id="test-user",
+            scopes=frozenset({"approve"}),
+            token_source="config",
+            permissions=frozenset(request.app.state.test_permissions),
+        )
+        return await call_next(request)
+
+
+def _client(tmp_db: Any, permissions: set[str]) -> TestClient:
+    app = Starlette(
+        routes=[
+            Mount(
+                "/v1",
+                routes=[
+                    Route("/api/personas", list_personas_endpoint),
+                    Route("/api/admin/personas", admin_list_personas),
+                    Route("/api/admin/personas", admin_create_persona, methods=["POST"]),
+                    Route("/api/admin/personas/{persona_id}", admin_get_persona),
+                    Route(
+                        "/api/admin/personas/{persona_id}",
+                        admin_update_persona,
+                        methods=["PATCH"],
+                    ),
+                ],
+            ),
+        ],
+        middleware=[Middleware(_InjectAuthMiddleware)],
+    )
+    app.state.test_permissions = permissions
+    # require_storage_or_503 reads the console's app-scoped handle; the picker
+    # endpoint reads the global registry (tmp_db initialized it) — point both
+    # at the same backend.
+    from turnstone.core.storage import get_storage
+
+    app.state.auth_storage = get_storage()
+    return TestClient(app)
+
+
+_ALL = {"persona.create", "persona.read", "persona.write"}
+
+
+@pytest.fixture
+def seeded(tmp_db: Any) -> str:
+    from turnstone.core.storage import get_storage
+
+    get_storage().create_persona(
+        {
+            "persona_id": "p1",
+            "name": "scribe",
+            "display_name": "Scribe",
+            "tool_allowlist": [],
+            "mcp_enabled": False,
+            "applies_to_kinds": ["interactive"],
+        }
+    )
+    return "p1"
+
+
+class TestRbac:
+    def test_admin_verbs_403_without_grant(self, tmp_db: Any, seeded: str) -> None:
+        c = _client(tmp_db, set())
+        assert c.get("/v1/api/admin/personas").status_code == 403
+        assert c.get("/v1/api/admin/personas/" + seeded).status_code == 403
+        assert (
+            c.post("/v1/api/admin/personas", json={"name": "x"}).status_code == 403
+        )
+        assert (
+            c.patch(
+                "/v1/api/admin/personas/" + seeded, json={"enabled": False}
+            ).status_code
+            == 403
+        )
+
+    def test_admin_verbs_succeed_with_grant(self, tmp_db: Any, seeded: str) -> None:
+        c = _client(tmp_db, _ALL)
+        assert c.get("/v1/api/admin/personas").status_code == 200
+        assert c.get("/v1/api/admin/personas/" + seeded).status_code == 200
+        created = c.post(
+            "/v1/api/admin/personas",
+            json={"name": "writer", "base_prompt": "W", "tool_allowlist": []},
+        )
+        assert created.status_code == 200
+        assert created.json()["tool_allowlist"] == []
+        patched = c.patch(
+            "/v1/api/admin/personas/" + seeded, json={"display_name": "Scribe 2"}
+        )
+        assert patched.status_code == 200
+        assert patched.json()["display_name"] == "Scribe 2"
+
+    def test_picker_needs_no_persona_perm(self, tmp_db: Any, seeded: str) -> None:
+        # Selection at creation must work for users with ZERO persona.*
+        # grants — the feed is authenticated-only, display fields only.
+        c = _client(tmp_db, set())
+        resp = c.get("/v1/api/personas")
+        assert resp.status_code == 200
+        rows = resp.json()["personas"]
+        assert [r["name"] for r in rows] == ["scribe"]
+        assert set(rows[0]) == {
+            "name",
+            "display_name",
+            "description",
+            "applies_to_kinds",
+            "is_default",
+        }
+
+    def test_picker_excludes_archived(self, tmp_db: Any, seeded: str) -> None:
+        from turnstone.core.storage import get_storage
+
+        get_storage().update_persona(seeded, enabled=False)
+        c = _client(tmp_db, set())
+        assert c.get("/v1/api/personas").json()["personas"] == []
+        # ...but the admin list still shows it (include_disabled).
+        admin = _client(tmp_db, _ALL)
+        rows = admin.get("/v1/api/admin/personas").json()["personas"]
+        assert [r["name"] for r in rows] == ["scribe"]
+        assert rows[0]["enabled"] is False
+
+
+class TestRouteContracts:
+    def test_invariant_violations_are_400(self, tmp_db: Any, seeded: str) -> None:
+        c = _client(tmp_db, _ALL)
+        # Duplicate slug.
+        assert (
+            c.post("/v1/api/admin/personas", json={"name": "scribe"}).status_code == 400
+        )
+        # Bad slug shape.
+        assert (
+            c.post("/v1/api/admin/personas", json={"name": "Not A Slug"}).status_code
+            == 400
+        )
+        # Default persona can't be archived.
+        c.patch("/v1/api/admin/personas/" + seeded, json={"is_default": True})
+        resp = c.patch("/v1/api/admin/personas/" + seeded, json={"enabled": False})
+        assert resp.status_code == 400
+        assert "archived" in resp.json()["error"]
+
+    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
+        assert (
+            c.patch("/v1/api/admin/personas/nope", json={"enabled": False}).status_code
+            == 404
+        )
+
+    def test_no_delete_route(self, tmp_db: Any, seeded: str) -> None:
+        c = _client(tmp_db, _ALL)
+        assert c.delete("/v1/api/admin/personas/" + seeded).status_code == 405

tests/test_persona_guards.py

@@ -0,0 +1,561 @@
+"""Guard suite for personas (#683) — the invariants the feature must hold.
+
+Each test pins one of the locked design decisions: personas shape the
+persona's own hands (composition + tool visibility) without ever weakening
+the approval path, task-agent identity, compaction mechanics, or the
+stamped-at-create isolation from later persona edits.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from turnstone.core.personas import PersonaSnapshot, snapshot_from_persona
+from turnstone.core.session import ChatSession
+from turnstone.core.storage import get_storage
+from turnstone.core.storage._utils import PERSONA_MUTABLE
+from turnstone.core.tools import TASK_AGENT_TOOLS
+from turnstone.core.workstream import WorkstreamKind
+
+
+def _snap(
+    *,
+    name: str = "guard",
+    prompt: str = "",
+    tools: frozenset[str] | None = None,
+    mcp: bool = True,
+    memory: bool = True,
+) -> PersonaSnapshot:
+    return PersonaSnapshot(name=name, prompt=prompt, tools=tools, mcp=mcp, memory=memory)
+
+
+def _session(mock_openai_client: Any, **kwargs: Any) -> ChatSession:
+    defaults: dict[str, Any] = dict(
+        client=mock_openai_client,
+        model="local-model",
+        ui=MagicMock(),
+        instructions=None,
+        temperature=0.5,
+        max_tokens=1000,
+        tool_timeout=10,
+    )
+    defaults.update(kwargs)
+    return ChatSession(**defaults)
+
+
+def _wire_names(session: ChatSession) -> list[str]:
+    from turnstone.core.providers._protocol import ModelCapabilities
+
+    caps = ModelCapabilities(supports_web_search=True)
+    with patch.object(session, "_get_capabilities", return_value=caps):
+        tools = session._get_active_tools() or []
+    return [t.get("function", {}).get("name") for t in tools]
+
+
+# ---------------------------------------------------------------------------
+# Guard 1 — rank guard: a gated tool call under ANY persona still hits the
+# normal approval path.  Persona visibility shapes what's advertised, never
+# what's approved.
+# ---------------------------------------------------------------------------
+
+
+class TestRankGuard:
+    def test_approval_path_unchanged_under_persona(self, tmp_db, mock_openai_client) -> None:
+        session = _session(
+            mock_openai_client,
+            persona_snapshot=_snap(tools=frozenset({"bash"})),
+        )
+        item = {
+            "call_id": "c1",
+            "func_name": "bash",
+            "execute": lambda _item: ("c1", "ok"),
+            "needs_approval": True,
+            "header": "test",
+            "preview": "",
+        }
+        with (
+            patch.object(session, "_prepare_tool", return_value=item),
+            patch.object(session, "_evaluate_intent"),
+            patch.object(session, "_emit_state"),
+            patch.object(session, "_init_system_messages"),
+            patch.object(session, "_check_cancelled"),
+        ):
+            session.ui.approve_tools.return_value = (True, None)
+            session._execute_tools([{"id": "c1", "function": {"name": "bash", "arguments": "{}"}}])
+        session.ui.approve_tools.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# Guard 2 — empty-toolset persona: NO tools block in the composed prompt,
+# zero tool definitions on the wire.  (The wire half also lives in
+# test_server_live.py::test_empty_toolset_persona_no_tools_on_wire, which
+# asserts the provider call itself.)
+# ---------------------------------------------------------------------------
+
+
+class TestEmptyToolset:
+    def test_prompt_has_no_tools_block_and_wire_is_empty(
+        self, tmp_db, mock_openai_client
+    ) -> None:
+        session = _session(mock_openai_client, persona_snapshot=_snap(tools=frozenset()))
+        prompt = session.system_messages[0]["content"]
+        # tools.md's IC block opener — self-suppressed on an empty envelope.
+        assert "read_file" not in prompt
+        assert "You have" not in prompt or "memories in scope" not in prompt
+        assert _wire_names(session) == []
+
+    def test_base_override_replaces_only_base(self, tmp_db, mock_openai_client) -> None:
+        session = _session(
+            mock_openai_client,
+            persona_snapshot=_snap(prompt="You are a scribe on a guard test.", tools=frozenset()),
+        )
+        prompt = session.system_messages[0]["content"]
+        assert "You are a scribe on a guard test." in prompt
+        # base.md's IC framing is REPLACED...
+        assert "read before you edit" not in prompt
+        # ...but CONTEXT still composes (the removed /creative fork dropped it).
+        assert "Current time:" in prompt or "Session context" in prompt or "User:" in prompt
+
+
+# ---------------------------------------------------------------------------
+# Guard 3 — tool_search escape hatch: included ⇒ soft set (discovered tools
+# join the visible wire); omitted ⇒ hard set (the pathway is disabled).
+# ---------------------------------------------------------------------------
+
+
+class TestToolSearchEscape:
+    def _mcp_client(self) -> MagicMock:
+        mcp = MagicMock()
+        mcp.get_tools.return_value = [
+            {
+                "type": "function",
+                "function": {"name": "mcp_widget", "description": "widget", "parameters": {}},
+            }
+        ]
+        mcp.resource_count_for_user.return_value = 0
+        mcp.prompt_count_for_user.return_value = 0
+        return mcp
+
+    def test_included_keeps_pathway_and_unions_discovered(
+        self, tmp_db, mock_openai_client
+    ) -> None:
+        session = _session(
+            mock_openai_client,
+            mcp_client=self._mcp_client(),
+            tool_search="on",
+            persona_snapshot=_snap(tools=frozenset({"read_file", "tool_search"})),
+        )
+        assert session._tool_search is not None
+        names = _wire_names(session)
+        assert set(names) == {"read_file", "tool_search"}
+        # Discovery expands the visible set — the allowlist unions with it.
+        session._tool_search.expand_visible(["mcp_widget"])
+        names = _wire_names(session)
+        assert "mcp_widget" in names
+
+    def test_omitted_is_hard(self, tmp_db, mock_openai_client) -> None:
+        session = _session(
+            mock_openai_client,
+            mcp_client=self._mcp_client(),
+            tool_search="on",
+            persona_snapshot=_snap(tools=frozenset({"read_file"})),
+        )
+        # The whole pathway is disabled — covers provider-native
+        # defer_loading mode, which has no synthetic name to filter.
+        assert session._tool_search is None
+        assert session._get_deferred_names() is None
+        assert set(_wire_names(session)) == {"read_file"}
+        # ...and it stays disabled across an MCP catalog refresh.
+        session._rebuild_tool_search()
+        assert session._tool_search is None
+
+
+# ---------------------------------------------------------------------------
+# Guard 4 — memory-off: no recall injection, memory tool hidden, memory
+# nudges suppressed; compaction mechanics stay untouched.
+# ---------------------------------------------------------------------------
+
+
+class TestMemoryOff:
+    def test_memory_levers(self, tmp_db, mock_openai_client) -> None:
+        session = _session(mock_openai_client, persona_snapshot=_snap(memory=False))
+        with patch.object(session, "_select_memory_candidates") as select:
+            session._init_system_messages()
+        select.assert_not_called()
+        assert "memory" not in _wire_names(session)
+        # Memory-directed nudges are suppressed; behavioural nudges stay.
+        session._memory_config.nudges = True
+        assert not session._nudges_enabled("start")
+        assert not session._nudges_enabled("tool_error")
+        assert session._nudges_enabled("repeat")
+        assert session._nudges_enabled("compaction_pending")
+
+    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.
+        hidden = _session(mock_openai_client, persona_snapshot=_snap(tools=frozenset()))
+        assert not hidden._persona_tool_visible("recall")
+        open_hands = _session(mock_openai_client, persona_snapshot=_snap())
+        assert open_hands._persona_tool_visible("recall")
+
+
+# ---------------------------------------------------------------------------
+# Guard 5 — MCP-off is session-wide: no merge into _tools OR _task_tools,
+# no listeners, refresh callback inert.
+# ---------------------------------------------------------------------------
+
+
+class TestMcpOff:
+    def test_session_wide_gate(self, tmp_db, mock_openai_client) -> None:
+        mcp = MagicMock()
+        mcp.get_tools.return_value = [
+            {"type": "function", "function": {"name": "mcp_widget", "parameters": {}}}
+        ]
+        session = _session(
+            mock_openai_client,
+            mcp_client=mcp,
+            persona_snapshot=_snap(mcp=False),
+        )
+        assert session._mcp_client is None
+        names = {t["function"]["name"] for t in session._tools if "function" in t}
+        assert "mcp_widget" not in names
+        task_names = {t["function"]["name"] for t in session._task_tools if "function" in t}
+        assert "mcp_widget" not in task_names
+        mcp.add_listener.assert_not_called()
+        mcp.add_resource_listener.assert_not_called()
+        mcp.add_prompt_listener.assert_not_called()
+        # A late catalog change can't re-merge.
+        session._on_mcp_tools_changed()
+        names_after = {t["function"]["name"] for t in session._tools if "function" in t}
+        assert "mcp_widget" not in names_after
+
+    def test_memory_off_does_not_touch_task_tools(self, tmp_db, mock_openai_client) -> None:
+        # The deliberate asymmetry with guard 5: memory-off shapes the
+        # persona's OWN hands; task agents keep their identity/envelope.
+        session = _session(mock_openai_client, persona_snapshot=_snap(memory=False))
+        assert [t["function"]["name"] for t in session._task_tools] == [
+            t["function"]["name"] for t in TASK_AGENT_TOOLS
+        ]
+
+
+# ---------------------------------------------------------------------------
+# Guard 6 — spawn: persona honored + validated at prep time; omitted means
+# the KIND default (resolved server-side at child creation), never the
+# parent's persona.
+# ---------------------------------------------------------------------------
+
+
+class TestSpawnPersona:
+    def _coord_session(self, mock_openai_client: Any) -> ChatSession:
+        return _session(
+            mock_openai_client,
+            kind=WorkstreamKind.COORDINATOR,
+            user_id="u1",
+            coord_client=MagicMock(),
+        )
+
+    def test_unknown_persona_is_clean_tool_error(self, tmp_db, mock_openai_client) -> None:
+        session = self._coord_session(mock_openai_client)
+        item = session._prepare_spawn_workstream("c1", {"persona": "nope"})
+        assert item.get("error")
+        assert "nope" in item["error"]
+
+    def test_kind_mismatch_is_clean_tool_error(self, tmp_db, mock_openai_client) -> None:
+        get_storage().create_persona(
+            {
+                "persona_id": "px",
+                "name": "coord-only",
+                "applies_to_kinds": ["coordinator"],
+            }
+        )
+        session = self._coord_session(mock_openai_client)
+        item = session._prepare_spawn_workstream("c1", {"persona": "coord-only"})
+        assert item.get("error")
+        assert "interactive" in item["error"]
+
+    def test_valid_persona_travels_to_spawn_body(self, tmp_db, mock_openai_client) -> None:
+        get_storage().create_persona(
+            {"persona_id": "py", "name": "scribe", "applies_to_kinds": ["interactive"]}
+        )
+        session = self._coord_session(mock_openai_client)
+        item = session._prepare_spawn_workstream("c1", {"persona": "scribe"})
+        assert not item.get("error")
+        assert item["persona"] == "scribe"
+        session._coord_client.spawn.return_value = {"ws_id": "w" * 32, "name": "child"}
+        with patch.object(session, "_report_tool_result"):
+            session._exec_spawn_workstream(item)
+        assert session._coord_client.spawn.call_args.kwargs["persona"] == "scribe"
+
+    def test_omitted_persona_is_not_inherited(self, tmp_db, mock_openai_client) -> None:
+        # The parent coordinator has its own persona; the child body must
+        # NOT carry it — the receiving node resolves the interactive
+        # default at child-creation time.
+        session = _session(
+            mock_openai_client,
+            kind=WorkstreamKind.COORDINATOR,
+            user_id="u1",
+            coord_client=MagicMock(),
+            persona_snapshot=_snap(name="executive"),
+        )
+        item = session._prepare_spawn_workstream("c1", {"initial_message": "go"})
+        assert item["persona"] == ""
+        session._coord_client.spawn.return_value = {"ws_id": "w" * 32, "name": "child"}
+        with patch.object(session, "_report_tool_result"):
+            session._exec_spawn_workstream(item)
+        assert session._coord_client.spawn.call_args.kwargs["persona"] == ""
+
+
+# ---------------------------------------------------------------------------
+# Guard 7 — task_agent has no persona parameter (sub-agents keep their own
+# identity; persona is a workstream-level concept).
+# ---------------------------------------------------------------------------
+
+
+def test_task_agent_schema_has_no_persona_param() -> None:
+    from turnstone.core.tools import TOOLS
+
+    task_agent = next(t for t in TOOLS if t["function"]["name"] == "task_agent")
+    assert "persona" not in task_agent["function"]["parameters"]["properties"]
+
+
+# ---------------------------------------------------------------------------
+# Guard 8 — immutability: nothing mutates a workstream's persona post-create.
+# The stamp is written from constructor attrs; config rewrites re-emit the
+# SAME stamp, and the persona row's slug is immutable.
+# ---------------------------------------------------------------------------
+
+
+class TestImmutability:
+    def test_save_config_reemits_same_stamp(self, tmp_db, mock_openai_client) -> None:
+        from turnstone.core.memory import load_workstream_config
+
+        snap = _snap(name="scribe", prompt="P", tools=frozenset({"read_file"}), memory=False)
+        session = _session(mock_openai_client, ws_id="w1" * 16, persona_snapshot=snap)
+        before = {
+            k: v for k, v in load_workstream_config("w1" * 16).items() if k.startswith("persona")
+        }
+        session.temperature = 0.9  # any config-touching change
+        session._save_config()
+        after = {
+            k: v for k, v in load_workstream_config("w1" * 16).items() if k.startswith("persona")
+        }
+        assert before == after == snap.to_config()
+
+    def test_persona_slug_is_immutable_in_storage(self) -> None:
+        assert "name" not in PERSONA_MUTABLE
+
+    def test_legacy_ws_never_gets_backstamped(self, tmp_db, mock_openai_client) -> None:
+        from turnstone.core.memory import load_workstream_config
+
+        session = _session(mock_openai_client, ws_id="w2" * 16)  # no persona
+        session._save_config()
+        keys = load_workstream_config("w2" * 16)
+        assert not any(k.startswith("persona") for k in keys)
+
+
+# ---------------------------------------------------------------------------
+# Guard 9 — the stamp survives rehydrate: SessionManager.open threads it
+# pre-construction (the same lane as the saved model alias).
+# ---------------------------------------------------------------------------
+# (resume()-adoption is covered in test_sessions.py::test_resume_restores_config)
+
+
+class TestRehydrateThreading:
+    def test_open_threads_snapshot_into_build_session(self) -> None:
+        from tests.test_session_manager import FakeAdapter, _make_manager
+
+        class RecordingAdapter(FakeAdapter):
+            def __init__(self) -> None:
+                super().__init__()
+                self.last_build_kwargs: dict[str, Any] = {}
+
+            def build_session(self, ws: Any, **kwargs: Any) -> Any:
+                self.last_build_kwargs = dict(kwargs)
+                return super().build_session(ws, **{"model": kwargs.get("model")})
+
+        adapter = RecordingAdapter()
+        mgr, _, storage = _make_manager(adapter)
+        ws = mgr.create(user_id="u1", persona="scribe")
+        ws_id = ws.id
+        snap = _snap(name="scribe", tools=frozenset(), mcp=False, memory=False)
+        storage.ws_config[ws_id] = snap.to_config()
+        mgr.close(ws_id)
+
+        reopened = mgr.open(ws_id)
+        assert reopened is not None
+        assert reopened.persona == "scribe"
+        assert adapter.last_build_kwargs["persona_snapshot"] == snap
+
+    def test_corrupt_stamp_fails_loudly_never_falls_back(self) -> None:
+        from tests.test_session_manager import _make_manager
+
+        mgr, _, storage = _make_manager()
+        ws = mgr.create(user_id="u1")
+        ws_id = ws.id
+        # Partial stamp = corruption (missing companions).
+        storage.ws_config[ws_id] = {"persona": "scribe"}
+        mgr.close(ws_id)
+        with pytest.raises(ValueError, match="corrupt persona snapshot"):
+            mgr.open(ws_id)
+
+
+# ---------------------------------------------------------------------------
+# Guard 10 — mandatory prompt policies compose under EVERY persona, including
+# empty-toolset ones; tool-gated policies drop with their tool.
+# ---------------------------------------------------------------------------
+
+
+class TestPolicyComposition:
+    def test_db_policy_rides_on_top_of_override(self) -> None:
+        from turnstone.prompts import ClientType, SessionContext, compose_system_message
+
+        ctx = SessionContext(
+            current_datetime="2026-07-02T10:00", timezone="UTC", username="guard"
+        )
+        policies = [
+            {"name": "mandatory", "content": "ALWAYS-ON-POLICY", "enabled": True},
+            {
+                "name": "gated",
+                "content": "BASH-GATED-POLICY",
+                "tool_gate": "bash",
+                "enabled": True,
+            },
+        ]
+        composed = compose_system_message(
+            ClientType.CLI,
+            ctx,
+            frozenset(),  # empty visible set — scribe-shaped
+            db_policies=policies,
+            base_override="You are a scribe.",
+        )
+        assert "You are a scribe." in composed
+        assert "ALWAYS-ON-POLICY" in composed
+        assert "BASH-GATED-POLICY" not in composed  # its tool is hidden
+
+
+# ---------------------------------------------------------------------------
+# Guard 11 — CLI --persona resolution: seed persona loads; unknown name
+# errors clearly at startup; --resume adopts the target's stamp.
+# ---------------------------------------------------------------------------
+
+
+class TestCliPersona:
+    def test_seed_persona_loads(self, tmp_db) -> None:
+        from turnstone.cli import resolve_cli_persona_kwargs
+
+        storage = get_storage()
+        storage.create_persona(
+            {
+                "persona_id": "p1",
+                "name": "writer",
+                "base_prompt": "W",
+                "tool_allowlist": [],
+                "mcp_enabled": False,
+                "applies_to_kinds": ["interactive"],
+            }
+        )
+        kwargs = resolve_cli_persona_kwargs(storage, "writer", None)
+        assert kwargs["persona"] == "writer"
+        assert kwargs["persona_snapshot"].tools == frozenset()
+
+    def test_unknown_name_exits(self, tmp_db, capsys) -> None:
+        from turnstone.cli import resolve_cli_persona_kwargs
+
+        with pytest.raises(SystemExit):
+            resolve_cli_persona_kwargs(get_storage(), "nope", None)
+        assert "not found or disabled" in capsys.readouterr().out
+
+    def test_kind_mismatch_exits(self, tmp_db) -> None:
+        from turnstone.cli import resolve_cli_persona_kwargs
+
+        storage = get_storage()
+        storage.create_persona(
+            {"persona_id": "p2", "name": "exec", "applies_to_kinds": ["coordinator"]}
+        )
+        with pytest.raises(SystemExit):
+            resolve_cli_persona_kwargs(storage, "exec", None)
+
+    def test_resume_adopts_target_stamp(self, tmp_db) -> None:
+        from turnstone.cli import resolve_cli_persona_kwargs
+        from turnstone.core.memory import save_workstream_config
+
+        snap = _snap(name="scribe", tools=frozenset(), mcp=False, memory=False)
+        save_workstream_config("t" * 32, snap.to_config())
+        kwargs = resolve_cli_persona_kwargs(get_storage(), None, "t" * 32)
+        assert kwargs["persona_snapshot"] == snap
+
+    def test_no_default_yields_legacy(self, tmp_db) -> None:
+        from turnstone.cli import resolve_cli_persona_kwargs
+
+        assert resolve_cli_persona_kwargs(get_storage(), None, None) == {}
+
+
+# ---------------------------------------------------------------------------
+# Guard 13 — template independence: edit AND archive the persona after
+# creating a workstream from it — the stamp is untouched.
+# ---------------------------------------------------------------------------
+
+
+class TestTemplateIndependence:
+    def test_edit_and_archive_leave_stamp_alone(self, tmp_db, mock_openai_client) -> None:
+        from turnstone.core.memory import load_workstream_config
+        from turnstone.core.personas import snapshot_from_config
+
+        storage = get_storage()
+        storage.create_persona(
+            {
+                "persona_id": "p1",
+                "name": "scribe",
+                "base_prompt": "ORIGINAL",
+                "tool_allowlist": [],
+                "mcp_enabled": False,
+                "memory_enabled": False,
+                "applies_to_kinds": ["interactive"],
+            }
+        )
+        row = storage.get_persona("p1")
+        assert row is not None
+        session = _session(
+            mock_openai_client, ws_id="w3" * 16, persona_snapshot=snapshot_from_persona(row)
+        )
+        original = session.system_messages[0]["content"]
+        assert "ORIGINAL" in original
+
+        storage.update_persona("p1", base_prompt="EDITED", tool_allowlist=None)
+        storage.update_persona("p1", enabled=False)
+
+        stamped = snapshot_from_config(load_workstream_config("w3" * 16))
+        assert stamped is not None
+        assert stamped.prompt == "ORIGINAL"
+        assert stamped.tools == frozenset()
+        # A fresh construction from the stamp reproduces the ORIGINAL prompt.
+        rehydrated = _session(mock_openai_client, ws_id="w3" * 16, persona_snapshot=stamped)
+        assert "ORIGINAL" in rehydrated.system_messages[0]["content"]
+        assert "EDITED" not in rehydrated.system_messages[0]["content"]
+
+
+# ---------------------------------------------------------------------------
+# Guard 15 — the collector's delta path carries persona on the rows it
+# builds from ws_created events (the saved-list twin lives in
+# test_saved_handler_unified.py).
+# ---------------------------------------------------------------------------
+
+
+def test_ws_created_event_shape_includes_persona() -> None:
+    import inspect
+
+    from turnstone.console import collector
+
+    src = inspect.getsource(collector)
+    # Both the snapshot-diff and SSE-relay ws_created builders must carry it.
+    assert src.count('"persona"') >= 3
+
+
+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()

turnstone/cli.py

@@ -855,6 +855,55 @@ def detect_model(client: Any, provider: str = "openai") -> tuple[str, int | None
 # ─── Main ──────────────────────────────────────────────────────────────────
 
 
+def resolve_cli_persona_kwargs(
+    storage: Any,
+    persona_arg: str | None,
+    resume_target: str | None,
+) -> dict[str, Any]:
+    """Resolve the persona stamp for a CLI session, pre-construction.
+
+    ``--resume`` adopts the TARGET workstream's stamped persona (the resumed
+    session must run from its stamp, not tonight's default); otherwise
+    ``--persona`` (or the interactive default persona) resolves against the
+    shelf.  A database with no personas seeded yields ``{}`` — an unstamped
+    legacy session, byte-identical behavior.
+
+    Unknown/disabled/kind-mismatched ``--persona`` names print a clear error
+    and ``sys.exit(1)`` — a startup misconfiguration must not silently start
+    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
+
+    if resume_target and storage is not None:
+        if persona_arg:
+            print(yellow("--persona is ignored with --resume (the stamped persona applies)"))
+        snap = snapshot_from_config(storage.load_workstream_config(resume_target) or {})
+        if snap is not None:
+            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"))
+            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
+        if default_row:
+            return {
+                "persona": default_row["name"],
+                "persona_snapshot": snapshot_from_persona(default_row),
+            }
+    return {}
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(
         description="Interactive CLI for OpenAI-compatible models with tool calling.",
@@ -1208,13 +1257,6 @@ def main() -> None:
 
     # Resolve the persona stamp BEFORE constructing the session — the four
     # levers apply inside ``ChatSession.__init__``, so resolution can't wait.
-    # ``--resume`` adopts the TARGET workstream's stamped persona (the
-    # resumed session must run from its stamp, not tonight's default);
-    # otherwise ``--persona`` (or the interactive default persona) resolves
-    # against the shelf.  A database with no personas seeded yields an
-    # unstamped legacy session — byte-identical behavior.
-    from turnstone.core.personas import snapshot_from_config, snapshot_from_persona
-
     cli_storage = _get_storage()
     resume_target: str | None = None
     if args.resume:
@@ -1225,37 +1267,7 @@ def main() -> None:
             print(red(f"Workstream not found: {args.resume}"))
             sys.exit(1)
 
-    persona_kwargs: dict[str, Any] = {}
-    if resume_target and cli_storage is not None:
-        if args.persona:
-            print(yellow("--persona is ignored with --resume (the stamped persona applies)"))
-        # A corrupt stamp raises here — fail loudly at startup rather than
-        # silently running the workstream under a default envelope.
-        snap = snapshot_from_config(cli_storage.load_workstream_config(resume_target) or {})
-        if snap is not None:
-            persona_kwargs = {"persona": snap.name, "persona_snapshot": snap}
-    elif args.persona:
-        row = cli_storage.get_persona_by_name(args.persona) if cli_storage else None
-        if not row or not row.get("enabled", False):
-            print(red(f"Persona not found or disabled: {args.persona}"))
-            sys.exit(1)
-        if "interactive" not in (row.get("applies_to_kinds") or []):
-            print(red(f"Persona '{args.persona}' does not apply to interactive sessions"))
-            sys.exit(1)
-        persona_kwargs = {
-            "persona": row["name"],
-            "persona_snapshot": snapshot_from_persona(row),
-        }
-    elif cli_storage is not None:
-        try:
-            default_row = cli_storage.get_default_persona("interactive")
-        except Exception:
-            default_row = None
-        if default_row:
-            persona_kwargs = {
-                "persona": default_row["name"],
-                "persona_snapshot": snapshot_from_persona(default_row),
-            }
+    persona_kwargs = resolve_cli_persona_kwargs(cli_storage, args.persona, resume_target)
 
     ws = manager.create(user_id="", **persona_kwargs)
     if args.skip_permissions and isinstance(ws.ui, TerminalUI):