"""The agent tools: when they are offered, and what stops them running.

Everything here goes through a real SSH server on 127.0.0.1, so the gate and the
approval loop are exercised against something that genuinely executes rather
than a stub that always agrees.
"""

from __future__ import annotations

import asyncio
import json as _json
import time

import pytest

from lembas.db.models import (
    KIND_AGENT,
    ROLE_ASSISTANT,
    Chat,
    Connection,
    Message,
    Model,
    SshProfile,
    User,
)
from lembas.services import generation as generation_service
from lembas.services import interaction, settings_store
from lembas.services import tools as tools_service
from lembas.services.agent import policy, session
from lembas.services.agent import ssh as ssh_service

asyncssh = pytest.importorskip("asyncssh")


# --- A machine to act on --------------------------------------------------------
class _Server(asyncssh.SSHServer):
    def begin_auth(self, username: str) -> bool:
        return False


async def _handler(process):
    command = process.command or ""
    process.stdout.write(f"ran: {command}\n")
    process.exit(0)


@pytest.fixture
async def machine(tmp_path):
    """A real sshd, and a profile pointing at it with its key already pinned."""
    project = tmp_path / "project"
    project.mkdir()

    server = await asyncssh.create_server(
        _Server,
        "127.0.0.1",
        0,
        server_host_keys=[asyncssh.generate_private_key("ssh-ed25519")],
        process_factory=_handler,
        sftp_factory=True,
    )
    port = next(iter(server.sockets)).getsockname()[1]
    line, fingerprint = await ssh_service.capture_host_key("127.0.0.1", port)
    try:
        yield {"port": port, "host_key": line, "fingerprint": fingerprint, "dir": str(project)}
    finally:
        server.close()
        await server.wait_closed()


def _setup(db, user_id, machine, *, mode=policy.MODE_MANUAL, enabled=True, kind=KIND_AGENT):
    """An agent chat pointed at the machine, with the feature switched on."""
    settings_store.update(db, {"enabled": enabled}, key=settings_store.AGENTS)

    profile = SshProfile(
        owner_id=user_id,
        name="Test box",
        host="127.0.0.1",
        port=machine["port"],
        username="tester",
        host_key=machine["host_key"],
        host_fingerprint=machine["fingerprint"],
        default_dir=machine["dir"],
    )
    db.add(profile)
    db.commit()

    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()

    chat = Chat(
        user_id=user_id,
        model_id="m",
        connection_id=connection.id,
        kind=kind,
        ssh_profile_id=profile.id,
        project_dir=machine["dir"],
        agent_mode=mode,
    )
    db.add(chat)
    db.commit()
    return chat, profile


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


# --- The gate --------------------------------------------------------------------
async def test_the_agent_tools_are_offered_to_an_agent_chat(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine)
    names = _offered(db, chat, db.get(User, user_id))

    assert {"shell_run", "file_read", "file_write", "file_list"} <= names


async def test_an_ordinary_chat_gets_none_of_them(db, user_id, machine):
    """A plain conversation can never shell out, whatever it is asked."""
    chat, _profile = _setup(db, user_id, machine, kind="chat")
    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))


async def test_nothing_is_offered_while_the_feature_is_off(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine, enabled=False)
    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))


async def test_nothing_is_offered_without_the_permission(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine)
    user = db.get(User, user_id)
    user.role = "user"  # administrators pass everything
    settings_store.update(db, {"default_permissions": {"tools.agent": False}})
    db.commit()

    assert "shell_run" not in _offered(db, chat, user)


async def test_nothing_is_offered_without_the_model_capability(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine)
    model = db.scalar(tools_service.select(Model))
    model.capabilities_json = {"tools": True, "tool_agent": False}
    db.commit()

    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))


async def test_a_disabled_connection_takes_the_tools_away(db, user_id, machine):
    """Offering a tool that fails on its first call is worse than not offering
    it, so every "no" collapses to an empty list."""
    chat, profile = _setup(db, user_id, machine)
    profile.enabled = False
    db.commit()

    assert "shell_run" not in _offered(db, chat, db.get(User, user_id))


async def test_somebody_elses_connection_is_not_reachable(db, user_id, machine):
    from lembas.security.passwords import hash_password

    chat, profile = _setup(db, user_id, machine)
    intruder = User(name="Sam", email="s@shire.test", password_hash=hash_password("x"))
    intruder.role = "user"
    db.add(intruder)
    db.commit()

    assert session.resolve(db, chat, intruder) is None


# --- Running something -----------------------------------------------------------
async def test_a_command_runs_on_the_machine(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    outcome = await tools_service.run_tool(
        context, "shell_run", '{"command": "echo hello"}'
    )
    assert outcome.event["status"] == "ok"
    assert "echo hello" in outcome.content


async def test_files_are_written_and_read_back(db, user_id, machine, tmp_path):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    await tools_service.run_tool(
        context, "file_write", '{"path": "note.txt", "content": "a mallorn tree"}'
    )
    assert (tmp_path / "project" / "note.txt").read_text() == "a mallorn tree"

    read = await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
    assert "mallorn" in read.content

    listed = await tools_service.run_tool(context, "file_list", "{}")
    assert "note.txt" in listed.content


# --- Changing part of a file -------------------------------------------------------
def _context(db, user_id, machine, **kwargs):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO, **kwargs)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    return tools_service.context_for(db, user, chat, tools=resolved)


async def test_editing_a_file_that_was_not_read_is_refused(db, user_id, machine, tmp_path):
    """Both halves matter. The wording is what the model acts on; that nothing
    was written is the actual guarantee."""
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)

    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps({"path": "note.txt", "patch": "@@ -1,2 +1,2 @@\n alpha\n-beta\n+BETA\n"}),
    )

    assert outcome.content.startswith("Read the file first!")
    assert outcome.event["status"] == "error"
    assert (tmp_path / "project" / "note.txt").read_text() == "alpha\nbeta\n"


async def test_reading_then_editing_writes_the_new_text(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\ngamma\n")
    context = _context(db, user_id, machine)

    await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps(
            {"path": "note.txt", "patch": "@@ -1,3 +1,3 @@\n alpha\n-beta\n+BETA\n gamma\n"}
        ),
    )

    assert outcome.event["status"] == "ok"
    assert (tmp_path / "project" / "note.txt").read_text() == "alpha\nBETA\ngamma\n"


async def test_a_relative_and_an_absolute_path_are_the_same_file(db, user_id, machine, tmp_path):
    """`./note.txt` read and `note.txt` edited has to count as having read it,
    or the check refuses the very thing it was meant to permit."""
    target = tmp_path / "project" / "note.txt"
    target.write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)

    await tools_service.run_tool(context, "file_read", '{"path": "./note.txt"}')
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps({"path": str(target), "patch": "@@ -1,2 +1,2 @@\n alpha\n-beta\n+BETA\n"}),
    )

    assert outcome.event["status"] == "ok", outcome.content


async def test_a_failed_hunk_names_it_and_writes_nothing(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)

    await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps({"path": "note.txt", "patch": "@@ -1,2 +1,2 @@\n nope\n-wrong\n+x\n"}),
    )

    assert outcome.event["status"] == "error"
    assert "Hunk 1" in outcome.content
    assert (tmp_path / "project" / "note.txt").read_text() == "alpha\nbeta\n"


async def test_a_write_counts_as_having_read_it(db, user_id, machine, tmp_path):
    """`_run_write` reads the old content for its diff anyway, so write-then-edit
    works in one reply without a second round trip."""
    context = _context(db, user_id, machine)

    await tools_service.run_tool(
        context, "file_write", '{"path": "new.txt", "content": "one\\ntwo\\n"}'
    )
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps({"path": "new.txt", "patch": "@@ -1,2 +1,2 @@\n one\n-two\n+TWO\n"}),
    )

    assert outcome.event["status"] == "ok", outcome.content
    assert (tmp_path / "project" / "new.txt").read_text() == "one\nTWO\n"


async def test_the_read_set_survives_an_approval(db, user_id, machine, tmp_path):
    """`as_approved` is `dataclasses.replace`, which copies field *references*,
    so the set is shared with the per-call copy a runner actually gets. That is
    wanted, and it is not obvious enough to leave unpinned."""
    from dataclasses import replace

    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)
    approved = replace(context, agent=context.agent.as_approved())

    await tools_service.run_tool(approved, "file_read", '{"path": "note.txt"}')

    assert context.agent.read_paths, "the read done under approval is not visible"


async def test_an_edit_that_changes_nothing_says_so(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)

    await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps({"path": "note.txt", "patch": "@@ -1,2 +1,2 @@\n alpha\n-beta\n+beta\n"}),
    )

    assert outcome.event["status"] == "ok"
    assert "changes nothing" in outcome.content


# --- The diff on the event ------------------------------------------------------------
async def test_an_edit_carries_a_diff(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\ngamma\n")
    context = _context(db, user_id, machine)

    await tools_service.run_tool(context, "file_read", '{"path": "note.txt"}')
    outcome = await tools_service.run_tool(
        context,
        "file_edit",
        _json.dumps(
            {"path": "note.txt", "patch": "@@ -1,3 +1,3 @@\n alpha\n-beta\n+BETA\n gamma\n"}
        ),
    )

    diff = outcome.event["diff"]
    assert "-beta" in diff
    assert "+BETA" in diff


async def test_writing_a_new_file_shows_it_as_all_additions(db, user_id, machine):
    """Which is what git does, and the right display."""
    context = _context(db, user_id, machine)

    outcome = await tools_service.run_tool(
        context, "file_write", '{"path": "fresh.txt", "content": "one\\ntwo\\n"}'
    )

    body = [
        line
        for line in outcome.event["diff"].split("\n")
        if line and not line.startswith(("@@", "+++", "---"))
    ]
    assert body and all(line.startswith("+") for line in body), body


async def test_overwriting_a_file_shows_what_changed(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "note.txt").write_text("alpha\nbeta\n")
    context = _context(db, user_id, machine)

    outcome = await tools_service.run_tool(
        context, "file_write", '{"path": "note.txt", "content": "alpha\\nBETA\\n"}'
    )

    assert "-beta" in outcome.event["diff"]
    assert "+BETA" in outcome.event["diff"]


async def test_a_file_too_big_to_read_is_written_without_a_diff(db, user_id, machine, tmp_path):
    """A truncated original would invent deletions of the tail, which is worse
    than showing no diff at all."""
    from lembas.services import settings_store as store

    store.update(db, {"max_output_bytes": 1024}, key=store.AGENTS)
    (tmp_path / "project" / "big.txt").write_text("x" * 4000)
    context = _context(db, user_id, machine)

    outcome = await tools_service.run_tool(
        context, "file_write", '{"path": "big.txt", "content": "small"}'
    )

    assert outcome.event["status"] == "ok"
    assert "diff" not in outcome.event


async def test_writing_a_file_drops_the_project_listing(db, user_id, machine):
    """Otherwise the model is shown a five-minute-old tree that it knows is
    wrong, and concludes the file it has just created does not exist.

    The TTL is for drift nobody can see coming. This is not that: it is this
    process changing the tree it has just described.
    """
    from lembas.services.agent import index as index_service

    chat, profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    index_service._CACHE[(profile.id, machine["dir"])] = index_service.ProjectIndex(
        paths=("stale.txt",), total=1, source="git", built_at=time.monotonic()
    )

    await tools_service.run_tool(
        context, "file_write", '{"path": "fresh.txt", "content": "hi"}'
    )

    assert index_service.cached(profile.id, machine["dir"]) is None


# --- The runner backstop ----------------------------------------------------------
async def test_a_runner_refuses_what_the_mode_forbids(db, user_id, machine):
    """`_authorise` is the real gate and runs first. This is the belt to that
    brace: a call arriving by some other path -- a retry, a re-run button --
    must not walk past it."""
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    outcome = await tools_service.run_tool(context, "shell_run", '{"command": "rm -rf /"}')
    assert outcome.event["status"] == "error"
    assert "not allowed" in outcome.content.lower()


async def test_plan_mode_still_reads(db, user_id, machine, tmp_path):
    (tmp_path / "project" / "readme.txt").write_text("contents")
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    outcome = await tools_service.run_tool(context, "file_read", '{"path": "readme.txt"}')
    assert outcome.event["status"] == "ok"
    assert "contents" in outcome.content


async def test_edit_mode_writes_but_will_not_run(db, user_id, machine):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_EDIT)
    user = db.get(User, user_id)
    resolved = tools_service.resolve_tools(db, chat, user)
    context = tools_service.context_for(db, user, chat, tools=resolved)

    wrote = await tools_service.run_tool(
        context, "file_write", '{"path": "x.txt", "content": "hi"}'
    )
    assert wrote.event["status"] == "ok"

    ran = await tools_service.run_tool(context, "shell_run", '{"command": "ls"}')
    assert ran.event["status"] == "error"


# --- The approval card in the loop -------------------------------------------------
def _chunk(name: str, arguments: str, *, call_id="c1", index=0):
    return {
        "choices": [
            {
                "delta": {
                    "tool_calls": [
                        {
                            "index": index,
                            "id": call_id,
                            "function": {"name": name, "arguments": arguments},
                        }
                    ]
                }
            }
        ]
    }


def _text(text: str) -> dict:
    return {"choices": [{"delta": {"content": text}}]}


def _stub_stream(rounds, seen):
    async def stream_chat(_endpoint, payload):
        seen.append(payload)
        for chunk in rounds[min(len(seen) - 1, len(rounds) - 1)]:
            yield chunk

    return stream_chat


async def _until_paused(generation, *, timeout: float = 3.0):
    deadline = asyncio.get_running_loop().time() + timeout
    while asyncio.get_running_loop().time() < deadline:
        if generation.pending is not None:
            return generation.pending
        await asyncio.sleep(0.01)
    raise AssertionError("the reply never paused for approval")


def _pending_reply(db, chat):
    db.add(Message(chat_id=chat.id, role="user", content="do it", complete=True))
    db.commit()
    assistant = Message(chat_id=chat.id, role=ROLE_ASSISTANT, content="", complete=False)
    db.add(assistant)
    db.commit()
    return assistant.id


async def test_a_command_waits_for_approval_and_the_card_shows_it(
    db, user_id, machine, monkeypatch
):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
    message_id = _pending_reply(db, chat)
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [[_chunk("shell_run", '{"command": "rm -rf /tmp/x"}')], [_text("Done.")]], []
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    task = asyncio.create_task(generation_service._run(generation))
    pending = await _until_paused(generation)

    assert pending.kind == interaction.KIND_APPROVAL
    item = pending.items[0]
    # The exact command, verbatim. A card that paraphrased it would be
    # approving something other than what runs.
    assert item.detail == "rm -rf /tmp/x"
    assert "Test box" in item.title
    assert "Manual" in item.reason

    pending.resolve(interaction.DENY)
    await task


async def test_denying_reaches_the_model_as_words_and_runs_nothing(
    db, user_id, machine, monkeypatch, tmp_path
):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
    message_id = _pending_reply(db, chat)
    payloads: list[dict] = []
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [
                [_chunk("file_write", '{"path": "never.txt", "content": "nope"}')],
                [_text("Understood.")],
            ],
            payloads,
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    task = asyncio.create_task(generation_service._run(generation))
    pending = await _until_paused(generation)
    pending.resolve(interaction.DENY)
    await task

    turns = [m for m in payloads[1]["messages"] if m.get("role") == "tool"]
    assert "declined" in turns[0]["content"].lower()
    # And it says not to look for another way round, because a model reads
    # "not allowed" as "not allowed like that" otherwise.
    assert "another way" in turns[0]["content"]
    assert not (tmp_path / "project" / "never.txt").exists()


async def test_allowing_runs_it(db, user_id, machine, monkeypatch, tmp_path):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
    message_id = _pending_reply(db, chat)
    payloads: list[dict] = []
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [
                [_chunk("file_write", '{"path": "yes.txt", "content": "written"}')],
                [_text("Done.")],
            ],
            payloads,
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    task = asyncio.create_task(generation_service._run(generation))
    pending = await _until_paused(generation)
    pending.resolve(interaction.ALLOW)
    await task

    assert (tmp_path / "project" / "yes.txt").read_text() == "written"


async def test_auto_mode_never_pauses(db, user_id, machine, monkeypatch, tmp_path):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [
                [_chunk("file_write", '{"path": "auto.txt", "content": "no asking"}')],
                [_text("Done.")],
            ],
            [],
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await asyncio.wait_for(generation_service._run(generation), timeout=10)

    assert generation.pending is None
    assert (tmp_path / "project" / "auto.txt").read_text() == "no asking"


async def test_the_credential_is_cleared_when_the_reply_ends(db, user_id, machine, monkeypatch):
    """A finished Generation lingers five minutes so late followers get the
    final frames. A private key should not linger with it."""
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)

    captured = {}
    original = tools_service.context_for

    def capture(*args, **kwargs):
        context = original(*args, **kwargs)
        captured["context"] = context
        return context

    monkeypatch.setattr(tools_service, "context_for", capture)
    monkeypatch.setattr(
        generation_service, "stream_chat", _stub_stream([[_text("Nothing to do.")]], [])
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    assert captured["context"].agent is not None
    assert captured["context"].agent.spec == {}, "the decrypted credential is dropped"


# --- The harness has to be able to name the machine ---------------------------
def test_the_registry_maps_the_agent_tools_to_their_family(db):
    """`harness._families` maps an offered tool *name* back to a family to
    decide which fragments apply, and it has no chat to resolve against. Without
    the agent tools listed here, `shell_run` resolves to no family and an agent
    chat is told nothing about the machine it is working on -- the same omission
    that cost custom tools their guidance once already."""
    book = tools_service.registry(db)
    for name in ("shell_run", "file_read", "file_write", "file_list"):
        assert name in book, name
        assert book[name].family == "agent"


async def test_the_harness_says_where_and_under_what_rules(db, user_id, machine):
    from lembas.services import harness

    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_MANUAL)
    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "Test box" in text, "which machine"
    assert machine["dir"] in text, "which directory"
    assert "Manual" in text, "what the mode permits"
    # The single most likely cause of "the agent seems stupid": `cd build`
    # followed by `make` fails silently otherwise.
    assert "fresh shell" in text


async def test_an_ordinary_chat_is_told_none_of_it(db, user_id, machine):
    from lembas.services import harness

    chat, _profile = _setup(db, user_id, machine, kind="chat")
    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas
    text = harness.compose(db, user, offered, chat)

    assert "Test box" not in text
    assert "fresh shell" not in text


async def test_an_agent_chat_is_told_its_real_round_budget(db, user_id, machine):
    """MAX_ROUNDS is three. An agent chat gets forty, and telling it three would
    be a false fact about its own budget on every turn."""
    from lembas.services import harness

    settings_store.update(db, {"max_steps": 25}, key=settings_store.AGENTS)
    chat, _profile = _setup(db, user_id, machine)
    user = db.get(User, user_id)
    values = harness.context_variables(
        db, user, tools_service.resolve_tools(db, chat, user).schemas, chat
    )
    assert values["max_rounds"] == "25"


async def test_an_agent_chat_gets_the_rounds_it_was_promised(db, user_id, machine, monkeypatch):
    """And is then allowed to use them, which is the half that was missing.

    The budget sizes the loop and names itself in the out-of-rounds message, and
    the harness above tells the model the same number. But the comparison that
    ends the loop read the global `MAX_ROUNDS` of three. So an agent chat
    allowed forty rounds stopped after three and reported that it had taken
    forty: two wrong answers to "why did it stop", with no way to tell them
    apart from the outside.
    """
    settings_store.update(db, {"max_steps": 5}, key=settings_store.AGENTS)
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    assistant = Message(chat_id=chat.id, role=ROLE_ASSISTANT, content="", complete=False)
    db.add(assistant)
    db.commit()

    payloads: list[dict] = []

    async def stream_chat(_endpoint, payload):
        payloads.append(payload)
        yield {
            "choices": [
                {"delta": {"tool_calls": [{"index": 0, "id": "c1", "function": {
                    "name": "file_list", "arguments": '{"path": "."}'}}]}}
            ]
        }

    monkeypatch.setattr(generation_service, "stream_chat", stream_chat)

    async def _no_title(*_args, **_kwargs):
        return ""

    monkeypatch.setattr("lembas.services.chat.generate_title", _no_title)

    generation = generation_service.Generation(chat_id=chat.id, message_id=assistant.id)
    await generation_service._run(generation)

    # Five rounds that may call tools, then the one that gives up.
    assert len(payloads) == 6
    assert "after 5 rounds" in generation.tool_events[-1]["error"]


async def test_a_reply_stops_when_it_has_written_too_much(db, user_id, machine, monkeypatch):
    """The bound that is meant to end a long piece of work.

    Steps are a runaway backstop now (200), so something has to say when enough
    has been written. Asserted on the loop, not on the wording: the count of
    requests must be far short of the step budget.
    """
    settings_store.update(
        db, {"max_steps": 200, "max_completion_tokens": 40}, key=settings_store.AGENTS
    )
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)

    payloads: list[dict] = []
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [[_text("x" * 400), _chunk("file_list", '{"path": "."}')]],
            payloads,
        ),
    )

    async def _no_title(*_args, **_kwargs):
        return ""

    monkeypatch.setattr("lembas.services.chat.generate_title", _no_title)

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    assert len(payloads) < 5, "it should have stopped long before the step backstop"
    assert "tokens" in generation.tool_events[-1]["error"]


async def test_the_token_ceiling_fires_on_an_endpoint_that_reports_no_usage(
    db, user_id, machine, monkeypatch
):
    """The half that would otherwise be silently broken.

    `generation.completion_tokens` is only populated when the endpoint sends a
    usage block, and llama.cpp, Ollama and friends never do -- the fallback
    estimate is computed once, in `_run`'s `finally:`, long after the loop that
    needs it. A ceiling reading only the reported figure would work on OpenAI
    and do nothing at all everywhere else. The stub above sends no usage, so
    this asserts the estimate path directly.
    """
    settings_store.update(
        db, {"max_steps": 200, "max_completion_tokens": 40}, key=settings_store.AGENTS
    )
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)

    payloads: list[dict] = []
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream([[_text("y" * 400), _chunk("file_list", '{"path": "."}')]], payloads),
    )

    async def _no_title(*_args, **_kwargs):
        return ""

    monkeypatch.setattr("lembas.services.chat.generate_title", _no_title)

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    assert not any("usage" in str(p) for p in payloads), "the stub reports no usage"
    assert "tokens" in generation.tool_events[-1]["error"]


async def test_a_zero_ceiling_means_no_ceiling(db, user_id, machine, monkeypatch):
    """Zero is how an administrator says "no limit", the same as index_chars.
    Read with `or 0` on the wrong side it would silently become 200_000."""
    settings_store.update(db, {"max_completion_tokens": 0}, key=settings_store.AGENTS)
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    user = db.get(User, user_id)
    context = session.resolve(db, chat, user)
    assert context.limits.completion_tokens == 0


# --- Interjecting while it works --------------------------------------------------
async def test_a_queued_prompt_is_taken_in_between_rounds(db, user_id, machine, monkeypatch):
    """The point of queueing in an agent chat: steering work already under way.

    An agent that has just finished one loop and is about to start another is
    exactly when "actually, do it the other way" is worth having, and making it
    wait for the whole reply would mean it arrives after the thing it was meant
    to change.
    """
    from lembas.services import chat as chat_service

    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)
    chat_service.create_message(
        db, chat, "user", "actually, check the other directory first", queued=True
    )

    payloads: list[dict] = []
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [[_chunk("file_list", '{"path": "."}')], [_text("Done.")]],
            payloads,
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    # Verbatim, in the user role, with nothing wrapped around it: this genuinely
    # is the person at the keyboard, and quoting it would teach the model that a
    # user turn can be a quotation -- the distinction `execute_plan` relies on.
    assert payloads[1]["messages"][-1] == {
        "role": "user",
        "content": "actually, check the other directory first",
    }


async def test_an_interjection_is_delivered_only_once(db, user_id, machine, monkeypatch):
    """Marked delivered before the request goes out, so a crash loses it rather
    than asking the same thing twice and letting an agent act on it twice."""
    from lembas.services import chat as chat_service

    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)
    waiting = chat_service.create_message(db, chat, "user", "one more thing", queued=True)

    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [
                [_chunk("file_list", '{"path": "."}')],
                [_chunk("file_list", '{"path": "src"}')],
                [_text("Done.")],
            ],
            [],
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    db.expire_all()
    assert db.get(Message, waiting.id).queued is False
    assert generation.injected_ids == [waiting.id]


async def test_the_reply_sorts_before_the_prompt_it_took_in(db, user_id, machine, monkeypatch):
    """Otherwise the next request reads "answer, then the question it answered",
    and a small model dutifully answers it a second time."""
    from lembas.services import chat as chat_service

    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_AUTO)
    message_id = _pending_reply(db, chat)
    waiting = chat_service.create_message(db, chat, "user", "and this", queued=True)

    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream([[_chunk("file_list", '{"path": "."}')], [_text("Done.")]], []),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await generation_service._run(generation)

    db.expire_all()
    reply = db.get(Message, message_id)
    assert reply.created_at > db.get(Message, waiting.id).created_at


# --- Plan mode's artifact ---------------------------------------------------------
def test_plan_submit_is_offered_only_in_plan_mode(db, user_id, machine):
    """It ends the reply. A model in Auto mode that proposed a plan instead of
    doing the work would be obeying the wrong instinct at the wrong moment."""
    user = db.get(User, user_id)
    planning, _p = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
    assert "plan_submit" in _offered(db, planning, user)

    planning.agent_mode = policy.MODE_AUTO
    db.commit()
    assert "plan_submit" not in _offered(db, planning, user)


async def test_a_plan_ends_the_reply_and_lands_on_the_message(
    db, user_id, machine, monkeypatch
):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
    message_id = _pending_reply(db, chat)

    plan = {"title": "Tidy the logs", "steps": ["Read the log", "Rotate it", "Restart"]}
    rounds = [
        [_chunk("plan_submit", _json.dumps(plan))],
        # The model gets one wordless round to explain itself. If it tries to
        # act in it -- as this one does -- there are no tools to act with.
        [_chunk("shell_run", '{"command": "rm -rf /"}'), _text("Here is what I would do.")],
    ]
    seen: list[dict] = []
    monkeypatch.setattr(generation_service, "stream_chat", _stub_stream(rounds, seen))

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await asyncio.wait_for(generation_service._run(generation), timeout=10)

    assert generation.plan == plan
    assert len(seen) == 2, "one round to plan, one to say what it proposed"
    assert "tools" not in seen[1], "the second round is offered nothing to act with"
    assert len(generation.tool_events) == 1, "the shell call had no tool to reach"

    db.expire_all()
    message = db.get(Message, message_id)
    assert message.plan_json == plan


async def test_a_plan_with_no_steps_is_sent_back(db, user_id, machine, monkeypatch):
    chat, _profile = _setup(db, user_id, machine, mode=policy.MODE_PLAN)
    message_id = _pending_reply(db, chat)
    monkeypatch.setattr(
        generation_service,
        "stream_chat",
        _stub_stream(
            [[_chunk("plan_submit", '{"title": "Nothing", "steps": []}')], [_text("Sorry.")]],
            [],
        ),
    )

    generation = generation_service.Generation(chat_id=chat.id, message_id=message_id)
    await asyncio.wait_for(generation_service._run(generation), timeout=10)

    assert generation.plan is None
    assert generation.tool_events[0]["status"] == "error"


# --- Carrying a plan out ------------------------------------------------------------
def test_executing_a_plan_switches_to_edit_and_quotes_it(client, db, registered, machine):
    """Edit, never Auto. The plan was written under a mode where every command
    stopped for approval, and a button that also removed the asking is not the
    button anybody pressed."""
    from sqlalchemy import select as _select

    user = db.scalar(_select(User))
    chat, _profile = _setup(db, user.id, machine, mode=policy.MODE_PLAN)
    message = Message(
        chat_id=chat.id,
        role=ROLE_ASSISTANT,
        content="Here is what I would do.",
        complete=True,
        plan_json={"title": "Tidy the logs", "steps": ["Read the log", "Rotate it"]},
    )
    db.add(message)
    db.commit()

    response = client.post(f"/api/chats/{chat.id}/messages/{message.id}/execute-plan")
    assert response.status_code == 200

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

    sent = db.scalars(
        _select(Message).where(Message.chat_id == chat.id, Message.role == "user")
    ).all()[-1]
    assert "Tidy the logs" in sent.content
    assert "Read the log" in sent.content
    # Quoted rather than stated. A plan whose text came out of a file the model
    # read must not arrive wearing the reader's authority.
    assert sent.content.lstrip().startswith("Carry out the plan you proposed above")
    assert "> " in sent.content


def test_a_message_with_no_plan_cannot_be_executed(client, db, registered, machine):
    from sqlalchemy import select as _select

    user = db.scalar(_select(User))
    chat, _profile = _setup(db, user.id, machine)
    message = Message(chat_id=chat.id, role=ROLE_ASSISTANT, content="hi", complete=True)
    db.add(message)
    db.commit()

    assert (
        client.post(f"/api/chats/{chat.id}/messages/{message.id}/execute-plan").status_code
        == 404
    )


# --- Rewind ---------------------------------------------------------------------------
def test_editing_a_turn_records_that_the_machine_did_not_rewind(
    client, db, registered, machine
):
    """The transcript goes back; the project directory does not. Deleting
    somebody's real working tree to match would be far worse than the
    inconsistency, so the model is told instead."""
    from sqlalchemy import select as _select

    user = db.scalar(_select(User))
    chat, _profile = _setup(db, user.id, machine)
    first = Message(chat_id=chat.id, role="user", content="do a thing", complete=True)
    db.add(first)
    db.commit()

    assert chat.rewound_at is None
    client.post(
        f"/api/chats/{chat.id}/messages/{first.id}/edit", data={"content": "do another thing"}
    )
    db.refresh(chat)
    assert chat.rewound_at is not None


def test_an_ordinary_chat_records_no_rewind(client, db, registered, machine):
    from sqlalchemy import select as _select

    user = db.scalar(_select(User))
    chat, _profile = _setup(db, user.id, machine, kind="chat")
    first = Message(chat_id=chat.id, role="user", content="hello", complete=True)
    db.add(first)
    db.commit()

    client.post(f"/api/chats/{chat.id}/messages/{first.id}/edit", data={"content": "hi"})
    db.refresh(chat)
    assert chat.rewound_at is None


async def test_the_harness_warns_after_a_rewind(db, user_id, machine):
    from datetime import UTC, datetime

    from lembas.services import harness

    chat, _profile = _setup(db, user_id, machine)
    user = db.get(User, user_id)
    offered = tools_service.resolve_tools(db, chat, user).schemas

    assert "was rewound" not in harness.compose(db, user, offered, chat)

    chat.rewound_at = datetime.now(UTC)
    db.commit()
    text = harness.compose(db, user, offered, chat)
    assert "was rewound" in text
    assert "still there" in text