LLeMbas/tests/test_chat_scope.py

"""What one chat may use, and the rule that it can only ever be less.

The security-shaped test here is `test_a_chat_cannot_widen_what_it_was_not_given`.
The scope is applied inside `resolve_tools` *after* the model's capabilities,
the reader's permissions and the instance configuration, so a crafted POST
turning something on reaches a tool those gates have already removed. Asserting
that against the UI path alone would prove nothing, so it is asserted against a
directly-written column.
"""

from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from lembas.db.models import Chat, Connection, Model, User
from lembas.services import settings_store
from lembas.services import tools as tools_service
from lembas.services.library import skills as skills_service


@pytest.fixture
def chat(db, user_id):
    connection = Connection(name="c", base_url="http://127.0.0.1:1", api_key_encrypted="")
    db.add(connection)
    db.commit()
    db.add(Model(connection_id=connection.id, model_id="m", capabilities_json={"tools": True}))
    db.commit()
    row = Chat(user_id=user_id, model_id="m", connection_id=connection.id)
    db.add(row)
    db.commit()
    return row


def _names(db, chat, user) -> set[str]:
    return set(tools_service.resolve_tools(db, chat, user).by_name)


# --- The route ------------------------------------------------------------------
def test_switching_a_family_off_writes_it_to_the_row(client: TestClient, db, chat, registered):
    response = client.post(
        f"/api/chats/{chat.id}/scope", data={"kind": "family", "name": "web_search"}
    )
    assert response.status_code == 204

    db.expire_all()
    assert db.get(Chat, chat.id).scope_json["families"]["web_search"] is False


def test_switching_it_back_on_removes_the_key(client: TestClient, db, chat, registered):
    """On is stored by *removing* the key, so absent stays the single
    representation of on and the column cannot grow a row per family per chat."""
    client.post(f"/api/chats/{chat.id}/scope", data={"kind": "family", "name": "notes"})
    client.post(
        f"/api/chats/{chat.id}/scope",
        data={"kind": "family", "name": "notes", "on": "true"},
    )

    db.expire_all()
    assert "families" not in db.get(Chat, chat.id).scope_json


def test_the_route_refuses_a_kind_it_does_not_know(client: TestClient, chat, registered):
    response = client.post(
        f"/api/chats/{chat.id}/scope", data={"kind": "everything", "name": "x"}
    )
    assert response.status_code == 400


def test_the_route_refuses_the_wrong_verb(client: TestClient, chat, registered):
    """The half of `tests/test_agent_mode.py`'s lesson that actually caught the
    bug: a control wired to a method a route does not serve fails silently."""
    assert client.get(f"/api/chats/{chat.id}/scope").status_code == 405


def test_somebody_elses_chat_is_not_reachable(client: TestClient, db, chat, registered):
    from lembas.security.passwords import hash_password

    other = User(name="Sam", email="s@shire.test", password_hash=hash_password("secret123"))
    db.add(other)
    db.commit()
    chat.user_id = other.id
    db.commit()

    response = client.post(
        f"/api/chats/{chat.id}/scope", data={"kind": "family", "name": "notes"}
    )
    assert response.status_code == 404


# --- What it does to the offer ------------------------------------------------------
def test_a_family_switched_off_is_not_offered(db, chat, user_id):
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    user = db.get(User, user_id)
    assert "web_search" in _names(db, chat, user)

    chat.scope_json = {"families": {"web_search": False}}
    db.commit()

    assert "web_search" not in _names(db, chat, user)


def test_switching_a_gate_off_takes_every_tool_in_it(db, chat, user_id):
    """A gate is one switch, not five. `notes` covers search, get, create, edit
    and delete -- which is the same reasoning the per-model capability
    checkboxes carry."""
    user = db.get(User, user_id)
    chat.scope_json = {"families": {"notes": False}}
    db.commit()

    offered = _names(db, chat, user)
    assert not [name for name in offered if name.startswith("notes_")]


def test_a_chat_cannot_widen_what_it_was_not_given(db, chat, user_id):
    """The one that matters. Scope is applied AFTER the gates and never instead
    of them, so writing `True` into the column reaches a tool the model's
    capabilities had already removed."""
    user = db.get(User, user_id)
    model = db.scalar(tools_service.select(Model))
    model.capabilities_json = {"tools": True, "tool_notes": False}
    chat.scope_json = {"families": {"notes": True}}
    db.commit()

    assert "notes_search" not in _names(db, chat, user)


def test_an_unknown_family_in_the_column_changes_nothing(db, chat, user_id):
    user = db.get(User, user_id)
    before = _names(db, chat, user)
    chat.scope_json = {"families": {"not-a-family": False}}
    db.commit()

    assert _names(db, chat, user) == before


# --- Skills -------------------------------------------------------------------------
@pytest.fixture
def skill(db, user_id):
    return skills_service.create(
        db,
        owner=db.get(User, user_id),
        name="weekly-report",
        description="When asked for the weekly report.",
        body="Do the thing.",
    )


def test_a_skill_switched_off_leaves_the_index(db, chat, user_id, skill):
    user = db.get(User, user_id)
    assert "weekly-report" in skills_service.index_block(db, user)
    assert "weekly-report" not in skills_service.index_block(
        db, user, exclude=["weekly-report"]
    )


def test_a_skill_switched_off_cannot_be_fetched_anyway(db, chat, user_id, skill):
    """Without this the narrowing is advisory: a model can name a skill it was
    never shown -- from an earlier turn, from a note -- and the runner would
    happily fetch it. Same rule as "what may be run is what was offered"."""
    import asyncio

    user = db.get(User, user_id)
    chat.scope_json = {"skills": {"weekly-report": False}}
    db.commit()

    context = tools_service.context_for(db, user, chat)
    outcome = asyncio.run(
        tools_service.run_tool(context, "skill_get", '{"name": "weekly-report"}')
    )
    assert outcome.event["status"] == "error"


def test_the_last_skill_switched_off_withdraws_skill_get(db, chat, user_id, skill):
    user = db.get(User, user_id)
    assert "skill_get" in _names(db, chat, user)

    chat.scope_json = {"skills": {"weekly-report": False}}
    db.commit()

    offered = _names(db, chat, user)
    assert "skill_get" not in offered
    assert "skill_create" in offered, "writing the first one is still possible"


# --- The zero-skills asymmetry --------------------------------------------------------
def test_with_no_skills_nothing_tells_the_model_to_read_one(db, chat, user_id):
    """The complaint this fixes. `tool.skills` was gated on the family alone, so
    a person with no skills got "read the full instructions with skill_get"
    above a list that was not there -- and got skill_get in the tools array, so
    the model spent a round finding out."""
    from lembas.services import harness

    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "skill_get" not in _names(db, chat, user)
    assert "skill_get" not in text
    assert "Skills available" not in text
    # The half that is most useful with none: you can save the first one.
    assert "save it with skill_create" in text


def test_with_a_skill_the_reading_guidance_comes_back(db, chat, user_id, skill):
    from lembas.services import harness

    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "skill_get" in text
    assert "weekly-report" in text
    assert "save it with skill_create" in text


# --- The tool list --------------------------------------------------------------------
def test_the_model_is_told_what_it_actually_has(db, chat, user_id):
    """`tool_names` was resolved and documented with no fragment reading it. A
    model that has to discover its own list by calling something and being told
    it does not exist spends a round finding out -- and with one round, that is
    the whole reply."""
    from lembas.services import harness

    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "The tools you have on this request are:" in text
    for name in tools_service.resolve_tools(db, chat, user).by_name:
        assert name in text


def test_a_family_switched_off_disappears_from_the_list_too(db, chat, user_id):
    from lembas.services import harness

    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    user = db.get(User, user_id)
    chat.scope_json = {"families": {"web_search": False}}
    db.commit()

    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "web_search" not in text


def test_no_tools_means_no_list(db, chat, user_id):
    from lembas.services import harness

    text = harness.compose(db, db.get(User, user_id), [])
    assert "The tools you have on this request" not in text


# --- The control that writes ------------------------------------------------------------
def test_the_verb_is_on_every_checkbox(client: TestClient, db, chat, registered):
    """The element carrying `name` has to be the element carrying the request.
    Two selects lost an entire release to getting this wrong -- their verb was
    on a form the event never reached, and the tests passed throughout because
    they asserted the markup rather than the property.

    `conftest.control_named` is the helper for this and wants exactly one match;
    there is one checkbox per family here, so the same check is made over all of
    them, which is the stronger claim anyway.
    """
    from html.parser import HTMLParser

    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    html = client.get(f"/chat/{chat.id}").text

    found: list[dict[str, str]] = []

    class Finder(HTMLParser):
        def handle_starttag(self, tag, attrs):
            got = {key: (value or "") for key, value in attrs}
            if got.get("name") == "on":
                found.append(got)

    Finder().feed(html)

    assert found, "the scope menu rendered no switches"
    for box in found:
        assert box.get("hx-post") == f"/api/chats/{chat.id}/scope"
        assert "kind" in box.get("hx-vals", ""), "and says which thing it is"


def test_the_menu_is_called_toggle(client: TestClient, db, chat, registered):
    """It was "What this chat can use", which described the contents rather than
    naming the control. The label is on the button and on the menu, and both are
    read aloud, so both have to say it."""
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    html = client.get(f"/chat/{chat.id}").text

    assert 'aria-label="Toggle"' in html
    assert "What this chat can use" not in html


def test_the_menu_no_longer_offers_to_type_an_at_sign(client: TestClient, db, chat, registered):
    """A menu you open in order to insert one character is a longer way round
    than the character. Typing `@` is untouched and is asserted elsewhere."""
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    html = client.get(f"/chat/{chat.id}").text

    assert "data-mention-open" not in html
    assert "Mention a file or a document" not in html


def test_with_nothing_to_narrow_there_is_no_button_at_all(
    client: TestClient, db, chat, registered
):
    """The guard used to be `has_scope or can upload`, because the mention row
    was something to show when there was no scope. With that gone the same guard
    would open an empty menu, which is worse than no button.

    A model with no `tools` capability is offered nothing, so there is nothing
    to switch off -- the honest way to reach an empty scope.
    """
    model = db.query(Model).filter_by(model_id="m").one()
    model.capabilities_json = {}
    db.commit()

    html = client.get(f"/chat/{chat.id}").text

    assert "picker__menu--scope" not in html