LLeMbas/tests/test_harness.py

"""The operational preamble, and how it sits beside the authored prompt."""

from __future__ import annotations

import pytest

from lembas.db.models import Chat, Connection, Model, User
from lembas.security.passwords import hash_password
from lembas.services import chat as chat_service
from lembas.services import harness, settings_store
from lembas.services import tools as tools_service
from lembas.services.library import memories as memories_service
from lembas.services.library import skills as skills_service


@pytest.fixture
def owner(db):
    user = User(name="Frodo", email="f@shire.test", password_hash=hash_password("x"))
    db.add(user)
    db.commit()
    return user


def _tools(*names):
    return [tools_service.REGISTRY[name].schema for name in names]


# --- Composition -------------------------------------------------------------
def test_no_tools_means_no_harness(db, owner):
    """An empty harness is worse than none: tokens that say only that there is
    nothing to say."""
    assert harness.compose(db, owner, []) == ""
    assert harness.compose(db, owner, None) == ""


def test_only_the_guidance_for_offered_tools_appears(db, owner):
    text = harness.compose(db, owner, _tools("web_search"))
    assert "Look things up" in text
    assert "You keep notes" not in text
    assert "Skills are procedures" not in text


def test_the_memory_block_is_included_when_memory_is_offered(db, owner):
    memories_service.add(db, owner=owner, content="Prefers metric units.")
    text = harness.compose(db, owner, _tools("memory_add"))
    assert "What you know about this person" in text
    assert "Prefers metric units." in text


def test_memories_are_absent_without_the_memory_tool(db, owner):
    """A model not given the memory tool has no business being told them."""
    memories_service.add(db, owner=owner, content="Prefers metric units.")
    text = harness.compose(db, owner, _tools("web_search"))
    assert "Prefers metric units." not in text


def test_the_skill_index_is_names_and_descriptions_only(db, owner):
    skills_service.create(
        db, owner=owner, name="weekly-report", description="When asked.", body="SECRET"
    )
    text = harness.compose(db, owner, _tools("skill_get"))
    assert "weekly-report: When asked." in text
    assert "SECRET" not in text


def test_an_empty_store_contributes_no_heading(db, owner):
    """And the guidance must not point at a heading that is not there: telling a
    model to consult an absent section is a good way to make it invent one."""
    text = harness.compose(db, owner, _tools("memory_add", "skill_get"))
    assert "### What you know about this person" not in text
    assert "### Skills available" not in text
    assert "was remembered earlier" not in text
    assert "You can remember durable facts" in text


def test_the_harness_is_capped(db, owner, monkeypatch):
    monkeypatch.setattr(harness, "MAX_HARNESS_CHARS", 200)
    for index in range(50):
        skills_service.create(
            db, owner=owner, name=f"skill-{index}", description="x" * 200, body="y"
        )
    assert len(harness.compose(db, owner, _tools("skill_get"))) <= 202


# --- Joining -----------------------------------------------------------------
def test_the_authored_prompt_comes_last():
    """It is closest to the conversation, and it is what the user actually
    wrote."""
    joined = harness.join("HARNESS", "AUTHORED")
    assert joined.index("HARNESS") < joined.index("AUTHORED")


def test_either_half_alone_is_returned_unchanged():
    assert harness.join("", "AUTHORED") == "AUTHORED"
    assert harness.join("HARNESS", "") == "HARNESS"
    assert harness.join("", "") == ""


# --- Through build_request ---------------------------------------------------
def _chat(db, owner, *, capabilities, model_prompt="", chat_prompt=""):
    connection = Connection(name="c", base_url="http://h", api_key_encrypted="")
    db.add(connection)
    db.commit()
    db.add(
        Model(
            connection_id=connection.id,
            model_id="m",
            capabilities_json=capabilities,
            system_prompt=model_prompt,
        )
    )
    db.commit()
    chat = Chat(
        user_id=owner.id, model_id="m", connection_id=connection.id, system_prompt=chat_prompt
    )
    db.add(chat)
    db.commit()
    return chat


def _system(body):
    first = body["messages"][0] if body["messages"] else {}
    return first.get("content", "") if first.get("role") == "system" else ""


def test_a_request_without_tools_has_no_harness_and_no_tools_key(db, owner):
    chat = _chat(db, owner, capabilities={})
    body = chat_service.build_request(db, chat, tools=[], user=owner)
    assert "tools" not in body
    assert "How to work" not in _system(body)


def test_the_harness_precedes_the_authored_prompt(db, owner):
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    chat = _chat(db, owner, capabilities={"tools": True}, chat_prompt="Speak as Gandalf.")
    offered = tools_service.enabled_tools(db, chat, owner)

    system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))
    assert system.index("How to work") < system.index("Speak as Gandalf.")


def test_precedence_between_the_authored_layers_is_untouched(db, owner):
    """The harness is a different axis. Exactly one authored layer still wins,
    and it is still the most specific one."""
    settings_store.update(db, {"system_prompt": "Instance."})
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)

    chat = _chat(
        db, owner, capabilities={"tools": True}, model_prompt="Model.", chat_prompt="Chat."
    )
    offered = tools_service.enabled_tools(db, chat, owner)
    system = _system(chat_service.build_request(db, chat, tools=offered, user=owner))

    assert "Chat." in system
    assert "Model." not in system
    assert "Instance." not in system
    # And the resolver on its own is unchanged.
    assert chat_service.effective_system_prompt(db, chat) == "Chat."


def test_a_model_prompt_wins_when_the_chat_has_none(db, owner):
    settings_store.update(db, {"system_prompt": "Instance."})
    chat = _chat(db, owner, capabilities={}, model_prompt="Model.")
    system = _system(chat_service.build_request(db, chat, user=owner))
    assert system == "Model."


def test_the_tools_array_rides_along(db, owner):
    settings_store.update(db, {"enabled": True}, key=settings_store.SEARCH)
    chat = _chat(db, owner, capabilities={"tools": True})
    offered = tools_service.enabled_tools(db, chat, owner)
    body = chat_service.build_request(db, chat, tools=offered, user=owner)
    assert body["tools"] == offered