"""The walkthrough (X6, M6.6b): an item's steps, its runs' turns, and each
turn's request read against the log it was made from."""

import hashlib
import sqlite3
from pathlib import Path

import pytest
from ui_fixtures import TOKEN, circling, recorded

from nisse_core.log.canonical import UserMessage
from nisse_core.models import calls
from nisse_ui import walk
from nisse_ui.app import App
from nisse_ui.sources import Source, StateReader, call_record


@pytest.fixture
def agent(tmp_path: Path) -> Source:
    source, store = recorded(tmp_path / "agent")
    store.close()
    return source


def page(app: App, path: str, status: int = 200) -> str:
    response = app.handle(path)
    assert response.status == status, response.body.decode()[:500]
    return response.body.decode()


def view(source: Source, call_id: int) -> walk.TurnView:
    with StateReader(source.state_db) as state, call_record(source) as reader:
        assert reader is not None
        found = walk.turn(state, reader, call_id)
    assert found is not None
    return found


def steps(source: Source, item: str | None) -> walk.Walkthrough:
    with StateReader(source.state_db) as state, call_record(source) as reader:
        return walk.walkthrough(state, reader, item)


# --- the way in -------------------------------------------------------------------


def test_the_sources_and_an_agents_items_are_listed(agent: Source) -> None:
    app = App([agent])
    home = page(app, "/")
    assert "/s/agent/" in home and ">1<" in home  # one work item
    items = page(app, "/s/agent/")
    assert "Fix the parser" in items and "/s/agent/item/w1" in items
    assert "/s/agent/item/-" in items  # the agent-level log


# --- the walkthrough --------------------------------------------------------------


def test_an_items_steps_hold_its_runs_turns_and_what_the_orchestrator_said(
    agent: Source,
) -> None:
    found = steps(agent, "w1")
    kinds = [kind for kind, _ in found.timeline]
    assert kinds == ["event", "event", "run"]  # the restatement, "go", the run
    [run] = [obj for kind, obj in found.timeline if kind == "run"]
    assert run.run["outcome"] == "wait"
    assert [t.number for t in run.turns] == [1, 2, 3]
    # The correction for the prose reply came during the run: it's shown
    # inside it, between the turns it fell between.
    inner = [kind for kind, _ in run.entries]
    assert inner == ["turn", "turn", "event", "turn"]
    assert any("prose" in told for told in run.turns[2].told)
    assert run.turns[0].reply.calls == (("read_file", '{"path": "parser.py"}'),)
    assert run.turns[1].reused == "280 of 340"
    html = page(App([agent]), "/s/agent/item/w1")
    assert "turn 3" in html and "reused 330 of 350" in html


def test_agent_level_calls_have_their_own_log(agent: Source) -> None:
    found = steps(agent, None)
    [run] = [obj for kind, obj in found.timeline if kind == "run"]
    assert (run.run["kind"], run.run["role"]) == ("classification", "classifier")
    assert "harmless" in page(App([agent]), "/s/agent/item/-")


def test_work_from_before_recording_has_its_steps_but_no_requests(
    tmp_path: Path,
) -> None:
    source, store = recorded(tmp_path / "a")
    # The agent restarts: a fresh, empty record; the log persists.
    calls.CallRecorder(source.calls_db).close()
    store.close()
    found = steps(source, "w1")
    assert found.before_record
    # The restatement, "go" and the correction: steps, with no runs to hold them.
    assert [kind for kind, _ in found.timeline] == ["event", "event", "event"]
    html = page(App([source]), "/s/agent/item/w1")
    assert "Recording began" in html
    assert "All 8 events" in html  # the log is all still there


def test_without_a_call_record_the_log_is_shown(tmp_path: Path) -> None:
    source, store = recorded(tmp_path / "a")
    store.close()
    source.calls_db.unlink()
    html = page(App([source]), "/s/agent/item/w1")
    assert "No call record" in html and "All 8 events" in html
    page(App([source]), "/s/agent/call/1", 404)


def test_a_run_going_in_circles_is_flagged(tmp_path: Path) -> None:
    source = circling(tmp_path / "c")
    found = steps(source, "pr-7")
    [run] = [obj for kind, obj in found.timeline if kind == "run"]
    [circle] = run.circles
    assert (circle.kind, circle.turns) == ("repeated reply", (0, 1, 2, 3))
    assert run.turns[0].flags == ["repeated reply"] and not run.turns[4].flags
    assert "Going in circles: repeated reply at turns 1, 2, 3, 4" in page(
        App([source]), "/s/circling/item/pr-7"
    )
    assert view(source, 2).flags == ["repeated reply"]


# --- one turn ---------------------------------------------------------------------


def test_each_message_is_shown_beside_the_log_events_it_came_from(
    agent: Source,
) -> None:
    turn = view(agent, 2)
    assert turn.match_note is None and turn.unmatched == []
    roles = [m.role for m in turn.messages]
    assert roles == ["system", "user", "assistant", "tool"]
    # The restatement and "go" fold into one user message.
    assert [s.id for s in turn.messages[1].events] == [1, 2]
    assert [s.id for s in turn.messages[2].events] == [3]
    assert [s.id for s in turn.messages[3].events] == [4]
    assert all(s.tags == ["unchanged"] for m in turn.messages for s in m.events)
    assert calls.NOTICE in turn.messages[0].text


def test_what_changed_since_the_turn_before_is_marked(agent: Source) -> None:
    first, second = view(agent, 1), view(agent, 2)
    assert not first.compared and not any(m.new for m in first.messages)
    assert second.compared
    assert [m.new for m in second.messages] == [False, False, True, True]
    assert [m.first_change for m in second.messages] == [False, False, True, False]
    assert not second.tools_changed
    assert (second.previous, second.following) == (1, 3)


def test_the_turn_page_shows_the_reply_reasoning_and_raw_json(agent: Source) -> None:
    html = page(App([agent]), "/s/agent/call/1")
    assert "First, read the parser." in html  # the reasoning
    assert "Request JSON, exactly as sent" in html
    assert "local-backing" in html  # the model that answered
    assert "reused" not in html or "0 of 300" in html


def test_how_rendering_changed_an_event_is_named() -> None:
    call = {"shortened": {"5": 1200}}
    logged = walk.Event(5, "t", "tool_result", {"result": "long"}, "w", "driver")
    sent = {"result": "lo"}
    assert walk._tags([5, "driver", "tool_result", "h"], sent, logged, call) == [
        "shortened by 1,200 characters"
    ]
    told = walk.Event(6, "t", "tool_call", {"tool": "x"}, "w", "helper")
    narrated = {"text": "x → ok", "source": "view:helper"}
    assert walk._tags(
        [6, "helper", "user_message", "h"], narrated, told, {"shortened": {}}
    ) == ["narrated from tool call by helper"]
    fact = walk.Event(8, "t", "fact", {"text": "agreed"}, "w", "orchestrator")
    as_said = {"text": "[orchestrator] agreed", "source": "view:orchestrator"}
    assert walk._tags(
        [8, "orchestrator", "user_message", "h"], as_said, fact, {"shortened": {}}
    ) == ["labelled for this role (a fact by orchestrator)"]
    said = walk.Event(
        7, "t", "user_message", {"text": "hi", "source": "m"}, "w", "operator"
    )
    labelled = {"text": "[operator, matrix] hi", "source": "view:operator"}
    assert walk._tags(
        [7, "operator", "user_message", "h"], labelled, said, {"shortened": {}}
    ) == ["labelled for this role"]
    assert walk._tags(
        [0, "harness", "fact", "h"], {"text": "x"}, None, {"shortened": {}}
    ) == ["made for this call, not in the log"]
    assert walk._tags(
        [9, "driver", "compaction", "h"], {"replaces": [2, 8]}, None, {"shortened": {}}
    ) == ["compaction of events 2–8"]


def test_a_reply_reads_from_either_route() -> None:
    chat = {
        "choices": [
            {
                "message": {
                    "content": "ok",
                    "reasoning_content": "hm",
                    "tool_calls": [{"function": {"name": "f", "arguments": "{}"}}],
                },
                "finish_reason": "tool_calls",
            }
        ]
    }
    assert walk.reply_of(chat) == walk.Reply("ok", "hm", (("f", "{}"),), "tool_calls")
    messages = {
        "content": [
            {"type": "thinking", "thinking": "hm"},
            {"type": "text", "text": "ok"},
            {"type": "tool_use", "name": "f", "input": {"a": 1}},
        ],
        "stop_reason": "tool_use",
    }
    assert walk.reply_of(messages) == walk.Reply(
        "ok", "hm", (("f", '{"a": 1}'),), "tool_use"
    )
    assert walk.reply_of("garbage") == walk.Reply()


# --- untrusted text and read-only -------------------------------------------------


def test_everything_shown_is_escaped_and_no_script_may_run(agent: Source) -> None:
    app = App([agent])
    for path in (
        "/s/agent/item/w1",
        "/s/agent/call/2",
        "/s/agent/call/3",
        "/s/agent/event/4",
    ):
        response = app.handle(path)
        html = response.body.decode()
        assert "<script>alert" not in html
        assert "&lt;script&gt;alert" in html
        assert "&lt;/pre&gt;&lt;/details&gt;" in html  # can't close the page's own
        csp = response.headers["Content-Security-Policy"]
        assert "default-src 'none'" in csp and "script-src" not in csp
        assert TOKEN not in html


def _digests(root: Path) -> dict[str, str]:
    """Each file's content, except SQLite's shared-memory index, which
    readers of a live WAL database take part in by design."""
    return {
        p.name: ""
        if p.name.endswith("-shm")
        else hashlib.sha256(p.read_bytes()).hexdigest()
        for p in sorted(root.iterdir())
        if p.is_file()
    }


def test_browsing_reads_only_and_leaves_no_lock_while_the_agent_writes(
    tmp_path: Path,
) -> None:
    source, store = recorded(tmp_path / "a")  # the store stays open, as live
    before = _digests(tmp_path / "a")
    app = App([source])
    for path in (
        "/",
        "/s/agent/",
        "/s/agent/item/w1",
        "/s/agent/item/-",
        "/s/agent/call/1",
        "/s/agent/call/2",
        "/s/agent/event/1",
    ):
        page(app, path)
    # No file changed, and none was created: the closed call record has no
    # -wal or -shm, and opening it made none.
    assert _digests(tmp_path / "a") == before
    assert not Path(f"{source.calls_db}-wal").exists()
    # The agent can still write at once: nothing here holds a lock.
    store.transcript("w1", "operator").append(UserMessage(text="more", source="m:2"))
    store.close()
    assert "more" in page(app, "/s/agent/item/w1")


# --- versions -----------------------------------------------------------------------


def test_a_store_of_an_unknown_version_is_named_not_guessed(agent: Source) -> None:
    with sqlite3.connect(agent.state_db) as conn:
        conn.execute("UPDATE meta SET value = '99' WHERE key = 'schema_version'")
    html = page(App([agent]), "/s/agent/item/w1", 409)
    assert "schema 99, not known to this page" in html
    assert "schema 99" in page(App([agent]), "/")


def test_a_call_record_of_an_unknown_layout_leaves_the_log_readable(
    agent: Source,
) -> None:
    with sqlite3.connect(agent.calls_db) as conn:
        conn.execute("UPDATE meta SET value = '2' WHERE key = 'layout_version'")
    html = page(App([agent]), "/s/agent/item/w1")
    assert "call record layout 2" in html and "All 8 events" in html
    page(App([agent]), "/s/agent/call/1", 409)


def test_item_data_that_doesnt_parse_still_renders(agent: Source) -> None:
    with sqlite3.connect(agent.state_db) as conn:
        conn.execute("UPDATE work_items SET data = 'not json', position = 'NEW_STATE'")
    html = page(App([agent]), "/s/agent/item/w1")
    assert "not json" in html and "NEW_STATE" in html
    assert "NEW_STATE" in page(App([agent]), "/s/agent/")


def test_unknown_paths_are_not_found(agent: Source) -> None:
    app = App([agent])
    page(app, "/s/nobody/", 404)
    page(app, "/s/agent/item/nope", 404)
    page(app, "/s/agent/call/999", 404)
    page(app, "/s/agent/event/999", 404)
    page(app, "/s/agent/call/x", 404)
    page(app, "/etc/passwd", 404)
    assert app.handle("/static/style.css").content_type.startswith("text/css")