"""Phase 70 E2E (Playwright, mock-only): the harness-aligned tool surface
(``ls`` / ``read(path)`` / ``grep(pattern, path?)``).

Story: ``.agents/user_stories/agent-document-tools.md`` (phase 70 reshapes
the tools that story delivered — owner decision 2026-09-03: "match
existing harnesses as much as possible", the pi.dev tool shapes).
Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_harness_aligned_tools.py -v --no-cov

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic marker flows in ``tests/e2e/mock_llm.py`` (phase 70: the
flows emit the NEW names with the NEW argument shapes):

* the READ flow (``use your tools`` (``TOOLS_TRIGGER``) + the HIGH
  prompt's ``<tools>`` section): ``ls`` (id ``call_0``, no arguments) →
  ``read`` on the JOINED combined ``source/path`` of the first catalog
  line (id ``call_1``) → the ``Read <source/path>. <quote>`` answer;
* the SEARCH flow (``search your documents`` (``SEARCH_TRIGGER``) + the
  ``<tools>`` section): ``grep`` with ``{"pattern": SEARCH_PATTERN}``
  (id ``call_0``) → the ``Found <matched line>`` answer.

The combined ``source/path`` string is the canonical document identity:
the mock joins the two labeled catalog fields itself (the catalog
format is unchanged), and the SSE ``tool`` frames carry exactly what the
model "passed" — ``read``'s combined path, ``grep``'s pattern, ``ls``'s
scope or null when unscoped (the phase-70 argument rule).

KB fixtures:

* READ flow — the ``test_agent_document_tools.py`` two-document pair
  (TRUNCATE-then-seed): ``Homelab/aws-route53.md`` seeded with one
  chunk whose embedding is the mock's own bag-of-words vector (the
  marker question cosines ≈0.69 against it, well past the E2E 0.30
  threshold, and it FTS-matches too → grounded) and
  ``Deployments/example-record-file.json`` indexed WITHOUT chunks (the
  catalog-first line the mock reads; never in the retrieval context).
* SEARCH flow — the phase-68 fixture (``tests/fixtures/search_docs/``)
  imported through the real importer, its line 6 carrying the sentinel
  ``reese-sentinel-42`` exactly once (``test_search_tool.py`` pattern).

Test → phase mapping (Playwright Mapping Rule):
1. ``test_read_flow_lines_answer_sources_no_raw_markup`` — the
   grounded READ turn: the UI shows the ``ls`` line (unscoped "🔎
   Listing documents", no argument) then the "📄 Reading <source/path>"
   line with the combined path in a ``<code>`` element, the answer
   streams and quotes the read document, the done-state sources
   include the read document, and NO raw tool markup (``<|…|>``,
   ``tool_call``) appears anywhere in the DOM — the live incident this
   phase fixes.
2. ``test_grep_flow_line_then_answer`` — the grounded SEARCH turn: the
   "🔎 Searching for <pattern>" line (sentinel in ``<code>``) then the
   matched-line answer.
3. ``test_wire_argument_rule_across_both_flows`` — the SSE wire across
   BOTH flows in one session: every ``tool`` frame's name is in
   {``ls``, ``read``, ``grep``} (no pre-phase-70 name ever reaches the
   client) and the argument rule holds — ``read`` → the combined path
   as passed, ``grep`` → the pattern, ``ls`` → null when unscoped.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import time
from datetime import UTC, datetime
from pathlib import Path
from threading import Thread
from typing import Any

from playwright.sync_api import Page, expect
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.config import Settings
from app.db import SessionLocal
from app.models import Chunk, Document
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import login
from tests.e2e.mock_llm import SEARCH_PATTERN, embed_text

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "search_docs"

# --------------------------------------------------------------------------
# READ flow — the two-document pair (cf. test_agent_document_tools.py)
# --------------------------------------------------------------------------

SEED_SOURCE = "Homelab"
SEED_PATH = "aws-route53.md"
SEED_SP = f"{SEED_SOURCE}/{SEED_PATH}"

READ_SOURCE = "Deployments"
READ_PATH = "example-record-file.json"
READ_SP = f"{READ_SOURCE}/{READ_PATH}"

#: The retrievable document (the grounded seed context): the repeated
#: record-file lines carry the marker question's key tokens — verified
#: ≈0.69 cosine against the mock's embeddings (E2E threshold 0.30) plus
#: FTS hits.
ROUTE53_CONTENT = (
    "# AWS Route 53 Notes\n\n"
    "## Record file\n\n"
    + (
        "The aws route53 hosted zone for reeselink keeps every record in "
        "reseelink.json — the exact JSON shape of reeselink.json is "
        "documented in example-record-file.json.\n"
    )
    * 10
    + "\n## Sync job\n\n"
    "A cron job pushes reeselink.json to the aws route53 hosted zone "
    "every fifteen minutes; the diff is applied through the route53 api.\n"
)

#: The read document (the catalog-first line the mock reads; no chunks,
#: so retrieval never puts it in context). Its FIRST line is longer than
#: 80 chars, so the mock's first-80-chars quote is newline-free.
RECORD_FILE_CONTENT = (
    '{ "version": 3, "comment": "ReeseLink hosted zone records — the exact '
    'JSON shape of reeselink.json",\n'
    '  "hosted_zone_id": "Z0RESEELINK01",\n'
    '  "record_sets": [\n'
    '    { "name": "www.reeselink.example", "type": "A", "ttl": 300,\n'
    '      "resource_records": [ { "value": "10.0.0.20" } ] },\n'
    '    { "name": "api.reeselink.example", "type": "CNAME", "ttl": 300,\n'
    '      "resource_records": [ { "value": "www.reeselink.example" } ] }\n'
    "  ]\n"
    "}\n"
)
assert "\n" not in RECORD_FILE_CONTENT[:80]  # the quote must stay one line

#: Carries ``TOOLS_TRIGGER`` (and nothing else — no multi-read, no
#: search, no other mock marker).
READ_QUESTION = (
    "Use your tools: what is the exact JSON shape of reeselink.json "
    "for my aws route53 hosted zone?"
)
for _other in (
    "read two documents",
    "search your documents",
    "write a long answer",
    "think in paragraphs",
    "think out loud",
    "show the end of your notes",
    "show me a table",
    "fail then answer",
    "always fail",
    "embed fail once",
    "pretend to think slowly",
):
    assert _other not in READ_QUESTION.lower(), _other

READ_ANSWER_PREFIX = f"Read {READ_SP}."
READ_ANSWER_QUOTE = RECORD_FILE_CONTENT[:80]


def _seed_read_pair(db: Session) -> None:
    """The two-document READ-flow KB (see the module docstring)."""
    md = Document(
        source=SEED_SOURCE,
        path=SEED_PATH,
        full_path=f"/tmp/{SEED_PATH}",
        title="AWS Route 53 Notes",
        content=ROUTE53_CONTENT,
        content_hash=hashlib.sha256(ROUTE53_CONTENT.encode()).hexdigest(),
        indexed_at=datetime.now(UTC),
    )
    db.add(md)
    db.flush()
    # One chunk carrying the mock's own embedding → genuine token
    # overlap between the marker question and this document (the only
    # retrievable document).
    db.add(
        Chunk(
            document_id=md.id,
            position=0,
            content=ROUTE53_CONTENT,
            embedding=embed_text(ROUTE53_CONTENT),
        )
    )
    db.add(
        Document(
            source=READ_SOURCE,
            path=READ_PATH,
            full_path=f"/tmp/{READ_PATH}",
            title="Example Record File",
            content=RECORD_FILE_CONTENT,
            content_hash=hashlib.sha256(RECORD_FILE_CONTENT.encode()).hexdigest(),
            indexed_at=datetime.now(UTC),
        )
    )


# --------------------------------------------------------------------------
# SEARCH flow — the phase-68 fixture (cf. test_search_tool.py)
# --------------------------------------------------------------------------

SEED_SOURCE_S = "search_docs"
SEED_PATH_S = "reese-notes.md"
SEED_SP_S = f"{SEED_SOURCE_S}/{SEED_PATH_S}"

#: The fixture's sentinel line (line 6) — the mock's grep matches it
#: exactly once; its ``text`` part is what the "Found …" answer quotes.
SENTINEL_LINE = f"The offsite vault passphrase marker is {SEARCH_PATTERN}."
FOUND_ANSWER = f"Found {SENTINEL_LINE[:80]}"

#: Carries ``SEARCH_TRIGGER`` and is on-topic (cosine ≈0.51 against the
#: fixture + FTS hits → HIGH gate, the ``<tools>`` section rides along).
SEARCH_QUESTION = (
    "Search your documents for the vault passphrase marker in my homelab "
    "kubernetes backup notes?"
)
assert SEARCH_PATTERN.lower() not in SEARCH_QUESTION.lower()


def _pin_fixture() -> None:
    """The fixture carries the sentinel on line 6, exactly once."""
    content = (FIXTURES / SEED_PATH_S).read_text(encoding="utf-8")
    lines = content.split("\n")
    assert lines[5] == SENTINEL_LINE, lines[5]
    assert sum(SEARCH_PATTERN in line for line in lines) == 1


# --------------------------------------------------------------------------
# DB seeding (TRUNCATE-then-seed / TRUNCATE-then-import)
# --------------------------------------------------------------------------


async def _import_search_fixtures(mock_port: int) -> ImportSummary:
    kwargs: dict[str, Any] = {
        "_env_file": None,
        "llm_base_url": f"http://127.0.0.1:{mock_port}/v1",
    }
    settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return await import_sources([FIXTURES], LLMClient(settings))


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread.

    Playwright's sync API keeps an asyncio loop running on the test
    thread, so ``asyncio.run`` cannot be called directly from a test
    body (the established house helper).
    """
    box: dict[str, Any] = {}

    def runner() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as e:  # noqa: BLE001 — re-raised on the test thread
            box["error"] = e

    t = Thread(target=runner)
    t.start()
    t.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _reset_db_read_pair() -> None:
    """Truncate the KB (plus the prompt-shaping tables), then seed the
    two-document READ-flow pair. ``steering_notes`` / ``kb_overview``
    are truncated too, so the HIGH prompt is exactly ``<relevance>`` +
    ``<documents>`` + ``<tools>`` — byte-stable prompts, byte-stable
    answers."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        db.commit()
        _seed_read_pair(db)
        db.commit()


def _reset_db_search_fixture(mock_port: int) -> None:
    """Truncate the KB (plus the prompt-shaping tables), then import the
    phase-68 search fixture through the real importer."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        db.commit()
    summary = _run_in_thread(_import_search_fixtures(mock_port))
    assert summary is not None and summary.added == 1, summary


# --------------------------------------------------------------------------
# Page helpers (the test_agent_document_tools.py pattern)
# --------------------------------------------------------------------------

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream
#: (a response clone read in the background) — wire-level assertions
#: for the ``tool`` frames, independent of the UI rendering.
SSE_HOOK = """
() => {
  if (window.__sseInstalled) return;
  window.__sseInstalled = true;
  window.__sseFrames = [];
  const origFetch = window.fetch;
  window.fetch = async function (...args) {
    const res = await origFetch.apply(this, args);
    try {
      const url = typeof args[0] === 'string' ? args[0] : args[0].url;
      if (url.includes('/api/chat')) {
        res.clone().text().then((bodyText) => {
          for (const block of bodyText.split('\\n\\n')) {
            const line = block.trim();
            if (line.startsWith('data: ')) {
              window.__sseFrames.push(line.slice(6));
            }
          }
        });
      }
    } catch (e) { /* non-clonable responses: ignored */ }
    return res;
  };
}
"""


def _install_sse_hook(page: Page) -> None:
    page.evaluate(SSE_HOOK)


def _drain_frames(page: Page) -> list[dict]:
    """One turn's SSE frames: wait for that turn's ``done`` frame, then
    return EVERY frame captured since the last drain (the hook's
    background read appends the whole stream at once after it closes, so
    clearing-and-reading is race-free per turn)."""
    deadline = time.monotonic() + 10.0
    while True:
        raw = page.evaluate(
            "() => { const f = window.__sseFrames || []; "
            "window.__sseFrames = []; return f; }"
        )
        parsed = [json.loads(line) for line in raw if line]
        if any(f.get("type") == "done" for f in parsed):
            return parsed
        if time.monotonic() > deadline:
            raise AssertionError(
                f"SSE hook captured no `done` frame (frames so far: "
                f"{len(parsed)}) — hook install failed?"
            )
        time.sleep(0.05)


def _tool_frames(frames: list[dict]) -> list[dict]:
    return [f for f in frames if f.get("type") == "tool"]


def _submit(page: Page, question: str) -> None:
    page.fill("#message-input", question)
    page.click("#send-btn")
    # The user bubble lands synchronously with the submit handler.
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)


def _wait_settled(page: Page) -> None:
    """The turn is complete: answer text in the bubble, button recovered.

    Phase 48: the label assertion carries the settle wait with an
    explicit timeout — the in-flight button is the enabled Stop control
    (never disabled), so ``to_be_enabled`` no longer blocks until the
    turn settles."""
    expect(page.locator(".msg.brain .bubble").last).not_to_have_text("", timeout=30_000)
    expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)


# --------------------------------------------------------------------------
# 1. The grounded READ turn: ls line → Reading line → quoted answer,
#    sources include the read doc, no raw tool markup anywhere in the DOM
# --------------------------------------------------------------------------


def test_read_flow_lines_answer_sources_no_raw_markup(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db_read_pair()
    login(page, app_url, next="/")
    _install_sse_hook(page)

    _submit(page, READ_QUESTION)
    _wait_settled(page)

    # The UI shows the ls line (UNSCOPED — no argument, no <code>) then
    # the "📄 Reading <source/path>" line with the COMBINED path in a
    # <code> element (the path is data, never markup).
    lines = page.locator(".msg.brain .tool-call")
    expect(lines).to_have_count(2)
    expect(lines.nth(0)).to_contain_text("Listing documents")
    expect(lines.nth(0).locator("code")).to_have_count(0)
    expect(lines.nth(1)).to_contain_text("Reading ")
    expect(lines.nth(1).locator("code")).to_have_text(READ_SP)

    # The answer streamed and quotes the read document (the mock's
    # deterministic echo: "Read <source/path>. <first 80 chars>").
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(READ_ANSWER_PREFIX)
    expect(bubble).to_contain_text(READ_ANSWER_QUOTE)

    # Wire level: ls then read — the phase-70 argument rule (ls
    # unscoped → null; read → the combined path as passed) — ahead of
    # the first delta.
    frames = _drain_frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "ls", "argument": None},
        {"type": "tool", "name": "read", "argument": READ_SP},
    ]
    first_delta = next(i for i, f in enumerate(frames) if f.get("type") == "delta")
    assert all(
        i < first_delta for i, f in enumerate(frames) if f.get("type") == "tool"
    )
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False
    # Done-state sources include the read document (retrieval doc first,
    # the agent's read doc after — the phase-37 extension contract).
    assert [(s["source"], s["path"]) for s in done["sources"]] == [
        (SEED_SOURCE, SEED_PATH),
        (READ_SOURCE, READ_PATH),
    ]

    # The live incident this phase fixes: NO raw tool markup anywhere in
    # the DOM — the model's trained wire shapes (<|tool_call_…|>,
    # "tool_calls", finish_reason) must never leak into the rendered
    # conversation.
    dom = page.locator("#messages").inner_html()
    for raw in ("<|", "tool_call", "tool_calls", "finish_reason"):
        assert raw not in dom, f"raw tool markup {raw!r} leaked into the DOM"


# --------------------------------------------------------------------------
# 2. The grounded SEARCH turn: the "🔎 Searching for <pattern>" line,
#    then the matched-line answer
# --------------------------------------------------------------------------


def test_grep_flow_line_then_answer(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _pin_fixture()
    page.set_default_timeout(30_000)
    _reset_db_search_fixture(mock_llm)
    login(page, app_url, next="/")
    _install_sse_hook(page)

    _submit(page, SEARCH_QUESTION)
    _wait_settled(page)

    # ONE tool line above the answer: "🔎 Searching for " + the sentinel
    # in a <code> element (the pattern is data, never markup).
    lines = page.locator(".msg.brain .tool-call")
    expect(lines).to_have_count(1)
    expect(lines.nth(0)).to_contain_text("Searching for")
    expect(lines.nth(0).locator("code")).to_have_text(SEARCH_PATTERN)

    # The answer quotes the MATCHED LINE — the grep result reached the
    # model and landed in the answer (the mock's deterministic echo).
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(FOUND_ANSWER)

    # Wire level: exactly ONE tool frame — grep carrying the PATTERN as
    # its argument (the phase-70 argument rule) — ahead of the first
    # delta; the turn is grounded.
    frames = _drain_frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "grep", "argument": SEARCH_PATTERN}
    ]
    first_delta = next(i for i, f in enumerate(frames) if f.get("type") == "delta")
    assert all(
        i < first_delta for i, f in enumerate(frames) if f.get("type") == "tool"
    )
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False


# --------------------------------------------------------------------------
# 3. The SSE wire across BOTH flows: every tool frame carries a
#    phase-70 name and the single-string argument rule
# --------------------------------------------------------------------------


def test_wire_argument_rule_across_both_flows(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _pin_fixture()
    page.set_default_timeout(30_000)
    _reset_db_read_pair()
    login(page, app_url, next="/")
    _install_sse_hook(page)

    # Turn 1 — the READ flow (ls → read on the combined path).
    _submit(page, READ_QUESTION)
    _wait_settled(page)
    read_frames = _drain_frames(page)

    # Turn 2 — re-seed the search fixture, then the SEARCH flow (grep
    # for the sentinel). The app's chat path is single-turn (system +
    # user message), so the first turn cannot influence this one.
    _reset_db_search_fixture(mock_llm)
    _submit(page, SEARCH_QUESTION)
    _wait_settled(page)
    search_frames = _drain_frames(page)

    read_tools = _tool_frames(read_frames)
    search_tools = _tool_frames(search_frames)
    # The ordered, combined tool-frame sequence across both flows: the
    # argument rule end-to-end — read → the combined path as passed,
    # grep → the pattern, ls → null when unscoped.
    assert read_tools + search_tools == [
        {"type": "tool", "name": "ls", "argument": None},
        {"type": "tool", "name": "read", "argument": READ_SP},
        {"type": "tool", "name": "grep", "argument": SEARCH_PATTERN},
    ]
    # No pre-phase-70 name ever reaches the client.
    for frame in read_tools + search_tools:
        assert frame["name"] in {"ls", "read", "grep"}, frame
        assert frame["argument"] is None or isinstance(frame["argument"], str)

    # And both turns answered (neither flow stalled at a tool round).
    assert next(f for f in read_frames if f["type"] == "done")["deflected"] is False
    assert next(f for f in search_frames if f["type"] == "done")["deflected"] is False