LLeMbas/tests/test_agent_mode.py

"""Choosing the approval mode: after a chat exists, and before one does.

The regression test at the top is the one that was missing. The mode select in
the topbar posted with `hx-post` against a route that only answers `PATCH`, so
every change returned 405 and the mode never moved -- and htmx surfaces nothing
on a failed request, so the control looked like it had worked. A control wired
to a method the route does not serve fails exactly this quietly, which is why
the assertion below reads the row rather than the response.
"""

from __future__ import annotations

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from lembas.db.models import KIND_AGENT, Chat, Connection, Model, SshProfile, User
from lembas.services import settings_store
from lembas.services.agent import policy

from .conftest import control_named


def _agent_chat(db, *, mode: str = policy.MODE_MANUAL) -> Chat:
    """An agent chat pointed at a profile that is never actually connected to.

    Nothing here opens a connection: changing the mode is a database write, and
    a real sshd would only make the test slower and flakier.

    A model is needed even though nothing generates: with none, `index.html`
    renders the "no models available" screen *instead of* the thread and the
    composer, so a test asserting on composer markup would pass against a page
    that does not contain a composer at all. That is not hypothetical -- it is
    how the first version of the assertion below came to be vacuous.
    """
    settings_store.update(db, {"enabled": True}, key=settings_store.AGENTS)
    user = db.scalars(select(User)).first()
    assert user is not None

    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"))
    db.commit()

    profile = SshProfile(
        owner_id=user.id,
        name="Test box",
        host="127.0.0.1",
        port=22,
        username="tester",
        host_key="host key",
        host_fingerprint="SHA256:x",
        default_dir="/project",
    )
    db.add(profile)
    db.commit()

    chat = Chat(
        user_id=user.id,
        model_id="m",
        connection_id=connection.id,
        kind=KIND_AGENT,
        ssh_profile_id=profile.id,
        project_dir="/project",
        agent_mode=mode,
    )
    db.add(chat)
    db.commit()
    return chat


def test_patching_the_mode_changes_it(client: TestClient, db, registered):
    chat = _agent_chat(db)

    response = client.patch(f"/api/chats/{chat.id}", data={"agent_mode": policy.MODE_PLAN})
    assert response.status_code == 204, response.text

    db.refresh(chat)
    assert chat.agent_mode == policy.MODE_PLAN


def test_the_mode_form_uses_a_method_the_route_serves(client: TestClient, db, registered):
    """The bug itself, stated as a test rather than as a comment.

    POST is not merely unhandled here, it is *silently* unhandled: htmx swallows
    the 405 and the select keeps showing whatever was clicked. Asserting the
    method is refused is what stops somebody reintroducing `hx-post` and finding
    the mode unchangeable again with nothing in the logs.
    """
    chat = _agent_chat(db)
    refused = client.post(f"/api/chats/{chat.id}", data={"agent_mode": policy.MODE_AUTO})
    assert refused.status_code == 405
    db.refresh(chat)
    assert chat.agent_mode == policy.MODE_MANUAL


def test_the_mode_select_carries_its_own_verb(client: TestClient, db, registered):
    """The request must hang off the element the event fires on.

    This is the invariant, and the previous version of this test did not check
    it. `hx-patch` lived on an empty sibling `<form>` that the select pointed at
    with `form="…"`, which was enough to make the markup look right and enough
    to make every assertion here pass -- while htmx bound the `change` listener
    to the form, and `change` fires on the select and bubbles to its *ancestors*
    only. The mode never once reached the database.

    So: assert on the control, by name, whichever element that turns out to be.
    """
    chat = _agent_chat(db)
    body = client.get(f"/chat/{chat.id}").text

    select = control_named(body, "agent_mode")
    assert select["hx-patch"] == f"/api/chats/{chat.id}"
    assert "hx-post" not in select
    # And the empty form still scopes the values, so the PATCH carries this
    # field alone rather than the whole composer -- `project_dir` in a PATCH is
    # a 409.
    assert select["form"] == "agent-mode-form"


def test_the_mode_is_offered_beside_the_composer_not_in_the_topbar(
    client: TestClient, db, registered
):
    """Where it is matters: it belongs where the message is written.

    Asserted on order rather than on a class name so it survives a restyle --
    what is being pinned is that the control comes after the thread, not what
    it looks like.
    """
    chat = _agent_chat(db)
    body = client.get(f"/chat/{chat.id}").text

    assert body.index('id="thread"') < body.index('id="agent-mode-form"')


# --- Chosen before the first word --------------------------------------------
def _profile_for(db) -> SshProfile:
    """A usable connection, and the feature switched on, with no chat yet."""
    settings_store.update(db, {"enabled": True}, key=settings_store.AGENTS)
    user = db.scalars(select(User)).first()
    assert user is not None
    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"))
    profile = SshProfile(
        owner_id=user.id,
        name="Test box",
        host="127.0.0.1",
        port=22,
        username="tester",
        host_key="host key",
        host_fingerprint="SHA256:x",
        default_dir="/project",
    )
    db.add(profile)
    db.commit()
    return profile


def test_a_chat_can_start_in_plan_mode(client: TestClient, db, registered):
    """The whole point: reaching Plan without first sending something in Manual."""
    profile = _profile_for(db)

    client.post(
        "/api/chats/start",
        data={
            "content": "have a look around",
            "kind": "agent",
            "ssh_profile_id": profile.id,
            "project_dir": "/project",
            "agent_mode": policy.MODE_PLAN,
        },
    )

    chat = db.scalars(select(Chat)).one()
    assert chat.kind == KIND_AGENT
    assert chat.agent_mode == policy.MODE_PLAN


def test_a_chat_started_with_no_mode_is_manual(client: TestClient, db, registered):
    """The column default still governs, so nobody's habits change."""
    profile = _profile_for(db)

    client.post(
        "/api/chats/start",
        data={"content": "hello", "kind": "agent", "ssh_profile_id": profile.id},
    )

    assert db.scalars(select(Chat)).one().agent_mode == policy.MODE_MANUAL


def test_an_unrecognised_mode_at_creation_is_ignored(client: TestClient, db, registered):
    profile = _profile_for(db)

    client.post(
        "/api/chats/start",
        data={
            "content": "hello",
            "kind": "agent",
            "ssh_profile_id": profile.id,
            "agent_mode": "root",
        },
    )

    assert db.scalars(select(Chat)).one().agent_mode == policy.MODE_MANUAL


def test_a_mode_on_a_plain_chat_is_ignored(client: TestClient, db, registered):
    """A plain chat has no approval loop, so a mode on one means nothing.

    It must not silently become an agent chat either: the connection is what
    decides that, and there is none here.
    """
    _profile_for(db)

    client.post("/api/chats/start", data={"content": "hello", "agent_mode": policy.MODE_AUTO})

    chat = db.scalars(select(Chat)).one()
    assert chat.kind != KIND_AGENT
    assert chat.agent_mode == policy.MODE_MANUAL


@pytest.mark.parametrize("wanted", ["", "sudo", "PLAN", "edit;auto"])
def test_an_unrecognised_mode_is_ignored(client: TestClient, db, registered, wanted):
    """Ignored rather than refused: an unknown mode is a bug in the sender, and
    failing the whole request would leave the reader with a chat they cannot
    change. The safe outcome is that nothing moves."""
    chat = _agent_chat(db, mode=policy.MODE_EDIT)

    client.patch(f"/api/chats/{chat.id}", data={"agent_mode": wanted})

    db.refresh(chat)
    assert chat.agent_mode == policy.MODE_EDIT


# --- Changing it while a reply is running ---------------------------------------
def test_the_mode_is_re_read_between_rounds(client: TestClient, db, registered):
    """The reported bug. The mode was snapshotted for the whole reply, so
    switching to Auto during a long agent reply went on asking about every
    call -- which looks exactly like a control that does not work, because for
    that reply it was one.

    Between rounds and not within one: a round's calls are authorised together,
    and switching must not retroactively approve what is already queued.
    """
    from lembas.db.models import User
    from lembas.services.agent import session as agent_session

    chat = _agent_chat(db, mode=policy.MODE_MANUAL)
    user = db.scalars(select(User)).first()
    agent = agent_session.resolve(db, chat, user)
    assert agent.mode == policy.MODE_MANUAL

    client.patch(f"/api/chats/{chat.id}", data={"agent_mode": policy.MODE_AUTO})
    # The route wrote through its own session; this one still holds the row it
    # loaded. `refresh` opens a fresh session in the real path, so this is the
    # test catching up rather than the behaviour under test.
    db.expire_all()

    agent_session.refresh(db, agent)
    assert agent.mode == policy.MODE_AUTO


def test_always_allow_reaches_the_reply_that_asked(client: TestClient, db, registered):
    """The same bug, in the place nobody reported because it is quieter: the
    verdict was accepted, written to the row, and then ignored for the rest of
    the reply that had just asked about it."""
    from lembas.db.models import User
    from lembas.services.agent import session as agent_session

    chat = _agent_chat(db, mode=policy.MODE_MANUAL)
    user = db.scalars(select(User)).first()
    agent = agent_session.resolve(db, chat, user)
    assert "pytest" not in agent.allow

    chat.scope_json = {**(chat.scope_json or {}), "allow": ["pytest"]}
    db.commit()

    agent_session.refresh(db, agent)
    assert "pytest" in agent.allow


def test_refreshing_keeps_what_this_reply_has_read(client: TestClient, db, registered):
    """`read_paths` is what `file_edit` checks before applying a patch, and it
    is a fact about this reply rather than about the row. Mutating in place is
    what keeps it -- and keeps the approved copy of a round, which holds field
    references rather than a copy."""
    from lembas.db.models import User
    from lembas.services.agent import session as agent_session

    chat = _agent_chat(db, mode=policy.MODE_MANUAL)
    user = db.scalars(select(User)).first()
    agent = agent_session.resolve(db, chat, user)
    agent.read_paths.add("/project/main.py")
    approved = agent.as_approved()

    agent_session.refresh(db, agent)

    assert "/project/main.py" in agent.read_paths
    assert "/project/main.py" in approved.read_paths


def test_an_unknown_mode_on_the_row_refreshes_to_manual(client: TestClient, db, registered):
    """A row that predates a rename has to fail towards asking, here as much as
    in `resolve`."""
    from lembas.db.models import User
    from lembas.services.agent import session as agent_session

    chat = _agent_chat(db, mode=policy.MODE_AUTO)
    user = db.scalars(select(User)).first()
    agent = agent_session.resolve(db, chat, user)

    chat.agent_mode = "reckless"
    db.commit()
    agent_session.refresh(db, agent)
    assert agent.mode == policy.MODE_MANUAL