diff --git a/tests/test_console_routing_proxy.py b/tests/test_console_routing_proxy.py index d2205f0f..804034d2 100644 --- a/tests/test_console_routing_proxy.py +++ b/tests/test_console_routing_proxy.py @@ -363,6 +363,22 @@ class TestClusterCreate: assert mock_post.call_args.kwargs["json"]["project_id"] == "proj-42" client.close() + def test_cluster_create_forwards_persona(self) -> None: + # The launcher's persona picker sends persona; the proxy selectively + # REBUILDS the forwarded body (it doesn't pass it through), so persona + # must be explicitly carried or the receiving node stamps its kind + # default instead of the operator's choice. + mock_post = _make_proxy_post(json_data={"ws_id": "p1ws"}) + client = TestClient(self._app_with_node(mock_post), raise_server_exceptions=False) + resp = client.post( + "/v1/api/cluster/workstreams/new", + json={"node_id": "node-a", "name": "j", "persona": "scribe"}, + headers=_TEST_AUTH_HEADERS, + ) + assert resp.status_code == 200 + assert mock_post.call_args.kwargs["json"]["persona"] == "scribe" + client.close() + # --------------------------------------------------------------------------- # Tests — route_proxy diff --git a/tests/test_governance_storage.py b/tests/test_governance_storage.py index 654d5f60..cea1c068 100644 --- a/tests/test_governance_storage.py +++ b/tests/test_governance_storage.py @@ -208,6 +208,16 @@ class TestRolePermissionOverrides: db.set_role_overrides("r1", {"approve", "model.skills.write"}, {"write"}) assert db.get_user_permissions("u1") == {"read", "approve", "model.skills.write"} + def test_get_user_permissions_applies_persona_write_overlay(self, db): + # persona.write is admin-default (migration 063), but the override layer + # can grant it to any NON-admin builtin role — the grant must flow + # through get_user_permissions like any other overlay perm. + db.create_role("r1", "editor", "Editor", "read,write", builtin=True, org_id="") + db.create_user("u1", "alice", "Alice", "$2b$hash") + db.assign_role("u1", "r1") + db.set_role_overrides("r1", {"persona.write"}, set()) + assert db.get_user_permissions("u1") == {"read", "write", "persona.write"} + def test_get_user_permissions_ignores_overlay_on_custom_role(self, db): # Overrides only apply to builtin rows. A custom role with stray # override rows (defensive case — should never happen via the API) diff --git a/tests/test_migration_063.py b/tests/test_migration_063.py index b8fc4188..62086333 100644 --- a/tests/test_migration_063.py +++ b/tests/test_migration_063.py @@ -176,9 +176,7 @@ class TestMigration063: stamped = { str(r[0]): str(r[1]) for r in conn.execute( - sa.text( - "SELECT ws_id, value FROM workstream_config WHERE key='persona'" - ) + sa.text("SELECT ws_id, value FROM workstream_config WHERE key='persona'") ).fetchall() } cols = conn.execute( @@ -217,3 +215,124 @@ class TestMigration063: assert "persona." not in _admin_perms(engine) finally: engine.dispose() + + def test_downgrade_purges_persona_config_keeps_creative_mode(self, tmp_path: Path) -> None: + # The downgrade's load-bearing contract (its own docstring): strip every + # persona* stamp the upgrade synthesized from a creative workstream, but + # leave creative_mode='True' intact so pre-063 code resumes it as + # creative again. + db_path = tmp_path / "063-down-creative.db" + cfg = _alembic_cfg(db_path) + command.upgrade(cfg, "062") + + engine = sa.create_engine(f"sqlite:///{db_path}") + try: + with engine.begin() as conn: + conn.execute( + sa.text( + "INSERT INTO workstreams (ws_id, name, state, created, updated) " + "VALUES ('ws-creative', 'ws-creative', 'closed', " + "'2026-01-01T00:00:00', '2026-01-01T00:00:00')" + ) + ) + conn.execute( + sa.text( + "INSERT INTO workstream_config (ws_id, key, value) " + "VALUES ('ws-creative', 'creative_mode', 'True')" + ) + ) + + command.upgrade(cfg, "063") + # Sanity: the upgrade actually stamped the five persona keys — else + # the downgrade assertion below would pass vacuously. + with engine.connect() as conn: + stamped = { + str(r[0]) + for r in conn.execute( + sa.text("SELECT key FROM workstream_config WHERE ws_id='ws-creative'") + ).fetchall() + } + assert { + "persona", + "persona_prompt", + "persona_tools", + "persona_mcp", + "persona_memory", + } <= stamped + + command.downgrade(cfg, "062") + with engine.connect() as conn: + keys = [ + str(r[0]) + for r in conn.execute( + sa.text("SELECT key FROM workstream_config WHERE ws_id='ws-creative'") + ).fetchall() + ] + creative = conn.execute( + sa.text( + "SELECT value FROM workstream_config " + "WHERE ws_id='ws-creative' AND key='creative_mode'" + ) + ).fetchone() + # Every persona* key is gone… + assert not any(k.startswith("persona") for k in keys) + # …while creative_mode='True' survives the round-trip. + assert creative is not None and str(creative[0]) == "True" + finally: + engine.dispose() + + def test_conversion_skips_workstream_with_existing_persona_key(self, tmp_path: Path) -> None: + # Idempotency guard (063 ~297-324): the conversion SELECT excludes any + # ws that already carries a persona key (NOT IN sub-select). A ws with + # BOTH creative_mode='True' AND a pre-existing persona stamp must upgrade + # without a PK collision on workstream_config(ws_id, key), leave exactly + # one persona row, and keep that stamp untouched. + db_path = tmp_path / "063-idempotent.db" + cfg = _alembic_cfg(db_path) + command.upgrade(cfg, "062") + + engine = sa.create_engine(f"sqlite:///{db_path}") + try: + with engine.begin() as conn: + conn.execute( + sa.text( + "INSERT INTO workstreams (ws_id, name, state, created, updated) " + "VALUES ('ws-both', 'ws-both', 'closed', " + "'2026-01-01T00:00:00', '2026-01-01T00:00:00')" + ) + ) + conn.execute( + sa.text( + "INSERT INTO workstream_config (ws_id, key, value) " + "VALUES ('ws-both', 'creative_mode', 'True')" + ) + ) + conn.execute( + sa.text( + "INSERT INTO workstream_config (ws_id, key, value) " + "VALUES ('ws-both', 'persona', 'scribe')" + ) + ) + + # No IntegrityError: the NOT IN guard skips ws-both, so the writer + # stamp is never re-INSERTed over the existing persona row. + command.upgrade(cfg, "063") + + with engine.connect() as conn: + persona_rows = conn.execute( + sa.text( + "SELECT value FROM workstream_config " + "WHERE ws_id='ws-both' AND key='persona'" + ) + ).fetchall() + row_persona = conn.execute( + sa.text("SELECT persona FROM workstreams WHERE ws_id='ws-both'") + ).fetchone() + # Exactly one stamp, and the pre-existing value is untouched. + assert len(persona_rows) == 1 + assert str(persona_rows[0][0]) == "scribe" + # The conversion's UPDATE never ran for this ws (not in creative_rows), + # so the row-projection column stays NULL — untouched, not 'writer'. + assert row_persona is not None and row_persona[0] is None + finally: + engine.dispose() diff --git a/tests/test_persona_endpoints.py b/tests/test_persona_endpoints.py index 3fe630d7..eb981f3b 100644 --- a/tests/test_persona_endpoints.py +++ b/tests/test_persona_endpoints.py @@ -83,11 +83,13 @@ _ALL = {"persona.create", "persona.read", "persona.write"} def seeded(tmp_db: Any) -> str: from turnstone.core.storage import get_storage + # Non-seed slug/display name: the migration ships a real ``scribe``, so a + # fixture named ``scribe`` would collide on a migrated DB. get_storage().create_persona( { "persona_id": "p1", - "name": "scribe", - "display_name": "Scribe", + "name": "test-scribe", + "display_name": "Test Scribe", "tool_allowlist": [], "mcp_enabled": False, "applies_to_kinds": ["interactive"], @@ -101,14 +103,9 @@ class TestRbac: 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.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 + 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: @@ -117,13 +114,11 @@ class TestRbac: 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": []}, + json={"name": "test-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"} - ) + 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" @@ -134,7 +129,7 @@ class TestRbac: resp = c.get("/v1/api/personas") assert resp.status_code == 200 rows = resp.json()["personas"] - assert [r["name"] for r in rows] == ["scribe"] + assert [r["name"] for r in rows] == ["test-scribe"] assert set(rows[0]) == { "name", "display_name", @@ -152,22 +147,17 @@ class TestRbac: # ...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 [r["name"] for r in rows] == ["test-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 - ) + # Duplicate slug (the seeded fixture owns ``test-scribe``). + assert c.post("/v1/api/admin/personas", json={"name": "test-scribe"}).status_code == 400 # Bad slug shape. - assert ( - c.post("/v1/api/admin/personas", json={"name": "Not A Slug"}).status_code - == 400 - ) + 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}) @@ -210,11 +200,142 @@ class TestRouteContracts: 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 - ) + 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 + + +class TestRbacCrossPerm: + """Single-permission clients pin each handler to its OWN persona.* verb. + + The success-path suite grants all three perms (``_ALL``), so a handler + accidentally wired to the wrong verb (read gating a write, say) still + passes there. A read-only and a write-only client expose that drift: read + can list/get but not create/patch, write can patch but not list. + """ + + def test_read_only_client(self, tmp_db: Any, seeded: str) -> None: + c = _client(tmp_db, {"persona.read"}) + assert c.get("/v1/api/admin/personas").status_code == 200 + assert c.get("/v1/api/admin/personas/" + seeded).status_code == 200 + post = c.post("/v1/api/admin/personas", json={"name": "test-new"}) + assert post.status_code == 403 + assert "persona.create" in post.json()["error"] + patch = c.patch("/v1/api/admin/personas/" + seeded, json={"display_name": "X"}) + assert patch.status_code == 403 + assert "persona.write" in patch.json()["error"] + + def test_write_only_client(self, tmp_db: Any, seeded: str) -> None: + c = _client(tmp_db, {"persona.write"}) + patch = c.patch("/v1/api/admin/personas/" + seeded, json={"display_name": "X2"}) + assert patch.status_code == 200 + assert patch.json()["display_name"] == "X2" + # persona.write does NOT satisfy the read gate on the list. + assert c.get("/v1/api/admin/personas").status_code == 403 + + +class TestArchiveAndDefaultFlipHttp: + """The archive + default-flip lifecycle end-to-end at the HTTP edge — the + layer the storage-level default tests can't see (route wiring + response + projection + the permless picker's enabled filter).""" + + def test_default_flip_demotes_incumbent(self, tmp_db: Any, seeded: str) -> None: + from turnstone.core.storage import get_storage + + # An incumbent interactive default alongside the (non-default) seeded + # persona; flipping the seeded one must demote the incumbent. + get_storage().create_persona( + { + "persona_id": "p2", + "name": "test-eng", + "display_name": "Test Eng", + "applies_to_kinds": ["interactive"], + "is_default": True, + } + ) + c = _client(tmp_db, _ALL) + resp = c.patch("/v1/api/admin/personas/" + seeded, json={"is_default": True}) + assert resp.status_code == 200 + assert resp.json()["is_default"] is True + # Exactly one default per kind after the flip — the incumbent demoted. + rows = c.get("/v1/api/admin/personas").json()["personas"] + defaults = [r["name"] for r in rows if r["is_default"]] + assert defaults == ["test-scribe"] + incumbent = get_storage().get_persona("p2") + assert incumbent is not None and incumbent["is_default"] is False + + def test_archive_non_default_hides_from_picker_keeps_in_admin( + self, tmp_db: Any, seeded: str + ) -> None: + c = _client(tmp_db, _ALL) + resp = c.patch("/v1/api/admin/personas/" + seeded, json={"enabled": False}) + assert resp.status_code == 200 + assert resp.json()["enabled"] is False + # Gone from the permless picker feed… + picker = _client(tmp_db, set()) + assert picker.get("/v1/api/personas").json()["personas"] == [] + # …but still present in the admin list (include_disabled). + rows = c.get("/v1/api/admin/personas").json()["personas"] + assert [r["name"] for r in rows] == ["test-scribe"] + assert rows[0]["enabled"] is False + + def test_unset_default_directly_is_400(self, tmp_db: Any, seeded: str) -> None: + c = _client(tmp_db, _ALL) + # Promote to default, then try to unset the flag directly. + c.patch("/v1/api/admin/personas/" + seeded, json={"is_default": True}) + resp = c.patch("/v1/api/admin/personas/" + seeded, json={"is_default": False}) + assert resp.status_code == 400 + assert "cannot unset is_default directly" in resp.json()["error"] + + +class TestOrgIdGuard: + def test_create_null_org_id_stored_empty(self, tmp_db: Any) -> None: + # An explicit JSON null org_id must persist as "" — ``str(None)`` would + # store the literal "None" and silently scope the persona to a bogus org. + from turnstone.core.storage import get_storage + + c = _client(tmp_db, _ALL) + resp = c.post("/v1/api/admin/personas", json={"name": "test-orgless", "org_id": None}) + assert resp.status_code == 200 + assert resp.json()["org_id"] == "" + stored = get_storage().get_persona(resp.json()["persona_id"]) + assert stored is not None and stored["org_id"] == "" + + +class TestProductionRoutes: + """The hand-built Starlette app in this module can't catch route-table + drift in ``console/server.create_app``. Introspect the real table.""" + + def test_persona_handlers_registered_with_methods(self) -> None: + from unittest.mock import MagicMock + + from starlette.routing import Mount, Route + + from turnstone.console.collector import ClusterCollector + from turnstone.console.server import create_app + + app = create_app(collector=ClusterCollector(storage=MagicMock())) + + def _walk(routes: Any, prefix: str = "") -> Any: + for r in routes: + if isinstance(r, Mount): + yield from _walk(r.routes, prefix + r.path) + elif isinstance(r, Route): + yield prefix + r.path, frozenset(r.methods or ()), r.endpoint.__name__ + + persona_routes = [row for row in _walk(app.routes) if "/personas" in row[0]] + reg = {(path, name): methods for path, methods, name in persona_routes} + + admin = "/v1/api/admin/personas" + admin_one = "/v1/api/admin/personas/{persona_id}" + assert "GET" in reg[(admin, "admin_list_personas")] + assert "POST" in reg[(admin, "admin_create_persona")] + assert "GET" in reg[(admin_one, "admin_get_persona")] + assert "PATCH" in reg[(admin_one, "admin_update_persona")] + # The permless picker feed is registered (creation surface). + assert "GET" in reg[("/v1/api/personas", "list_personas_endpoint")] + # Archive-only contract: NO DELETE anywhere on the persona surface. + all_methods: set[str] = set().union(*reg.values()) + assert "DELETE" not in all_methods diff --git a/tests/test_persona_guards.py b/tests/test_persona_guards.py index 0e10265e..4bef5eb8 100644 --- a/tests/test_persona_guards.py +++ b/tests/test_persona_guards.py @@ -65,28 +65,35 @@ def _wire_names(session: ChatSession) -> list[str]: class TestRankGuard: def test_approval_path_unchanged_under_persona(self, tmp_db, mock_openai_client) -> None: + # bash is IN the persona's allowlist, so it survives the visibility + # filter — but the REAL preparer must still derive needs_approval + # from the tool (not a patched stub), and the gated call must still + # route through ui.approve_tools. Persona visibility shapes what's + # advertised, never what's approved. 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": "", + tc = { + "id": "c1", + "function": {"name": "bash", "arguments": json.dumps({"command": "echo hi"})}, } + # The real _prepare_bash dispatch derives the approval flag. + prepared = session._prepare_tool(tc) + assert prepared["needs_approval"] is True + session.ui.approve_tools.return_value = (True, None) with ( - patch.object(session, "_prepare_tool", return_value=item), + patch.object(session, "_exec_bash", return_value=("c1", "ok")) as exec_bash, 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._execute_tools([tc]) session.ui.approve_tools.assert_called_once() + # The item the gate actually saw carried the derived flag. + gated = session.ui.approve_tools.call_args.args[0] + assert gated[0]["needs_approval"] is True + exec_bash.assert_called_once() # --------------------------------------------------------------------------- @@ -98,15 +105,21 @@ class TestRankGuard: class TestEmptyToolset: - def test_prompt_has_no_tools_block_and_wire_is_empty( - self, tmp_db, mock_openai_client - ) -> None: + 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) == [] + # Even with memories IN SCOPE, the "memories in scope" advisory must + # not compose — the empty toolset hides the memory tool, and the + # preamble must never point the model at a tool the wire omits. The + # prior `"You have" not in prompt or ...` disjunction was vacuous + # (no memory was ever in scope, so the branch was unreachable). + fake = [{"memory_id": "m1", "name": "n", "scope": "user", "scope_id": "u", "content": "c"}] + with patch.object(session, "_select_memory_candidates", return_value=(fake, "recency")): + session._init_system_messages() + assert "memories in scope" not in session.system_messages[0]["content"] def test_base_override_replaces_only_base(self, tmp_db, mock_openai_client) -> None: session = _session( @@ -140,9 +153,7 @@ class TestToolSearchEscape: 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: + 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(), @@ -208,6 +219,50 @@ class TestToolSearchEscape: session._rebuild_tool_search() assert session._tool_search is None + def test_soft_set_expansion_recomposes_prompt(self, tmp_db, mock_openai_client) -> None: + # A soft set composed the prompt against the pre-expansion visible + # names, so tool-gated policy segments for a just-discovered tool + # were dropped. Expanding a NEW name via tool_search must recompose + # so the operator's guidance lands with the tool — but a repeat + # (already-expanded) discovery must NOT pay the recompose again. + 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 + widget = next( + t + for t in session._tool_search.get_deferred_tools() + if t["function"]["name"] == "mcp_widget" + ) + with patch.object(session._tool_search, "search", return_value=[widget]): + with patch.object(session, "_init_system_messages") as recompose: + session._exec_tool_search({"query": "widget", "call_id": "c1"}) + recompose.assert_called_once() + # Second discovery of the same name adds nothing new — no recompose. + with patch.object(session, "_init_system_messages") as recompose_again: + session._exec_tool_search({"query": "widget", "call_id": "c2"}) + recompose_again.assert_not_called() + + def test_legacy_expansion_does_not_recompose(self, tmp_db, mock_openai_client) -> None: + # Without a persona set (_persona_tools is None) the prompt is + # composed against the full catalog, so a tool_search expansion + # needn't recompose — the soft-set recompose is persona-specific. + session = _session(mock_openai_client, mcp_client=self._mcp_client(), tool_search="on") + assert session._tool_search is not None + assert session._persona_tools is None + widget = next( + t + for t in session._tool_search.get_deferred_tools() + if t["function"]["name"] == "mcp_widget" + ) + with patch.object(session._tool_search, "search", return_value=[widget]): + with patch.object(session, "_init_system_messages") as recompose: + session._exec_tool_search({"query": "widget", "call_id": "c1"}) + recompose.assert_not_called() + # --------------------------------------------------------------------------- # Guard 4 — memory-off: no recall injection, memory tool hidden, memory @@ -250,6 +305,108 @@ class TestMemoryOff: open_hands = _session(mock_openai_client, persona_snapshot=_snap()) assert open_hands._persona_tool_visible("recall") + @staticmethod + def _drive_advised_compaction(session: ChatSession) -> tuple[Any, Any]: + """Drive a REAL advised-stop compaction + auto-resume through send(). + + Mirrors ``test_cooperative_compaction.py``'s end-to-end recipe: the + model pauses mid-task (latching ``_compaction_advised``), the turn is + over-threshold, a real summary is produced via ``_utility_completion``, + and the loop hands a ``compaction_resume`` user turn back. Returns the + ``_utility_completion`` and ``_append_user_turn`` spies so callers can + assert the spill happened and which resume-nudge variant fired. + """ + from types import SimpleNamespace + + from turnstone.core.trajectory import turns_from_dicts + + session.messages = turns_from_dicts( + [ + {"role": "user", "content": "do the task"}, + {"role": "assistant", "content": "on it"}, + ] + ) + session._msg_tokens = [5, 5] + session._title_generated = True + session.compact_max_tokens = 100 + session._system_tokens = 0 + summary = SimpleNamespace(content="## Open tasks\nfinish it", finish_reason="stop") + n = {"i": 0} + + def stream(*_a: Any, **_k: Any) -> dict[str, str]: + n["i"] += 1 + if n["i"] == 1: + session._compaction_advised = True # advisory fired this turn + return {"role": "assistant", "content": "pausing to compact"} + return {"role": "assistant", "content": "all done"} + + def est(*_a: Any, **_k: Any) -> int: + return 9_999 if n["i"] <= 1 else 10 # over threshold only on the stop turn + + with ( + patch.object(session, "_create_stream_with_retry", return_value=iter([])), + patch.object(session, "_stream_response", side_effect=stream), + patch.object(session, "_full_messages", return_value=[]), + patch.object(session, "_update_token_table"), + patch.object(session, "_print_status_line"), + patch.object(session, "_emit_state"), + patch.object(session, "_estimated_prompt_tokens", side_effect=est), + patch.object(session, "_utility_completion", return_value=summary) as uc, + patch.object(session, "_append_user_turn", wraps=session._append_user_turn) as resume, + patch("turnstone.core.session.save_message"), + ): + session.send("go") + return uc, resume + + def test_real_compaction_recall_pointer_when_recall_visible( + self, tmp_db, mock_openai_client + ) -> None: + # memory-off hides only the memory tool — recall stays visible, so an + # actual compaction spills a summary AND the resume nudge points at + # recall (NUDGE_COMPACTION_RESUME). + from tests._session_helpers import make_session + from turnstone.core.metacognition import NUDGE_COMPACTION_RESUME + + session = make_session( + client=mock_openai_client, + context_window=10_000, + max_tokens=1_000, + tool_timeout=10, + persona_snapshot=_snap(memory=False), + ) + assert session._persona_tool_visible("recall") + uc, resume = self._drive_advised_compaction(session) + assert uc.call_count >= 1 # a real summary was produced (the spill) + resume_calls = [ + c for c in resume.call_args_list if c.kwargs.get("source") == "compaction_resume" + ] + assert len(resume_calls) == 1 + assert resume_calls[0].args[0] == NUDGE_COMPACTION_RESUME + + def test_real_compaction_recall_pointer_when_recall_hidden( + self, tmp_db, mock_openai_client + ) -> None: + # scribe-shaped empty toolset hides recall — the spill still happens + # but the resume nudge switches to the no-recall variant. + from tests._session_helpers import make_session + from turnstone.core.metacognition import NUDGE_COMPACTION_RESUME_NO_RECALL + + session = make_session( + client=mock_openai_client, + context_window=10_000, + max_tokens=1_000, + tool_timeout=10, + persona_snapshot=_snap(tools=frozenset()), + ) + assert not session._persona_tool_visible("recall") + uc, resume = self._drive_advised_compaction(session) + assert uc.call_count >= 1 + resume_calls = [ + c for c in resume.call_args_list if c.kwargs.get("source") == "compaction_resume" + ] + assert len(resume_calls) == 1 + assert resume_calls[0].args[0] == NUDGE_COMPACTION_RESUME_NO_RECALL + # --------------------------------------------------------------------------- # Guard 5 — MCP-off is session-wide: no merge into _tools OR _task_tools, @@ -467,6 +624,149 @@ class TestRehydrateThreading: with pytest.raises(ValueError, match="corrupt persona snapshot"): mgr.open(ws_id) # retry reproduces the loud error, not RuntimeError + def test_stamp_survives_compaction_then_resume(self, tmp_db, mock_openai_client) -> None: + # Compaction rewrites the conversation, never the persona stamp (it + # lives in workstream_config). After a real compaction on a stamped + # workstream, a fresh resume re-adopts all five levers intact. + from types import SimpleNamespace + + from turnstone.core.memory import ( + register_workstream, + save_message, + save_workstream_config, + ) + from turnstone.core.trajectory import turns_from_dicts + + snap = _snap( + name="scribe", prompt="P", tools=frozenset({"read_file"}), mcp=False, memory=False + ) + ws_id = "w" * 32 + register_workstream(ws_id) + save_message(ws_id, "user", "do the thing") + save_message(ws_id, "assistant", "did the thing") + save_workstream_config(ws_id, snap.to_config()) + + session = _session(mock_openai_client, ws_id=ws_id, persona_snapshot=snap) + session.messages = turns_from_dicts( + [ + {"role": "user", "content": "do the thing"}, + {"role": "assistant", "content": "did the thing"}, + ] + ) + session._msg_tokens = [5, 5] + session.compact_max_tokens = 100 + session._system_tokens = 0 + summary = SimpleNamespace(content="## Decisions\ndense", finish_reason="stop") + with patch.object(session, "_utility_completion", return_value=summary): + assert session._compact_messages(auto=True) is True + + fresh = _session(mock_openai_client) + assert fresh.resume(ws_id) + assert fresh._persona_name == "scribe" + assert fresh._persona_prompt == "P" + assert fresh._persona_tools == frozenset({"read_file"}) + assert fresh._persona_mcp is False + assert fresh._persona_memory is False + + +# --------------------------------------------------------------------------- +# Guard 9, resume-adoption lane — a mid-session ``resume()`` parses the target +# stamp BEFORE mutating session state: a corrupt stamp leaves this session +# intact; an MCP-on stamp is refused by an MCP-gated-off session; an MCP-off +# stamp narrows the live MCP surface in place. +# --------------------------------------------------------------------------- + + +class TestResumeAdoption: + @staticmethod + def _mcp() -> MagicMock: + mcp = MagicMock() + mcp.get_tools.return_value = [ + {"type": "function", "function": {"name": "mcp_widget", "parameters": {}}} + ] + mcp.resource_count_for_user.return_value = 0 + mcp.prompt_count_for_user.return_value = 0 + return mcp + + def test_corrupt_target_stamp_leaves_session_intact(self, tmp_db, mock_openai_client) -> None: + from turnstone.core.memory import ( + load_workstream_config, + register_workstream, + save_message, + save_workstream_config, + ) + + a_id, b_id = "a" * 32, "b" * 32 + register_workstream(a_id) + save_message(a_id, "user", "hi from A") + session = _session(mock_openai_client) + assert session.resume(a_id) # this session lives on A + before_msgs = list(session.messages) + + register_workstream(b_id) + save_message(b_id, "user", "hi from B") + save_workstream_config(b_id, {"persona": "scribe"}) # partial = corrupt + b_config_before = load_workstream_config(b_id) + + with pytest.raises(ValueError, match="corrupt persona snapshot"): + session.resume(b_id) + # Parse-before-mutate: identity + history untouched, still on A. + assert session._ws_id == a_id + assert session.messages == before_msgs + # ...and a later config save writes A's row, never repairs B's stamp + # with a persona the operator never chose for B. + session._save_config() + assert load_workstream_config(b_id) == b_config_before + + def test_mcp_on_stamp_refused_when_gated_off(self, tmp_db, mock_openai_client) -> None: + from turnstone.core.memory import ( + register_workstream, + save_message, + save_workstream_config, + ) + + # A real client was withheld by the persona gate → _mcp_gated_off. + session = _session( + mock_openai_client, mcp_client=self._mcp(), persona_snapshot=_snap(mcp=False) + ) + assert session._mcp_gated_off is True + b_id = "b" * 32 + register_workstream(b_id) + save_message(b_id, "user", "hi") + save_workstream_config(b_id, _snap(name="scribe", mcp=True).to_config()) + with pytest.raises(ValueError, match="open the workstream fresh"): + session.resume(b_id) + + def test_mcp_off_stamp_narrows_in_place(self, tmp_db, mock_openai_client) -> None: + from turnstone.core.memory import ( + register_workstream, + save_message, + save_workstream_config, + ) + + mcp = self._mcp() + session = _session(mock_openai_client, mcp_client=mcp) # legacy: MCP live + assert session._mcp_client is mcp + assert "mcp_widget" in {t["function"]["name"] for t in session._tools if "function" in t} + + b_id = "b" * 32 + register_workstream(b_id) + save_message(b_id, "user", "hi") + save_workstream_config(b_id, _snap(name="scribe", mcp=False).to_config()) + assert session.resume(b_id) + # Surface dropped in place: client gone, no MCP tools on either set. + assert session._mcp_client is None + assert "mcp_widget" not in { + t["function"]["name"] for t in session._tools if "function" in t + } + assert "mcp_widget" not in { + t["function"]["name"] for t in session._task_tools if "function" in t + } + # ...and the three listeners were deregistered on the way out. + mcp.remove_listener.assert_called() + mcp.remove_resource_listener.assert_called() + mcp.remove_prompt_listener.assert_called() + # --------------------------------------------------------------------------- # Guard 10 — mandatory prompt policies compose under EVERY persona, including @@ -478,9 +778,7 @@ 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" - ) + ctx = SessionContext(current_datetime="2026-07-02T10:00", timezone="UTC", username="guard") policies = [ {"name": "mandatory", "content": "ALWAYS-ON-POLICY", "enabled": True}, { @@ -606,18 +904,57 @@ class TestTemplateIndependence: # --------------------------------------------------------------------------- # 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). +# test_saved_handler_unified.py). Both ws_created builders are exercised +# behaviourally: the poll-diff additions path (_reconcile_node) and the +# SSE-relay path (_apply_delta). Precedent: test_console.py::TestCollectorDelta. # --------------------------------------------------------------------------- -def test_ws_created_event_shape_includes_persona() -> None: - import inspect +def _persona_collector() -> Any: + from tests._coord_test_helpers import MockStorage + from turnstone.console.collector import ClusterCollector - from turnstone.console import collector + return ClusterCollector(storage=MockStorage(), discovery_interval=999) - src = inspect.getsource(collector) - # Both the snapshot-diff and SSE-relay ws_created builders must carry it. - assert src.count('"persona"') >= 3 + +def test_ws_created_reconcile_lane_carries_persona() -> None: + # Poll-diff additions: a freshly-appeared workstream becomes a ws_created + # event AND is stored on the node snapshot — persona rides both. + from turnstone.console.collector import NodeSnapshot + + c = _persona_collector() + node = NodeSnapshot(node_id="node-a", server_url="http://a:8080") + c._nodes["node-a"] = node + pending = c._reconcile_node( + "node-a", + node, + [{"id": "ws1", "name": "n", "state": "idle", "kind": "interactive", "persona": "scribe"}], + ) + created = [e for e in pending if e["type"] == "ws_created"] + assert len(created) == 1 + assert created[0]["persona"] == "scribe" + assert node.workstreams["ws1"]["persona"] == "scribe" + + +def test_ws_created_apply_delta_lane_carries_persona() -> None: + # SSE relay: a node's ws_created delta fans out to console listeners AND + # seeds node.workstreams — persona rides the emitted row and the store. + import queue + + from turnstone.console.collector import NodeSnapshot + + c = _persona_collector() + c._nodes["node-a"] = NodeSnapshot(node_id="node-a", server_url="http://a:8080") + q: queue.Queue[dict[str, Any]] = queue.Queue() + c.register_listener(q) + c._apply_delta( + "node-a", + {"type": "ws_created", "ws_id": "ws1", "name": "new", "persona": "scribe"}, + ) + event = q.get_nowait() + assert event["type"] == "ws_created" + assert event["persona"] == "scribe" + assert c._nodes["node-a"].workstreams["ws1"]["persona"] == "scribe" def test_snapshot_roundtrip_via_json() -> None: @@ -697,9 +1034,7 @@ class TestForkAdoptsStamp: ), session_factory=_session_factory, ) - mgr = SessionManager( - adapter, storage=get_storage(), max_active=10, event_emitter=adapter - ) + mgr = SessionManager(adapter, storage=get_storage(), max_active=10, event_emitter=adapter) handler = make_create_handler( SessionEndpointConfig( permission_gate=None, @@ -729,7 +1064,15 @@ class TestForkAdoptsStamp: app.state.global_listeners = [] app.state.global_listeners_lock = threading.Lock() - yield TestClient(app, raise_server_exceptions=False), mgr + try: + yield TestClient(app, raise_server_exceptions=False), mgr + finally: + # Close the sessions this manager created, then release the + # process-global WebUI queue so it can't leak into the next test + # (precedent: test_coord_rich_ws_state_payload.py teardown). + for ws in mgr.list_all(): + mgr.close(ws.id) + WebUI._global_queue = None def _seed_default(self) -> None: get_storage().create_persona( @@ -787,3 +1130,179 @@ class TestForkAdoptsStamp: assert ws is not None and ws.session is not None assert ws.session._persona_name == "" # unstamped, not the default assert not ws.persona + + +# --------------------------------------------------------------------------- +# Guard 6 (receiving side) — a fresh create resolves the body ``persona`` at +# creation time and stamps it: no separate persona permission is required +# (workstreams.create alone suffices); a kind-mismatch / unknown persona is a +# loud 4xx; an omitted persona takes the kind default; a FAILED default lookup +# fails closed (503) rather than silently widening to the stock envelope. +# --------------------------------------------------------------------------- + + +class TestCreateStampsPersona: + @pytest.fixture() + def _create_app(self, tmp_db): + """The production ``make_create_handler`` behind the production + permission gate — the caller carries ONLY ``workstreams.create`` (no + service scope, no persona-specific permission), so a successful stamp + proves persona needs no dedicated permission.""" + import queue + import threading + + from starlette.applications import Starlette + from starlette.middleware import Middleware + from starlette.middleware.base import BaseHTTPMiddleware + from starlette.routing import Mount, Route + from starlette.testclient import TestClient + + from turnstone.core.adapters.interactive_adapter import InteractiveAdapter + from turnstone.core.auth import AuthResult + from turnstone.core.session_manager import SessionManager + from turnstone.core.session_routes import SessionEndpointConfig, make_create_handler + from turnstone.server import ( + WebUI, + _interactive_create_build_kwargs, + _interactive_create_post_install, + _interactive_create_validate_request, + _interactive_manager_lookup, + _interactive_tenant_check, + ) + + class _Auth(BaseHTTPMiddleware): + async def dispatch(self, request: Any, call_next: Any) -> Any: + request.state.auth_result = AuthResult( + user_id="test-user", + scopes=frozenset(), # no service scope — no bypass + token_source="config", + permissions=frozenset({"workstreams.create"}), + ) + return await call_next(request) + + def _session_factory(ui: Any, model_alias: Any = None, ws_id: Any = None, **kw: Any): + return ChatSession( + client=MagicMock(), + model=model_alias or "test-model", + ui=ui, + instructions=None, + temperature=0.5, + max_tokens=1000, + tool_timeout=10, + ws_id=ws_id, + persona_snapshot=kw.get("persona_snapshot"), + ) + + gq: queue.Queue[dict[str, Any]] = queue.Queue() + WebUI._global_queue = gq + adapter = InteractiveAdapter( + global_queue=gq, + ui_factory=lambda ws: WebUI( + ws_id=ws.id, + user_id=ws.user_id, + kind=ws.kind, + parent_ws_id=ws.parent_ws_id, + ), + session_factory=_session_factory, + ) + mgr = SessionManager(adapter, storage=get_storage(), max_active=10, event_emitter=adapter) + handler = make_create_handler( + SessionEndpointConfig( + permission_gate=None, + manager_lookup=_interactive_manager_lookup, + tenant_check=_interactive_tenant_check, + not_found_label="Workstream not found", + audit_action_prefix="workstream", + create_supports_attachments=True, + create_supports_user_id_override=True, + create_validate_request=_interactive_create_validate_request, + create_build_kwargs=_interactive_create_build_kwargs, + create_post_install=_interactive_create_post_install, + ), + accepted_permissions=("workstreams.create", "admin.coordinator"), + ) + app = Starlette( + routes=[ + Mount( + "/v1", + routes=[Route("/api/workstreams/new", handler, methods=["POST"])], + ) + ], + middleware=[Middleware(_Auth)], + ) + app.state.workstreams = mgr + app.state.skip_permissions = True + app.state.global_queue = gq + app.state.global_listeners = [] + app.state.global_listeners_lock = threading.Lock() + + try: + yield TestClient(app, raise_server_exceptions=False), mgr + finally: + for ws in mgr.list_all(): + mgr.close(ws.id) + WebUI._global_queue = None + + @staticmethod + def _seed(name: str, kinds: list[str], **extra: Any) -> None: + get_storage().create_persona( + {"persona_id": "p_" + name, "name": name, "applies_to_kinds": kinds, **extra} + ) + + def test_create_stamps_persona_with_create_permission_only(self, _create_app) -> None: + client, mgr = _create_app + self._seed("scribe", ["interactive"], base_prompt="S", tool_allowlist=[], mcp_enabled=False) + resp = client.post("/v1/api/workstreams/new", json={"persona": "scribe"}) + assert resp.status_code == 200, resp.text + ws = mgr.get(resp.json()["ws_id"]) + assert ws is not None and ws.session is not None + assert ws.persona == "scribe" + assert ws.session._persona_name == "scribe" + + def test_create_kind_mismatch_is_400(self, _create_app) -> None: + client, mgr = _create_app + self._seed("orchestrator", ["coordinator"]) # coordinator-only + resp = client.post("/v1/api/workstreams/new", json={"persona": "orchestrator"}) + assert resp.status_code == 400 + assert "does not apply to kind" in resp.json()["error"] + + def test_create_unknown_persona_is_400(self, _create_app) -> None: + client, mgr = _create_app + resp = client.post("/v1/api/workstreams/new", json={"persona": "nonexistent"}) + assert resp.status_code == 400 + assert "not found or disabled" in resp.json()["error"] + + def test_create_omitted_persona_stamps_default(self, _create_app) -> None: + client, mgr = _create_app + self._seed("engineer", ["interactive"], is_default=True) + resp = client.post("/v1/api/workstreams/new", json={"name": "x"}) + assert resp.status_code == 200, resp.text + ws = mgr.get(resp.json()["ws_id"]) + assert ws is not None and ws.session is not None + assert ws.session._persona_name == "engineer" + + def test_create_default_lookup_failure_is_503(self, _create_app) -> None: + # Fail-closed: a storage blip during default resolution must not + # degrade to the stock (unstamped) envelope — the operator may have + # promoted a restricted persona to default, and silently widening it + # is the failure mode this lane guards against. + client, mgr = _create_app + storage = get_storage() + with patch.object( + type(storage), "get_default_persona", side_effect=RuntimeError("db down") + ): + resp = client.post("/v1/api/workstreams/new", json={"name": "x"}) + assert resp.status_code == 503 + assert "persona resolution unavailable" in resp.json()["error"] + + def test_create_clean_none_default_is_unstamped_legacy(self, _create_app) -> None: + # No persona in the body and no default configured — a clean ``None`` + # (not a lookup failure) still creates, unstamped, byte-identical to a + # legacy pre-persona workstream. + client, mgr = _create_app + resp = client.post("/v1/api/workstreams/new", json={"name": "x"}) + assert resp.status_code == 200, resp.text + ws = mgr.get(resp.json()["ws_id"]) + assert ws is not None and ws.session is not None + assert ws.session._persona_name == "" + assert not ws.persona diff --git a/tests/test_persona_storage.py b/tests/test_persona_storage.py index 47310218..060fe201 100644 --- a/tests/test_persona_storage.py +++ b/tests/test_persona_storage.py @@ -13,6 +13,7 @@ from __future__ import annotations from typing import Any import pytest +import sqlalchemy as sa def _mk(backend: Any, name: str, **over: Any) -> dict[str, Any]: @@ -32,8 +33,10 @@ def _mk(backend: Any, name: str, **over: Any) -> dict[str, Any]: class TestPersonaCRUD: def test_create_and_get_defaults(self, backend: Any) -> None: - p = _mk(backend, "scribe") - assert p["display_name"] == "Scribe" + # Non-seed slug (the migration seeds a real "scribe"); the display name + # is name.title(), so a hyphenated slug title-cases each segment. + p = _mk(backend, "test-scribe") + assert p["display_name"] == "Test-Scribe" assert p["base_prompt"] is None assert p["tool_allowlist"] is None assert p["mcp_enabled"] is True @@ -48,16 +51,16 @@ class TestPersonaCRUD: assert backend.get_default_persona("interactive") is None def test_get_by_name(self, backend: Any) -> None: - _mk(backend, "writer", base_prompt="You write.") - p = backend.get_persona_by_name("writer") + _mk(backend, "test-writer", base_prompt="You write.") + p = backend.get_persona_by_name("test-writer") assert p is not None - assert p["persona_id"] == "id-writer" + assert p["persona_id"] == "id-test-writer" assert p["base_prompt"] == "You write." def test_duplicate_name_rejected(self, backend: Any) -> None: - _mk(backend, "scribe") + _mk(backend, "test-scribe") with pytest.raises(ValueError, match="already exists"): - backend.create_persona({"persona_id": "other", "name": "scribe"}) + backend.create_persona({"persona_id": "other", "name": "test-scribe"}) def test_missing_identity_rejected(self, backend: Any) -> None: with pytest.raises(ValueError, match="persona_id and name"): @@ -194,3 +197,139 @@ class TestPersonaDefaults: # falls back to unstamped legacy creation. _mk(backend, "p") assert backend.get_default_persona("interactive") is None + + +class TestPersonaStorageHardening: + """Serializer size caps, corrupt-row reads, the serialize-before-invariant + ordering, and the single-default backstop — the storage edge every future + ingress (SDK-direct, admin CLI) inherits, so it rejects rather than + truncates or decodes garbage.""" + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("display_name", "x" * 129), + ("description", "x" * 1025), + ("base_prompt", "x" * 32769), + ], + ) + def test_capped_field_over_limit_raises(self, backend: Any, field: str, value: str) -> None: + # Each operator-authored text field is bounded; one char over its cap is + # a ValueError naming the field, not a silent truncation. + with pytest.raises(ValueError, match=field): + _mk(backend, "capped", **{field: value}) + + def test_allowlist_too_many_entries_raises(self, backend: Any) -> None: + with pytest.raises(ValueError, match="tool_allowlist"): + _mk(backend, "big-list", tool_allowlist=[f"t{i}" for i in range(513)]) + + def test_allowlist_entry_too_long_raises(self, backend: Any) -> None: + with pytest.raises(ValueError, match="tool_allowlist"): + _mk(backend, "long-entry", tool_allowlist=["x" * 257]) + + def test_corrupt_allowlist_read_raises_naming_persona(self, backend: Any) -> None: + # A row whose tool_allowlist JSON parses but is the wrong shape (an + # object where a list-of-strings is required) must fail loudly on read, + # naming the persona — never decode into a garbage envelope that masks a + # broken invariant. + _mk(backend, "corrupt-row") + with backend._engine.begin() as conn: + conn.execute( + sa.text("UPDATE personas SET tool_allowlist = :bad WHERE persona_id = :pid"), + {"bad": '{"not": "a list"}', "pid": "id-corrupt-row"}, + ) + with pytest.raises(ValueError, match="id-corrupt-row"): + backend.get_persona("id-corrupt-row") + with pytest.raises(ValueError, match="id-corrupt-row"): + backend.list_personas() + + def test_update_none_kinds_raises_value_error_not_type_error(self, backend: Any) -> None: + # applies_to_kinds=None (an explicit JSON null from an + # UpdatePersonaRequest) reaches storage; validating BEFORE the invariant + # checks surfaces the serializer's precise ValueError instead of a + # TypeError escaping the route's 400 mapping as a 500. pytest.raises on + # ValueError alone would let a TypeError propagate and fail the test. + _mk(backend, "upd-none") + with pytest.raises(ValueError, match="applies_to_kinds"): + backend.update_persona("id-upd-none", applies_to_kinds=None, is_default=True) + + def test_duplicate_name_insert_race_maps_to_value_error(self, backend: Any) -> None: + # TOCTOU: two concurrent creates both pass the name pre-check, then one + # loses the UNIQUE(name) INSERT. The loser's IntegrityError must surface + # as the same "already exists" ValueError the pre-check raises (one 400 + # shape), never an opaque 500. Force the race window by blanking the + # pre-check's result for a name that really exists, so the INSERT hits a + # genuine constraint violation. + import contextlib + + _mk(backend, "racer") # the winner row is really present now + real_conn = backend._conn + + class _NoRow: + def fetchone(self) -> None: + return None + + class _PrecheckMiss: + # Delegates to a real connection but blanks the FIRST result + # (create_persona's name pre-check) so the code proceeds to INSERT. + def __init__(self, conn: Any) -> None: + self._conn = conn + self._blanked = False + + def execute(self, *args: Any, **kwargs: Any) -> Any: + result = self._conn.execute(*args, **kwargs) + if not self._blanked: + self._blanked = True + return _NoRow() + return result + + def __getattr__(self, name: str) -> Any: + return getattr(self._conn, name) + + @contextlib.contextmanager + def _racing_conn() -> Any: + with real_conn() as conn: + yield _PrecheckMiss(conn) + + backend._conn = _racing_conn + try: + with pytest.raises(ValueError, match="already exists"): + backend.create_persona({"persona_id": "racer-2", "name": "racer"}) + finally: + backend._conn = real_conn + + def test_single_default_backstop_rolls_back( + self, backend: Any, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Manufacture two enabled interactive defaults directly (bypassing the + # demotion the normal path enforces), then suppress the in-txn demotion + # to model a promotion that slipped past serialization — the exact + # concurrent state the post-promote backstop exists to catch. Its + # ValueError must roll the whole transaction back (the promotion must + # NOT stick). + now = "2026-01-01T00:00:00" + with backend._engine.begin() as conn: + for pid in ("mfg-d1", "mfg-d2"): + conn.execute( + sa.text( + "INSERT INTO personas (persona_id, name, display_name, " + "description, base_prompt, tool_allowlist, mcp_enabled, " + "memory_enabled, applies_to_kinds, is_default, enabled, " + "org_id, created_by, created, updated) VALUES " + "(:pid, :pid, '', '', NULL, NULL, 1, 1, :kinds, 1, 1, " + "'', '', :now, :now)" + ), + {"pid": pid, "kinds": '["interactive"]', "now": now}, + ) + _mk(backend, "promotee") # a third: enabled, interactive, non-default + monkeypatch.setattr( + type(backend).__module__ + "._validate_and_clear_default_persona", + lambda *a, **k: None, + ) + with pytest.raises(ValueError, match="concurrent default"): + backend.update_persona("id-promotee", is_default=True) + # The backstop rolled the txn back: the promotion did not commit, and the + # manufactured pair still hold their (illegally duplicated) default flag. + assert backend.get_persona("id-promotee")["is_default"] is False + assert backend.get_persona("mfg-d1")["is_default"] is True + assert backend.get_persona("mfg-d2")["is_default"] is True diff --git a/tests/test_saved_handler_unified.py b/tests/test_saved_handler_unified.py index e7a1ab86..1dd4a064 100644 --- a/tests/test_saved_handler_unified.py +++ b/tests/test_saved_handler_unified.py @@ -358,6 +358,31 @@ async def test_single_kind_saved_unchanged(monkeypatch: pytest.MonkeyPatch) -> N } +async def test_saved_row_maps_persona_value(monkeypatch: pytest.MonkeyPatch) -> None: + """The saved-row builder must map the persona SLUG through to the row + dict — key-presence alone (asserted above) wouldn't catch a positional + column mix-up in the 18-tuple unpack. A distinct project_id/owner/persona + triple (adjacent tuple slots 15/16/17) pins the persona value to the + right column: an off-by-one onto owner or project_id fails the assert.""" + coord = [ + _row( + "c" * 32, + updated="2026-03-01T00:00:00", + kind="coordinator", + project_id="proj-x", + owner="alice", + persona="scribe", + ) + ] + _patch_storage(monkeypatch, coord_rows=coord, interactive_rows=[]) + + handler = make_saved_handler(_coord_cfg()) + rows = (await _body(await handler(_request())))["workstreams"] + row = rows[0] + assert row["persona"] == "scribe" + assert row["project_id"] == "proj-x" # adjacent slot maps distinctly + + async def test_single_kind_saved_500s_on_missing_list_kind(monkeypatch: pytest.MonkeyPatch) -> None: """The single-kind misconfig guard is unchanged by the extraction.""" _patch_storage(monkeypatch, coord_rows=[], interactive_rows=[]) diff --git a/tests/test_shell_js.py b/tests/test_shell_js.py index dbd4cc8d..ba5d1c48 100644 --- a/tests/test_shell_js.py +++ b/tests/test_shell_js.py @@ -27,6 +27,7 @@ _SHELL_CSS = _SHARED / "shell.css" _CONSOLE_INDEX = _ROOT / "turnstone/console/static/index.html" _CONSOLE_APP = _ROOT / "turnstone/console/static/app.js" _CONSOLE_ADMIN = _ROOT / "turnstone/console/static/admin.js" +_UI_INDEX = _ROOT / "turnstone/ui/static/index.html" _RAIL_JS = _SHARED / "rail.js" @@ -157,6 +158,35 @@ def test_console_index_loads_shell_module_and_caps() -> None: assert "TURNSTONE_SHELL_CAPS" in body, "console index must set the shell capability flags" +def test_persona_picker_surfaces_wired() -> None: + """The persona creation/authoring surfaces are wired the same way every + other feature is — losing an id or the shared data-layer script tag + silently drops the picker without a JS error. + + The standalone server UI carries BOTH creation pickers (the quick-create + ``dashboard-persona`` select and the full ``new-ws-persona`` dialog select) + plus the shared ``personas.js`` data layer; the console carries the same + data layer plus the admin authoring ``persona-shelf`` dialog, mounted by + admin.js. (The console launcher's own picker is a composer OPTION field, + not a static id — see test_console_launcher_routes_by_kind.) + """ + ui_index = _UI_INDEX.read_text(encoding="utf-8") + assert 'id="new-ws-persona"' in ui_index, "the new-ws dialog must carry the persona select" + assert 'id="dashboard-persona"' in ui_index, "the quick-create persona select must exist" + assert '