"""A model's own character, and what it makes of the person in front of it. Two things in one table, and the discriminator is a nullable column — so the assertions that matter most are about the boundary between them: an instance-wide persona must not be reachable as somebody's reflection, and one account's reflection must never be visible or deletable by another. A model-written note about a person that the person cannot read is the thing this must not become. The safety story for self-modification is a record and a way back rather than a gate, which is `SkillRevision`'s argument; the revision tests are where that is pinned. """ from __future__ import annotations import json import pytest from sqlalchemy import select from lembas.db.models import ( AUTHOR_MODEL, AUTHOR_USER, ROLE_USER, Chat, Connection, Model, Persona, User, ) from lembas.services import harness as harness_service from lembas.services import personas as personas_service from lembas.services import settings_store from lembas.services import tools as tools_service from lembas.services.crypto import encrypt @pytest.fixture(autouse=True) def personality_allowed(db, registered): settings_store.update(db, {"default_permissions": {"tools.persona": True}}) connection = Connection( name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("") ) db.add(connection) db.commit() for index, name in enumerate(("test-model", "other-model")): db.add( Model( connection_id=connection.id, model_id=name, display_name=name, position=index, capabilities_json={"tools": True}, ) ) db.commit() def _user(db) -> User: return db.scalars(select(User).order_by(User.created_at)).first() def _second_user(db) -> User: """A row directly, the way `test_sharing.py` makes its three accounts.""" from lembas.security.passwords import hash_password user = User( name="Sam", email="s@example.test", password_hash=hash_password("x"), role="user" ) db.add(user) db.commit() return user def _chat(db, model_id: str = "test-model", user: User | None = None) -> Chat: chat = Chat(user_id=(user or _user(db)).id, title="t", model_id=model_id) db.add(chat) db.commit() return chat async def _run(db, chat: Chat, name: str, args: dict): user = db.get(User, chat.user_id) resolved = tools_service.resolve_tools(db, chat, user) context = tools_service.context_for(db, user, chat, tools=resolved) return await tools_service.run_tool(context, name, json.dumps(args)) # --- The two halves are not the same row -------------------------------------- def test_a_persona_and_a_reflection_are_separate_rows_for_one_model(db): user = _user(db) personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") personas_service.write(db, model_key="test-model", owner=user, content="They test things.") assert personas_service.block(db, "test-model", None) == "I am terse." assert personas_service.block(db, "test-model", user) == "They test things." def test_a_missing_persona_does_not_fall_back_to_a_reflection(db): """They answer different questions. A fallback between them would put "what it makes of you" where "who it is" belongs, in the first person.""" user = _user(db) personas_service.write(db, model_key="test-model", owner=user, content="They test things.") assert personas_service.block(db, "test-model", None) == "" def test_each_model_keeps_its_own_read_of_the_same_person(db): user = _user(db) personas_service.write(db, model_key="test-model", owner=user, content="Impatient.") personas_service.write(db, model_key="other-model", owner=user, content="Thorough.") assert personas_service.block(db, "test-model", user) == "Impatient." assert personas_service.block(db, "other-model", user) == "Thorough." def test_one_accounts_reflection_is_invisible_to_another(db): """The whole reason the reflection is keyed on the person and not only on the model. On an instance with two accounts, inheriting somebody else's is both wrong and a disclosure.""" first = _user(db) second = _second_user(db) personas_service.write(db, model_key="test-model", owner=first, content="Writes tests.") assert personas_service.block(db, "test-model", second) == "" assert [row.content for row in personas_service.reflections_for(db, second)] == [] assert [row.content for row in personas_service.reflections_for(db, first)] == [ "Writes tests." ] # --- Writing, keeping, and going back ----------------------------------------- def test_every_change_keeps_what_was_there(db): personas_service.write(db, model_key="test-model", owner=None, content="First.") personas_service.write( db, model_key="test-model", owner=None, content="Second.", note="thought again" ) row = personas_service.get(db, "test-model", None) assert row.content == "Second." assert [r.content for r in row.revisions] == ["First."] assert row.revisions[0].note == "thought again" def test_writing_the_same_text_again_keeps_no_revision(db): """Otherwise a model that rewrites itself identically every turn fills the history and pushes the real "before" out of it.""" personas_service.write(db, model_key="test-model", owner=None, content="Same.") personas_service.write(db, model_key="test-model", owner=None, content="Same.") assert personas_service.get(db, "test-model", None).revisions == [] def test_reverting_keeps_the_text_it_replaced(db): """An undo that cannot be undone is a second way to lose the same work.""" personas_service.write(db, model_key="test-model", owner=None, content="First.") personas_service.write(db, model_key="test-model", owner=None, content="Second.") row = personas_service.get(db, "test-model", None) personas_service.revert(db, row, row.revisions[0]) # The session is built with `expire_on_commit=False`, so a committed change # is not visible through an object already loaded here until it is expired. db.expire_all() row = personas_service.get(db, "test-model", None) assert row.content == "First." assert "Second." in [r.content for r in row.revisions] assert row.author == AUTHOR_USER def test_the_history_is_bounded(db): for index in range(personas_service.MAX_REVISIONS + 8): personas_service.write(db, model_key="test-model", owner=None, content=f"v{index}") db.expire_all() row = personas_service.get(db, "test-model", None) assert len(row.revisions) <= personas_service.MAX_REVISIONS def test_an_over_long_text_is_trimmed_rather_than_refused(db): """`memories.py`'s rule: a write the model could not have known was too long should not cost it the turn.""" row = personas_service.write( db, model_key="test-model", owner=None, content="x" * 5000 ) assert len(row.content) == personas_service.MAX_PERSONA_CHARS def test_a_reflection_is_held_to_the_shorter_limit(db): row = personas_service.write( db, model_key="test-model", owner=_user(db), content="y" * 5000 ) assert len(row.content) == personas_service.MAX_VIEW_CHARS def test_the_row_survives_the_model_row_being_replaced(db): """Keyed on the model's own id and not on the `Model` primary key, because "Test & refresh" deletes a model the endpoint has stopped listing and gives it a new primary key when it returns. A personality must not be collateral.""" personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") row = db.scalar(select(Model).where(Model.model_id == "test-model")) connection_id = row.connection_id db.delete(row) db.commit() db.add(Model(connection_id=connection_id, model_id="test-model")) db.commit() assert personas_service.block(db, "test-model", None) == "I am terse." # --- What the tools write ----------------------------------------------------- async def test_persona_write_can_only_rewrite_the_answering_model(db): """There is deliberately no argument naming a model: the key is the model this reply is being written by, so a call cannot reach another one's.""" chat = _chat(db, "test-model") outcome = await _run(db, chat, "persona_write", {"content": "I am blunt.", "why": "learnt"}) assert outcome.event["status"] == "ok" assert personas_service.block(db, "test-model", None) == "I am blunt." assert personas_service.block(db, "other-model", None) == "" async def test_persona_write_is_recorded_as_the_models_own_work(db): chat = _chat(db) await _run(db, chat, "persona_write", {"content": "Mine."}) assert personas_service.get(db, "test-model", None).author == AUTHOR_MODEL async def test_an_empty_persona_write_is_refused_rather_than_erasing(db): """It replaces rather than appends, so an empty call would be a wipe — and a model that has been talked into one turn of nonsense should not be able to end its own character in it.""" personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") chat = _chat(db) outcome = await _run(db, chat, "persona_write", {"content": " "}) assert outcome.event["status"] == "error" assert personas_service.block(db, "test-model", None) == "I am terse." async def test_impression_write_is_keyed_on_the_person_as_well_as_the_model(db): chat = _chat(db) await _run(db, chat, "impression_write", {"content": "They want the short answer."}) user = _user(db) assert personas_service.block(db, "test-model", user) == "They want the short answer." # Not the model's own persona, which is the row next to it. assert personas_service.block(db, "test-model", None) == "" async def test_an_empty_impression_write_clears_it(db): """The opposite of the persona, on purpose: "I have no standing view of this person" is a legitimate state, and "I have no character" is not.""" chat = _chat(db) await _run(db, chat, "impression_write", {"content": "Something."}) await _run(db, chat, "impression_write", {"content": ""}) assert personas_service.block(db, "test-model", _user(db)) == "" async def test_the_tool_is_offered_only_with_the_capability_and_the_permission(db): chat = _chat(db) user = _user(db) assert "persona_write" in tools_service.resolve_tools(db, chat, user).by_name model = db.scalar(select(Model).where(Model.model_id == "test-model")) model.capabilities_json = {"tools": True, "tool_persona": False} db.commit() assert "persona_write" not in tools_service.resolve_tools(db, chat, user).by_name model.capabilities_json = {"tools": True} user.role = ROLE_USER settings_store.update(db, {"default_permissions": {"tools.persona": False}}) db.commit() assert "persona_write" not in tools_service.resolve_tools(db, chat, user).by_name # --- What reaches the prompt -------------------------------------------------- def _values(db, chat: Chat, *, families: list[str]) -> dict[str, str]: offered = [ tool.schema for tool in tools_service.registry(db).values() if tools_service.gate_of(tool.family) in families ] return harness_service.context_variables(db, db.get(User, chat.user_id), offered, chat) def test_both_variables_are_gated_on_the_family(db): """A model that may not keep either has no business being handed them, and the query should not happen at all on an instance that does not use this.""" user = _user(db) personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") personas_service.write(db, model_key="test-model", owner=user, content="Impatient.") chat = _chat(db) without = _values(db, chat, families=["memory"]) assert without["persona"] == "" assert without["person_view"] == "" with_it = _values(db, chat, families=["persona"]) assert with_it["persona"] == "I am terse." assert with_it["person_view"] == "Impatient." def test_a_switched_off_persona_reads_as_absent(db): row = personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") row.enabled = False db.commit() assert personas_service.block(db, "test-model", None) == "" def test_the_fragments_vanish_when_there_is_nothing_to_say(db): chat = _chat(db) preamble = harness_service.compose( db, _user(db), [ tool.schema for tool in tools_service.registry(db).values() if tools_service.gate_of(tool.family) == "persona" ], chat, ) assert "Who you are" not in preamble assert "What you have made of them" not in preamble def test_the_fragments_carry_the_texts_when_there_are_some(db): user = _user(db) personas_service.write(db, model_key="test-model", owner=None, content="I argue back.") personas_service.write(db, model_key="test-model", owner=user, content="Likes brevity.") chat = _chat(db) preamble = harness_service.compose( db, user, [ tool.schema for tool in tools_service.registry(db).values() if tools_service.gate_of(tool.family) == "persona" ], chat, ) assert "I argue back." in preamble assert "Likes brevity." in preamble # The persona comes before the impression: a fact the person stated should be # read before an opinion the model formed about them. assert preamble.index("I argue back.") < preamble.index("Likes brevity.") # --- The screens -------------------------------------------------------------- def test_the_person_can_read_and_delete_what_a_model_makes_of_them(client, db, registered): """The whole reason writing one is acceptable. A model-written note about somebody that they cannot see is not something this should hold.""" user = _user(db) personas_service.write(db, model_key="test-model", owner=user, content="Wants brevity.") page = client.get("/settings") assert "Wants brevity." in page.text assert "What models make of you" in page.text row = personas_service.reflections_for(db, user)[0] client.post(f"/api/library/reflections/{row.id}/delete", follow_redirects=False) db.expire_all() assert personas_service.reflections_for(db, user) == [] def test_nobody_can_delete_somebody_elses_reflection(client, db): second = _second_user(db) row = personas_service.write( db, model_key="test-model", owner=second, content="Theirs." ) response = client.post(f"/api/library/reflections/{row.id}/delete", follow_redirects=False) assert response.status_code == 404 db.expire_all() assert personas_service.get(db, "test-model", second) is not None def test_a_models_own_persona_cannot_be_deleted_from_the_settings_page(client, db): """`owner_id IS NULL` is the instance's, not this person's. An id from that half arriving at the reader's route must be refused on ownership rather than found by existence.""" row = personas_service.write(db, model_key="test-model", owner=None, content="Instance.") response = client.post(f"/api/library/reflections/{row.id}/delete", follow_redirects=False) assert response.status_code == 404 db.expire_all() assert personas_service.block(db, "test-model", None) == "Instance." def test_an_administrator_can_read_write_and_revert_a_persona(client, db): model = db.scalar(select(Model).where(Model.model_id == "test-model")) client.post( f"/admin/models/{model.id}/persona", data={"content": "I am terse."}, follow_redirects=False, ) client.post( f"/admin/models/{model.id}/persona", data={"content": "I am not terse at all."}, follow_redirects=False, ) db.expire_all() row = personas_service.get(db, "test-model", None) assert row.content == "I am not terse at all." assert row.author == AUTHOR_USER page = client.get(f"/admin/models/{model.id}/edit") assert "I am not terse at all." in page.text assert "Earlier personalities" in page.text client.post( f"/admin/models/{model.id}/persona/revert", data={"revision_id": row.revisions[0].id}, follow_redirects=False, ) db.expire_all() assert personas_service.get(db, "test-model", None).content == "I am terse." def test_a_revision_of_another_model_cannot_be_restored_onto_this_one(client, db): """Checked against this persona rather than merely existing, or an id from another model's history transplants its personality.""" personas_service.write(db, model_key="other-model", owner=None, content="Theirs first.") personas_service.write(db, model_key="other-model", owner=None, content="Theirs second.") personas_service.write(db, model_key="test-model", owner=None, content="Mine.") foreign = personas_service.get(db, "other-model", None).revisions[0] model = db.scalar(select(Model).where(Model.model_id == "test-model")) response = client.post( f"/admin/models/{model.id}/persona/revert", data={"revision_id": foreign.id}, follow_redirects=False, ) assert response.status_code == 404 db.expire_all() assert personas_service.block(db, "test-model", None) == "Mine." def test_clearing_the_persona_from_the_admin_page_removes_it(client, db): model = db.scalar(select(Model).where(Model.model_id == "test-model")) personas_service.write(db, model_key="test-model", owner=None, content="I am terse.") client.post(f"/admin/models/{model.id}/persona", data={"content": ""}, follow_redirects=False) db.expire_all() assert personas_service.get(db, "test-model", None) is None assert db.scalars(select(Persona)).all() == []