LLeMbas/tests/test_personality.py

"""A model's personality with one person, and what it makes of them.

Both are per (model, person), in two tables — so the assertions that matter most
are about the boundaries between them: the administrator's default must not leak
*into* somebody who has their own, one account's personality and impression must
be invisible and undeletable to another, and a personality must not be reachable
through the impression route or the other way round.

A model-written text about a person that the person cannot read is the thing this
must not become, so the settings page is tested as part of the feature rather than
as decoration.

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. An impression deliberately has no history — see its own docstring.
"""

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,
    Impression,
    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_personality_and_an_impression_are_separate_rows(db):
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=user, content="I am terse.")
    personas_service.write_impression(
        db, model_key="test-model", owner=user, content="They test things."
    )

    assert personas_service.block(db, "test-model", user) == "I am terse."
    assert personas_service.view_block(db, "test-model", user) == "They test things."


def test_a_personality_is_each_persons_own(db):
    """The change asked for in 1.5.0: a character is something a model works out
    with somebody, so two people do not share one."""
    first = _user(db)
    second = _second_user(db)
    personas_service.write(db, model_key="test-model", owner=first, content="Blunt with them.")
    personas_service.write(db, model_key="test-model", owner=second, content="Careful here.")

    assert personas_service.block(db, "test-model", first) == "Blunt with them."
    assert personas_service.block(db, "test-model", second) == "Careful here."


def test_a_person_without_one_of_their_own_gets_the_default(db):
    """What makes the administrator's default mean anything."""
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=None, content="The default.")
    assert personas_service.block(db, "test-model", user) == "The default."


def test_the_default_stops_applying_once_somebody_has_their_own(db):
    """A starting point and not a layer. Two personalities at once would
    contradict each other and nobody could tell which was losing -- the same
    reasoning that makes system prompts replace rather than stack."""
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=None, content="The default.")
    personas_service.write(db, model_key="test-model", owner=user, content="Mine.")

    assert personas_service.block(db, "test-model", user) == "Mine."
    # And the default is untouched, for everybody who has not got their own.
    assert personas_service.block(db, "test-model", _second_user(db)) == "The default."


def test_a_missing_personality_does_not_fall_back_to_an_impression(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_impression(
        db, model_key="test-model", owner=user, content="They test things."
    )
    assert personas_service.block(db, "test-model", user) == ""


def test_each_model_keeps_its_own_read_of_the_same_person(db):
    user = _user(db)
    personas_service.write_impression(
        db, model_key="test-model", owner=user, content="Impatient."
    )
    personas_service.write_impression(
        db, model_key="other-model", owner=user, content="Thorough."
    )

    assert personas_service.view_block(db, "test-model", user) == "Impatient."
    assert personas_service.view_block(db, "other-model", user) == "Thorough."


def test_one_accounts_impression_is_invisible_to_another(db):
    first = _user(db)
    second = _second_user(db)
    personas_service.write_impression(
        db, model_key="test-model", owner=first, content="Writes tests."
    )

    assert personas_service.view_block(db, "test-model", second) == ""
    assert [row.content for row in personas_service.impressions_for(db, second)] == []
    assert [row.content for row in personas_service.impressions_for(db, first)] == [
        "Writes tests."
    ]


def test_one_accounts_personality_is_invisible_to_another(db):
    first = _user(db)
    second = _second_user(db)
    personas_service.write(db, model_key="test-model", owner=first, content="Mine alone.")

    assert [row.content for row in personas_service.personas_of(db, second)] == []
    assert [row.content for row in personas_service.personas_of(db, first)] == ["Mine alone."]


# --- 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_an_impression_is_held_to_the_shorter_limit(db):
    """Shorter on purpose: it is a standing impression, not a file."""
    row = personas_service.write_impression(
        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 or a person: both are
    taken from the context, so a call cannot reach another model's character or
    somebody else's copy of this one's."""
    user = _user(db)
    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", user) == "I am blunt."
    assert personas_service.get(db, "other-model", user) is None


async def test_persona_write_never_touches_the_default(db):
    """A model editing everybody's starting point from inside one conversation is
    a much larger thing than editing its own character."""
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=None, content="The default.")
    chat = _chat(db, "test-model")

    await _run(db, chat, "persona_write", {"content": "Mine now."})

    assert personas_service.block(db, "test-model", user) == "Mine now."
    assert personas_service.get(db, "test-model", None).content == "The default."


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", _user(db)).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."""
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=user, 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", user) == "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.view_block(db, "test-model", user) == "They want the short answer."
    # Not the personality, which is a row in the other table.
    assert personas_service.get(db, "test-model", user) is 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.view_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=user, content="I am terse.")
    personas_service.write_impression(
        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):
    user = _user(db)
    row = personas_service.write(db, model_key="test-model", owner=user, content="I am terse.")
    row.enabled = False
    db.commit()
    assert personas_service.block(db, "test-model", user) == ""


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=user, content="I argue back.")
    personas_service.write_impression(
        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_both(client, db, registered):
    """The whole reason writing either is acceptable. Model-written text 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="Blunt with them.")
    personas_service.write_impression(
        db, model_key="test-model", owner=user, content="Wants brevity."
    )

    page = client.get("/settings")
    assert "Who each model is with you" in page.text
    assert "Blunt with them." in page.text
    assert "What models make of you" in page.text
    assert "Wants brevity." in page.text

    persona = personas_service.personas_of(db, user)[0]
    impression = personas_service.impressions_for(db, user)[0]
    client.post(f"/api/library/personalities/{persona.id}/delete", follow_redirects=False)
    client.post(f"/api/library/impressions/{impression.id}/delete", follow_redirects=False)
    db.expire_all()
    assert personas_service.personas_of(db, user) == []
    assert personas_service.impressions_for(db, user) == []


def test_deleting_a_personality_falls_back_to_the_default(client, db):
    """Which is what makes offering the delete reasonable: it is a reset, not the
    loss of the model's character."""
    user = _user(db)
    personas_service.write(db, model_key="test-model", owner=None, content="The default.")
    personas_service.write(db, model_key="test-model", owner=user, content="Mine.")

    row = personas_service.personas_of(db, user)[0]
    client.post(f"/api/library/personalities/{row.id}/delete", follow_redirects=False)

    db.expire_all()
    assert personas_service.block(db, "test-model", user) == "The default."


def test_nobody_can_delete_somebody_elses(client, db):
    second = _second_user(db)
    persona = personas_service.write(
        db, model_key="test-model", owner=second, content="Theirs."
    )
    impression = personas_service.write_impression(
        db, model_key="test-model", owner=second, content="Theirs too."
    )

    assert client.post(
        f"/api/library/personalities/{persona.id}/delete", follow_redirects=False
    ).status_code == 404
    assert client.post(
        f"/api/library/impressions/{impression.id}/delete", follow_redirects=False
    ).status_code == 404

    db.expire_all()
    assert personas_service.get(db, "test-model", second) is not None
    assert personas_service.impression(db, "test-model", second) is not None


def test_the_default_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/personalities/{row.id}/delete", follow_redirects=False
    )

    assert response.status_code == 404
    db.expire_all()
    assert personas_service.get(db, "test-model", None).content == "Instance."


def test_a_personality_cannot_be_deleted_through_the_impression_route(client, db):
    """Two tables, two routes, and an id from one must not resolve in the other --
    `db.get` on the wrong class returns None, which is the answer that matters."""
    user = _user(db)
    row = personas_service.write(db, model_key="test-model", owner=user, content="Mine.")

    response = client.post(
        f"/api/library/impressions/{row.id}/delete", follow_redirects=False
    )

    assert response.status_code == 404
    db.expire_all()
    assert personas_service.block(db, "test-model", user) == "Mine."


def test_an_administrator_can_read_write_and_revert_the_default(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 defaults" 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_default_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() == []


def test_clearing_the_default_leaves_everybodys_own_alone(client, db):
    """They diverged from it; removing the starting point is not removing them."""
    user = _user(db)
    model = db.scalar(select(Model).where(Model.model_id == "test-model"))
    personas_service.write(db, model_key="test-model", owner=None, content="The default.")
    personas_service.write(db, model_key="test-model", owner=user, content="Mine.")

    client.post(f"/admin/models/{model.id}/persona", data={"content": ""}, follow_redirects=False)

    db.expire_all()
    assert personas_service.block(db, "test-model", user) == "Mine."
    assert db.scalars(select(Impression)).all() == []