"""Integration: the phase-105 hidden-folders flag through the real import
pipeline.

Phase 105 (TODO.md L3 — "…a toggle per input … to allow indexing hidden
.folders."): ``import_sources`` gains ``include_hidden_by_root`` (same
``str(root)`` keying as phase 89's ``ignore_by_root``). A1
(owner-confirmed 2026-09-14): flag ON admits dot-prefixed components —
hidden files are indexed, embedded, and summarized exactly like visible
files — while ``EXCLUDED_DIRS`` stay excluded in both states. A2: a file
indexed with the flag ON leaves the KB on the next ``prune=True`` run
with the flag OFF (the untouched ``seen`` set does the work). A4: no map
→ byte-identical to pre-phase-105. Mirrors the fixture-tree + mock-LLM
pattern of ``test_importer_ignore.py``: a ``tmp_path`` source dir run
through the real ``import_sources`` into the compose Postgres, with
:class:`tests.fakes.FakeEmbedder` as the deterministic LLM stand-in.
"""
from __future__ import annotations

import asyncio
from pathlib import Path

from sqlalchemy import select, text
from sqlalchemy.orm import Session

from app.models import Chunk, Document
from app.rag.importer import import_sources
from tests.fakes import FakeEmbedder

NAME = "HiddenFix"


def _write(root: Path, rel: str, content: str) -> None:
    path = root / rel
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def _tree(tmp_path: Path, name: str = NAME) -> Path:
    """One visible md, a hidden dir (md + non-markdown yaml), an excluded dir."""
    root = tmp_path / name
    _write(root, "visible.md", "# Visible\n\nvisible body\n")
    _write(root, ".hidden/note.md", "# Note\n\nHIDDEN-MD-CONTENT\n")
    _write(root, ".hidden/data.yaml", "key: HIDDEN-YAML-VALUE\n")
    _write(root, ".venv/junk.md", "# Junk\n\nEXCLUDED-CONTENT\n")
    return root


def _reset(db: Session) -> None:
    # House cleanup pattern (tests/integration/test_importer_ignore.py).
    db.execute(text("TRUNCATE chunks, documents, query_log"))
    db.commit()


def test_hidden_paths_not_indexed_by_default(db: Session, tmp_path: Path) -> None:
    # A4: no map → today's behavior, byte-identical — the hidden files
    # never produce a Document/Chunk row and are never embedded.
    _reset(db)
    root = _tree(tmp_path)
    llm = FakeEmbedder()

    summary = asyncio.run(import_sources([root], llm, session=db))
    assert summary.files == 1
    assert summary.added == 1
    assert summary.errors == 0
    assert summary.summaries == 0
    assert summary.summary_errors == 0

    docs = db.scalars(select(Document)).all()
    assert {(d.source, d.path) for d in docs} == {(NAME, "visible.md")}
    for rel in (".hidden/note.md", ".hidden/data.yaml", ".venv/junk.md"):
        assert not any(d.path == rel for d in docs)
    for texts in llm.calls:  # every embed batch
        for t in texts:
            assert "HIDDEN-MD-CONTENT" not in t and "HIDDEN-YAML-VALUE" not in t
            assert "EXCLUDED-CONTENT" not in t
    # The hidden yaml never reached the lite model.
    assert not llm.chat_calls
    _reset(db)


def test_hidden_paths_indexed_when_flag_on(db: Session, tmp_path: Path) -> None:
    # A1: with the map, hidden files are indexed, embedded, and
    # summarized exactly like visible files — EXCLUDED_DIRS stay out.
    _reset(db)
    root = _tree(tmp_path)
    llm = FakeEmbedder()

    summary = asyncio.run(
        import_sources(
            [root], llm, session=db, include_hidden_by_root={str(root): True}
        )
    )
    assert summary.files == 3
    assert summary.added == 3
    assert summary.errors == 0
    assert summary.summary_errors == 0

    docs = db.scalars(select(Document)).all()
    assert {(d.source, d.path) for d in docs} == {
        (NAME, "visible.md"),
        (NAME, ".hidden/note.md"),
        (NAME, ".hidden/data.yaml"),
    }
    # EXCLUDED_DIRS is excluded in BOTH states (A1).
    assert not any(d.path == ".venv/junk.md" for d in docs)
    chunks = db.scalars(select(Chunk)).all()
    assert not any("EXCLUDED-CONTENT" in c.content for c in chunks)
    # The hidden md was embedded like any visible md (no summary — the
    # markdown path skips the lite model).
    note = db.scalar(select(Document).where(Document.path == ".hidden/note.md"))
    assert note is not None and note.summary is None
    assert any("HIDDEN-MD-CONTENT" in c.content for c in chunks)
    # The hidden yaml went through the FULL non-markdown path (phase 30):
    # a stored summary plus one embedded is_summary chunk on top of the
    # content chunks.
    yaml_doc = db.scalar(select(Document).where(Document.path == ".hidden/data.yaml"))
    assert yaml_doc is not None and yaml_doc.summary is not None
    yaml_chunks = [
        c for c in chunks if c.document_id == yaml_doc.id
    ]
    assert any(c.is_summary for c in yaml_chunks)
    assert any(not c.is_summary for c in yaml_chunks)
    assert any("HIDDEN-YAML-VALUE" in c.content for c in yaml_chunks)
    # Only the yaml reached the lite model.
    assert summary.summaries == 1
    assert len(llm.chat_calls) == 1
    user = next(m["content"] for m in llm.chat_calls[0] if m["role"] == "user")
    assert "HIDDEN-YAML-VALUE" in user
    _reset(db)


def test_flag_off_re_run_prunes_hidden_docs(db: Session, tmp_path: Path) -> None:
    # A2: the owner flips the flag off — the next prune run walks with
    # the default rules, the hidden files never enter ``seen``, and their
    # rows (summary chunk included) leave the index.
    _reset(db)
    root = _tree(tmp_path)
    llm = FakeEmbedder()

    s1 = asyncio.run(
        import_sources(
            [root], llm, session=db, include_hidden_by_root={str(root): True}
        )
    )
    assert (s1.files, s1.added) == (3, 3)
    assert (
        db.scalar(select(Document).where(Document.path == ".hidden/note.md"))
        is not None
    )

    s2 = asyncio.run(import_sources([root], llm, session=db, prune=True))
    assert s2.files == 1
    assert s2.unchanged == 1  # visible.md untouched
    assert s2.pruned == 2  # both hidden documents
    assert (
        db.scalar(select(Document).where(Document.path == ".hidden/note.md"))
        is None
    )
    assert (
        db.scalar(select(Document).where(Document.path == ".hidden/data.yaml"))
        is None
    )
    assert (
        db.scalar(select(Document).where(Document.path == "visible.md")) is not None
    )
    # The summary chunk rows went with their documents (cascade).
    assert not any(
        "HIDDEN-YAML-VALUE" in c.content for c in db.scalars(select(Chunk)).all()
    )
    _reset(db)


def test_progress_total_agrees_with_walk_in_both_states(
    db: Session, tmp_path: Path
) -> None:
    # The phase-64 pre-walk uses the same per-root flag as the loop, so
    # ``total`` agrees with the walk in both states.
    _reset(db)
    root = _tree(tmp_path)
    llm = FakeEmbedder()
    on_totals: set[int] = set()

    def on_progress(source: str, rel: str, done: int, total: int) -> None:
        on_totals.add(total)

    s_on = asyncio.run(
        import_sources(
            [root],
            llm,
            session=db,
            progress=on_progress,
            include_hidden_by_root={str(root): True},
        )
    )
    assert s_on.files == 3
    assert on_totals == {3}  # hidden files counted in the denominator

    _reset(db)
    llm2 = FakeEmbedder()
    off_totals: set[int] = set()

    def off_progress(source: str, rel: str, done: int, total: int) -> None:
        off_totals.add(total)

    s_off = asyncio.run(
        import_sources([root], llm2, session=db, progress=off_progress)
    )
    assert s_off.files == 1
    assert off_totals == {1}  # visible only, the pre-phase-105 count
    _reset(db)


def test_unlisted_root_stays_hidden(db: Session, tmp_path: Path) -> None:
    # The map is per-root, not global: listing one root as True leaves
    # the other root exactly as pre-phase-105.
    _reset(db)
    root_a = _tree(tmp_path, name="HiddenFixA")
    root_b = _tree(tmp_path, name="HiddenFixB")
    llm = FakeEmbedder()

    summary = asyncio.run(
        import_sources(
            [root_a, root_b],
            llm,
            session=db,
            include_hidden_by_root={str(root_a): True},
        )
    )
    # A: 3 (flag on) · B: 1 (unlisted → False)
    assert summary.files == 4
    docs = db.scalars(select(Document)).all()
    assert {(d.source, d.path) for d in docs} == {
        ("HiddenFixA", "visible.md"),
        ("HiddenFixA", ".hidden/note.md"),
        ("HiddenFixA", ".hidden/data.yaml"),
        ("HiddenFixB", "visible.md"),
    }
    assert not any(
        d.source == "HiddenFixB" and d.path == ".hidden/note.md" for d in docs
    )
    _reset(db)