brain-of-reese/tests/integration/test_docs_api.py

"""Integration tests: GET /api/docs — empty shape + populated shape —
and GET /api/docs/tree (phase 97, task 02 — the full recursive KB tree
the RAG view renders: shape, ordering, counts, summaries, the 403 gate,
the superset rule, and the stat-walk equivalence with ``GET /api/docs``).

Also the BYTES routes and their content-payload affordances: the
phase-122 (task 04) image route (``GET /api/documents/{id}/image``
— exact bytes, the extension's Content-Type, the 404 matrix, public
per phase 124 A4) and the phase-148 (task 02) PDF page route
(``GET /api/documents/{id}/pages/{n}`` — the same trust model and 404
posture for a page-qualifying document's rendered page PNGs) + the
content endpoint's ``image_url`` / ``page_urls`` keys (present for the
qualifying row, ABSENT otherwise — never ``null``), plus the phase-163
(task 01) ``page_strip_mode`` class signal (``"primary"`` /``"additive"``
for a stamped strip, ABSENT for a pre-phase row and for every
non-qualifying document).

Also the phase-167 (task 03) per-level drill route
``GET /api/docs/tree/node`` (the node shape, the 403/404 matrix, the
pin that its children equal the same level of the unlimited
``/api/docs/tree?depth=8`` response) and the ETag revalidation of the
three catalogue reads (200 + validator, conditional 304 with no body,
the admin-PATCH bust — the LOCKED A4 proof — coexisting with gzip).

Uses the real compose Postgres (``db`` fixture) and FastAPI's TestClient.
"""
from __future__ import annotations

import asyncio
import base64
import gzip
import hashlib
import inspect
import itertools
import logging
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import delete, func, select, text

import app.api.docs as docs_api
import app.rag.importer as rag_importer
from app.config import Settings
from app.core import tokens as token_service
from app.core.security_headers import CSP
from app.main import app as fastapi_app
from app.models import Chunk, Document, FolderSummary, GitSource, SourcesMeta
from app.rag import git_sources as rag_git_sources
from app.rag.folder_summaries import missing_folder_summaries
from app.rag.importer import import_sources
from app.rag.llm import LLMError
from app.rag.summarizer import DESCRIBE_PROMPT
from tests.fakes import FakeEmbedder

_TREE_TABLES = "chunks, documents, folder_summaries, git_sources"


def _truncate_tree_tables(db) -> None:
    db.execute(text(f"TRUNCATE {_TREE_TABLES}"))
    db.commit()


_counter = itertools.count()


def _seed_doc(db, source: str, path: str, title: str, n_chunks: int, indexed_at: datetime) -> None:
    """One indexed document with *n_chunks* content chunks (unique hash)."""
    doc = Document(
        source=source,
        path=path,
        full_path=f"/tmp/{source}/{path}",
        title=title,
        content=f"# {title}\n\nBody.",
        content_hash=f"{next(_counter):064d}",
        indexed_at=indexed_at,
    )
    db.add(doc)
    db.flush()
    db.add_all(
        Chunk(document_id=doc.id, position=i, content=f"chunk {i}", embedding=[0.01] * 768)
        for i in range(n_chunks)
    )


def _tree_file_nodes(sources) -> list[dict]:
    """Every file node of a tree response, walked recursively."""
    files: list[dict] = []
    for source in sources:
        for child in source["children"]:
            if child["kind"] == "file":
                files.append(child)
            else:
                files.extend(_tree_file_nodes([child]))
    return files


def _tree_pending_keys(sources) -> set[tuple[str, str]]:
    """Every ``(source, folder_path)`` flagged ``summary_pending`` in a
    tree response — the SOURCE root rides ``folder_path = ""`` (the
    ``folder_summaries`` convention); walked recursively over the whole
    tree (the D3 set the cross-check compares against
    :func:`missing_folder_summaries`)."""
    keys: set[tuple[str, str]] = set()

    def _walk(source_name: str, node: dict) -> None:
        if node.get("summary_pending"):
            keys.add((source_name, node["path"] if node.get("kind") == "folder" else ""))
        for child in node.get("children", ()):
            _walk(source_name, child)

    for source in sources:
        _walk(source["name"], source)
    return keys


def test_docs_empty_shape(admin_client, db) -> None:
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    r = admin_client.get("/api/docs")
    assert r.status_code == 200
    assert r.json() == {"documents": []}


def test_docs_populated_shape_sorted_with_chunk_counts(admin_client, db) -> None:
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    now = datetime.now(UTC)
    k8s = Document(
        source="Homelab",
        path="kubernetes.md",
        full_path="/tmp/kubernetes.md",
        title="Kubernetes Homelab Cluster",
        content="# Kubernetes Homelab Cluster\n\nTalos on 3 nodes.",
        content_hash="a" * 64,
        indexed_at=now,
    )
    empty = Document(
        source="Deployments",
        path="empty.md",
        full_path="/tmp/empty.md",
        title="No Chunks Yet",
        content="two-phase: doc exists, embeddings pending",
        content_hash="b" * 64,
        indexed_at=now,
    )
    db.add_all([empty, k8s])
    db.flush()
    db.add_all(
        Chunk(document_id=k8s.id, position=i, content=f"chunk {i}", embedding=[0.01] * 768)
        for i in range(3)
    )
    db.commit()

    r = admin_client.get("/api/docs")
    assert r.status_code == 200
    body = r.json()
    # Ordered by (source, path): Deployments < Homelab.
    assert [d["path"] for d in body["documents"]] == ["empty.md", "kubernetes.md"]
    by_path = {d["path"]: d for d in body["documents"]}

    k = by_path["kubernetes.md"]
    assert k["source"] == "Homelab"
    assert k["title"] == "Kubernetes Homelab Cluster"
    assert k["chunks"] == 3
    datetime.fromisoformat(k["indexed_at"])  # raises if not valid ISO-8601
    uuid.UUID(k["id"])  # raises if not a valid UUID
    assert by_path["empty.md"]["chunks"] == 0  # outerjoin → zero, not missing

    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


def test_docs_response_matches_schema_shape(admin_client, db) -> None:
    r = admin_client.get("/api/docs")
    assert r.status_code == 200
    body = r.json()
    assert set(body) == {"documents"}
    for d in body["documents"]:
        # Wire-additive (phase 106, task 05): the pre-date keys are all
        # still there, joined by ``created_at`` (the document's creation
        # date — the RAG view's ``Created`` column).
        assert set(d) == {
            "id", "source", "path", "title", "chunks", "created_at", "indexed_at"
        }
        assert isinstance(d["chunks"], int) and d["chunks"] >= 0
        datetime.fromisoformat(d["created_at"])  # raises if not ISO-8601


# --------------------------------------------------------------------
# GET /api/docs/tree (phase 97, task 02).
# --------------------------------------------------------------------


def test_docs_tree_populated_shape_order_counts_summaries(admin_client, db) -> None:
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    # Registry order (added_at) is Homelab → Deployments — deliberately
    # NOT alphabetical (the registry order leads, the phase-97 rule).
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    db.add(
        GitSource(
            url="https://github.com/reese/Deployments.git",
            kind="git",
            added_at=base + timedelta(hours=1),
        )
    )
    _seed_doc(db, "Homelab", "k8s/talos.md", "Talos", 3, base)
    _seed_doc(db, "Homelab", "k8s/cluster.md", "Cluster", 2, base)
    _seed_doc(db, "Homelab", "k8s/helm/charts.md", "Charts", 1, base)
    _seed_doc(db, "Homelab", "root-note.md", "Root note", 4, base)
    _seed_doc(db, "Deployments", "deploy-a.md", "Deploy A", 1, base)
    _seed_doc(db, "Deployments", "deploy-b.md", "Deploy B", 0, base)
    db.add_all(
        [
            FolderSummary(source="Homelab", folder_path="", summary="Homelab docs."),
            FolderSummary(source="Homelab", folder_path="k8s", summary="K8s stuff."),
            # A stored row for a folder with NO indexed descendants —
            # e.g. a manual row surviving the prune below the 2-doc
            # minimum (phase 97, task 01): the tree's folders come from
            # INDEXED paths (the existence rule), not from summary rows.
            FolderSummary(source="Deployments", folder_path="orphan", summary="Ghost."),
        ]
    )
    db.commit()

    # Phase 167 (task 02): this shape pin walks the FULL tree, so it
    # asks for the whole build explicitly — the endpoint's default is
    # the view's first paint (depth=1), depth=8 is the entire tree.
    r = admin_client.get("/api/docs/tree?depth=8")
    assert r.status_code == 200
    body = r.json()
    # ``totals`` joins the top-level keys (phase 167, additive).
    assert set(body) == {"sources", "totals"}
    assert body["totals"] == {
        "documents": 6,
        "chunks": 11,
        "last_indexed": base.isoformat(),
    }
    sources = body["sources"]
    assert [s["name"] for s in sources] == ["Homelab", "Deployments"]

    homelab, deployments = sources
    # Wire-additive (phase 98, task 03): the pre-pending keys are all
    # still there, joined by ``summary_pending`` — and (phase 106,
    # task 05) by ``updated_at`` (the subtree's max document date, D9).
    assert set(homelab) == {
        "name", "documents", "updated_at", "summary", "summary_pending", "children"
    }
    assert homelab["documents"] == 4  # the whole recursive count
    assert homelab["summary"] == "Homelab docs."  # the (source, "") row
    assert homelab["summary_pending"] is False  # the stored root row covers it
    assert deployments["summary"] is None  # no stored root row
    assert deployments["documents"] == 2

    # Homelab: subfolder first (path order among folders), then the
    # root files; the nested folder k8s/helm recurses one level deeper.
    k8s, root_note = homelab["children"]
    assert k8s["kind"] == "folder"
    assert k8s["path"] == "k8s"
    assert k8s["documents"] == 3  # talos + cluster + charts (subtree)
    assert k8s["summary"] == "K8s stuff."
    helm, cluster, talos = k8s["children"]
    assert (helm["kind"], helm["path"], helm["documents"], helm["summary"]) == (
        "folder",
        "k8s/helm",
        1,
        None,
    )
    assert [
        (c["kind"], c["path"], c["title"], c["chunks"])
        for c in (cluster, talos)
    ] == [("file", "k8s/cluster.md", "Cluster", 2), ("file", "k8s/talos.md", "Talos", 3)]
    chart = helm["children"][0]
    assert (chart["kind"], chart["path"], chart["title"], chart["chunks"]) == (
        "file",
        "k8s/helm/charts.md",
        "Charts",
        1,
    )
    datetime.fromisoformat(chart["indexed_at"])  # valid ISO-8601
    assert (root_note["kind"], root_note["path"], root_note["title"], root_note["chunks"]) == (
        "file",
        "root-note.md",
        "Root note",
        4,
    )

    # Deployments: flat — direct files in catalog order, and the
    # ``orphan`` summary row does NOT create a folder node.
    assert [(c["kind"], c["path"]) for c in deployments["children"]] == [
        ("file", "deploy-a.md"),
        ("file", "deploy-b.md"),
    ]

    _truncate_tree_tables(db)


def test_docs_tree_403_anonymous(client, db) -> None:
    """Admin-only, like ``GET /api/docs``: anonymous → 403
    ``admin only`` (the RAG view's anonymous gate never fetches the
    tree)."""
    _truncate_tree_tables(db)
    r = client.get("/api/docs/tree")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}


def test_docs_tree_empty_registry_and_catalog(
    admin_client, db, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Nothing registered, nothing indexed → ``{"sources": []}``.

    "Empty registry" means the ``git_sources`` table AND the
    ``BOR_GIT_SOURCES`` env fallback are both empty — the dev ``.env``
    names a source, so the resolver's ``get_settings`` is patched with
    a fresh ``Settings(_env_file=None, git_sources="")`` (the
    ``test_sync_api`` / ``test_git_sources_api`` pattern: the dev
    ``.env`` never leaks in)."""
    monkeypatch.setattr(
        rag_git_sources,
        "get_settings",
        lambda: Settings(_env_file=None, git_sources=""),  # pyright: ignore[reportCallIssue]
    )
    _truncate_tree_tables(db)
    r = admin_client.get("/api/docs/tree")
    assert r.status_code == 200
    # Phase 167 (task 02): ``totals`` rides even the empty tree —
    # 0 documents, 0 chunks, no last-indexed stamp.
    assert r.json() == {
        "sources": [],
        "totals": {"documents": 0, "chunks": 0, "last_indexed": None},
    }


def test_docs_tree_indexed_only_source_after_registered(admin_client, db) -> None:
    """The superset rule: a registered 0-document source still lists
    (first — the registry leads), and indexed-only sources (documents
    whose source is not in ``git_sources`` — ad-hoc imports, removed
    but not-yet-pruned sources) trail in alphabetical order."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Alpha.git", kind="git", added_at=base))
    _seed_doc(db, "Zeta", "z1.md", "Z1", 1, base)
    _seed_doc(db, "Midx", "m1.md", "M1", 2, base)
    db.commit()

    r = admin_client.get("/api/docs/tree")
    assert r.status_code == 200
    sources = r.json()["sources"]
    assert [s["name"] for s in sources] == ["Alpha", "Midx", "Zeta"]
    alpha, midx, zeta = sources
    assert (alpha["documents"], alpha["children"], alpha["summary"]) == (0, [], None)
    assert alpha["summary_pending"] is False  # 0 documents — never pending
    assert midx["documents"] == 1
    assert midx["summary_pending"] is True  # 1 document — the ≥ 1 minimum, no row
    assert zeta["documents"] == 1
    assert zeta["summary_pending"] is True  # 1 document — the ≥ 1 minimum, no row
    assert [c["path"] for c in midx["children"]] == ["m1.md"]

    _truncate_tree_tables(db)


def test_docs_tree_summary_pending_on_source_and_folder_nodes(admin_client, db) -> None:
    """The endpoint returns ``summary_pending`` on SOURCE + FOLDER
    nodes (phase 98, task 03 — D3): true iff the recursive count is
    ≥ 1 AND no stored row — the NESTED 1-doc folder included (the ≥ 1
    rule); false WITH a stored row (any — the endpoint cannot tell AI
    from manual). File nodes carry no flag."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    # k8s → 3 documents (talos + cluster + charts), NO stored row → pending
    _seed_doc(db, "Homelab", "k8s/talos.md", "Talos", 1, base)
    _seed_doc(db, "Homelab", "k8s/cluster.md", "Cluster", 1, base)
    # k8s/helm → 1 document — a candidate at the ≥ 1 rule (no row)
    _seed_doc(db, "Homelab", "k8s/helm/charts.md", "Charts", 1, base)
    # wiki → 2 documents, WITH a stored row → not pending
    _seed_doc(db, "Homelab", "wiki/one.md", "One", 1, base)
    _seed_doc(db, "Homelab", "wiki/two.md", "Two", 1, base)
    db.add_all(
        [
            # The 5-doc source root IS covered → the source node is not pending
            FolderSummary(source="Homelab", folder_path="", summary="Homelab docs."),
            FolderSummary(source="Homelab", folder_path="wiki", summary="Wiki pages."),
        ]
    )
    db.commit()

    # Phase 167 (task 02): the nested ``k8s/helm`` node needs the full
    # build — depth=8 keeps this pin walking every level (the default
    # is the view's first paint only).
    r = admin_client.get("/api/docs/tree?depth=8")
    assert r.status_code == 200
    (homelab,) = r.json()["sources"]
    # Phase 106 (task 05): ``updated_at`` joins the source node keys.
    assert set(homelab) == {
        "name", "documents", "updated_at", "summary", "summary_pending", "children"
    }
    assert homelab["summary"] == "Homelab docs."
    assert homelab["summary_pending"] is False
    # Direct subfolders in path order: k8s < wiki.
    k8s, wiki = [c for c in homelab["children"] if c["kind"] == "folder"]
    assert k8s["path"] == "k8s"
    assert k8s["summary"] is None
    assert k8s["summary_pending"] is True  # 3 docs, no stored row
    helm = k8s["children"][0]
    assert (helm["kind"], helm["path"]) == ("folder", "k8s/helm")
    assert helm["summary_pending"] is True  # 1 doc — pending at the ≥ 1 rule
    assert wiki["summary"] == "Wiki pages."
    assert wiki["summary_pending"] is False  # 2 docs, but a stored row covers it
    # File nodes carry no pending flag at all (the file table has no
    # description column — D3's file exclusion).
    for node in (k8s, wiki, helm):
        for child in node["children"]:
            if child["kind"] == "file":
                assert "summary_pending" not in child

    _truncate_tree_tables(db)


def test_docs_tree_pending_set_equals_missing_folder_summaries(admin_client, db) -> None:
    """The D3 cross-check (ONE concept end to end): with a PARTIAL
    summary table (some rows deleted — the phase-96 gap-fill pattern),
    the set of ``(source, folder_path)`` flagged pending in the fetched
    tree (source root = ``""``) equals
    :func:`missing_folder_summaries` — the marker can never drift from
    the gap-fill."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Alpha.git", kind="git", added_at=base))
    db.add(
        GitSource(
            url="https://github.com/reese/Beta.git",
            kind="git",
            added_at=base + timedelta(hours=1),
        )
    )
    db.add(
        GitSource(
            url="https://github.com/reese/Gamma.git",
            kind="git",
            added_at=base + timedelta(hours=2),
        )
    )
    # Alpha: a/b holds 2 docs, c holds 1 (a candidate at the ≥ 1
    # rule), the root holds 4 — candidates (Alpha, ""), (Alpha, "a"),
    # (Alpha, "a/b"), (Alpha, "c").
    _seed_doc(db, "Alpha", "a/b/c1.md", "C1", 1, base)
    _seed_doc(db, "Alpha", "a/b/c2.md", "C2", 1, base)
    _seed_doc(db, "Alpha", "c/solo.md", "Solo", 1, base)
    _seed_doc(db, "Alpha", "top.md", "Top", 1, base)
    # Beta: x holds 2 docs, the root holds 2 — candidates
    # (Beta, ""), (Beta, "x").
    _seed_doc(db, "Beta", "x/one.md", "One", 1, base)
    _seed_doc(db, "Beta", "x/two.md", "Two", 1, base)
    # Gamma: registered, 0 documents — no candidates at all.
    # Seed every candidate row, then DELETE two of them (the phase-96
    # direct-row-deletion pattern — a fail-soft miss / cleared row).
    db.add_all(
        FolderSummary(source=s, folder_path=f, summary=t)
        for (s, f), t in {
            ("Alpha", ""): "Alpha root.",
            ("Alpha", "a"): "Alpha a.",
            ("Alpha", "a/b"): "Alpha a b.",
            ("Alpha", "c"): "Alpha c.",
            ("Beta", ""): "Beta root.",
            ("Beta", "x"): "Beta x.",
        }.items()
    )
    db.commit()
    db.execute(
        delete(FolderSummary).where(
            FolderSummary.source == "Alpha", FolderSummary.folder_path == "a"
        )
    )
    db.execute(
        delete(FolderSummary).where(
            FolderSummary.source == "Beta", FolderSummary.folder_path == ""
        )
    )
    db.commit()

    # Phase 167 (task 02): the cross-check walks EVERY node, so it
    # requests the full tree (the default depth ships only the view's
    # first level — truncated nodes still carry their own markers).
    r = admin_client.get("/api/docs/tree?depth=8")
    assert r.status_code == 200
    pending = _tree_pending_keys(r.json()["sources"])
    # THE cross-check: the marker set IS the gap-fill's candidate set.
    assert pending == set(missing_folder_summaries(db))
    # And the explicit expectation (the test is readable without the
    # helper): exactly the two deleted keys, root riding "".
    assert pending == {("Alpha", "a"), ("Beta", "")}
    # Never flagged: the 0-document registered source (Gamma) and every
    # node that still holds a row — incl. the 1-doc Alpha/c, whose
    # stored row covers it (the ≥ 1 rule makes it a candidate, but a
    # candidate WITH a row is not missing).
    assert ("Alpha", "c") not in pending
    assert not any(name == "Gamma" for name, _ in pending)
    assert ("Alpha", "a/b") not in pending
    assert ("Beta", "x") not in pending
    assert ("Alpha", "") not in pending

    _truncate_tree_tables(db)


# --------------------------------------------------------------------
# PATCH /api/folders/summary (phase 97, task 03) — the admin
# folder-description editor: update / create / source-root / clear /
# double-clear, the 404s, the 403s, and the round-trip through
# ``GET /api/docs/tree`` (task 02).
# --------------------------------------------------------------------

#: A deliberately old stamp: a save must ADVANCE ``updated_at`` past it.
OLDER_STAMP = datetime(2020, 1, 1, tzinfo=UTC)


def _seed_folder_pair(db, source: str, folder: str, base: datetime) -> None:
    """Two documents under ``source/folder`` — a summarizable folder
    (≥ 1 doc at the current minimum; two keep the pair realistic)."""
    _seed_doc(db, source, f"{folder}/one.md", f"{folder} one", 1, base)
    _seed_doc(db, source, f"{folder}/two.md", f"{folder} two", 2, base)


def _get_folder_row(db, source: str, folder: str) -> FolderSummary | None:
    return db.scalar(
        select(FolderSummary).where(
            FolderSummary.source == source, FolderSummary.folder_path == folder
        )
    )


def test_folder_summary_update_ai_row_flips_manual_and_advances_stamp(
    admin_client: TestClient, db
) -> None:
    """Update an AI-written row: the text is replaced, ``manually_edited``
    flips to true, ``updated_at`` advances past the seeded stamp — and a
    second save with different text updates IN PLACE (same PK, one
    row). The round-trip: ``GET /api/docs/tree`` shows the new text on
    the folder node."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_folder_pair(db, "Homelab", "k8s", base)
    db.add(
        FolderSummary(
            source="Homelab",
            folder_path="k8s",
            summary="AI text.",
            manually_edited=False,
            updated_at=OLDER_STAMP,
        )
    )
    db.commit()
    db.expire_all()

    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "k8s", "summary": "  Owner words.  "},
    )
    assert r.status_code == 200, r.text
    assert set(r.json()) == {"source", "folder_path", "summary"}
    assert r.json() == {"source": "Homelab", "folder_path": "k8s", "summary": "Owner words."}

    db.expire_all()
    row = _get_folder_row(db, "Homelab", "k8s")
    assert row is not None
    assert row.summary == "Owner words."  # stripped before storing
    assert row.manually_edited is True  # a manual row from this save on
    assert row.updated_at > OLDER_STAMP  # fresh stamp, not the seeded one

    # The task-02 tree shows the owner's text on the folder node.
    tree = admin_client.get("/api/docs/tree").json()
    k8s = next(c for c in tree["sources"][0]["children"] if c["kind"] == "folder")
    assert (k8s["path"], k8s["summary"]) == ("k8s", "Owner words.")

    # A second save updates in place — same row, no second PK.
    r2 = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "k8s", "summary": "Owner words v2."},
    )
    assert r2.status_code == 200, r2.text
    assert r2.json()["summary"] == "Owner words v2."
    db.expire_all()
    assert db.scalar(select(func.count()).select_from(FolderSummary)) == 1  # one row
    row = _get_folder_row(db, "Homelab", "k8s")
    assert row is not None
    assert row.summary == "Owner words v2."
    assert row.manually_edited is True

    _truncate_tree_tables(db)


def test_folder_summary_create_where_no_row_exists(
    admin_client: TestClient, db
) -> None:
    """A manual description can be CREATED where no row exists — a
    folder the generator has not written yet (its fail-soft miss, or
    a pre-sync state): insert, not update, with
    ``manually_edited = true``."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_doc(db, "Homelab", "solo/only.md", "Only", 1, base)  # 1-doc folder
    db.commit()
    db.expire_all()
    # No AI row yet (the generator has not run in this test).
    assert _get_folder_row(db, "Homelab", "solo") is None

    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "solo", "summary": "One-off scripts."},
    )
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "Homelab",
        "folder_path": "solo",
        "summary": "One-off scripts.",
    }
    db.expire_all()
    row = _get_folder_row(db, "Homelab", "solo")
    assert row is not None
    assert row.summary == "One-off scripts."
    assert row.manually_edited is True

    _truncate_tree_tables(db)


def test_folder_summary_root_save_on_registered_zero_doc_source(
    admin_client: TestClient, db
) -> None:
    """``folder_path = ""`` is valid for ANY allowed source — a registered
    0-document source (nothing indexed yet) still accepts a root
    description: the source check is registered-OR-indexed."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Empty.git", kind="git", added_at=base))
    db.commit()

    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Empty", "folder_path": "", "summary": "Docs incoming."},
    )
    assert r.status_code == 200, r.text
    assert r.json() == {"source": "Empty", "folder_path": "", "summary": "Docs incoming."}
    db.expire_all()
    row = _get_folder_row(db, "Empty", "")
    assert row is not None
    assert row.manually_edited is True

    _truncate_tree_tables(db)


def test_folder_summary_source_root_round_trips_through_tree(
    admin_client: TestClient, db
) -> None:
    """The source root (``folder_path: ""``) round-trips through the
    task-02 tree endpoint: save → the source node carries the text;
    clear → it is null again."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_doc(db, "Homelab", "a/b.md", "B", 1, base)
    _seed_doc(db, "Homelab", "c.md", "C", 1, base)
    db.commit()

    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "", "summary": "All the homelab docs."},
    )
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "Homelab",
        "folder_path": "",
        "summary": "All the homelab docs.",
    }
    tree = admin_client.get("/api/docs/tree").json()
    assert tree["sources"][0]["summary"] == "All the homelab docs."
    db.expire_all()
    assert _get_folder_row(db, "Homelab", "") is not None

    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "", "summary": "   "},
    )
    assert r.status_code == 200, r.text
    assert r.json() == {"source": "Homelab", "folder_path": "", "summary": None}
    tree = admin_client.get("/api/docs/tree").json()
    assert tree["sources"][0]["summary"] is None

    _truncate_tree_tables(db)


def test_folder_summary_clear_deletes_row_and_double_clear_is_noop(
    admin_client: TestClient, db
) -> None:
    """Empty/whitespace clears: the row is deleted — AI-written OR
    manual, either way it is gone (the next KB-changing sync regenerates
    an AI row: the reset path) — and the response ``summary`` is null.
    A second clear with no row is a 200 no-op. The task-02 tree shows
    null on both folder nodes after the clears."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_folder_pair(db, "Homelab", "k8s", base)
    _seed_folder_pair(db, "Homelab", "manual", base)
    db.add(
        FolderSummary(
            source="Homelab",
            folder_path="k8s",
            summary="AI text.",
            manually_edited=False,
            updated_at=OLDER_STAMP,
        )
    )
    db.add(
        FolderSummary(
            source="Homelab",
            folder_path="manual",
            summary="Owner text.",
            manually_edited=True,
            updated_at=OLDER_STAMP,
        )
    )
    db.commit()
    db.expire_all()

    for folder, blank in (("k8s", "  "), ("manual", "")):  # whitespace, then empty
        r = admin_client.patch(
            "/api/folders/summary",
            json={"source": "Homelab", "folder_path": folder, "summary": blank},
        )
        assert r.status_code == 200, r.text
        assert r.json() == {"source": "Homelab", "folder_path": folder, "summary": None}
    db.expire_all()
    assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0  # both rows gone

    # A second clear (no row) is a 200 no-op.
    r = admin_client.patch(
        "/api/folders/summary",
        json={"source": "Homelab", "folder_path": "k8s", "summary": ""},
    )
    assert r.status_code == 200, r.text
    assert r.json() == {"source": "Homelab", "folder_path": "k8s", "summary": None}
    db.expire_all()
    assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0  # still nothing

    # The task-02 tree shows null on both folder nodes after the clears.
    tree = admin_client.get("/api/docs/tree").json()
    nodes = {
        c["path"]: c for c in tree["sources"][0]["children"] if c["kind"] == "folder"
    }
    assert nodes["k8s"]["summary"] is None
    assert nodes["manual"]["summary"] is None

    _truncate_tree_tables(db)


def test_folder_summary_404_unknown_source_folder_and_traversal(
    admin_client: TestClient, db
) -> None:
    """404s, in check order: an unknown source (neither registered nor
    indexed) wins over the folder check — including for a folder that
    EXISTS under another source; an unknown folder, a traversal folder
    path, and a file merely sharing a folder's name (no indexed
    descendant) are all ``folder not found``. Nothing is written."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_folder_pair(db, "Homelab", "real", base)
    _seed_doc(db, "Homelab", "note", "A file named like a folder", 1, base)
    db.commit()

    cases = (
        # (source, folder_path, detail)
        ("Ghost", "", "source not found"),  # unknown source, root
        ("Ghost", "real", "source not found"),  # folder exists ELSEWHERE — source wins
        ("Homelab", "nope", "folder not found"),  # unknown folder
        ("Homelab", "../../etc", "folder not found"),  # traversal: no prefix match
        ("Homelab", "note", "folder not found"),  # file named like a folder, no descendants
        ("Homelab", "real/", "folder not found"),  # trailing slash: strict prefix rule
    )
    for source, folder, detail in cases:
        r = admin_client.patch(
            "/api/folders/summary",
            json={"source": source, "folder_path": folder, "summary": "whatever"},
        )
        assert r.status_code == 404, (source, folder, r.status_code)
        assert r.json() == {"detail": detail}, (source, folder)

    db.expire_all()
    assert db.scalar(select(func.count()).select_from(FolderSummary)) == 0  # nothing written

    _truncate_tree_tables(db)


def test_folder_summary_403_anonymous_and_token_user(
    client: TestClient, db
) -> None:
    """The ``require_admin`` gate: an anonymous caller AND a live
    access-token user who is not the admin (phase 79 token users exist)
    both get 403 ``admin only`` — the endpoint gate is the API-level
    defense in depth, not the RAG view's render gate."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_folder_pair(db, "Homelab", "k8s", base)
    db.execute(text("TRUNCATE api_tokens"))
    db.commit()
    body = {"source": "Homelab", "folder_path": "k8s", "summary": "x"}
    try:
        # Anonymous (the shared ``client`` is unsigned in this module).
        r = client.patch("/api/folders/summary", json=body)
        assert r.status_code == 403
        assert r.json() == {"detail": "admin only"}

        # A live access-token user who is not the admin (phase 79):
        # signed in via the public token login, the row is active.
        _row, plaintext = token_service.create_token(db, "pin-holder")
        db.commit()
        holder = TestClient(fastapi_app)
        s = holder.post("/api/token-auth", json={"token": plaintext})
        assert s.status_code == 204, s.text
        r = holder.patch("/api/folders/summary", json=body)
        assert r.status_code == 403
        assert r.json() == {"detail": "admin only"}
    finally:
        db.execute(text("TRUNCATE api_tokens"))
        db.commit()
        _truncate_tree_tables(db)


def test_folder_summary_patch_never_constructs_an_llm_client() -> None:
    """Source pin (the house pattern): a folder description is NEVER
    embedded — no chunk, no retrieval role beyond the ``ls`` line — so
    the PATCH handler must never touch the LLM client (the deliberate
    contrast with the phase-57 document-summary re-embed)."""
    src = inspect.getsource(docs_api.update_folder_summary)
    assert "LLMClient" not in src


def test_docs_tree_stat_walk_equivalence_with_flat_list(admin_client, db) -> None:
    """The RAG view computes its KB-wide stat cards by walking the
    in-memory tree — the walk must yield EXACTLY what
    ``GET /api/docs`` reports for the same data (the stat-card values
    are unchanged by the redesign)."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_doc(db, "Homelab", "a/b/c.md", "C", 3, base)
    _seed_doc(db, "Homelab", "a/d.md", "D", 2, base)
    _seed_doc(db, "Homelab", "top.md", "Top", 5, base)
    _seed_doc(db, "Homelab", "solo.md", "Solo", 0, base)
    db.commit()

    # Phase 167 (task 02): the walk needs EVERY file node, so it asks
    # for the full tree; the server ``totals`` must equal the same walk
    # over the flat list (the stat cards cannot drift either way).
    tree = admin_client.get("/api/docs/tree?depth=8").json()
    flat = admin_client.get("/api/docs").json()
    assert tree["totals"]["documents"] == len(flat["documents"])
    assert tree["totals"]["chunks"] == sum(d["chunks"] for d in flat["documents"])
    assert tree["totals"]["last_indexed"] == max(d["indexed_at"] for d in flat["documents"])

    files = _tree_file_nodes(tree["sources"])
    assert len(files) == len(flat["documents"])  # document count
    assert sum(f["chunks"] for f in files) == sum(d["chunks"] for d in flat["documents"])
    # The tree's per-source counts match the flat list's per-source counts.
    for source in tree["sources"]:
        flat_count = sum(
            1 for d in flat["documents"] if d["source"] == source["name"]
        )
        assert source["documents"] == flat_count
        source_files = _tree_file_nodes([source])
        assert sum(f["chunks"] for f in source_files) == sum(
            d["chunks"] for d in flat["documents"] if d["source"] == source["name"]
        )

    _truncate_tree_tables(db)

# ---------------------------------------------------------------------------
# Phase 122, task 02 — image ingest end-to-end (the full ``import_sources``
# pipeline against the real DB; task 06 finalizes the phase-122 suite here
# with the image route + content-endpoint shapes).
# ---------------------------------------------------------------------------

#: A real 1×1 transparent PNG — the importer is content-agnostic (it
#: never parses the image), but a well-formed fixture keeps the tests
#: honest about what a real upload looks like.
PNG_1X1 = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAAC0lEQVR4nGP4DwQACfsD/fteaysAAAAASUVORK5CYII="
)


def _image_llm[
    ImageLLM: FakeEmbedder
](tmp_path, llm_cls: type[ImageLLM] = FakeEmbedder, **kwargs) -> ImageLLM:
    """A fake LLM with the phase-122 image knobs (toggle ON by default;
    the image dir defaults under *tmp_path* unless overridden).
    *llm_cls* (task 03) may be the mock vision client (``_MockVisionLLM``
    below) for the no-seam-patch end-to-end path — the PEP 695 type
    parameter keeps the helper's return type honest (``chat_models``).
    """
    kwargs.setdefault("_env_file", None)
    kwargs.setdefault("images", True)
    kwargs.setdefault("image_dir", str(tmp_path / "images"))
    llm = llm_cls()
    llm.settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return llm


def _patch_description(monkeypatch, description) -> None:
    """Pin the task-02 seam (``rag_importer._describe_or_skip``) to
    *description*. Task 03 fills the seam with the CHAT model's vision
    call — these end-to-end mechanics do not change with it. Phase 125
    (task 03) widened the seam with the optional ``context`` kwarg (the
    markdown reference context) — the fake accepts (and ignores) it:
    these fixtures carry no image references, so the seam gets
    ``context=None``."""

    async def _fake(llm, *, data, source, rel, full_path, context=None):
        return description

    monkeypatch.setattr(rag_importer, "_describe_or_skip", _fake)


def _cleanup_source(db, source: str) -> None:
    for doc in db.scalars(select(Document).where(Document.source == source)).all():
        db.delete(doc)
    db.commit()


def test_import_sources_images_on_indexes_image_docs(
    db, tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """``images=True`` end-to-end: a standalone image in a source becomes
    a Document — bytes digested, the persistent copy in ``image_dir``
    (``<doc-id>.png``), ``content`` = the (mock) description, and ONLY
    that text is embedded (the content chunks + the phase-30 summary
    chunk, 768 dims) — while a text doc in the same source stays a
    plain text doc."""
    source_root = tmp_path / "imgsource"
    source_root.mkdir()
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    (source_root / "notes.md").write_text("# Notes\n\nBody text.\n", encoding="utf-8")
    llm = _image_llm(tmp_path)
    _patch_description(monkeypatch, "A network diagram of the homelab VLANs.")
    try:
        summary = asyncio.run(
            import_sources([source_root], llm, session=db, prune=True)
        )
        assert (summary.files, summary.added, summary.images_failed) == (2, 2, 0)
        assert summary.formats == {"md": 1, "png": 1}

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None, "the image file must become a document"
        assert doc.is_image is True and doc.image_path is not None
        assert doc.content == "A network diagram of the homelab VLANs."
        assert doc.title == "diagram"  # the non-markdown stem rule
        copy = Path(doc.image_path)
        assert copy.parent == Path(llm.settings.image_dir)
        assert copy.name == f"{doc.id}.png"
        assert copy.read_bytes() == PNG_1X1

        # The ONLY embedded text is the description (the embedding model
        # never sees pixels): every content chunk carries it, the
        # phase-30 summary chunk exists + is embedded, 768 dims. Task
        # 03: the description IS the summary (stored verbatim — no
        # ``lite`` call, no pointer line), so the summary chunk mirrors
        # ``doc.content`` exactly.
        chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
        content_chunks = [c for c in chunks if not c.is_summary]
        assert [c.content for c in content_chunks] == [doc.content]
        assert doc.summary == doc.content
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
        assert summary_chunks[0].content == doc.content
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768

        # The text doc is untouched by the image machinery.
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "notes.md"
            )
        )
        assert md is not None and md.is_image is False and md.image_path is None
    finally:
        _cleanup_source(db, source_root.name)


class _MockVisionLLM(FakeEmbedder):
    """The phase-122 mock VISION client (task 03): the CHAT model
    answers the multimodal describe call (the image bytes' data URL)
    with a fixed, retrieval-oriented description; text (``lite``) calls
    keep the ``FakeEmbedder`` behaviour. The REAL
    ``rag_importer._describe_or_skip`` → ``summarizer.describe_image``
    chain runs end-to-end against it (no seam patch); every chat
    call's model is recorded (``chat_models``)."""

    DESCRIPTION = (
        "A network diagram of the homelab VLANs: the core switch, the "
        "router, and three labeled subnets."
    )

    def __init__(self) -> None:
        super().__init__()
        self.chat_models: list[str | None] = []

    async def chat(self, messages, model=None):
        self.chat_calls.append(list(messages))
        self.chat_models.append(model)
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            # The phase-122 describe call — the multimodal message.
            return self.DESCRIPTION
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


def test_import_sources_mock_vision_client_end_to_end(db, tmp_path) -> None:
    """Task 03 end-to-end (NO seam patch): a fixture PNG through the
    mock vision client — the real ``_describe_or_skip`` →
    ``describe_image`` → CHAT-model call — yields a doc whose
    ``content`` == ``summary`` == the description, with its
    ``is_summary`` chunk embedded, and whose ONLY embedded text is that
    description (the embedding model never sees pixels)."""
    source_root = tmp_path / "visione2e"
    source_root.mkdir()
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    llm = _image_llm(tmp_path, _MockVisionLLM)
    try:
        summary = asyncio.run(
            import_sources([source_root], llm, session=db)
        )
        assert (summary.files, summary.added, summary.images_failed) == (1, 1, 0)
        # The describe call went to the CHAT model (LOCKED A3), once.
        assert llm.chat_models == [llm.settings.llm_chat_model]

        # The wire shape: the multimodal user message — the fixed
        # prompt's text part + the image's data-URL part.
        (message,) = llm.chat_calls[0]
        assert message["role"] == "user"
        content: Any = message["content"]  # the multimodal part list
        assert content[0] == {"type": "text", "text": DESCRIBE_PROMPT}
        assert content[1]["type"] == "image_url"
        assert content[1]["image_url"]["url"].startswith("data:image/png;base64,")

        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None, "the image file must become a document"
        assert doc.is_image is True and doc.image_path is not None
        assert doc.content == _MockVisionLLM.DESCRIPTION
        assert doc.summary == _MockVisionLLM.DESCRIPTION  # task 03: verbatim

        # The ONLY embedded text of the doc is the description — twice
        # (the one content chunk + the is_summary chunk), 768 dims.
        chunks = db.scalars(select(Chunk).where(Chunk.document_id == doc.id)).all()
        summary_chunks = [c for c in chunks if c.is_summary]
        assert len(summary_chunks) == 1 and summary_chunks[0].position == -1
        assert all(c.content == _MockVisionLLM.DESCRIPTION for c in chunks)
        for c in chunks:
            assert c.embedding is not None and len(c.embedding) == 768
        embedded = [t for batch in llm.calls for t in batch]
        assert embedded == [_MockVisionLLM.DESCRIPTION] * 2
    finally:
        _cleanup_source(db, source_root.name)


class _MockVisionFailsLLM(FakeEmbedder):
    """A NON-VISION chat model (LOCKED A3's honest failure): the
    multimodal describe call raises (the SDK errors — a chat model
    without vision rejects the ``image_url`` part), text (``lite``)
    calls keep the ``FakeEmbedder`` behaviour."""

    async def chat(self, messages, model=None):
        self.chat_calls.append(list(messages))
        user = next((m["content"] for m in messages if m.get("role") == "user"), "")
        if isinstance(user, list):
            raise LLMError("simulated non-vision chat model (test sentinel)")
        first = user.split()
        return "Summary of " + (first[0] if first else "<empty>")


def test_import_sources_failing_vision_skips_image_keeps_sync_green(
    db, tmp_path, caplog: pytest.LogCaptureFixture
) -> None:
    """LOCKED A3 fail-soft end-to-end (the task-06 integration pin):
    a fixture PNG through a NON-VISION chat model — the real seam, no
    patch — skips the image doc (``images_failed == 1``, NO row, NO
    orphan copy — not even the image dir) while the TEXT doc in the
    same source is indexed as usual (the sync completes, no row
    mutation anywhere for the failed image)."""
    source_root = tmp_path / "visionfail"
    source_root.mkdir()
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    (source_root / "notes.md").write_text("# Notes\n\nBody.\n", encoding="utf-8")
    llm = _image_llm(tmp_path, _MockVisionFailsLLM)
    try:
        with caplog.at_level(logging.INFO, logger="app.importer"):
            summary = asyncio.run(
                import_sources([source_root], llm, session=db)
            )
        assert (summary.files, summary.added, summary.images_failed) == (2, 1, 1)
        # The image: no row, no copy (the dir itself was never created).
        assert (
            db.scalar(
                select(Document).where(
                    Document.source == source_root.name, Document.path == "diagram.png"
                )
            )
            is None
        )
        assert not Path(llm.settings.image_dir).expanduser().exists()
        # The text doc indexed as usual (the sync stayed green).
        md = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "notes.md"
            )
        )
        assert md is not None and md.is_image is False
        # The importer's warning names the document (the PLAN §9 signal).
        warnings = [
            r
            for r in caplog.records
            if r.name == "app.importer" and "image description failed" in r.getMessage()
        ]
        assert len(warnings) == 1 and warnings[0].levelno == logging.WARNING
        assert f"source={source_root.name} path=diagram.png" in warnings[0].getMessage()
    finally:
        _cleanup_source(db, source_root.name)


def test_import_sources_images_off_ignores_and_prune_guard_protects(
    db, tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """``images=False`` (the default) end-to-end: the walk ignores the
    image file entirely (not counted, no row, no copy), and a
    ``prune=True`` run MUST NOT delete a pre-existing image doc — the
    LOCKED prune guard (invisible to the walk ≠ deleted)."""
    source_root = tmp_path / "imgsrc_off"
    source_root.mkdir()
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    _patch_description(monkeypatch, "A network diagram.")
    try:
        llm_on = _image_llm(tmp_path)
        s_on = asyncio.run(import_sources([source_root], llm_on, session=db))
        assert s_on.added == 1
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None
        copy = Path(doc.image_path)
        assert copy.exists()

        # Toggle OFF (a fresh fake on the code defaults): the walk is
        # blind to the file, and prune protects the pre-existing image
        # doc + its copy.
        llm_off = FakeEmbedder()  # Settings(_env_file=None) → images False
        s_off = asyncio.run(
            import_sources([source_root], llm_off, session=db, prune=True)
        )
        assert (s_off.files, s_off.added, s_off.pruned) == (0, 0, 0)
        assert db.scalar(select(Document).where(Document.id == doc.id)) is not None
        assert copy.exists(), "the copy survives with the doc"
    finally:
        _cleanup_source(db, source_root.name)


def test_import_sources_toggle_on_prunes_deleted_image_with_copy(
    db, tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Toggle ON, the image file deleted: the normal prune runs — the
    doc row AND its ``image_dir`` copy are removed (the copy's
    lifecycle is tied to the row)."""
    source_root = tmp_path / "imgsrc_prune"
    source_root.mkdir()
    (source_root / "diagram.png").write_bytes(PNG_1X1)
    llm_on = _image_llm(tmp_path)
    _patch_description(monkeypatch, "A network diagram.")
    try:
        asyncio.run(import_sources([source_root], llm_on, session=db))
        doc = db.scalar(
            select(Document).where(
                Document.source == source_root.name, Document.path == "diagram.png"
            )
        )
        assert doc is not None
        copy = Path(doc.image_path)
        assert copy.exists()

        (source_root / "diagram.png").unlink()
        s = asyncio.run(
            import_sources([source_root], llm_on, session=db, prune=True)
        )
        assert (s.pruned, s.added, s.unchanged) == (1, 0, 0)
        assert (
            db.scalar(select(Document).where(Document.source == source_root.name))
            is None
        )
        assert not copy.exists(), "the image_dir copy is deleted with the doc"
    finally:
        _cleanup_source(db, source_root.name)


# ---------------------------------------------------------------------------
# Phase 122 (task 04) — the image BYTES route + the content/tree wire
# affordance (the serve side of the image-document contract).
# ---------------------------------------------------------------------------


def _seed_image_doc(
    db,
    tmp_path: Path,
    *,
    source: str = "ImgSrc",
    path: str = "pic.png",
    data: bytes = PNG_1X1,
    content: str = "A red square on a white background.",
    image_path: str | None = "auto",
    doc_id: str | None = None,
) -> Document:
    """One ``is_image`` document row with its persistent copy (the
    importer's ``image_dir`` layout) under *tmp_path*; ``image_path``
    ``"auto"`` writes the copy, ``None`` leaves the row without a copy
    (the lost-copy corner)."""
    doc = Document(
        id=uuid.UUID(doc_id) if doc_id else uuid.uuid4(),
        source=source,
        path=path,
        full_path=f"/tmp/{path}",
        title=path.rsplit(".", 1)[0],
        content=content,
        content_hash=hashlib.sha256(data).hexdigest(),
        indexed_at=datetime.now(UTC),
        created_at=datetime(2026, 1, 1, tzinfo=UTC),
        summary=content,
        is_image=True,
    )
    if image_path == "auto":
        copy = tmp_path / f"{doc.id}{Path(path).suffix}"
        copy.write_bytes(data)
        doc.image_path = str(copy)
    elif image_path is not None:
        doc.image_path = image_path
    db.add(doc)
    db.flush()
    db.add_all(
        [
            Chunk(
                document_id=doc.id, position=0, content=content, embedding=[0.01] * 768
            ),
            Chunk(
                document_id=doc.id,
                position=-1,
                content=content,
                embedding=[0.01] * 768,
                is_summary=True,
            ),
        ]
    )
    db.commit()
    return doc


def _cleanup_kb(db) -> None:
    """Truncate the KB tables. The settle commit matters: SQLAlchemy
    does NOT autoflush pending ORM objects before a raw ``text()``
    statement — a pending chunks INSERT flushed *after* the TRUNCATE
    would FK-violate (the document row is already gone), so any
    pending state is committed (then truncated) first."""
    db.commit()
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


@pytest.mark.parametrize(
    ("ext", "mime"),
    [
        (".png", "image/png"),
        (".jpg", "image/jpeg"),
        (".jpeg", "image/jpeg"),
        (".webp", "image/webp"),
        (".gif", "image/gif"),
        (".bmp", "image/bmp"),
    ],
)
def test_image_route_serves_exact_bytes_with_content_type(
    admin_client: TestClient, db, tmp_path: Path, ext: str, mime: str
) -> None:
    """The serve contract: the route streams the EXACT stored bytes
    (a per-extension sentinel — a mix-up between the six formats is
    caught) with the extension's ``Content-Type`` (the
    ``IMAGE_MIMES`` map — one map, one truth) and
    ``Cache-Control: private, max-age=3600`` (content-hashed bytes —
    long enough, bustable by re-upload)."""
    _cleanup_kb(db)
    try:
        data = PNG_1X1 + ext.encode("ascii")  # per-extension sentinel bytes
        doc = _seed_image_doc(db, tmp_path, path=f"pic{ext}", data=data)
        r = admin_client.get(f"/api/documents/{doc.id}/image")
        assert r.status_code == 200
        assert r.headers["content-type"] == mime  # exact type per extension
        assert r.headers["cache-control"] == "private, max-age=3600"
        assert r.content == data  # the exact uploaded bytes, nothing else
    finally:
        _cleanup_kb(db)


def test_image_route_404_matrix(admin_client: TestClient, db, tmp_path: Path) -> None:
    """Every non-servable case is 404 ``document not found`` (the
    router's unknown-document shape — the same detail string the
    content endpoint uses): a missing id, a MALFORMED id (an unparseable
    string maps here, not to a 422 — a guessed id is an unknown
    document), a text doc, an image doc whose ``image_path`` is NULL,
    and a row whose copy was lost on disk (defensive — the row exists,
    the bytes don't)."""
    _cleanup_kb(db)
    try:
        _seed_doc(db, "TextSrc", "note.md", "Note", 1, datetime.now(UTC))
        db.commit()  # the module's _seed_doc leaves the row uncommitted
        text_doc_id = db.scalar(
            select(Document.id).where(
                Document.source == "TextSrc", Document.path == "note.md"
            )
        )
        no_copy = _seed_image_doc(db, tmp_path, path="nopy.png", image_path=None)
        lost = _seed_image_doc(db, tmp_path, path="lost.png")
        assert lost.image_path is not None  # the "auto" copy was written
        Path(lost.image_path).unlink()  # the copy is lost (the row remains)

        for doc_id in (
            str(uuid.uuid4()),  # missing id
            "not-a-uuid",  # malformed id → 404, not 422
            str(text_doc_id),  # text doc
            str(no_copy.id),  # image doc, image_path NULL
            str(lost.id),  # image doc, copy lost
        ):
            r = admin_client.get(f"/api/documents/{doc_id}/image")
            assert r.status_code == 404, doc_id
            assert r.json() == {"detail": "document not found"}, doc_id
    finally:
        _cleanup_kb(db)


def test_image_route_is_public_like_the_shared_chat_content(
    admin_client: TestClient, db, tmp_path: Path
) -> None:
    """Phase 124 (LOCKED A4 — the answer figure is "visible … on the
    shared page"): the BYTES route is PUBLIC (the phase-123
    chat-image precedent — the image is part of a shared chat's
    rendered content, and the shared chat is the app's ONLY anonymous
    surface, PLAN A10; the unguessable ``uuid4`` document id IS the
    credential, no enumeration value): an ANONYMOUS caller (a FRESH
    client — no session) gets the bytes with the image Content-Type,
    and a signed-in caller gets the same. The document's TEXT (the
    content endpoint) stays user-gated — only the bytes are public."""
    _cleanup_kb(db)
    try:
        doc = _seed_image_doc(db, tmp_path)
        anonymous = TestClient(fastapi_app)
        r = anonymous.get(f"/api/documents/{doc.id}/image")
        assert r.status_code == 200
        assert r.headers["content-type"].startswith("image/")
        assert r.content == PNG_1X1
        # The signed-in client (admin) gets the same bytes.
        r_admin = admin_client.get(f"/api/documents/{doc.id}/image")
        assert r_admin.status_code == 200
        assert r_admin.content == PNG_1X1
    finally:
        _cleanup_kb(db)


def test_content_endpoint_exposes_the_image_affordance(
    admin_client: TestClient, db, tmp_path: Path
) -> None:
    """The content endpoint (the viewer's data source): ``is_image`` is
    ALWAYS present (text doc: false — the one new key; the wire shape
    gains nothing else), and ``image_url`` — the bytes route's path —
    is present for an image doc and ABSENT for a text doc (never null,
    the ``DocContent`` omission rule)."""
    _cleanup_kb(db)
    try:
        _seed_doc(db, "TextSrc", "note.md", "Note", 1, datetime.now(UTC))
        db.commit()  # the app's endpoint session reads committed data only
        r = admin_client.get(
            "/api/documents/content", params={"source": "TextSrc", "path": "note.md"}
        )
        assert r.status_code == 200
        body = r.json()
        assert body["is_image"] is False
        assert "image_url" not in body  # absent — never null (text doc)

        doc = _seed_image_doc(db, tmp_path, source="ImgSrc", path="pic.png")
        r = admin_client.get(
            "/api/documents/content", params={"source": "ImgSrc", "path": "pic.png"}
        )
        assert r.status_code == 200
        body = r.json()
        assert body["is_image"] is True
        assert body["image_url"] == f"/api/documents/{doc.id}/image"
        assert body["content"] == body["summary"]  # the description (task 03)
    finally:
        _cleanup_kb(db)


# ---------------------------------------------------------------------------
# Phase 148 (task 02) — the PDF page BYTES route + the content payload's
# ``page_urls`` affordance (phase 163 task 01 adds the ``page_strip_mode``
# class signal beside it)
# ---------------------------------------------------------------------------


def _page_settings(pages_dir: Path) -> Settings:
    """Fresh settings (the ``.env`` file ignored) with the phase-148
    ``pdf_page_dir`` pinned to *pages_dir* — the page route reads the
    setting at request time, and the copies under test never land in
    the real home (the ``get_settings`` monkeypatch pattern)."""
    return Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        pdf_page_dir=str(pages_dir),
    )


def _seed_page_doc(
    db,
    pages_dir: Path,
    *,
    source: str = "PdfSrc",
    path: str = "besta.pdf",
    page_count: int | None = 2,
    write_copies: bool = True,
    page_strip_mode: str | None = None,
) -> Document:
    """One PDF document row with ``page_count`` + its rendered page
    copies (the importer's ``pdf_page_dir`` layout —
    ``<doc-id>-p{N}.png``, N 1-based); page N is the ``PNG_1X1 + N``
    sentinel bytes, so a page mix-up between the two copies is caught.
    ``page_count=None`` is a non-qualifying PDF (no copies);
    ``write_copies=False`` leaves a qualifying row without its files
    (the lost-copy corner); ``page_strip_mode`` (phase 163) stamps the
    strip CLASS — ``None`` is the pre-phase row."""
    doc = Document(
        id=uuid.uuid4(),
        source=source,
        path=path,
        full_path=f"/tmp/{path}",
        title=path.rsplit(".", 1)[0],
        content="[Page 1 illustration: the parts tray] [Page 2 illustration: step 1]",
        content_hash=hashlib.sha256(path.encode("utf-8")).hexdigest(),
        indexed_at=datetime.now(UTC),
        created_at=datetime(2026, 1, 1, tzinfo=UTC),
        summary="IKEA BESTA frame assembly manual (page descriptions).",
        page_count=page_count,
        page_strip_mode=page_strip_mode,
    )
    db.add(doc)
    db.flush()
    if write_copies and page_count:
        pages_dir.mkdir(parents=True, exist_ok=True)
        for n in range(1, page_count + 1):
            (pages_dir / f"{doc.id}-p{n}.png").write_bytes(PNG_1X1 + str(n).encode("ascii"))
    db.add_all(
        [
            Chunk(
                document_id=doc.id, position=0, content=doc.content, embedding=[0.01] * 768
            ),
            Chunk(
                document_id=doc.id,
                position=-1,
                content=doc.summary,
                embedding=[0.01] * 768,
                is_summary=True,
            ),
        ]
    )
    db.commit()
    return doc


def test_page_route_serves_exact_bytes_with_png_type(
    admin_client: TestClient, db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The serve contract: each page of a qualifying doc streams the
    EXACT stored PNG bytes (a per-page sentinel — a page mix-up is
    caught) as ``image/png`` (a page is always a PNG —
    ``render_pdf_pages``'s ``tobytes("png")``) with
    ``Cache-Control: private, max-age=3600`` (the image route's
    header — content-addressed by the document id + page number).
    The path is built ONLY from the row's id + ``n`` (the row
    derives it; ``n`` is the int path parameter) — the copies
    written under the tmp ``pdf_page_dir`` are what come back."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        monkeypatch.setattr(docs_api, "get_settings", lambda: _page_settings(pages_dir))
        doc = _seed_page_doc(db, pages_dir, page_count=2)
        for n in (1, 2):
            r = admin_client.get(f"/api/documents/{doc.id}/pages/{n}")
            assert r.status_code == 200, n
            assert r.headers["content-type"] == "image/png"
            assert r.headers["cache-control"] == "private, max-age=3600"
            assert r.content == PNG_1X1 + str(n).encode("ascii")  # the exact stored bytes
    finally:
        _cleanup_kb(db)


def test_page_route_404_matrix(
    admin_client: TestClient, db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Every non-servable case is 404 ``document not found`` (the
    router's unknown-document shape — the same detail string the
    content endpoint uses): a missing id, a MALFORMED id (an
    unparseable string maps here, not to a 422 — the image route's
    mapping: a guessed id is an unknown document), a text doc, an
    image doc, a PDF row whose ``page_count`` is NULL (a
    non-qualifying doc — the route knows nothing of its pages),
    ``n`` outside ``1..page_count`` (``n=0``/``n=-1`` below the range
    — the int parameter makes fractions impossible — and ``n=3``
    above a 2-page doc), and a qualifying row whose copy was lost on
    disk (defensive — the row exists, the bytes don't; the viewer's
    onerror note covers it). The int parameter is the no-traversal
    pin: a NON-int ``n`` 422s at the router, and no request value
    ever reaches a filesystem path (the out-of-range 404s show the
    path is bounded by the ROW's extent)."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        monkeypatch.setattr(docs_api, "get_settings", lambda: _page_settings(pages_dir))
        _seed_doc(db, "TextSrc", "note.md", "Note", 1, datetime.now(UTC))
        db.commit()  # the module's _seed_doc leaves the row uncommitted
        text_doc_id = db.scalar(
            select(Document.id).where(Document.source == "TextSrc", Document.path == "note.md")
        )
        img = _seed_image_doc(db, tmp_path, path="pic.png")
        plain = _seed_page_doc(db, pages_dir, path="texty.pdf", page_count=None)
        lost = _seed_page_doc(db, pages_dir, path="lost.pdf")
        assert (pages_dir / f"{lost.id}-p1.png").is_file()  # the copies were written
        (pages_dir / f"{lost.id}-p1.png").unlink()  # the copy is lost (the row remains)
        qualifying = _seed_page_doc(db, pages_dir, path="besta.pdf")

        for doc_id, n in (
            (str(uuid.uuid4()), 1),  # missing id
            ("abc", 1),  # malformed id → 404, not 422
            (str(text_doc_id), 1),  # text doc
            (str(img.id), 1),  # image doc
            (str(plain.id), 1),  # PDF, page_count NULL (non-qualifying)
            (str(lost.id), 1),  # qualifying doc, copy lost on disk
            (str(qualifying.id), 0),  # below range
            (str(qualifying.id), -1),  # negative — the int parse allows it, the range rejects it
            (str(qualifying.id), 3),  # above range (page_count 2)
        ):
            r = admin_client.get(f"/api/documents/{doc_id}/pages/{n}")
            assert r.status_code == 404, (doc_id, n)
            assert r.json() == {"detail": "document not found"}, (doc_id, n)

        # The int parameter (the no-traversal pin): a NON-int n is a
        # router-level 422 — no other ``n`` shape reaches the handler.
        r = admin_client.get(f"/api/documents/{qualifying.id}/pages/abc")
        assert r.status_code == 422
    finally:
        _cleanup_kb(db)


def test_page_route_is_public_like_the_shared_chat_content(
    admin_client: TestClient, db, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The image route's phase-124 A4 trust model verbatim: a shared
    chat can cite a page FIGURE (the viewer's strip, task 03), so the
    BYTES serve ANONYMOUSLY — the unguessable ``uuid4`` document id IS
    the credential, no enumeration value (PLAN A10: the shared chats
    are the app's ONLY anonymous surface). An ANONYMOUS caller (a
    FRESH client — no session) gets the page PNG, and a signed-in
    caller gets the same. The document's TEXT (the content endpoint)
    stays user-gated — only the page bytes are public."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        monkeypatch.setattr(docs_api, "get_settings", lambda: _page_settings(pages_dir))
        doc = _seed_page_doc(db, pages_dir)
        anonymous = TestClient(fastapi_app)
        r = anonymous.get(f"/api/documents/{doc.id}/pages/1")
        assert r.status_code == 200
        assert r.headers["content-type"] == "image/png"
        assert r.content == PNG_1X1 + b"1"
        # The signed-in client (admin) gets the same bytes.
        r_admin = admin_client.get(f"/api/documents/{doc.id}/pages/1")
        assert r_admin.status_code == 200
        assert r_admin.content == PNG_1X1 + b"1"
    finally:
        _cleanup_kb(db)


def test_content_endpoint_exposes_the_page_urls_affordance(
    admin_client: TestClient, db, tmp_path: Path
) -> None:
    """The content endpoint (the viewer's data source — task 03's
    strip reads from it): a page-qualifying PDF (``page_count`` set)
    carries ``page_urls`` = the EXACT 1-based route list over
    ``1..page_count`` (the count == ``page_count`` — a per-page render
    gap is still listed; the route 404s that N); text docs, image
    docs, and non-qualifying PDFs (``page_count`` NULL) keep the key
    ABSENT from the JSON — never ``null`` (the ``image_url``
    omission rule — a non-qualifying doc's payload is byte-identical
    to pre-phase)."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        doc = _seed_page_doc(db, pages_dir, page_count=3)
        r = admin_client.get(
            "/api/documents/content", params={"source": "PdfSrc", "path": "besta.pdf"}
        )
        assert r.status_code == 200
        body = r.json()
        assert body["page_urls"] == [
            f"/api/documents/{doc.id}/pages/1",
            f"/api/documents/{doc.id}/pages/2",
            f"/api/documents/{doc.id}/pages/3",
        ]  # the exact 1-based list, count == page_count

        _seed_doc(db, "TextSrc", "note.md", "Note", 1, datetime.now(UTC))
        db.commit()  # the app's endpoint session reads committed data only
        r = admin_client.get(
            "/api/documents/content", params={"source": "TextSrc", "path": "note.md"}
        )
        assert r.status_code == 200
        assert "page_urls" not in r.json()  # absent — never null (text doc)

        _seed_image_doc(db, tmp_path, source="ImgSrc", path="pic.png")
        r = admin_client.get(
            "/api/documents/content", params={"source": "ImgSrc", "path": "pic.png"}
        )
        assert r.status_code == 200
        assert "page_urls" not in r.json()  # absent — never null (image doc)

        _seed_page_doc(db, pages_dir, path="texty.pdf", page_count=None)
        r = admin_client.get(
            "/api/documents/content", params={"source": "PdfSrc", "path": "texty.pdf"}
        )
        assert r.status_code == 200
        assert "page_urls" not in r.json()  # absent — never null (non-qualifying PDF)
        # Phase 163 (task 01): the class key follows the same rule — a
        # pre-phase row (mode NULL) omits it while keeping page_urls.
        assert "page_strip_mode" not in r.json()
    finally:
        _cleanup_kb(db)


@pytest.mark.parametrize("mode", ["primary", "additive"])
def test_content_endpoint_carries_the_strip_class(
    admin_client: TestClient, db, tmp_path: Path, mode: str
) -> None:
    """Phase 163 (task 01, LOCKED A2/A3) — the real-DB round-trip of the
    strip CLASS: a stamped row (``page_strip_mode`` = ``"primary"`` — the
    phase-133 fallback class, the strip IS the view — or ``"additive"`` —
    the phase-140 pictorial class, text + a Text/Pages toggle) serves the
    class VERBATIM beside the untouched ``page_urls`` list (the viewer's
    branch signal, task 04); every other key stands where phase 148 left
    it."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        _seed_page_doc(db, pages_dir, page_count=3, page_strip_mode=mode)
        r = admin_client.get(
            "/api/documents/content", params={"source": "PdfSrc", "path": "besta.pdf"}
        )
        assert r.status_code == 200
        body = r.json()
        assert body["page_strip_mode"] == mode
        assert len(body["page_urls"]) == 3  # the strip rides along, unchanged
    finally:
        _cleanup_kb(db)


def test_content_strip_class_never_travels_without_the_strip(
    admin_client: TestClient, db, tmp_path: Path
) -> None:
    """Phase 163 (task 01) — the two strip keys are ONE block keyed off
    ``page_count``: a row with a stray ``page_strip_mode`` and no strip
    (``page_count`` NULL — the corner a partial repair could leave)
    serializes NEITHER key, byte-identically to a plain non-qualifying
    document (the viewer must never be handed a class for pages that do
    not exist)."""
    _cleanup_kb(db)
    try:
        pages_dir = tmp_path / "pages"
        _seed_page_doc(
            db,
            pages_dir,
            path="stray.pdf",
            page_count=None,
            write_copies=False,
            page_strip_mode="additive",
        )
        r = admin_client.get(
            "/api/documents/content", params={"source": "PdfSrc", "path": "stray.pdf"}
        )
        assert r.status_code == 200
        body = r.json()
        assert "page_strip_mode" not in body
        assert "page_urls" not in body
        # The exact pre-phase key set (the byte-identity pin).
        assert set(body) == {
            "source",
            "path",
            "title",
            "format",
            "summary",
            "created_at",
            "content",
            "indexed_at",
            "chunks",
            "is_image",
        }
    finally:
        _cleanup_kb(db)


def _tree_file_nodes_all(sources) -> list[dict]:
    """Every file node of a tree response, walked recursively (all
    sources — env-registered 0-document sources may join the response
    when the ``git_sources`` table is truncated, and they carry no
    file nodes; the assertions below hold over whatever files exist).
    """
    files: list[dict] = []

    def _walk(node: dict) -> None:
        for child in node.get("children", ()):
            if child["kind"] == "file":
                files.append(child)
            else:
                _walk(child)

    for source in sources:
        _walk(source)
    return files


def test_tree_image_file_node_affordance_and_text_node_byte_identical(
    admin_client: TestClient, db, tmp_path: Path
) -> None:
    """The tree (the RAG view's single fetch): an image doc's file node
    carries the thumbnail affordance (``is_image`` true,
    ``image_url`` = the bytes route's path, ``summary`` verbatim — the
    RAG view's thumbnail ``alt``); EVERY text file node keeps the
    pre-phase wire shape byte-identically (the six keys — no
    ``is_image``/``image_url``/``summary`` — the phase's
    byte-identical criterion: the fields are row-driven, so a KB with
    no image rows serializes exactly as pre-phase)."""
    _truncate_tree_tables(db)
    try:
        base = datetime.now(UTC)
        _seed_doc(db, "MixedSrc", "a.md", "A", 1, base)
        img = _seed_image_doc(db, tmp_path, source="MixedSrc", path="pic.png")

        r = admin_client.get("/api/docs/tree")  # both seeds committed (_seed_image_doc)
        assert r.status_code == 200
        files = {f["path"]: f for f in _tree_file_nodes_all(r.json()["sources"])}

        # Text node: byte-identical pre-phase wire shape (no image keys).
        assert set(files["a.md"]) == {
            "kind", "path", "title", "chunks", "created_at", "indexed_at"
        }

        # Image node: the affordance rides the node.
        pic = files["pic.png"]
        assert pic["is_image"] is True
        assert pic["image_url"] == f"/api/documents/{img.id}/image"
        assert pic["summary"] == "A red square on a white background."
    finally:
        _truncate_tree_tables(db)


def test_tree_with_no_image_rows_is_byte_identical(admin_client: TestClient, db) -> None:
    """The pre-phase KB (no ``is_image`` rows): the image-docs map is
    empty and EVERY file node serializes in the pre-phase shape — the
    row-driven fields introduce no wire change at all (the phase's
    byte-identical criterion, the toggle irrelevant)."""
    _truncate_tree_tables(db)
    try:
        base = datetime.now(UTC)
        _seed_doc(db, "PlainSrc", "x.md", "X", 2, base)
        db.commit()  # the app's endpoint session reads committed data only
        r = admin_client.get("/api/docs/tree")
        assert r.status_code == 200
        files = _tree_file_nodes_all(r.json()["sources"])
        assert len(files) == 1  # the seeded doc (env sources carry no files)
        assert set(files[0]) == {
            "kind", "path", "title", "chunks", "created_at", "indexed_at"
        }
    finally:
        _truncate_tree_tables(db)


# ---------------------------------------------------------------------------
# Phase 167 (task 01): transport gzip on the real app — the /api/docs/tree
# payload (the TODO L5 2 MB defect) compresses, small responses stay
# identity, and the A20 security-header + A19 page contracts survive the
# new layer sitting between SecurityHeaders (outermost) and Caching.
# ---------------------------------------------------------------------------


def _seed_gzip_corpus(db, count: int = 50) -> None:
    """>= 50 indexed docs — the tree response lands well above the 860-byte
    floor (the production corpus that motivates the phase is ~1 k docs)."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    for i in range(count):
        _seed_doc(db, "gzip-corpus", f"docs/sub-{i % 5}/file-{i:02d}.md", f"Doc {i:02d}", 1, base)
    db.commit()  # the endpoint session reads committed data only


def test_docs_tree_ships_gzip_and_decompresses_to_the_documented_json(
    admin_client: TestClient, db
) -> None:
    """Accept-Encoding: gzip → Content-Encoding: gzip + Vary, and the raw
    wire bytes decompress to EXACTLY the identity bytes (the documented
    tree JSON, 50 file nodes) — ~an order of magnitude smaller."""
    _seed_gzip_corpus(db)
    # Phase 167 (task 02): the 50-file corpus must ride the wire for
    # this byte pin, so the full tree is requested explicitly (the
    # endpoint default ships one level, which would fall under the
    # compression floor).
    try:
        identity = admin_client.get(
            "/api/docs/tree?depth=8", headers={"Accept-Encoding": "identity"}
        )
        assert identity.status_code == 200
        assert "content-encoding" not in identity.headers
        assert len(identity.content) >= 860  # the seed is above the floor

        r = admin_client.get(
            "/api/docs/tree?depth=8", headers={"Accept-Encoding": "gzip"}
        )
        assert r.status_code == 200
        assert r.headers["content-encoding"] == "gzip"
        assert "accept-encoding" in r.headers.get("vary", "").lower()
        # httpx decodes transparently: same documented JSON.
        sources = r.json()["sources"]
        assert len(_tree_file_nodes_all(sources)) == 50
        # The RAW stream is a real gzip of the identity body.
        with admin_client.stream(
            "GET", "/api/docs/tree?depth=8", headers={"Accept-Encoding": "gzip"}
        ) as streamed:
            raw = b"".join(streamed.iter_raw())
        assert gzip.decompress(raw) == identity.content
        assert len(raw) < len(identity.content) // 2  # repetitive JSON: >> 2x
    finally:
        _truncate_tree_tables(db)


def test_gzipped_response_still_carries_the_a20_headers(
    admin_client: TestClient, db
) -> None:
    """The outer-layer invariant (A20): gzip sits INSIDE
    SecurityHeadersMiddleware, so a gzipped response still carries the
    CSP, nosniff and the no-framing headers."""
    _seed_gzip_corpus(db, 60)
    try:
        # depth=8 (phase 167, task 02): the whole tree stays above the
        # 860-byte floor — the point here is a GZIPPED response.
        r = admin_client.get("/api/docs/tree?depth=8", headers={"Accept-Encoding": "gzip"})
        assert r.headers["content-encoding"] == "gzip"
        assert r.headers["content-security-policy"] == CSP
        assert r.headers["x-content-type-options"] == "nosniff"
        assert r.headers["x-frame-options"] == "DENY"
    finally:
        _truncate_tree_tables(db)


def test_small_api_response_stays_identity(client: TestClient) -> None:
    """A tiny /api/* response (under the 860-byte floor) is NOT compressed
    even when the client offers gzip — the minimum-size exemption."""
    r = client.get("/api/health", headers={"Accept-Encoding": "gzip"})
    assert r.status_code == 200
    assert "content-encoding" not in r.headers
    assert r.json()["status"] == "ok"


def test_html_page_contract_survives_gzip(client: TestClient) -> None:
    """A19 through the new layer: the rewritten HTML page (>= the floor, so
    it compresses) STILL serves no-cache with the ?v= asset rewrite and no
    validators — and the contract holds AFTER decompression."""
    r = client.get("/", headers={"Accept-Encoding": "gzip"})
    assert r.status_code == 200
    assert r.headers["content-encoding"] == "gzip"  # 95 KB page, far above the floor
    assert r.headers["cache-control"] == "no-cache"
    assert "etag" not in r.headers
    assert "last-modified" not in r.headers
    assert r.text.count("?v=") >= 1  # the caching rewrite ran, gzip preserved it


# ---------------------------------------------------------------------------
# Phase 167 (task 02): the depth-limited tree + server totals (D2/A2) — the
# default depth is the RAG view's first paint (sources + ONE level), an
# explicit depth builds deeper, out-of-range values are rejected, and an
# untruncated node keeps its pre-phase-167 wire shape.
# ---------------------------------------------------------------------------


def _seed_depth_corpus(db) -> datetime:
    """Four documents nested one level deeper each (root, ``a``, ``a/b``,
    ``a/b/c``) with DISTINCT ``indexed_at`` stamps, plus one stored
    summary on ``a`` — a fixture whose truncated nodes still carry a
    recursive count, and whose level rows carry a description and a
    pending marker."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    _seed_doc(db, "Homelab", "root.md", "Root", 1, base)
    _seed_doc(db, "Homelab", "a/top.md", "Top", 2, base + timedelta(hours=1))
    _seed_doc(db, "Homelab", "a/b/mid.md", "Mid", 3, base + timedelta(hours=2))
    _seed_doc(db, "Homelab", "a/b/c/deep.md", "Deep", 4, base + timedelta(hours=3))
    db.add(FolderSummary(source="Homelab", folder_path="a", summary="A desc."))
    db.commit()  # the app's endpoint session reads committed data only
    return base


def test_docs_tree_default_depth_ships_one_level_with_totals(admin_client, db) -> None:
    """``GET /api/docs/tree`` (no param) is the view's first paint:
    sources + ONE level. The folder with unloaded children carries
    ``truncated: true`` + ``children: []`` AND its recursive count;
    ``totals`` matches ``GET /api/docs`` (documents, chunks, the newest
    ``indexed_at``) — the stat cards stay honest on a partial tree."""
    base = _seed_depth_corpus(db)
    try:
        r = admin_client.get("/api/docs/tree")
        assert r.status_code == 200
        body = r.json()
        assert set(body) == {"sources", "totals"}
        (homelab,) = body["sources"]
        assert homelab["documents"] == 4  # the whole recursive source count
        assert "truncated" not in homelab  # depth >= 1: never truncated
        folders = [c for c in homelab["children"] if c["kind"] == "folder"]
        files = [c for c in homelab["children"] if c["kind"] == "file"]
        assert [f["path"] for f in files] == ["root.md"]  # direct files ship
        (a,) = folders
        assert a["path"] == "a"
        assert a["truncated"] is True
        assert a["children"] == []
        assert a["documents"] == 3  # the recursive count, no children built
        assert a["summary"] == "A desc."  # the level row shows the description
        assert a["summary_pending"] is False  # the stored row covers it
        # Nothing deeper is on the wire (the 2 MB payload stays home).
        assert [f["path"] for f in _tree_file_nodes_all(body["sources"])] == ["root.md"]
        # Totals from the SAME catalogue rows (no extra query).
        assert body["totals"] == {
            "documents": 4,
            "chunks": 10,
            "last_indexed": (base + timedelta(hours=3)).isoformat(),
        }
        flat = admin_client.get("/api/docs").json()
        assert body["totals"]["documents"] == len(flat["documents"])
        assert body["totals"]["chunks"] == sum(d["chunks"] for d in flat["documents"])
        assert body["totals"]["last_indexed"] == max(
            d["indexed_at"] for d in flat["documents"]
        )
    finally:
        _truncate_tree_tables(db)


def test_docs_tree_depth_three_builds_deeper_and_truncates_the_rest(
    admin_client, db
) -> None:
    """``?depth=3``: ``a`` and ``a/b`` are built (no ``truncated`` key),
    ``a/b``'s subfolder ``a/b/c`` is truncated — while ``a/b``'s OWN
    direct files still ship: the budget limits DEPTH, never the row set
    of the levels it does send. The phase-98 pending rule is unchanged
    on truncated nodes (``a/b/c``: 1 doc, no stored row → pending)."""
    _seed_depth_corpus(db)
    try:
        body = admin_client.get("/api/docs/tree?depth=3").json()
        (homelab,) = body["sources"]
        (a,) = [c for c in homelab["children"] if c["kind"] == "folder"]
        assert "truncated" not in a  # expanded at this depth
        a_children = {c["path"]: c for c in a["children"]}
        assert set(a_children) == {"a/b", "a/top.md"}
        a_b = a_children["a/b"]
        assert "truncated" not in a_b
        a_b_c, mid = a_b["children"]  # subfolder first, then the direct file
        assert (a_b_c["kind"], a_b_c["path"], a_b_c["truncated"]) == (
            "folder",
            "a/b/c",
            True,
        )
        assert (a_b_c["documents"], a_b_c["children"]) == (1, [])
        assert a_b_c["summary"] is None
        assert a_b_c["summary_pending"] is True  # 1 doc, no stored row
        assert (mid["kind"], mid["path"]) == ("file", "a/b/mid.md")
        assert {f["path"] for f in _tree_file_nodes_all(body["sources"])} == {
            "root.md",
            "a/top.md",
            "a/b/mid.md",
        }  # deep.md never reaches the wire at this depth
    finally:
        _truncate_tree_tables(db)


def test_docs_tree_depth_validation_matrix(admin_client, db) -> None:
    """``depth`` is bounded (1..8) and typed: ``0`` and ``9`` are out of
    range and ``abc`` is not an int — all 422 before any build runs;
    the bounds themselves answer 200."""
    _seed_depth_corpus(db)
    try:
        for bad in ("0", "9", "abc"):
            r = admin_client.get(f"/api/docs/tree?depth={bad}")
            assert r.status_code == 422, bad
        assert admin_client.get("/api/docs/tree?depth=1").status_code == 200
        assert admin_client.get("/api/docs/tree?depth=8").status_code == 200
    finally:
        _truncate_tree_tables(db)


def test_docs_tree_full_build_keeps_the_pre_phase_node_wire_shape(
    admin_client, db
) -> None:
    """The wire-shape pin: on a FULL (``depth=8``) build nothing is
    truncated, so NO node carries the ``truncated`` key — the source,
    folder and file key sets are exactly the pre-phase-167 ones (the
    omission serializer keeps every existing consumer byte-identical);
    the additive top-level ``totals`` is the only body change."""
    _seed_depth_corpus(db)
    try:
        body = admin_client.get("/api/docs/tree?depth=8").json()
        assert set(body) == {"sources", "totals"}
        (homelab,) = body["sources"]
        assert set(homelab) == {
            "name", "documents", "updated_at", "summary", "summary_pending", "children"
        }
        seen_folders = seen_files = 0
        stack = list(homelab["children"])
        while stack:
            node = stack.pop()
            if node["kind"] == "folder":
                seen_folders += 1
                assert set(node) == {
                    "kind", "path", "documents", "updated_at", "summary",
                    "summary_pending", "children",
                }
                stack.extend(node["children"])
            else:
                seen_files += 1
                assert set(node) == {
                    "kind", "path", "title", "chunks", "created_at", "indexed_at"
                }
        assert (seen_folders, seen_files) == (3, 4)
    finally:
        _truncate_tree_tables(db)


# ---------------------------------------------------------------------------
# Phase 167 (task 03): GET /api/docs/tree/node — the drill's per-level
# fetch — and the ETag revalidation of the three catalogue reads
# (D3, LOCKED A3/A4).
# ---------------------------------------------------------------------------


def _find_child(node: dict, path: str) -> dict:
    """The child of *node* whose ``path`` equals *path* (the level walk
    the drill does client-side, expressed once for the assertions)."""
    return next(child for child in node["children"] if child["path"] == path)


def test_tree_node_source_root_is_the_kb_tree_source_shape(
    admin_client: TestClient, db
) -> None:
    """``?source=Homelab`` (``path`` omitted = ``""``; the explicit
    ``path=`` answers the identical node): the SOURCE shape — the full
    key set, never ``truncated``, the whole recursive source count —
    with the root level only: the direct file plus the subfolder
    flagged for the next drill."""
    _seed_depth_corpus(db)
    try:
        r = admin_client.get("/api/docs/tree/node", params={"source": "Homelab"})
        assert r.status_code == 200
        node = r.json()
        assert "truncated" not in node  # the node's OWN children are always built
        assert set(node) == {
            "name",
            "documents",
            "updated_at",
            "summary",
            "summary_pending",
            "children",
        }
        assert node["name"] == "Homelab"
        assert node["documents"] == 4  # the whole recursive source count
        assert node["summary_pending"] is True  # no stored (Homelab, "") row
        (a,) = [c for c in node["children"] if c["kind"] == "folder"]
        (root,) = [c for c in node["children"] if c["kind"] == "file"]
        assert (a["path"], a["truncated"], a["children"], a["documents"]) == (
            "a",
            True,
            [],
            3,
        )
        assert root["path"] == "root.md"
        # The explicit empty path is the SAME node as the omitted one.
        r2 = admin_client.get(
            "/api/docs/tree/node", params={"source": "Homelab", "path": ""}
        )
        assert r2.status_code == 200
        assert r2.json() == node
    finally:
        _truncate_tree_tables(db)


def test_tree_node_nested_folder_fields_and_depth_one_children(
    admin_client: TestClient, db
) -> None:
    """``?source=Homelab&path=a&depth=1``: the KbTreeFolder shape with
    every field the level row renders — recursive count, recursive
    ``updated_at``, ``summary``, ``summary_pending`` (phase 98 D3) — and
    its DIRECT children only: the deeper subfolder flagged
    ``truncated`` (the next drill's target), the direct file present."""
    _seed_depth_corpus(db)
    try:
        r = admin_client.get(
            "/api/docs/tree/node", params={"source": "Homelab", "path": "a"}
        )
        assert r.status_code == 200
        node = r.json()
        assert node["kind"] == "folder"
        assert node["path"] == "a"
        assert "truncated" not in node  # this node's children WERE built
        assert node["documents"] == 3  # recursive — the phase-94 count rule
        assert node["updated_at"] is not None  # recursive max (phase 106 D9)
        assert node["summary"] == "A desc."
        assert node["summary_pending"] is False  # the stored row covers it
        a_b, top = node["children"]  # subfolder first, then the direct file
        assert (a_b["kind"], a_b["path"], a_b["truncated"], a_b["children"]) == (
            "folder",
            "a/b",
            True,
            [],
        )
        assert a_b["documents"] == 2  # mid.md + deep.md, recursive
        assert a_b["summary_pending"] is True  # docs but no stored row
        assert (top["kind"], top["path"]) == ("file", "a/top.md")
    finally:
        _truncate_tree_tables(db)


def test_tree_node_depth_two_builds_one_level_deeper(
    admin_client: TestClient, db
) -> None:
    """``&depth=2`` spends one more level: ``a/b`` arrives BUILT (no
    ``truncated`` key, its direct file present) and only its own
    subfolder stays truncated — the same budget semantics the full tree
    applies, answered for one node."""
    _seed_depth_corpus(db)
    try:
        r = admin_client.get(
            "/api/docs/tree/node",
            params={"source": "Homelab", "path": "a", "depth": 2},
        )
        assert r.status_code == 200
        a_b = _find_child(r.json(), "a/b")
        assert "truncated" not in a_b  # expanded at this depth
        a_b_c, mid = a_b["children"]
        assert (a_b_c["path"], a_b_c["truncated"], a_b_c["documents"]) == (
            "a/b/c",
            True,
            1,
        )
        assert (mid["kind"], mid["path"]) == ("file", "a/b/mid.md")
    finally:
        _truncate_tree_tables(db)


def test_tree_node_404_unknown_source_folder_and_file_path(
    admin_client: TestClient, db
) -> None:
    """404 ``{"detail": "unknown node"}`` for a source that is neither
    registered nor indexed, a folder failing the phase-94 existence
    rule, a FILE's path (a document's OWN path is not a folder), and
    traversal strings — which have no filesystem surface to hit: the
    node is derived from catalogue ROWS alone (the row-only rule, the
    phase-10 traversal argument reused)."""
    _seed_depth_corpus(db)
    try:
        for params in (
            {"source": "nope"},  # neither registered nor indexed
            {"source": "Homelab", "path": "nope"},  # no indexed path starts with it
            {"source": "Homelab", "path": "a/top.md"},  # a FILE's own path
            {"source": "Homelab", "path": "../Homelab"},  # traversal is not a folder
            {"source": "..", "path": ""},  # nor is it a source
        ):
            r = admin_client.get("/api/docs/tree/node", params=params)
            assert r.status_code == 404, params
            assert r.json() == {"detail": "unknown node"}
    finally:
        _truncate_tree_tables(db)


def test_tree_node_403_anonymous(client: TestClient, db) -> None:
    """The same gate as ``/api/docs/tree`` — anonymous 403
    ``{"detail": "admin only"}`` (the RAG view never fetches for guests,
    the endpoint must not lean on that)."""
    _seed_depth_corpus(db)
    try:
        r = client.get("/api/docs/tree/node", params={"source": "Homelab"})
        assert r.status_code == 403
        assert r.json() == {"detail": "admin only"}
    finally:
        _truncate_tree_tables(db)


def test_tree_node_validation_matrix(admin_client: TestClient, db) -> None:
    """``source`` is required and ``depth`` keeps the tree route's
    bounds (``ge=1, le=8``): a missing source, ``depth=0``, ``depth=9``
    and ``depth=abc`` all 422 before anything is built."""
    _seed_depth_corpus(db)
    try:
        assert admin_client.get("/api/docs/tree/node").status_code == 422
        for bad in ("0", "9", "abc"):
            r = admin_client.get(
                "/api/docs/tree/node", params={"source": "Homelab", "depth": bad}
            )
            assert r.status_code == 422, bad
    finally:
        _truncate_tree_tables(db)


def test_tree_node_children_equal_the_full_tree_level(
    admin_client: TestClient, db
) -> None:
    """ONE concept, two routes, pinned equal: at the same budget the
    node route answers EXACTLY what ``GET /api/docs/tree`` carries at
    the same level — the source root at the default budget AND the full
    one, and the nested ``a`` folder on the full build — the JSON
    compared whole, so the drill can never paint a node the tree would
    not have produced."""
    _seed_depth_corpus(db)
    try:
        for depth in (1, 8):
            tree = admin_client.get(f"/api/docs/tree?depth={depth}").json()
            (source_tree,) = tree["sources"]
            node_root = admin_client.get(
                "/api/docs/tree/node", params={"source": "Homelab", "depth": depth}
            )
            assert node_root.status_code == 200
            assert node_root.json() == source_tree
        full = admin_client.get("/api/docs/tree?depth=8").json()
        a_tree = _find_child(full["sources"][0], "a")
        node_a = admin_client.get(
            "/api/docs/tree/node",
            params={"source": "Homelab", "path": "a", "depth": 8},
        )
        assert node_a.status_code == 200
        assert node_a.json() == a_tree
    finally:
        _truncate_tree_tables(db)


_CATALOGUE_READ_URLS = (
    "/api/docs",
    "/api/docs/tree",
    "/api/docs/tree/node?source=Homelab&path=a",
)


def test_catalogue_reads_carry_etag_and_answer_304(
    admin_client: TestClient, db
) -> None:
    """The ETag matrix on all three catalogue reads (A4): the first GET
    → 200 + weak ``ETag`` + ``Cache-Control: private, must-revalidate``;
    the re-send with ``If-None-Match`` → 304 with an EMPTY body and
    BOTH headers (RFC 9110 §8.8.3 — a 304 repeats the validator); the
    strong form of the weak tag, a comma-separated list and ``*`` all
    revalidate; a wrong tag is a plain 200 carrying the CURRENT tag."""
    _seed_depth_corpus(db)
    try:
        for url in _CATALOGUE_READ_URLS:
            first = admin_client.get(url)
            assert first.status_code == 200, url
            tag = first.headers["etag"]
            assert tag.startswith('W/"'), url
            assert first.headers["cache-control"] == "private, must-revalidate", url

            hit = admin_client.get(url, headers={"If-None-Match": tag})
            assert hit.status_code == 304, url
            assert hit.content == b"", url
            assert hit.headers["etag"] == tag, url
            assert hit.headers["cache-control"] == "private, must-revalidate", url

            strong = tag[2:]  # the same tag WITHOUT the weakness marker
            assert admin_client.get(
                url, headers={"If-None-Match": strong}
            ).status_code == 304, url
            assert admin_client.get(
                url, headers={"If-None-Match": f'W/"stale-tag", {strong}'}
            ).status_code == 304, url
            assert admin_client.get(url, headers={"If-None-Match": "*"}).status_code == (
                304
            ), url

            miss = admin_client.get(url, headers={"If-None-Match": 'W/"not-the-tag"'})
            assert miss.status_code == 200, url
            assert miss.headers["etag"] == tag, url
            assert miss.json() == first.json(), url
    finally:
        _truncate_tree_tables(db)


def _seed_bust_corpus(db) -> datetime:
    """The depth corpus with DETERMINISTIC, distinct ``created_at``
    stamps (the default seeding's subtree max is insert-time
    microseconds): ``a/b/c/deep.md`` IS the max of ``a``, so editing its
    date provably MOVES every answer that carries ``a``'s
    ``updated_at`` — which is what the bust test below asserts."""
    _truncate_tree_tables(db)
    base = datetime.now(UTC)
    dates = {
        "root.md": base - timedelta(days=30),
        "a/top.md": base - timedelta(days=20),
        "a/b/mid.md": base - timedelta(days=10),
        "a/b/c/deep.md": base - timedelta(days=1),  # the subtree max of "a"
    }
    db.add(GitSource(url="https://github.com/reese/Homelab.git", kind="git", added_at=base))
    for i, (path, created) in enumerate(dates.items()):
        doc = Document(
            source="Homelab",
            path=path,
            full_path=f"/tmp/Homelab/{path}",
            title=path,
            content=f"# {path}\n\nBody.",
            content_hash=f"{next(_counter):064d}",  # unique per row
            indexed_at=base,
            created_at=created,
        )
        db.add(doc)
        db.flush()
        db.add(
            Chunk(document_id=doc.id, position=0, content=f"chunk {i}", embedding=[0.01] * 768)
        )
    db.add(FolderSummary(source="Homelab", folder_path="a", summary="A desc."))
    db.commit()  # the app's endpoint session reads committed data only
    return base


def test_etag_busts_on_folder_summary_and_document_date(
    admin_client: TestClient, db
) -> None:
    """The LOCKED A4 proof: the admin PATCHes mutate the tree WITHOUT
    bumping ``sources_meta.version`` (asserted around the edits — a
    version-derived tag would keep answering 304 here). After
    ``PATCH /api/folders/summary`` AND after ``PATCH
    /api/documents/date`` the tag CHANGES and the conditional GET with
    the OLD tag answers a fresh 200 carrying the new data."""
    _seed_bust_corpus(db)
    try:
        version_before = db.scalar(select(func.max(SourcesMeta.version)))
        tree_url = "/api/docs/tree"
        node_url = "/api/docs/tree/node?source=Homelab&path=a"
        old_tree = admin_client.get(tree_url).headers["etag"]
        old_node = admin_client.get(node_url).headers["etag"]

        r = admin_client.patch(
            "/api/folders/summary",
            json={"source": "Homelab", "folder_path": "a", "summary": "Handwritten."},
        )
        assert r.status_code == 200
        t = admin_client.get(tree_url, headers={"If-None-Match": old_tree})
        assert t.status_code == 200 and t.headers["etag"] != old_tree
        assert _find_child(t.json()["sources"][0], "a")["summary"] == "Handwritten."
        n = admin_client.get(node_url, headers={"If-None-Match": old_node})
        assert n.status_code == 200 and n.headers["etag"] != old_node
        assert n.json()["summary"] == "Handwritten."

        # The date edit moves the subtree max (deep.md is "a"'s max in
        # this corpus) — the tree, the node AND /api/docs all change.
        old_tree2 = admin_client.get(tree_url).headers["etag"]
        old_docs = admin_client.get("/api/docs").headers["etag"]
        deep_max_before = _find_child(
            admin_client.get(tree_url).json()["sources"][0], "a"
        )["updated_at"]
        r = admin_client.patch(
            "/api/documents/date",
            json={"source": "Homelab", "path": "a/b/c/deep.md", "date": "2020-01-01"},
        )
        assert r.status_code == 200
        t2 = admin_client.get(tree_url, headers={"If-None-Match": old_tree2})
        assert t2.status_code == 200 and t2.headers["etag"] != old_tree2
        deep_max_after = _find_child(t2.json()["sources"][0], "a")["updated_at"]
        assert deep_max_after != deep_max_before  # the body really moved
        d2 = admin_client.get("/api/docs", headers={"If-None-Match": old_docs})
        assert d2.status_code == 200 and d2.headers["etag"] != old_docs
        deep = next(d for d in d2.json()["documents"] if d["path"] == "a/b/c/deep.md")
        assert deep["created_at"].startswith("2020-01-01")
        # A4's other half: NEITHER PATCH bumped the KB generation (that
        # would mark every saved chat stale — phase 53).
        assert db.scalar(select(func.max(SourcesMeta.version))) == version_before
    finally:
        _truncate_tree_tables(db)


def test_gzipped_catalogue_200_carries_etag_and_conditional_304(
    admin_client: TestClient, db
) -> None:
    """The compression layer and the validator coexist: the gzipped 200
    of the tree still carries the ``ETag`` (the header survives the
    encoding), and the conditional request answers a body-less 304 —
    no ``Content-Encoding`` on a response with no body."""
    _seed_gzip_corpus(db)
    try:
        # depth=8: the whole 50-doc tree stays above the 860-byte floor
        # (the default depth-1 body would ride under it, identity).
        r = admin_client.get("/api/docs/tree?depth=8", headers={"Accept-Encoding": "gzip"})
        assert r.status_code == 200
        assert r.headers["content-encoding"] == "gzip"
        tag = r.headers["etag"]
        assert tag.startswith('W/"')
        assert r.headers["cache-control"] == "private, must-revalidate"
        hit = admin_client.get(
            "/api/docs/tree?depth=8",
            headers={"Accept-Encoding": "gzip", "If-None-Match": tag},
        )
        assert hit.status_code == 304
        assert hit.content == b""
        assert "content-encoding" not in hit.headers
        assert hit.headers["etag"] == tag
        assert hit.headers["cache-control"] == "private, must-revalidate"
    finally:
        _truncate_tree_tables(db)