phase: 167_api_payload_diet

All gates green. Final report:

**Verified (phase 167 final pass — all 6 tasks already in `complete/`)**
- Live re-measure of the headline criterion on the **1,000-doc live-replica corpus** (restored `data/bor_live_kb_replica.dump.sql` into a scratch DB `bor_audit`, alembic 0011→0027, dropped after; the shared dev DB was never touched): `GET /api/docs/tree` default depth=1 = **8,474 identity → 1,128 B with `Content-Encoding: gzip`** (< 40 KB bar; ~2 MB defect fixed), repeat load **304 / 0 bytes**; depth=8 265,147→23,515; node drill 20,844→1,065; `/api/docs` 246,366→45,104; `/api/documents/content` 892,124→60,776 (`LOOK`, accepted-with-why). Reproduces `AUDIT.md` to within 1 byte.
- Transport contracts proven live: `POST /api/chat` **no** `Content-Encoding` + frames arriving live + budget line `… enc=identity duration_ms=7380 stream=1`; HTML still `no-cache` with the `?v=` rewrite (and gzipped); A20 headers on 200 **and** 304; ETag identical across identity/gzip; `depth=0|9|abc`→422; anonymous tree/node→403; unknown folder/source→404 `unknown node`.

**Gates (all 100% green)**
- `uv run pytest` → **3980 passed**; `uv run pytest --cov=app --cov-report=term-missing` → **TOTAL 99%** (`compression.py`, `http_cache.py`, `response_budget.py` 100%; `api/docs.py` 99%, one defensive dedupe line).
- `uv run ruff check .` → All checks passed; `uv run pyright` → 0 errors.
- Phase + regression E2E **in isolation** (`--junitxml`, exit codes): `test_kb_tree_lazy_load` 11, `test_kb_tree` 8, `test_kb_tree_nav` 4, `test_navbar_refresh` 7, `test_document_dates` 6, `test_git_source_dates` 4, `test_sync_summary_visibility` 3, `test_table_ux_polish` 6, `test_sources_midstream_bug` 6, `test_ls_tree_drilldown` 3, `test_smoke` 3 — **0 failures/errors** (the earlier `-q -q` run had simply suppressed pytest's summary line).

**Documentation** — README.md + AGENTS.md reviewed in full (1,436 / 80 lines):
- Fixed (pre-existing, phase-167-exposed): README *Caching & deploys* claimed "One Starlette middleware (`app/core/caching.py`) applies the rule" while the section now describes three transport layers → rewritten to name caching/compression/http_cache, the real outermost→innermost order, and why `bytes=` is uncompressed.
- AGENTS.md rule 5 (UI Structure Check) gained the phase-167 clause: a lazily-drilled level keeps the current level rendered, its in-flight `role="status"` / `role="alert"`+Retry row lives in the same bounded scroll panel, and stat cards read server `totals` (never a walk of a depth-limited tree).
- Cross-checked and consistent: every `BOR_*` in `.env.example` is in the README table, every `Settings` field is documented, all 15 README-cited API routes exist in the app, no stale "one fetch / drill client-side" claim remains anywhere.

**Notable decisions / deviations**
- `compression.py` adds a documented `_RebufferSmallResponses` seam beyond the D1 sketch (the caching layer's `BaseHTTPMiddleware` re-framing otherwise defeats Starlette's `minimum_size` exemption and inflates tiny responses) — integration-pinned, math still Starlette's.
- A node-route **404 = vanished folder → honest reset to top level** (per `00_phase.md` D4/D6), not task-06's "alert + Retry"; 500 shows alert+Retry with the level preserved — both pinned, nothing weakened.
- Retrieval battery not measurable (shared E2E DB holds no corpus), but `git diff --name-only -- app/rag` is empty, so retrieval is unchanged by construction. `AUDIT.md` left byte-identical (it lives under `todo/`, off-limits to this pass); no git index/history commands run.

**Next pending phase:** none — `167_api_payload_diet` is the only entry in `.agents/phases/todo/`; the harness moves it to `complete/` and commits.

tests/unit/test_compression.py

@@ -0,0 +1,497 @@
+"""Unit tests: ``app.core.compression.SelectiveGZipMiddleware`` (phase 167, D1, LOCKED A1).
+
+Pins the three contracts of the selective-gzip layer on a tiny ASGI app —
+
+* a ~50 KB JSON route compresses (``Content-Encoding: gzip``, the body
+  ``gzip.decompress``es to the original, ``Vary`` mentions
+  ``accept-encoding``), an ``Accept-Encoding: identity`` client gets
+  identity bytes, a ~50-byte route stays under the floor, and a response
+  that already set ``content-encoding`` is forwarded byte-identical (never
+  double-encoded — the fronting-nginx contract);
+
+* the EXACT-path skip: a route named ``/api/chat`` is NOT compressed and
+  streams frame-by-frame (the raw-ASGI interleaving proof: every frame
+  reaches the client's ``send`` BEFORE the app yields the next one — a
+  buffering middleware fails this deterministically — plus the
+  ``list(stream)`` client idiom the existing SSE tests use), while
+  ``/api/chat-images/x`` shares the prefix and MUST compress;
+
+* non-HTTP scopes (``lifespan``/``websocket``) pass straight through.
+
+Plus the config kill switch: ``compress_min_bytes=0`` → ``create_app``
+adds no gzip layer at all; ``-1`` → the validator refuses startup naming
+the field; the middleware sits INSIDE ``SecurityHeadersMiddleware`` (A20)
+and OUTSIDE ``CachingMiddleware``.
+"""
+from __future__ import annotations
+
+import asyncio
+import gzip
+import json
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from pydantic import ValidationError
+from starlette.applications import Starlette
+from starlette.middleware.gzip import GZipMiddleware
+from starlette.requests import Request
+from starlette.responses import Response, StreamingResponse
+from starlette.routing import Route
+from starlette.types import Message, Receive, Scope, Send
+
+import app.main as main_mod
+from app.config import Settings
+from app.core.compression import DEFAULT_SKIP_PATHS, SelectiveGZipMiddleware
+
+
+def _settings(**kwargs: Any) -> Settings:
+    """Build Settings without reading a .env file (the test_config pattern)."""
+    kwargs.setdefault("_env_file", None)
+    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)
+
+
+# ~50 KB of highly compressible JSON — the /api/docs/tree shape class.
+_BIG_BODY = json.dumps(
+    {
+        "items": [
+            {"n": i, "path": f"source/docs/file-{i}.md", "title": f"Doc {i}"}
+            for i in range(1000)
+        ]
+    }
+).encode()
+
+# ~50 bytes — comfortably under the 860-byte floor (compression would inflate).
+_SMALL_BODY = json.dumps({"ok": True, "note": "intentionally tiny body"}).encode("utf-8")
+
+# Five SSE frames (~2 KB total — ABOVE the floor, so a missing skip cannot
+# hide behind the minimum_size exemption).
+_SSE_FRAMES = [
+    f"event: thinking\ndata: {'x' * 400}\n\n".encode() for _ in range(5)
+]
+
+_PRE_ENCODED_BODY = gzip.compress(_BIG_BODY)
+
+
+def _inner_app() -> Starlette:
+    async def big(request: Request) -> Response:
+        return Response(_BIG_BODY, media_type="application/json")
+
+    async def small(request: Request) -> Response:
+        return Response(_SMALL_BODY, media_type="application/json")
+
+    async def chat(request: Request) -> StreamingResponse:
+        async def frames():
+            for frame in _SSE_FRAMES:
+                yield frame
+
+        return StreamingResponse(frames(), media_type="text/event-stream")
+
+    async def chat_images(request: Request) -> Response:
+        # Shares the /api/chat PREFIX but is a different endpoint — the
+        # skip must be EXACT, so this one compresses.
+        return Response(_BIG_BODY, media_type="application/json")
+
+    async def pre_encoded(request: Request) -> Response:
+        # A body that arrived already encoded (e.g. from a future CDN
+        # layer): the responder skips it, never double-encodes.
+        return Response(
+            _PRE_ENCODED_BODY,
+            media_type="application/json",
+            headers={"Content-Encoding": "gzip"},
+        )
+
+    return Starlette(
+        routes=[
+            Route("/big", big),
+            Route("/small", small),
+            Route("/api/chat", chat, methods=["POST"]),
+            Route("/api/chat-images/x", chat_images),
+            Route("/pre-encoded", pre_encoded),
+        ]
+    )
+
+
+def _client(minimum_size: int = 860) -> TestClient:
+    return TestClient(
+        SelectiveGZipMiddleware(_inner_app(), minimum_size=minimum_size)
+    )
+
+
+# ---------------------------------------------------------------------------
+# The gzip path (Accept-Encoding negotiation, the floor, the pre-encoded skip)
+# ---------------------------------------------------------------------------
+
+
+def test_large_json_compresses_decompresses_and_sets_vary() -> None:
+    client = _client()
+    response = client.get("/big", headers={"Accept-Encoding": "gzip"})
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert "accept-encoding" in response.headers.get("vary", "").lower()
+    # httpx transparently decodes: the documented JSON arrives whole.
+    assert response.content == _BIG_BODY
+    assert len(response.content) > 860  # the seed really is above the floor
+
+    # The RAW wire bytes are a real gzip stream of the original body.
+    with client.stream("GET", "/big", headers={"Accept-Encoding": "gzip"}) as streamed:
+        raw = b"".join(list(streamed.iter_raw()))
+    assert gzip.decompress(raw) == _BIG_BODY
+    assert len(raw) < len(_BIG_BODY)  # ~8-12x on this JSON; order of magnitude suffices
+
+
+def test_identity_client_is_not_compressed() -> None:
+    client = _client()
+    response = client.get("/big", headers={"Accept-Encoding": "identity"})
+    assert response.status_code == 200
+    assert "content-encoding" not in response.headers
+    assert response.content == _BIG_BODY
+
+
+def test_small_response_stays_under_the_floor() -> None:
+    client = _client()
+    response = client.get("/small", headers={"Accept-Encoding": "gzip"})
+    assert response.status_code == 200
+    assert "content-encoding" not in response.headers
+    assert response.content == _SMALL_BODY
+
+
+def test_pre_encoded_body_is_never_double_encoded() -> None:
+    client = _client()
+    with client.stream("GET", "/pre-encoded", headers={"Accept-Encoding": "gzip"}) as streamed:
+        assert streamed.headers["content-encoding"] == "gzip"
+        raw = b"".join(list(streamed.iter_raw()))
+    # Byte-identical passthrough of the already-encoded body (Starlette's
+    # "skip when content-encoding is set" rule — the nginx coexistence pin).
+    assert raw == _PRE_ENCODED_BODY
+    assert gzip.decompress(raw) == _BIG_BODY
+
+
+# ---------------------------------------------------------------------------
+# The EXACT-path skip (A15: the SSE stream is never compressed, never buffered)
+# ---------------------------------------------------------------------------
+
+
+def test_default_skip_paths_constant() -> None:
+    assert DEFAULT_SKIP_PATHS == ("/api/chat",)
+
+
+def test_chat_route_is_not_compressed_and_streams_whole() -> None:
+    """The client-level pin (the ``list(stream)`` idiom the SSE tests use):
+    a route named /api/chat carries NO Content-Encoding and its raw bytes
+    are exactly the SSE frames, concatenated, above the size floor."""
+    client = _client()
+    with client.stream(
+        "POST", "/api/chat", json={"message": "hi"}, headers={"Accept-Encoding": "gzip"}
+    ) as streamed:
+        assert streamed.status_code == 200
+        assert "content-encoding" not in streamed.headers
+        chunks = list(streamed.iter_raw())
+    assert b"".join(chunks) == b"".join(_SSE_FRAMES)
+    assert sum(len(c) for c in chunks) > 860  # the skip beats the floor, not size
+
+
+def test_skipped_stream_arrives_frame_by_frame_uncompressed() -> None:
+    """The frame-by-frame proof (deterministic, no timing): every frame
+    reaches the client's ``send`` BEFORE the app produces the next one.
+    A buffering middleware (BaseHTTPMiddleware, or a gzip pass) yields
+    all ``app:*`` events first — this strict interleaving fails then."""
+    events: list[str] = []  # shared: the app logs a yield, send logs delivery
+    sent: list[Message] = []
+
+    async def sse_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/event-stream")],
+            }
+        )
+        for i, frame in enumerate(_SSE_FRAMES):
+            events.append(f"app:frame-{i}")
+            await send({"type": "http.response.body", "body": frame, "more_body": True})
+        events.append("app:final")
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    async def recording_send(message: Message) -> None:
+        events.append(f"client:{message['type']}")
+        sent.append(message)
+
+    async def receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    middleware = SelectiveGZipMiddleware(sse_app, minimum_size=860)
+    asyncio.run(
+        middleware(
+            {"type": "http", "path": "/api/chat", "headers": [(b"accept-encoding", b"gzip")]},
+            receive,
+            recording_send,
+        )
+    )
+
+    expected = ["client:http.response.start"]
+    for i in range(len(_SSE_FRAMES)):
+        expected += [f"app:frame-{i}", "client:http.response.body"]
+    expected += ["app:final", "client:http.response.body"]
+    assert events == expected
+
+    start = sent[0]
+    headers = {k.decode().lower() for k, _ in start["headers"]}  # pyright: ignore[reportGeneralTypeIssues]
+    assert "content-encoding" not in headers
+    body = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
+    assert body == b"".join(_SSE_FRAMES)
+
+
+def _raw_harness(
+    *,
+    path: str,
+    app: Any,
+) -> list[Message]:
+    """Drive the middleware at the raw ASGI level; return the exact
+    messages that reached the client's ``send``."""
+    sent: list[Message] = []
+
+    async def recording_send(message: Message) -> None:
+        sent.append(message)
+
+    async def receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    scope: Scope = {
+        "type": "http",
+        "path": path,
+        "headers": [(b"accept-encoding", b"gzip")],
+    }
+    middleware = SelectiveGZipMiddleware(app, minimum_size=860)
+    asyncio.run(middleware(scope, receive, recording_send))
+    return sent
+
+
+def test_same_stream_at_a_non_skipped_path_goes_through_gzip() -> None:
+    """The skip is PATH-driven: the identical streaming app mounted at a
+    non-skipped path is compressed by the composed GZipMiddleware."""
+
+    async def json_stream_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"application/json")],
+            }
+        )
+        for i, frame in enumerate(_SSE_FRAMES):
+            await send(
+                {"type": "http.response.body", "body": frame, "more_body": i < len(_SSE_FRAMES) - 1}
+            )
+
+    sent = _raw_harness(path="/api/other-stream", app=json_stream_app)
+    start_headers = {
+        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
+    }
+    assert start_headers.get("content-encoding") == "gzip"
+    raw = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
+    assert gzip.decompress(raw) == b"".join(_SSE_FRAMES)
+
+
+def test_reframed_small_body_stays_identity_under_the_floor() -> None:
+    """The CachingMiddleware seam (starlette 1.x ``BaseHTTPMiddleware``
+    re-frames every response it passes up: chunk with ``more_body=True``
+    + empty terminator): a body whose ``content-length`` is below the
+    floor is re-buffered into ONE buffered message so Starlette's own
+    small-response exemption fires — identity bytes, no ``Vary`` append."""
+    payload = b'{"status":"ok","version":"0.1.0"}'  # 33 bytes — the /api/health class
+
+    async def reframed_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    (b"content-type", b"application/json"),
+                    (b"content-length", str(len(payload)).encode()),
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": payload, "more_body": True})
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    sent = _raw_harness(path="/api/health", app=reframed_app)
+    start_headers = {
+        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
+    }
+    assert "content-encoding" not in start_headers
+    assert "vary" not in start_headers  # the exemption adds nothing
+    body = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
+    assert body == payload
+
+
+def test_lying_content_length_that_outgrows_the_floor_compresses() -> None:
+    """A body that outgrows the floor despite a small ``content-length``
+    stops the hold and streams through the normal compressing path — the
+    buffer is bounded by the floor and correctness never depends on the
+    header being honest."""
+
+    async def lying_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    (b"content-type", b"application/json"),
+                    (b"content-length", b"5"),  # a lie
+                ],
+            }
+        )
+        await send({"type": "http.response.body", "body": _BIG_BODY, "more_body": True})
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    sent = _raw_harness(path="/api/lias", app=lying_app)
+    start_headers = {
+        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
+    }
+    assert start_headers.get("content-encoding") == "gzip"
+    raw = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
+    assert gzip.decompress(raw) == _BIG_BODY
+
+
+def test_non_body_message_while_holding_flushes_untouched() -> None:
+    """``http.response.pathsend`` (or anything unexpected) arriving while
+    the hold is active: the held messages and the message itself are
+    forwarded byte-for-byte — the hold never swallows a frame."""
+
+    async def pathsend_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    (b"content-type", b"text/plain"),
+                    (b"content-length", b"4"),
+                ],
+            }
+        )
+        await send({"type": "http.response.pathsend", "path": "/tmp/file"})
+
+    sent = _raw_harness(path="/assets/file", app=pathsend_app)
+    assert [m["type"] for m in sent] == ["http.response.start", "http.response.pathsend"]
+    start_headers = {k.decode().lower() for k, _ in sent[0]["headers"]}  # pyright: ignore[reportGeneralTypeIssues]
+    assert "content-encoding" not in start_headers
+
+
+def test_chat_images_prefix_still_compresses() -> None:
+    """The exact-match pin: /api/chat-images/x shares the /api/chat
+    prefix and MUST go through the gzip path."""
+    client = _client()
+    response = client.get("/api/chat-images/x", headers={"Accept-Encoding": "gzip"})
+    assert response.status_code == 200
+    assert response.headers["content-encoding"] == "gzip"
+    assert response.content == _BIG_BODY
+
+
+# ---------------------------------------------------------------------------
+# Non-HTTP scopes pass straight through
+# ---------------------------------------------------------------------------
+
+
+def test_non_http_scopes_pass_through_untouched() -> None:
+    seen: list[str] = []
+
+    async def probe_app(scope: Scope, receive: Receive, send: Send) -> None:
+        seen.append(scope["type"])
+
+    async def receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    async def send(message: Message) -> None:  # pragma: no cover - app sends nothing
+        raise AssertionError("no ASGI messages expected")
+
+    middleware = SelectiveGZipMiddleware(probe_app, minimum_size=860)
+    for scope_type in ("lifespan", "websocket"):
+        asyncio.run(
+            middleware({"type": scope_type, "path": "/api/anything"}, receive, send)
+        )
+    assert seen == ["lifespan", "websocket"]
+
+
+def test_composes_starlette_gzip_with_minimum_size() -> None:
+    """The middleware reuses Starlette's GZipMiddleware, it does not fork
+    the compression math (D1)."""
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None: ...
+
+    middleware = SelectiveGZipMiddleware(app, minimum_size=1234)
+    assert isinstance(middleware.gzip, GZipMiddleware)
+    assert middleware.gzip.minimum_size == 1234
+    assert middleware.skip_paths == DEFAULT_SKIP_PATHS
+
+
+# ---------------------------------------------------------------------------
+# Config: the BOR_COMPRESS_MIN_BYTES kill switch and the loud negative
+# ---------------------------------------------------------------------------
+
+
+def _middleware_names(app: Any) -> list[str]:
+    """Middleware class names, outermost first (``user_middleware`` order)."""
+    return [getattr(mw.cls, "__name__", "") for mw in app.user_middleware]
+
+
+def test_compress_min_bytes_defaults_to_the_nginx_floor(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("BOR_COMPRESS_MIN_BYTES", raising=False)
+    assert _settings().compress_min_bytes == 860
+
+
+def test_compress_min_bytes_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("BOR_COMPRESS_MIN_BYTES", "2048")
+    assert _settings().compress_min_bytes == 2048
+    monkeypatch.setenv("BOR_COMPRESS_MIN_BYTES", "0")  # the kill switch is legal
+    assert _settings().compress_min_bytes == 0
+
+
+def test_compress_min_bytes_negative_refuses_startup_naming_the_field() -> None:
+    with pytest.raises(ValidationError, match="compress_min_bytes must be >= 0"):
+        _settings(compress_min_bytes=-1)
+
+
+def test_create_app_wires_gzip_inside_security_headers(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A20 holds: SecurityHeaders stays outermost (index 0 — starlette's
+    last add_middleware is the outermost layer), gzip sits between it and
+    the caching layer, so an already-rewritten HTML page compresses and
+    the /api/* caching pass-through happens before compression.
+
+    Phase 167 task 05 adds ``ResponseBudgetMiddleware`` between gzip and
+    caching — INSIDE the compressor on purpose, so its ``bytes=`` counts the
+    body the ROUTE produced while its ``enc=`` still reads the encoding the
+    compressor put on the wire (see tests/unit/test_response_budget.py)."""
+    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 860)
+    app2 = main_mod.create_app()
+    assert _middleware_names(app2) == [
+        "SecurityHeadersMiddleware",
+        "SelectiveGZipMiddleware",
+        "ResponseBudgetMiddleware",
+        "CachingMiddleware",
+        "SessionMiddleware",
+    ]
+    # The app still serves.
+    assert TestClient(app2).get("/api/health").status_code == 200
+
+
+def test_create_app_adds_no_gzip_layer_when_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """BOR_COMPRESS_MIN_BYTES=0 removes the middleware entirely (the kill
+    switch): the gzip class is absent from app.user_middleware."""
+    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 0)
+    app2 = main_mod.create_app()
+    names = _middleware_names(app2)
+    assert "SelectiveGZipMiddleware" not in names
+    assert names == [
+        "SecurityHeadersMiddleware",
+        # Phase 167 task 05: the byte-budget line is not optional.
+        "ResponseBudgetMiddleware",
+        "CachingMiddleware",
+        "SessionMiddleware",
+    ]

tests/unit/test_frontend_router.py

@@ -572,11 +572,13 @@ def test_rag_view_refetches_on_reshow() -> None:
     """Phase 77 task 02 (+ phase 97 task 04, + phase 99 task 02):
     the RAG (knowledge base) view re-fetches on a user-initiated
     re-show — sources.js listens and re-runs ``loadTree()`` (the
-    phase-97 catalog load: ONE fetch of ``GET /api/docs/tree``), now
-    AFTER the phase-99 history alignment (adopt the entry's kb / reset
-    to the top — a block-arrow listener, hence the ``});`` close in
-    the pin helper below). The load is race-tokened (phase 79) and
-    the RE-ENTRANT render — ``renderLevel`` clears BOTH row containers
+    catalog load — ONE fetch of ``GET /api/docs/tree?depth=1`` since
+    phase 167's payload diet; the view loads the catalogue from the
+    TREE endpoint, not the flat one), now AFTER the phase-99 history
+    alignment (adopt the entry's kb / reset to the top — a block-arrow
+    listener, hence the ``});`` close in the pin helper below). The load
+    is race-tokened (phase 79) and the RE-ENTRANT render —
+    ``renderLevel`` clears BOTH row containers
     at the TOP before filling them (the History pattern from task 01,
     extended to the folders table) — so a refresh from a populated
     level into a sparser (or empty) result replaces the rows instead
@@ -588,7 +590,9 @@ def test_rag_view_refetches_on_reshow() -> None:
     load = js.find("async function loadTree()")
     assert load != -1, "loadTree must exist"
     body = js[load : js.find("\n  }", load)]
-    assert 'fetch("/api/docs/tree")' in body, "the load must fetch the tree endpoint"
+    assert 'fetch("/api/docs/tree?depth=1")' in body, (
+        "the load must fetch the tree endpoint (depth-limited, phase 167)"
+    )
     assert "++loadSeq" in body, "the race token stays (phase 79)"
     render = js.find("function renderLevel()")
     assert render != -1, "renderLevel must exist"

tests/unit/test_http_cache.py

@@ -0,0 +1,110 @@
+"""Unit tests: ``app/core/http_cache.py`` (phase 167, task 03).
+
+The three pure helpers behind the catalogue reads' revalidation
+(LOCKED A4): the weak content-hash tag (shape, stability, the bytes-vs-
+str equality), the tolerant ``If-None-Match`` evaluator (``None`` /
+empty / ``*`` / single / comma-list / weak↔strong both directions /
+whitespace / the prefix-is-not-a-match pin), and the ONE header pair.
+"""
+from __future__ import annotations
+
+import re
+
+from app.core.http_cache import (
+    REVALIDATE_CACHE_CONTROL,
+    etag_matches,
+    revalidate_headers,
+    weak_etag,
+)
+
+_TAG_RE = re.compile(r'W/"[0-9a-f]{32}"')
+
+
+def test_weak_etag_shape_is_weak_32_hex() -> None:
+    """``W/"<sha256 hexdigest[:32]>"`` — the prefix, the 32-hex-char body,
+    the total 36-char length (the file's truncation convention)."""
+    tag = weak_etag("hello")
+    assert _TAG_RE.fullmatch(tag), tag
+    assert len(tag) == 36
+    assert tag.startswith('W/"') and tag.endswith('"')
+
+
+def test_weak_etag_is_deterministic_and_content_sensitive() -> None:
+    assert weak_etag("same body") == weak_etag("same body")
+    assert weak_etag("same body") != weak_etag("same body!")
+    assert weak_etag("") != weak_etag(" ")
+
+
+def test_weak_etag_bytes_and_str_forms_agree() -> None:
+    """The hash is over the UTF-8 bytes either way — the endpoint may
+    hash the ``str`` and serve its encoding without diverging."""
+    text = '{"tree": "héllo — ünïcode"}'
+    assert weak_etag(text) == weak_etag(text.encode("utf-8"))
+
+
+def test_weak_etag_truncation_matches_the_sha256_prefix() -> None:
+    import hashlib
+
+    body = "exact served bytes"
+    expected = hashlib.sha256(body.encode("utf-8")).hexdigest()[:32]
+    assert weak_etag(body) == f'W/"{expected}"'
+
+
+def test_etag_matches_none_and_empty_never_match() -> None:
+    """No conditional header (or an empty/whitespace-only one) is an
+    unconditional GET → the route must answer 200."""
+    assert etag_matches(None, 'W/"x"') is False
+    assert etag_matches("", 'W/"x"') is False
+    assert etag_matches("   ", 'W/"x"') is False
+
+
+def test_etag_matches_star() -> None:
+    """RFC 9110 §13.3.3: ``*`` matches anything the server holds."""
+    assert etag_matches("*", 'W/"x"') is True
+    assert etag_matches(" * ", 'W/"anything"') is True
+
+
+def test_etag_matches_single_candidate() -> None:
+    assert etag_matches('W/"x"', 'W/"x"') is True
+    assert etag_matches('W/"x"', 'W/"y"') is False
+
+
+def test_etag_matches_comma_separated_list() -> None:
+    header = 'W/"a", W/"b", W/"c"'
+    assert etag_matches(header, 'W/"a"') is True
+    assert etag_matches(header, 'W/"c"') is True
+    assert etag_matches(header, 'W/"d"') is False
+    # no space after the comma — a real client sends both shapes
+    assert etag_matches('W/"a",W/"b"', 'W/"b"') is True
+
+
+def test_etag_matches_weak_strong_equivalence_both_directions() -> None:
+    """``W/"x"`` matches ``"x"`` whichever side carries the weakness
+    marker (the stored form must not decide the outcome)."""
+    assert etag_matches('"x"', 'W/"x"') is True  # client strong, ours weak
+    assert etag_matches('W/"x"', '"x"') is True  # client weak, ours strong
+
+
+def test_etag_matches_is_whitespace_tolerant() -> None:
+    assert etag_matches('  W/"x"  ', 'W/"x"') is True
+    assert etag_matches('W/"a" ,  "x" ', 'W/"x"') is True
+    assert etag_matches('W/ "x"', 'W/"x"') is True
+
+
+def test_etag_matches_prefix_is_not_a_match() -> None:
+    """A tag that merely SHARES a prefix with ours is a mismatch — the
+    comparison is the full 32-hex-char digest, not ``startswith``."""
+    ours = "abcdef0123456789abcdef0123456789"
+    shorter = ours[:-1]
+    assert etag_matches(f'W/"{shorter}"', f'W/"{ours}"') is False
+    assert etag_matches(f'"{shorter}"', f'W/"{ours}"') is False
+
+
+def test_revalidate_headers_is_the_one_pair() -> None:
+    """The 200 AND the 304 carry exactly this pair (A19's read-only API
+    arm: private + must-revalidate beside the validator)."""
+    assert revalidate_headers('W/"x"') == {
+        "ETag": 'W/"x"',
+        "Cache-Control": "private, must-revalidate",
+    }
+    assert REVALIDATE_CACHE_CONTROL == "private, must-revalidate"

tests/unit/test_kb_tree_builder.py

@@ -11,8 +11,18 @@ file metadata, and the ``group_folder_listing`` cross-check property
 """
 from __future__ import annotations
 
-from app.api.docs import TreeDocRow, build_kb_tree
+import pytest
+from pydantic import ValidationError
+
+from app.api.docs import (
+    TreeDocRow,
+    _folder_counts,
+    _kb_totals,
+    _subtree_max,
+    build_kb_tree,
+)
 from app.rag.agent import group_folder_listing
+from app.schemas import KbTree, KbTreeFile, KbTreeTotals
 
 T0 = "2026-09-01T08:00:00+00:00"
 T1 = "2026-09-02T08:00:00+00:00"
@@ -633,3 +643,284 @@ def test_image_affordance_absent_without_the_map() -> None:
     assert set(file.model_dump()) == {
         "kind", "path", "title", "chunks", "created_at", "indexed_at"
     }
+
+
+# ---------------------------------------------------------------------------
+# Phase 167 (task 02, D2/A2) — the depth budget: a node whose children were
+# NOT built says so (``truncated``) while keeping every field the Sources
+# view renders (recursive count, recursive updated_at — the D9 max via the
+# _folder_counts max-date map, summary, summary_pending), the untruncated
+# wire shape stays byte-identical, and the catalogue totals ride the
+# response (``_kb_totals`` — the stat cards cannot be summed client-side
+# off a truncated tree).
+# ---------------------------------------------------------------------------
+
+#: The depth fixture: one chain four levels deep, distinct created dates
+#: (C0 < C1 < C2 < C3 — the DEEPEST file holds the max, the D9 trap) and
+#: distinct indexed stamps.
+DEPTH_ROWS: list[TreeDocRow] = [
+    ("S", "root.md", "Root", 1, T0, C0),
+    ("S", "a/top.md", "Top", 2, T1, C1),
+    ("S", "a/b/mid.md", "Mid", 3, T1, C2),
+    ("S", "a/b/c/deep.md", "Deep", 4, T2, C3),
+]
+
+DEPTH_SUMMARIES = {("S", "a"): "A desc."}  # a/b and a/b/c stay un-stored
+
+
+def _all_nodes(node):
+    """The node plus every descendant node (folders and files)."""
+    yield node
+    for child in getattr(node, "children", []):
+        if child.kind == "folder":
+            yield from _all_nodes(child)
+        else:
+            yield child
+
+
+def _all_folder_nodes(node):
+    """Every FOLDER descendant (the SOURCE node itself is excluded — it
+    carries ``name``, not ``path``, and has no ``kind``)."""
+    return [n for n in _all_nodes(node) if getattr(n, "kind", None) == "folder"]
+
+
+def _paths_under(node) -> set[str]:
+    """Paths of every descendant (folders and files); the source node
+    has no path of its own."""
+    out: set[str] = set()
+    for child in getattr(node, "children", []):
+        out.add(child.path)
+        if child.kind == "folder":
+            out |= _paths_under(child)
+    return out
+
+
+def test_depth_none_is_byte_identical_to_the_unlimited_build() -> None:
+    """``depth=None`` (the builder default) reproduces today's tree
+    byte-for-byte: the default call, the explicit None, and an
+    over-deep budget all serialize identically, and NO node carries the
+    truncated key (the omission rule)."""
+    default = [s.model_dump() for s in build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES)]
+    explicit = [
+        s.model_dump() for s in build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=None)
+    ]
+    over_deep = [
+        s.model_dump() for s in build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=8)
+    ]
+    assert default == explicit == over_deep
+    (source,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES)
+    assert all("truncated" not in n.model_dump() for n in _all_nodes(source))
+    # The full chain IS built: every level of the fixture is present.
+    assert _paths_under(source) == {
+        "root.md",
+        "a",
+        "a/top.md",
+        "a/b",
+        "a/b/mid.md",
+        "a/b/c",
+        "a/b/c/deep.md",
+    }
+
+
+def test_depth_one_ships_sources_plus_one_level_with_truncated_flags() -> None:
+    """``depth=1``: the source's direct children are built (direct file
+    ``root.md`` included) and the subfolder with unloaded contents
+    carries the FULL truncation field set — ``truncated is True``,
+    ``children == []``, the RECURSIVE count, the RECURSIVE max date
+    (the deep C3 — D9 survives without the children), the stored
+    ``summary`` and the ``summary_pending`` flag (rule unchanged)."""
+    (source,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=1)
+    assert [f.path for f in _file_nodes(source)] == ["root.md"]
+    (a,) = _folder_nodes(source)
+    assert a.path == "a"
+    assert a.truncated is True
+    assert a.model_dump()["truncated"] is True
+    assert a.children == []
+    assert a.documents == 3  # recursive — the count rule is depth-proof
+    assert a.updated_at == C3  # the DEEP max, no built children to derive from
+    assert a.summary == "A desc."  # the level row shows a subfolder's description
+    assert a.summary_pending is False  # the stored row covers it (phase 98)
+    # Nothing deeper is built anywhere.
+    assert _paths_under(source) == {"root.md", "a"}
+
+
+def test_depth_two_builds_two_levels_and_truncates_the_deeper_one() -> None:
+    """``depth=2``: level 1 (``a``) is expanded UNTRUNCATED (its direct
+    file ``a/top.md`` ships), level 2 (``a/b``) is the truncated node —
+    the pending rule applies to it (2 docs, no stored row →
+    ``summary_pending`` true with ``summary None``), and neither its
+    own file (``a/b/mid.md``) nor the third level appear: the budget
+    removes the LEVELS below it, level by level."""
+    (source,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=2)
+    (a,) = _folder_nodes(source)
+    assert "truncated" not in a.model_dump()  # untruncated: key absent
+    assert [f.path for f in _file_nodes(a)] == ["a/top.md"]
+    (a_b,) = _folder_nodes(a)
+    assert a_b.truncated is True
+    assert a_b.children == []
+    assert (a_b.documents, a_b.updated_at) == (2, C3)  # mid + deep, max C3
+    assert a_b.summary is None
+    assert a_b.summary_pending is True  # the phase-98 rule rides truncated nodes
+    assert _paths_under(source) == {
+        "root.md",
+        "a",
+        "a/top.md",
+        "a/b",
+    }
+
+
+def test_source_node_is_never_truncated_at_any_depth() -> None:
+    """The view's top level is source → level 1, so the budget never
+    marks the SOURCE node truncated (at any depth, empty sources
+    included) — its wire shape never carries the key."""
+    (ref,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES)
+    for depth in (1, 2, 3, 8):
+        full, empty = build_kb_tree(
+            ["S", "Empty"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=depth
+        )
+        assert full.truncated is False
+        assert "truncated" not in full.model_dump()
+        assert empty.truncated is False
+        assert "truncated" not in empty.model_dump()
+        assert empty.children == []
+        # Direct children always built: the source's level is complete.
+        assert [c.path for c in full.children] == [c.path for c in ref.children]
+
+
+def test_truncated_key_never_touches_the_file_node_shape() -> None:
+    """A FILE node has no children to truncate: the field does not even
+    exist on the class, and no dump ever carries it (the phase-122 text
+    wire shape stays byte-identical at every depth)."""
+    assert "truncated" not in KbTreeFile.model_fields
+    for depth in (1, 2, 8):
+        (source,) = build_kb_tree(["S"], DEPTH_ROWS, {}, depth=depth)
+        file_nodes = [n for n in _all_nodes(source) if getattr(n, "kind", None) == "file"]
+        assert file_nodes  # the fixture has files at every shipped level
+        for node in file_nodes:
+            assert "truncated" not in node.model_dump()
+
+
+def test_max_dates_map_agrees_with_subtree_max_at_every_depth() -> None:
+    """One concept, two derivations, cross-pinned: the
+    ``_folder_counts`` max-date map equals ``_subtree_max`` over the
+    FULLY expanded children for every folder (the expanded builder path
+    stays ``_subtree_max``, so a divergence fails loudly) — and every
+    node present in a depth-limited tree carries the SAME
+    ``updated_at`` as the full build, truncated or not (the phase-106
+    D9 column survives the payload diet)."""
+    file_rows = [
+        (path, title, chunks, indexed, created)
+        for _source, path, title, chunks, indexed, created in DEPTH_ROWS
+    ]
+    _folders, _counts, max_dates = _folder_counts(file_rows)
+    (full,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES)
+    full_dates = {n.path: n.updated_at for n in _all_folder_nodes(full)}
+    assert full_dates == max_dates  # the cross-pin at full expansion
+    for node in _all_folder_nodes(full):
+        # The expanded node's derivation equals the aggregate map.
+        assert _subtree_max(node.children) == max_dates[node.path]
+    for depth in (1, 2, 3):
+        (limited,) = build_kb_tree(["S"], DEPTH_ROWS, DEPTH_SUMMARIES, depth=depth)
+        for node in _all_folder_nodes(limited):
+            assert node.updated_at == max_dates[node.path] == full_dates[node.path]
+        # The source max also spans the UNSENT levels.
+        assert limited.updated_at == C3
+
+
+def test_truncation_keeps_the_phase_94_and_98_invariants_at_every_depth() -> None:
+    """Existence rule, ordering, counts, summaries, pending markers and
+    dates: every node a limited tree DOES send is field-for-field the
+    node the full tree sends (the depth budget removes depth only — it
+    never changes what a shipped level says), and the top level is
+    identical at every depth."""
+    (full,) = build_kb_tree(["S"], CROSS_ROWS, CROSS_SUMMARIES)
+    full_map = {
+        n.path: (n.documents, n.summary_pending, n.updated_at, n.summary)
+        for n in _all_folder_nodes(full)
+    }
+    for depth in (1, 2, 3):
+        (limited,) = build_kb_tree(["S"], CROSS_ROWS, CROSS_SUMMARIES, depth=depth)
+        # Same level-1 rows, same order, at every depth.
+        assert [c.path for c in limited.children] == [c.path for c in full.children]
+        for node in _all_folder_nodes(limited):
+            assert (
+                node.documents,
+                node.summary_pending,
+                node.updated_at,
+                node.summary,
+            ) == full_map[node.path]
+            # Existence rule: a node is present ⟺ a catalogue prefix.
+            assert node.path in full_map
+        # Every truncated node is exactly the truncation contract.
+        for node in _all_folder_nodes(limited):
+            if node.truncated:
+                assert node.children == []
+            else:
+                assert node.model_dump().get("truncated") is None
+
+
+def test_truncated_nodes_keep_the_image_affordance_at_their_level() -> None:
+    """Phase 122's affordance is row-driven, not depth-driven: a DIRECT
+    image file of the source root still carries the thumbnail keys at
+    ``depth=1``; the truncated folder below simply does not list its
+    own rows (they arrive with the drill)."""
+    rows = [
+        ("S", "pic.png", "pic", 1, T0, C0),
+        ("S", "d/x.png", "x", 1, T0, C0),
+    ]
+    images = {
+        ("S", "pic.png"): (DOC_ID, "A red square."),
+        ("S", "d/x.png"): (DOC_ID, "A blue circle."),
+    }
+    (source,) = build_kb_tree(["S"], rows, {}, images, depth=1)
+    (pic,) = _file_nodes(source)
+    assert pic.is_image is True
+    assert pic.model_dump()["image_url"] == f"/api/documents/{DOC_ID}/image"
+    assert pic.model_dump()["summary"] == "A red square."
+    (d,) = _folder_nodes(source)
+    assert d.truncated is True
+    assert d.children == []
+
+
+# ---------------------------------------------------------------------------
+# _kb_totals (phase 167, task 02) — the stat-card totals the endpoint ships
+# beside any depth of tree (no extra query, empty-KB honest).
+# ---------------------------------------------------------------------------
+
+
+def test_kb_totals_math_over_the_catalogue_rows() -> None:
+    """documents = the row count, chunks = the sum, last_indexed = the
+    MAX indexed stamp (lexicographic over one ISO shape = chronological
+    — the same rule as the D9 max)."""
+    totals = _kb_totals(DEPTH_ROWS)
+    assert totals.documents == 4
+    assert totals.chunks == 10  # 1 + 2 + 3 + 4
+    assert totals.last_indexed == T2  # the newest indexed_at
+
+
+def test_kb_totals_empty_kb_is_zero_and_none() -> None:
+    """The empty KB: 0 documents, 0 chunks, ``last_indexed is None`` —
+    the stat cards render the designed empty state, never a crash."""
+    totals = _kb_totals([])
+    assert totals == KbTreeTotals(documents=0, chunks=0, last_indexed=None)
+    assert totals.model_dump() == {
+        "documents": 0,
+        "chunks": 0,
+        "last_indexed": None,
+    }
+
+
+def test_kb_tree_totals_default_and_floor_validation() -> None:
+    """``KbTree.totals`` is an additive top-level key with an honest
+    empty default (an older call shape still serializes), and the
+    ge=0 bounds refuse negative totals loudly (the endpoint computes
+    them — a negative is a bug, never data)."""
+    assert KbTree(sources=[]).model_dump()["totals"] == {
+        "documents": 0,
+        "chunks": 0,
+        "last_indexed": None,
+    }
+    with pytest.raises(ValidationError):
+        KbTreeTotals(documents=-1, chunks=0)
+    with pytest.raises(ValidationError):
+        KbTreeTotals(documents=0, chunks=-1)

tests/unit/test_kb_tree_ui.py

@@ -5,10 +5,13 @@ The RAG (Knowledge base) view lists the catalog the way the agent's
 top, then per level the subfolders (recursive count + the STORED
 description) and the level's files (the ``#docs-table`` — phase 106,
 task 08, D8, added the Created column BEFORE Indexed, making it the
-6-column contract). The view's catalog load is now
-``loadTree()``: ONE fetch of ``GET /api/docs/tree`` (the full recursive
-tree in a single payload — task 02), then client-side drill navigation
-(no per-level fetch, no URL change).
+6-column contract). The view's catalog load is ``loadTree()``: ONE
+fetch of ``GET /api/docs/tree?depth=1`` (sources + one level + the
+server's ``totals``) plus ONE ``GET /api/docs/tree/node`` fetch per
+not-yet-loaded level — phase 167 (task 04, LOCKED A3 — the TODO L5
+2 MB payload fix) SUPERSEDED the phase-97 "one full-tree fetch, drill
+purely client-side" contract; the drill UX (breadcrumb, no URL change)
+is unchanged.
 
 This module pins the source-level contract (the house pattern of
 ``tests/unit/test_sync_button.py`` — read the frontend files as text,
@@ -24,8 +27,14 @@ suite, task 06):
   ``#stat-cards`` and BEFORE the file table (now the 6-column
   Source | Path | Title | Chunks | Created | Indexed head — phase 106
   (task 08, D8) added the Created column BEFORE Indexed);
-* the exact ``/api/docs/tree`` fetch (and the flat ``/api/docs`` fetch
-  is gone from the view module);
+* the exact ``/api/docs/tree?depth=1`` load fetch + the single
+  ``/api/docs/tree/node`` drill fetch (the flat ``/api/docs`` fetch is
+  gone from the view module), the level cache (declared once, cleared
+  by every load, consulted by ``needsLevelFetch``, written only after a
+  200), the §7.4 drill states (in-flight ``aria-busy``/``aria-disabled``
+  + one ``role="status"`` row, ``role="alert"`` failure with a real
+  Retry ``<button>``, the node 404 → top-level reset, the race token
+  extended to the drill), and the ``totals``-first stat cards;
 * the drill state (``current`` / ``kbTree``), the state transitions
   (source row click, folder row click, breadcrumb up, top reset), and
   the aria-current last segment;
@@ -239,18 +248,25 @@ def test_rag_view_ships_the_tree_skeleton_in_order() -> None:
         assert f">{column}</th>" in doc_head.group(1), f"#docs-table head must keep {column!r}"
 
 
-# ---------- the tree load: the ONE fetch + the race token ----------
+# ---------- the tree load: the depth-1 fetch, the node fetch, the token ----------
 
 
 def test_sources_js_fetches_the_tree_endpoint_only() -> None:
-    """The catalog load is `loadTree()`: ONE fetch of the tree endpoint.
-    The flat `GET /api/docs` fetch is GONE from the view module (the
-    endpoint itself stays — the API surface is untouched, task 02)."""
+    """Phase 167 (task 04, A3 — supersedes the phase-97 "ONE full-tree
+    fetch, zero per-level fetches" pin): the catalog load is ONE
+    `fetch("/api/docs/tree?depth=1")` and the drill has exactly ONE
+    node-fetch call site (`fetch("/api/docs/tree/node?…`, reached only
+    through fetchLevel). The flat `GET /api/docs` fetch stays GONE from
+    the view module (the endpoint itself stays — task 02)."""
     js = _js()
-    assert 'fetch("/api/docs/tree")' in js, "the load must fetch /api/docs/tree"
-    assert js.count('fetch("/api/docs/tree")') == 1, (
-        "the tree is the view's SINGLE catalog fetch (zero per-level fetches)"
+    code = _code(js)
+    assert code.count('fetch("/api/docs/tree?depth=1")') == 1, (
+        "the load is ONE depth=1 request (the 2 MB first-payload fix)"
     )
+    assert code.count('fetch("/api/docs/tree/node?') == 1, (
+        "exactly ONE per-level fetch call site — every drill routes through it"
+    )
+    assert 'fetch("/api/docs/tree")' not in js, "the depth-less load URL is gone"
     assert 'fetch("/api/docs")' not in js, "the flat /api/docs fetch must be gone"
 
 
@@ -301,8 +317,9 @@ def test_source_row_click_drills_into_the_source_root() -> None:
     body = js[js.find("function makeSourceRow(") : js.find("function makeFolderRow(")]
     assert 'link.className = "folder-link"' in body, "the source row uses the row drill link"
     assert "link.textContent = s.name" in body
-    assert "goTo({ source: s.name, folder: \"\" })" in body, (
-        "a source row drills to the source ROOT (folder \"\")"
+    assert "goTo({ source: s.name, folder: \"\" }, link)" in body, (
+        "a source row drills to the source ROOT (folder \"\") — the link "
+        "rides the phase-167 in-flight state"
     )
     assert "countTd.textContent = String(s.documents)" in body, "the recursive count"
     assert 'makeDescCell(s, s.name, "", s.name)' in body, (
@@ -326,8 +343,9 @@ def test_folder_row_click_drills_into_the_folder() -> None:
         "the label is the last segment (the breadcrumb language)"
     )
     assert "link.title = current.source + \"/\" + f.path" in body, "full path on hover"
-    assert "goTo({ source: current.source, folder: f.path })" in body, (
-        "a folder row drills to that folder"
+    assert "goTo({ source: current.source, folder: f.path }, link)" in body, (
+        "a folder row drills to that folder — the link rides the "
+        "phase-167 in-flight state"
     )
     assert "countTd.textContent = String(f.documents)" in body
     assert (
@@ -370,19 +388,24 @@ def test_breadcrumb_links_go_up_and_the_top_reset() -> None:
 
 def test_drill_navigation_pushes_state_only_entries() -> None:
     """Phase 99 (task 02, D2): every user-initiated drill is a history
-    entry — ``goTo`` is the push variant of ``applyTarget``: when the
-    target DIFFERS from the current level (a re-click of the current
-    row/segment pushes NO duplicate entry), it FIRST records a
-    STATE-ONLY entry — ``history.pushState({ view: "rag", kb: target },
-    "")`` — the second arg is ``""`` and there is no third (the URL
-    stays the shell's pathname: no new route, the phase-76 deep-link
-    surface is untouched) — then sets ``current`` + re-renders. The
-    drill still never fetches (the tree is in memory) and the module
-    never ``replaceState``s."""
+    entry — ``applyTarget(target, push)``: when the target DIFFERS
+    from the current level (a re-click of the current row/segment
+    pushes NO duplicate entry), it FIRST records a STATE-ONLY entry —
+    ``history.pushState({ view: "rag", kb: target }, "")`` — the second
+    arg is ``""`` and there is no third (the URL stays the shell's
+    pathname: no new route, the phase-76 deep-link surface is
+    untouched) — then sets ``current`` + re-renders. The module never
+    ``replaceState``s. Phase 167 (task 04, A3 — supersedes the "the
+    drill never fetches" pin): ``goTo`` hands the target to
+    ``enterFolder`` with ``push=true``, so the push lands only when the
+    level is IN HAND (a failed drill records nothing) — ``applyTarget``
+    and ``goTo`` themselves stay fetch-free; the drill's only fetch
+    lives in ``fetchLevel`` (pinned below)."""
     js = _js()
-    start = js.find("function applyTarget(")
-    assert start != -1, "applyTarget must exist (the two-step drill)"
-    body = js[start : js.find("function makeSourceRow(")]
+    assert js.find("function applyTarget(") != -1, (
+        "applyTarget must exist (the two-step drill)"
+    )
+    body = _fn(js, "applyTarget")
     gate_i = body.find(
         "if (push && (target.source !== current.source || target.folder !== current.folder)) {"
     )
@@ -394,14 +417,191 @@ def test_drill_navigation_pushes_state_only_entries() -> None:
     )
     # STATE-ONLY: exactly ONE pushState call site — the "" second arg,
     # no third (no URL change).
-    assert _code(body).count("pushState") == 1, (
+    assert _code(js).count("pushState") == 1, (
         "ONE pushState call site — the state-only drill entry"
     )
     assert "replaceState" not in _code(js), "no replaceState anywhere in the module"
     # goTo is the push variant — every drill call site keeps calling it.
-    go = body[body.find("function goTo(") :]
-    assert "applyTarget(target, true)" in go, "goTo pushes (the user-initiated drill)"
-    assert "fetch(" not in body, "the drill never fetches (client-side only)"
+    go = _fn(js, "goTo")
+    assert "enterFolder(target, link, true)" in go, (
+        "goTo pushes (the user-initiated drill, phase 167 via enterFolder)"
+    )
+    assert "fetch(" not in _code(body), "applyTarget stays a pure state transition"
+    assert "fetch(" not in _code(go), "goTo delegates the fetch to enterFolder only"
+
+
+# ---------- phase 167 (task 04): the lazy drill (cache, states, reset) ----------
+
+
+def test_level_cache_declared_cleared_and_consulted() -> None:
+    """The level cache (phase 167, D4): ONE module-scoped Map keyed
+    `source + "/" + folder`; EVERY successful load clears it (a refresh
+    invalidates every drilled level — the sync/upload/re-show call
+    sites keep working unchanged) BEFORE the never-stale reset (the
+    reset re-derives known-ness from the NEW tree); `needsLevelFetch`
+    consults it (the second visit to a folder is free — zero network);
+    the write sits behind the ok gate (only a 200 caches, never the
+    404/failure arms) and follows the merge (children ride the node
+    before the level is marked loaded)."""
+    js = _js()
+    assert js.count("const levelCache = new Map();") == 1, (
+        "ONE module-scoped level cache"
+    )
+    load = _fn(js, "loadTree")
+    clear_i = load.find("levelCache.clear();")
+    reset_i = load.find("resetVanishedLocation();")
+    assert 0 < clear_i < reset_i, "clear the cache BEFORE the reset (new tree)"
+    needs = _fn(js, "needsLevelFetch")
+    assert "node.truncated === true" in needs, "only a TRUNCATED node needs a fetch"
+    assert "levelCache.has(" in needs, "a cached level never refetches"
+    fetch = _fn(js, "fetchLevel")
+    merge_i = fetch.find("Object.assign(targetNode, merged);")
+    set_i = fetch.find('levelCache.set(target.source + "/" + target.folder, true);')
+    ok_gate = fetch.find("if (response === null || !response.ok) {")
+    vanish_i = fetch.find("if (vanished) {")
+    assert -1 < vanish_i < ok_gate < merge_i < set_i, (
+        "404 arm → ok gate → merge → cache: only a 200 caches, after the merge"
+    )
+
+
+def test_drill_race_token_extends_to_the_node_fetch() -> None:
+    """Phase 79 carried over (task 04): fetchLevel captures the
+    loadSeq token (plus a navSeq bumped by every applyTarget) and a
+    response may only paint while NEITHER a newer load NOR a newer
+    navigation happened — pinned twice (after the fetch, after the
+    json read). A new loadTree invalidates the in-flight drill the
+    moment it starts, and rehydrates the user's own truncated level
+    through the node route BEFORE it paints (a refresh never shows a
+    half-empty level)."""
+    js = _js()
+    fetch = _fn(js, "fetchLevel")
+    assert "const my = loadSeq;" in fetch, "the phase-79 token extends to the drill"
+    assert "const nav = navSeq;" in fetch, "a navigation invalidates the drill too"
+    assert fetch.count("if (my !== loadSeq || nav !== navSeq) return false;") == 2, (
+        "a stale drill response never paints (checked after the fetch and the json)"
+    )
+    load = _fn(js, "loadTree")
+    assert load.find("drillPending = null;") < load.find("await fetch"), (
+        "a new load invalidates the in-flight drill immediately"
+    )
+    assert "if (needsLevelFetch(currentLevelNode(), current.source)) {" in load, (
+        "the user's own truncated level is REHYDRATED through the node route"
+    )
+    assert "let navSeq = 0;" in js
+    # …and the ADOPTED navigation (popstate) rehydrates the same way:
+    # a forward/back entry can point at a level the client has not
+    # fetched since the last load (ensureCurrentLevel, after the adopt).
+    pop = js[js.find('window.addEventListener("popstate"') :]
+    pop = pop[: pop.find("});")]
+    adopt_i = pop.find("applyTarget(kb, false)")
+    ensure_i = pop.find("ensureCurrentLevel();")
+    assert -1 < adopt_i < ensure_i, "adopt the entry, THEN rehydrate its level"
+    ensure = _fn(js, "ensureCurrentLevel")
+    assert "if (drillPending) return;" in ensure
+    assert "if (needsLevelFetch(currentLevelNode(), current.source)) {" in ensure
+
+
+def test_drill_pending_guards_and_aria_inflight_state() -> None:
+    """§7.4 in-flight: enterFolder early-returns while a drill is
+    pending (a double click cannot start a second fetch), sets the
+    drillPending state BEFORE awaiting fetchLevel, and applies the
+    target only once the level is in hand. fetchLevel marks the
+    clicked link aria-busy + aria-disabled BEFORE the fetch and clears
+    both on every settle path (the control never stays disabled — the
+    failure path re-enables)."""
+    js = _js()
+    enter = _fn(js, "enterFolder")
+    guard_i = enter.find("if (drillPending) return;")
+    pending_i = enter.find(
+        "drillPending = { source: target.source, folder: target.folder };"
+    )
+    fetch_i = enter.find("await fetchLevel(target, node, link)")
+    apply_i = enter.find("applyTarget(target, push);")
+    assert -1 < guard_i < pending_i < fetch_i < apply_i, (
+        "guard → in-flight state → fetch → apply (only when the level is in hand)"
+    )
+    fetch = _fn(js, "fetchLevel")
+    busy_i = fetch.find('link.setAttribute("aria-busy", "true")')
+    disabled_i = fetch.find('link.setAttribute("aria-disabled", "true")')
+    call_i = fetch.find("response = await fetch(")
+    clear_busy = fetch.find('link.removeAttribute("aria-busy")')
+    clear_disabled = fetch.find('link.removeAttribute("aria-disabled")')
+    assert -1 < busy_i < call_i and -1 < disabled_i < call_i, (
+        "the busy state rides the row link BEFORE the fetch"
+    )
+    assert call_i < clear_busy and call_i < clear_disabled, (
+        "every settle path re-enables the control (§7.4)"
+    )
+
+
+def test_drill_inflight_status_and_failure_alert_rows() -> None:
+    """The ONE in-flight row of the folders panel (phase 167, D4): the
+    loading row is a role="status" "Loading <folder>…" line while the
+    level the user is LEAVING stays rendered; the failure row is a
+    role="alert" "Couldn't load <folder> — " line with a REAL
+    <button type="button">Retry</button> whose click re-runs the SAME
+    enterFolder fetch. Neither builder calls renderLevel (the current
+    level never blanks); both clear the previous drill row first (ONE
+    row at a time) and textContent only (the house rule)."""
+    js = _js()
+    loading = _fn(js, "showDrillLoading")
+    assert "clearDrillRow();" in loading
+    assert 'note.setAttribute("role", "status");' in loading
+    assert 'note.textContent = "Loading " + name + "…"' in loading
+    assert 'row.className = "kb-level-loading"' in loading
+    assert "foldersTbody.appendChild(row)" in loading, "the row lives in the folders panel"
+    assert "renderLevel" not in _code(loading), "the leaving level stays rendered"
+    fail = _fn(js, "showDrillFailure")
+    assert "clearDrillRow();" in fail
+    assert 'note.setAttribute("role", "alert");' in fail
+    assert '"Couldn\'t load " + target.folder.split("/").pop() + " — "' in fail
+    assert 'retry.type = "button";' in fail, "Retry is a REAL button"
+    assert 'retry.textContent = "Retry";' in fail
+    assert (
+        "retry.addEventListener(\"click\", () => enterFolder(target, link, true));" in fail
+    ), "Retry re-runs the SAME drill fetch"
+    assert "renderLevel" not in _code(fail), "the failure keeps the current level rendered"
+    clear = _fn(js, "clearDrillRow")
+    assert "drillRowEl.remove();" in clear and "drillRowEl = null;" in clear
+
+
+def test_vanished_node_resets_to_the_top_level() -> None:
+    """The node route's 404 IS the "the folder vanished" signal (task
+    04, D4): the drill resets `current` to the TOP level and re-renders
+    (no stale breadcrumb), caches NOTHING for the failed level, and
+    never applies the vanished target (enterFolder returns on false).
+    A network/non-OK failure takes the alert arm instead — the current
+    level stays rendered (pinned above)."""
+    js = _js()
+    fetch = _fn(js, "fetchLevel")
+    assert "response.status === 404" in fetch, "the 404 is detected"
+    vanish = fetch[fetch.find("if (vanished) {") : fetch.find(
+        "if (response === null || !response.ok) {"
+    )]
+    assert "current = { source: null, folder: null };" in vanish, (
+        "the 404 resets to the top level (the phase-97 never-stale contract)"
+    )
+    assert "renderLevel();" in vanish, "the reset renders — no stale breadcrumb"
+    assert "return false;" in vanish, "the vanished target is never applied"
+    assert "levelCache" not in vanish, "the failing level is NOT cached"
+    enter = _fn(js, "enterFolder")
+    assert "if (!loaded) return;" in enter, "enterFolder applies only on a real merge"
+
+
+def test_module_docstring_documents_the_lazy_drill() -> None:
+    """The house per-phase module-note convention: the Phase 167
+    (task 04) section records the depth-1 load, the per-level node
+    fetch + cache, the §7.4 drill states, the extended race token, the
+    404 reset and the totals-first stat cards — citing TODO L5."""
+    js = _js()
+    header = js[: js.find("import { fetchIsAdmin }")]
+    assert "Phase 167 (task 04, D4/A3" in header
+    assert "TODO L5" in header, "the 2 MB payload defect is documented"
+    assert "LAZY DRILL" in header
+    assert "/api/docs/tree?depth=1" in header
+    assert "levelCache" in header
+    assert "loadSeq token EXTENDS" in header, "the race-token narrative stays accurate"
+    assert "folderKnownIn" in header, "the existence-rule change is documented"
 
 
 # ---------- the level rendering ----------
@@ -507,12 +707,39 @@ def test_render_clears_both_row_containers_before_filling() -> None:
         assert clear_docs < body.find(append), f"{append} must come after the clears"
 
 
-# ---------- the stat cards: the whole-tree walk ----------
+# ---------- the stat cards: server totals, walk as the fallback ----------
+
+
+def test_stat_cards_prefer_the_server_totals() -> None:
+    """Phase 167 (task 04, D2/D4): a depth-limited tree CANNOT be
+    summed client-side, so treeStats() reads the server's `totals`
+    (documents / chunks / last_indexed) FIRST; the phase-97 recursive
+    walk stays only as the defensive no-`totals` fallback (a tree
+    served by a pre-phase process — the no-crash path). The stat-card
+    rendering (String(st.docs)/String(st.totalChunks)/fmtDate(st.last)
+    with the "–" null idiom) is UNCHANGED."""
+    js = _js()
+    body = _fn(js, "treeStats")
+    totals_i = body.find("const totals = kbTree.totals;")
+    guard_i = body.find('if (totals && typeof totals === "object") {')
+    docs_i = body.find("docs: totals.documents,")
+    chunks_i = body.find("totalChunks: totals.chunks,")
+    last_i = body.find('last: totals.last_indexed || "",')
+    walk_i = body.find("const walk = (node) => {")
+    assert -1 < totals_i < guard_i < docs_i < chunks_i < last_i < walk_i, (
+        "totals first, the recursive walk ONLY as the no-totals fallback"
+    )
+    render = js[js.find("function renderLevel(") : js.find("function renderEmpty(")]
+    assert "const st = treeStats();" in render
+    assert 'st.last ? fmtDate(st.last) : "–"' in render, (
+        "the exact fmtDate formatting call is unchanged at every level"
+    )
 
 
 def test_stat_cards_walk_the_whole_tree() -> None:
-    """The KB-wide stat cards walk the WHOLE tree (every file node at
-    any depth): document count, chunks sum, max indexed_at (fmtDate
+    """The no-`totals` fallback walk (phase 97, kept defensive since
+    phase 167): walk the WHOLE held tree (every file node at any
+    depth): document count, chunks sum, max indexed_at (fmtDate
     reuse) — the values identical to the former flat walk. A
     0-document registered source contributes nothing (and still lists
     its row — the ls invariant)."""
@@ -577,21 +804,30 @@ def test_module_docstring_documents_the_semantic_change() -> None:
 
 def test_reset_to_top_when_the_location_vanished() -> None:
     """PLAN §7.4: after a re-fetch, if the current source is no longer
-    in the tree (unregistered/pruned) OR the current folder no longer
-    exists under it, `current` RESETS to the top level BEFORE rendering
-    — no stale breadcrumb, no stale block. The folder existence check
-    mirrors the phase-94 rule: some indexed path starts with
-    `folder + "/"` (the source root — \"\" — always exists)."""
+    in the tree (unregistered/pruned) OR the current folder is no
+    longer KNOWN under it, `current` RESETS to the top level BEFORE
+    rendering — no stale breadcrumb, no stale block. Phase 167 (task
+    04, D4): the phase-94 full-tree path-prefix walk becomes
+    folderKnownIn — a folder is known iff its WHOLE ancestor chain
+    exists in the levels the client HOLDS (the depth-1 tree + the
+    drilled-and-merged levels; a broken chain — gone OR simply not
+    loaded that deep — resets honestly; the node route's 404 is the
+    drill path's own vanished-signal, pinned separately)."""
     js = _js()
-    body = js[js.find("function resetVanishedLocation()") : js.find("function folderExistsIn(")]
+    body = _fn(js, "resetVanishedLocation")
     assert "if (current.source === null) return;" in body, "the top level never vanishes"
     assert "kbTree.sources.find((s) => s.name === current.source)" in body
     assert body.count("current = { source: null, folder: null };") == 2, (
         "both vanishing arms reset to the top level"
     )
-    exists = js[js.find("function folderExistsIn(") : js.find("function currentLevelNode(")]
-    assert 'if (folderPath === "") return true;' in exists, "the source root always exists"
-    assert "p.startsWith(folderPath + \"/\")" in exists, "the phase-94 existence rule"
+    assert "folderKnownIn(src, current.folder)" in body, (
+        "the reset consults KNOWN-ness (the held levels), never a full-tree walk"
+    )
+    known = _fn(js, "folderKnownIn")
+    assert 'if (folderPath === "") return true;' in known, "the source root always exists"
+    assert "if (!next) return false;" in known, "a broken ancestor chain is not known"
+    assert "collect" not in _code(known), "the recursive path-collection walk is gone"
+    assert "folderExistsIn" not in js, "the old full-tree existence walker is gone"
     # The reset runs on the success path of the load, before the render
     # (pinned in the load-order test — here: it is defined and called).
     assert "resetVanishedLocation();" in js[js.find("async function loadTree()"):]
@@ -736,6 +972,7 @@ def test_styles_carry_the_tree_classes_with_no_new_hue() -> None:
         ".kb-crumb-current",
         ".kb-crumb-sep",
         ".kb-level",
+        ".kb-level-loading",
         ".kb-folders-table",
         ".folder-link",
     ):

tests/unit/test_response_budget.py

@@ -0,0 +1,702 @@
+"""Unit tests: ``app.core.response_budget`` — the byte-budget log line
+(phase 167, task 05, D5, LOCKED A5).
+
+Pins the contract of the line, the level cut, and the pass-through rules
+on a tiny ASGI app (the raw-ASGI interleaving harness of
+``test_compression.py`` where ordering matters, ``TestClient`` where the
+composed stack matters):
+
+* the EXACT shape ``api method=… path=… status=… bytes=… enc=…
+  duration_ms=…`` for a large 200 at INFO and a small 200 at DEBUG;
+* a 304 logs ``bytes=0`` (the revalidation win stays visible);
+* a non-``/api/`` path and a non-HTTP scope log NOTHING;
+* a stream logs ONE line with ``stream=1``, ``bytes`` = the sum of the
+  chunks, emitted when the stream COMPLETES (never mid-flight) — and the
+  frames are never buffered;
+* ``enc`` reads the WIRE encoding (``gzip`` through the real compression
+  layer, ``identity`` for a client that does not accept it) while
+  ``bytes`` stays the size the ROUTE produced;
+* an app exception is logged (finally-style, bytes as counted) and
+  re-raised UNCHANGED;
+* the config knob: ``0`` logs every ``/api/*`` response at INFO, a
+  negative value refuses startup naming the field, and the middleware sits
+  INSIDE ``SelectiveGZipMiddleware`` so ``bytes=`` is the route's size.
+"""
+from __future__ import annotations
+
+import asyncio
+import gzip
+import json
+import logging
+import re
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+from pydantic import ValidationError
+from starlette.applications import Starlette
+from starlette.responses import Response
+from starlette.routing import Route
+from starlette.types import Message, Receive, Scope, Send
+
+import app.core.response_budget as budget_mod
+import app.main as main_mod
+from app.config import Settings
+from app.core.compression import SelectiveGZipMiddleware
+from app.core.response_budget import (
+    API_PATH_PREFIX,
+    UNKNOWN_STATUS,
+    ResponseBudgetMiddleware,
+)
+
+#: One line, fully anchored — the shape README documents and the
+#: ``grep ' :: api '`` workflow parses.
+LINE_RE = re.compile(
+    r"^api method=(?P<method>[A-Z]+) path=(?P<path>\S+) status=(?P<status>\d+) "
+    r"bytes=(?P<bytes>\d+) enc=(?P<enc>\S+) duration_ms=(?P<ms>\d+)$"
+)
+
+#: ~28 KB of tree-shaped JSON.
+_BODY = json.dumps(
+    {
+        "sources": [
+            {
+                "name": "src",
+                "documents": i,
+                "children": [{"path": f"d/{i}/{n}.md"} for n in range(20)],
+            }
+            for i in range(200)
+        ]
+    }
+).encode()
+
+_SMALL_BODY = b'{"ok":true}'
+
+#: Four SSE-class frames, each comfortably above the gzip floor.
+_FRAMES = [f"data: {'x' * 500}\n\n".encode() for _ in range(4)]
+
+_LOGGER_NAME = "app.response_budget"
+
+
+def _settings(**kwargs: Any) -> Settings:
+    """Build Settings without reading a .env file (the test_config pattern)."""
+    kwargs.setdefault("_env_file", None)
+    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)
+
+
+def _api_lines(caplog: pytest.LogCaptureFixture) -> list[tuple[int, str]]:
+    """``(levelno, message)`` for this module's logger only, in order."""
+    return [
+        (record.levelno, record.getMessage())
+        for record in caplog.records
+        if record.name == _LOGGER_NAME
+    ]
+
+
+class _Counter(logging.Handler):
+    """Counts emitted lines live, so an app can observe "has the line been
+    written yet?" from INSIDE the response — the ordering proof."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.count = 0
+
+    def emit(self, record: logging.LogRecord) -> None:
+        self.count += 1
+
+
+def _http_scope(path: str, method: str = "GET") -> Scope:
+    return {"type": "http", "method": method, "path": path, "headers": []}
+
+
+async def _receive() -> Message:
+    return {"type": "http.disconnect"}
+
+
+async def _send(message: Message) -> None:
+    return None
+
+
+def _raw_json_app(body: bytes, *, status: int = 200, media_type: str = "application/json") -> Any:
+    """The smallest ASGI app that answers one buffered body — the raw-ASGI
+    level is where the WIRE bytes (post-compression) are observable."""
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": status,
+                "headers": [(b"content-type", media_type.encode())],
+            }
+        )
+        await send({"type": "http.response.body", "body": body, "more_body": False})
+
+    return app
+
+
+def _raw_harness(middleware_app: Any, *, path: str, accept: str) -> list[Message]:
+    """Drive a middleware stack at the raw ASGI level; return the exact
+    messages that reached the client's ``send``."""
+    sent: list[Message] = []
+
+    async def recording_send(message: Message) -> None:
+        sent.append(message)
+
+    asyncio.run(
+        middleware_app(
+            {
+                "type": "http",
+                "method": "GET",
+                "path": path,
+                "headers": [(b"accept-encoding", accept.encode())],
+            },
+            _receive,
+            recording_send,
+        )
+    )
+    return sent
+
+
+def _inner_app(
+    *,
+    body: bytes = _BODY,
+    status: int = 200,
+    path: str = "/api/docs/tree",
+    media_type: str = "application/json",
+) -> Starlette:
+    async def respond(request: Any) -> Response:
+        return Response(body, status_code=status, media_type=media_type)
+
+    async def boom(request: Any) -> Response:
+        raise RuntimeError("kaboom")
+
+    return Starlette(
+        routes=[
+            Route(path, respond, methods=["GET", "POST"]),
+            Route("/api/boom", boom, methods=["GET"]),
+            Route("/sources.html", respond, methods=["GET"]),
+        ]
+    )
+
+
+def _client(app: Any, *, threshold: int = 1024) -> TestClient:
+    return TestClient(ResponseBudgetMiddleware(app, threshold=threshold))
+
+
+# ---------------------------------------------------------------------------
+# The line's exact shape, and the level the budget decides
+# ---------------------------------------------------------------------------
+
+
+def test_large_response_logs_one_info_line_in_the_exact_shape(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        client = _client(_inner_app(), threshold=1024)
+        assert client.get("/api/docs/tree").status_code == 200
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    level, message = lines[0]
+    assert level == logging.INFO
+    match = LINE_RE.match(message)
+    assert match is not None, message
+    assert match.group("method") == "GET"
+    assert match.group("path") == "/api/docs/tree"
+    assert match.group("status") == "200"
+    assert int(match.group("bytes")) == len(_BODY)
+    assert match.group("enc") == "identity"
+    assert int(match.group("ms")) >= 0
+
+
+def test_small_response_logs_the_same_line_at_debug(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        client = _client(_inner_app(body=_SMALL_BODY), threshold=1024)
+        assert client.get("/api/docs/tree").status_code == 200
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    level, message = lines[0]
+    assert level == logging.DEBUG
+    assert f"bytes={len(_SMALL_BODY)}" in message
+
+
+def test_budget_is_the_inclusive_cut(caplog: pytest.LogCaptureFixture) -> None:
+    """``bytes >= budget`` is INFO, one byte less is DEBUG — the boundary the
+    README and ``.env.example`` state."""
+
+    async def exactly(request: Any) -> Response:
+        return Response(b"x" * 100)
+
+    async def one_less(request: Any) -> Response:
+        return Response(b"x" * 99)
+
+    app = Starlette(
+        routes=[
+            Route("/api/a", exactly, methods=["GET"]),
+            Route("/api/b", one_less, methods=["GET"]),
+        ]
+    )
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        client = _client(app, threshold=100)
+        client.get("/api/a")
+        client.get("/api/b")
+
+    assert [level for level, _ in _api_lines(caplog)] == [logging.INFO, logging.DEBUG]
+
+
+def test_threshold_defaults_to_the_settings_value(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """No ``threshold`` argument — the way ``create_app`` wires it — so the
+    live ``log_large_response_bytes`` decides per response."""
+    monkeypatch.setattr(
+        budget_mod, "get_settings", lambda: _settings(log_large_response_bytes=5)
+    )
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        TestClient(ResponseBudgetMiddleware(_inner_app(body=_SMALL_BODY))).get(
+            "/api/docs/tree"
+        )
+
+    assert [level for level, _ in _api_lines(caplog)] == [logging.INFO]
+
+
+def test_zero_budget_logs_every_api_response_at_info(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """``BOR_LOG_LARGE_RESPONSE_BYTES=0`` is the loud mode: even an 11-byte
+    body is INFO (the operator's one-off investigation setting)."""
+    monkeypatch.setattr(
+        budget_mod, "get_settings", lambda: _settings(log_large_response_bytes=0)
+    )
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        TestClient(ResponseBudgetMiddleware(_inner_app(body=_SMALL_BODY))).get(
+            "/api/docs/tree"
+        )
+
+    # 11 bytes, budget 0 -> INFO (the same body is DEBUG at any positive
+    # budget: see test_small_response_logs_the_same_line_at_debug).
+    assert [level for level, _ in _api_lines(caplog)] == [logging.INFO]
+
+
+# ---------------------------------------------------------------------------
+# Statuses: a 304 logs 0 bytes, an error is counted like any other response
+# ---------------------------------------------------------------------------
+
+
+def test_304_logs_zero_bytes(caplog: pytest.LogCaptureFixture) -> None:
+    """The revalidation win stays in the log — a bodiless 304 is a line with
+    ``bytes=0``, not a missing line."""
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        client = _client(_inner_app(body=b"", status=304), threshold=1024)
+        assert client.get("/api/docs/tree").status_code == 304
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    match = LINE_RE.match(lines[0][1])
+    assert match is not None
+    assert match.group("status") == "304"
+    assert match.group("bytes") == "0"
+    assert lines[0][0] == logging.DEBUG
+
+
+def test_error_status_is_logged_like_any_other_response(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        client = _client(_inner_app(body=b'{"detail":"admin only"}', status=403))
+        assert client.get("/api/docs/tree").status_code == 403
+
+    match = LINE_RE.match(_api_lines(caplog)[0][1])
+    assert match is not None and match.group("status") == "403"
+
+
+# ---------------------------------------------------------------------------
+# What is NOT measured
+# ---------------------------------------------------------------------------
+
+
+def test_non_api_path_logs_nothing(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        # budget 0: ANY /api line would show, so silence is the whole pin.
+        client = _client(_inner_app(), threshold=0)
+        assert client.get("/sources.html").status_code == 200
+
+    assert _api_lines(caplog) == []
+
+
+def test_non_http_scopes_pass_through_silently(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``lifespan`` / ``websocket`` scopes reach the app untouched and write
+    no line — there is no response to measure."""
+    seen: list[str] = []
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        seen.append(scope["type"])
+
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        for scope_type in ("lifespan", "websocket"):
+            asyncio.run(
+                ResponseBudgetMiddleware(app, threshold=0)(
+                    {"type": scope_type, "path": "/api/x"}, _receive, _send
+                )
+            )
+
+    assert seen == ["lifespan", "websocket"]
+    assert _api_lines(caplog) == []
+
+
+# ---------------------------------------------------------------------------
+# The stream: one line, stream=1, counted total, emitted at completion
+# ---------------------------------------------------------------------------
+
+
+def test_stream_logs_once_at_completion_with_stream_flag_and_total_bytes(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The SSE contract of the line: nothing while frames are in flight, one
+    ``stream=1`` line when the final chunk completes, ``bytes`` = the sum of
+    the frames (NOT one frame)."""
+    counter = _Counter()
+    logger = logging.getLogger(_LOGGER_NAME)
+    logger.addHandler(counter)
+    in_flight: list[int] = []
+
+    async def sse_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/event-stream")],
+            }
+        )
+        for frame in _FRAMES:
+            await send(
+                {"type": "http.response.body", "body": frame, "more_body": True}
+            )
+            in_flight.append(counter.count)  # after each frame is delivered
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+        in_flight.append(counter.count)
+
+    try:
+        with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+            asyncio.run(
+                ResponseBudgetMiddleware(sse_app, threshold=1)(
+                    _http_scope("/api/chat", "POST"), _receive, _send
+                )
+            )
+    finally:
+        logger.removeHandler(counter)
+
+    # Every in-flight probe except the last saw zero lines; the completion
+    # is what emitted.
+    assert in_flight == [0] * len(_FRAMES) + [1]
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    message = lines[0][1]
+    assert message.endswith(" stream=1")
+    match = LINE_RE.match(message.removesuffix(" stream=1"))
+    assert match is not None
+    assert match.group("method") == "POST"
+    assert match.group("path") == "/api/chat"
+    assert int(match.group("bytes")) == sum(len(frame) for frame in _FRAMES)
+
+
+def test_stream_frames_are_never_buffered() -> None:
+    """Zero-buffering proof (the interleaving idiom of
+    ``test_compression.py``): every frame reaches the client's ``send``
+    before the app produces the next one — a ``BaseHTTPMiddleware`` here
+    would emit all ``app:*`` events first."""
+    events: list[str] = []
+
+    async def sse_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/event-stream")],
+            }
+        )
+        for i, frame in enumerate(_FRAMES):
+            events.append(f"app:frame-{i}")
+            await send(
+                {"type": "http.response.body", "body": frame, "more_body": True}
+            )
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+
+    async def recording_send(message: Message) -> None:
+        events.append(f"client:{message['type']}")
+
+    asyncio.run(
+        ResponseBudgetMiddleware(sse_app, threshold=0)(
+            _http_scope("/api/chat", "POST"), _receive, recording_send
+        )
+    )
+
+    expected = ["client:http.response.start"]
+    for i in range(len(_FRAMES)):
+        expected += [f"app:frame-{i}", "client:http.response.body"]
+    expected += ["client:http.response.body"]
+    assert events == expected
+
+
+# ---------------------------------------------------------------------------
+# enc= reads the wire, bytes= stays the route's size
+# ---------------------------------------------------------------------------
+
+
+def _stack(*, threshold: int = 1024) -> TestClient:
+    """response-budget INSIDE the selective-gzip layer — the order
+    ``create_app`` builds."""
+    return TestClient(
+        SelectiveGZipMiddleware(
+            ResponseBudgetMiddleware(_inner_app(), threshold=threshold),
+            minimum_size=860,
+        )
+    )
+
+
+def test_enc_gzip_is_reported_through_the_compression_layer(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The two halves of the line, measured against the wire: the client
+    receives a gzipped body, the log says ``enc=gzip`` and still counts the
+    ROUTE's uncompressed bytes (the position inside the compressor is what
+    buys that)."""
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        sent = _raw_harness(
+            SelectiveGZipMiddleware(
+                ResponseBudgetMiddleware(_raw_json_app(_BODY), threshold=1024),
+                minimum_size=860,
+            ),
+            path="/api/docs/tree",
+            accept="gzip",
+        )
+
+    wire = b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
+    assert gzip.decompress(wire) == _BODY  # the transfer really was gzip
+    match = LINE_RE.match(_api_lines(caplog)[-1][1])
+    assert match is not None
+    assert match.group("enc") == "gzip"
+    assert int(match.group("bytes")) == len(_BODY) > len(wire)
+
+
+def test_enc_identity_when_the_client_does_not_accept_gzip(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        response = _stack().get(
+            "/api/docs/tree", headers={"Accept-Encoding": "identity"}
+        )
+
+    assert "content-encoding" not in response.headers
+    match = LINE_RE.match(_api_lines(caplog)[-1][1])
+    assert match is not None and match.group("enc") == "identity"
+    assert int(match.group("bytes")) == len(_BODY)
+
+
+def test_enc_reports_a_pre_encoded_response(caplog: pytest.LogCaptureFixture) -> None:
+    """A route that encodes its own body (the fronting-proxy case) is
+    reported as such — the header on the wire is the answer."""
+
+    async def encoded(request: Any) -> Response:
+        return Response(
+            gzip.compress(_BODY),
+            media_type="application/json",
+            headers={"Content-Encoding": "gzip"},
+        )
+
+    app = Starlette(routes=[Route("/api/encoded", encoded, methods=["GET"])])
+    with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
+        TestClient(ResponseBudgetMiddleware(app, threshold=1024)).get("/api/encoded")
+
+    match = LINE_RE.match(_api_lines(caplog)[-1][1])
+    assert match is not None and match.group("enc") == "gzip"
+
+
+# ---------------------------------------------------------------------------
+# The app raising: logged, and re-raised unchanged
+# ---------------------------------------------------------------------------
+
+
+def test_app_exception_logs_and_reraises_unchanged(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An app that dies before it can answer still writes one line (nothing
+    was counted, status 500 — what the outer error layer will answer) and the
+    exception propagates UNCHANGED."""
+
+    async def dying_app(scope: Scope, receive: Receive, send: Send) -> None:
+        raise RuntimeError("kaboom")
+
+    with (
+        caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME),
+        pytest.raises(RuntimeError, match="kaboom"),
+    ):
+        asyncio.run(
+            ResponseBudgetMiddleware(dying_app, threshold=0)(
+                _http_scope("/api/docs/tree"), _receive, _send
+            )
+        )
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    match = LINE_RE.match(lines[0][1])
+    assert match is not None
+    assert match.group("status") == str(UNKNOWN_STATUS)
+    assert match.group("bytes") == "0"
+
+
+def test_a_500_produced_by_the_framework_is_counted_like_a_response(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Through a real app the unhandled exception becomes a 500 WITH a body
+    (Starlette's ``Internal Server Error`` text): the line then reports that
+    status and those bytes — one line either way, never zero."""
+    with (
+        caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME),
+        pytest.raises(RuntimeError, match="kaboom"),
+    ):
+        _client(_inner_app(), threshold=0).get("/api/boom")
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    match = LINE_RE.match(lines[0][1])
+    assert match is not None
+    assert match.group("status") == "500"
+    assert int(match.group("bytes")) == len(b"Internal Server Error")
+
+
+def test_exception_mid_stream_logs_the_bytes_counted_so_far(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A stream that dies after two frames still writes ONE line with the
+    bytes it actually got out (finally-style) — and the exception is not
+    swallowed."""
+
+    async def dying_stream_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/event-stream")],
+            }
+        )
+        await send({"type": "http.response.body", "body": _FRAMES[0], "more_body": True})
+        await send({"type": "http.response.body", "body": _FRAMES[1], "more_body": True})
+        raise RuntimeError("stream died")
+
+    with (
+        caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME),
+        pytest.raises(RuntimeError, match="stream died"),
+    ):
+        asyncio.run(
+            ResponseBudgetMiddleware(dying_stream_app, threshold=0)(
+                _http_scope("/api/chat", "POST"), _receive, _send
+            )
+        )
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    message = lines[0][1]
+    assert message.endswith(" stream=1")
+    assert f"bytes={len(_FRAMES[0]) + len(_FRAMES[1])}" in message
+
+
+def test_a_cancelled_stream_still_logs_one_line_with_what_got_out(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The client-walked-away path (the live audit hit it: an SSE turn cut
+    mid-answer): the app unwinds with ``CancelledError`` and no completing
+    body, the ``finally`` writes ONE line with the bytes that actually got
+    out, and the cancellation propagates (the middleware never swallows it)."""
+
+    async def cancelled_stream_app(scope: Scope, receive: Receive, send: Send) -> None:
+        await send(
+            {
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [(b"content-type", b"text/event-stream")],
+            }
+        )
+        await send({"type": "http.response.body", "body": _FRAMES[0], "more_body": True})
+        raise asyncio.CancelledError  # what the server does on a client disconnect
+
+    with (
+        caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME),
+        pytest.raises(asyncio.CancelledError),
+    ):
+        asyncio.run(
+            ResponseBudgetMiddleware(cancelled_stream_app, threshold=0)(
+                _http_scope("/api/chat", "POST"), _receive, _send
+            )
+        )
+
+    lines = _api_lines(caplog)
+    assert len(lines) == 1
+    assert lines[0][1].endswith(" stream=1")
+    assert f"bytes={len(_FRAMES[0])}" in lines[0][1]
+
+
+# ---------------------------------------------------------------------------
+# Config + wiring
+# ---------------------------------------------------------------------------
+
+
+def test_log_large_response_bytes_defaults_to_64_kb() -> None:
+    assert _settings().log_large_response_bytes == 65536
+
+
+def test_log_large_response_bytes_env_override(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("BOR_LOG_LARGE_RESPONSE_BYTES", "4096")
+    assert _settings().log_large_response_bytes == 4096
+    monkeypatch.setenv("BOR_LOG_LARGE_RESPONSE_BYTES", "0")  # loud mode is legal
+    assert _settings().log_large_response_bytes == 0
+
+
+def test_negative_budget_refuses_startup_naming_the_field() -> None:
+    with pytest.raises(ValidationError, match="log_large_response_bytes must be >= 0"):
+        _settings(log_large_response_bytes=-1)
+
+
+def test_middleware_sits_inside_gzip_outside_caching(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The documented stack (README + ``app/main.py``): SecurityHeaders
+    outermost → gzip → response-budget → caching → session. That position is
+    what makes ``bytes=`` the ROUTE's size while ``enc=`` still names the
+    wire encoding."""
+    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 860)
+    names = [
+        getattr(mw.cls, "__name__", "") for mw in main_mod.create_app().user_middleware
+    ]
+    assert names == [
+        "SecurityHeadersMiddleware",
+        "SelectiveGZipMiddleware",
+        "ResponseBudgetMiddleware",
+        "CachingMiddleware",
+        "SessionMiddleware",
+    ]
+
+
+def test_budget_is_wired_even_with_compression_off(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``BOR_COMPRESS_MIN_BYTES=0`` removes gzip only — the budget line is
+    not optional (LOCKED A5: measure everything, reject nothing), so the
+    measurement survives the compression kill switch."""
+    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 0)
+    names = [
+        getattr(mw.cls, "__name__", "") for mw in main_mod.create_app().user_middleware
+    ]
+    assert "SelectiveGZipMiddleware" not in names
+    assert "ResponseBudgetMiddleware" in names
+
+
+def test_api_path_prefix_constant() -> None:
+    assert API_PATH_PREFIX == "/api/"