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, prompts, 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_tool_guidance(db, owner):
    """The core fragments still go out -- a model with no tools has no clock
    either, and telling it the date is not "tokens that say nothing". What it
    must not get is instructions about tools it was never offered."""
    text = harness.compose(db, owner, [])
    assert "Today is" in text
    assert "You have tools" not in text
    assert "Look things up" not in text
    assert harness.compose(db, owner, None) == text


def test_clearing_the_core_fragments_restores_an_empty_harness(db, owner):
    """The behaviour change is a default, not a rule: an administrator who wants
    nothing sent to a tool-less model can still have exactly that."""
    prompts.save(
        db,
        {f.key: "" for f in prompts.BUILTIN if f.group == prompts.GROUP_CORE},
    )
    assert harness.compose(db, owner, []) == ""


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_carries_no_tool_guidance_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 "You have tools" 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.endswith("Model.")
    assert "Instance." not in system
    assert chat_service.effective_system_prompt(db, chat) == "Model."


def test_the_seam_line_only_appears_when_there_is_something_to_hand_over_to(db, owner):
    """It introduces the authored prompt. With no authored prompt it would be
    pointing at nothing, which is the failure the whole `requires` idea exists
    to avoid."""
    bare = _chat(db, owner, capabilities={})
    assert "was written by whoever set up" not in _system(
        chat_service.build_request(db, bare, user=owner)
    )

    authored = _chat(db, owner, capabilities={}, chat_prompt="Speak as Gandalf.")
    assert "was written by whoever set up" in _system(
        chat_service.build_request(db, authored, user=owner)
    )


def test_an_administrators_wording_replaces_the_default(db, owner):
    prompts.save(db, {"core.today": "The date is {{today}}, more or less."})
    text = harness.compose(db, owner, [])
    assert "more or less." in text
    assert "Your training data stops well before this" not in text


def test_the_attached_files_are_named_and_explained(db, owner):
    from lembas.db.models import Attachment

    chat = _chat(db, owner, capabilities={})
    db.add(
        Attachment(
            user_id=owner.id,
            chat_id=chat.id,
            filename="report.txt",
            stored_name="x.txt",
            media_type="text/plain",
            size_bytes=10,
            kind="text",
        )
    )
    db.commit()

    text = harness.compose(db, owner, [], chat)
    assert "report.txt" in text
    assert "<document name=" in text


def test_a_chat_with_no_attachments_says_nothing_about_documents(db, owner):
    chat = _chat(db, owner, capabilities={})
    assert "<document name=" not in harness.compose(db, owner, [], chat)


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