brain-of-reese/tests/e2e/mock_llm.py

"""Deterministic OpenAI-compatible mock for E2E tests (aipi stand-in).

Implements just enough of the aipi surface:

* ``GET  /v1/models``
* ``POST /v1/embeddings`` — real bag-of-words vectors (768-dim, L2-normed).
  Because similarity is *genuine token overlap*, the relevance threshold
  behaves the same way it will in production: related questions score high,
  unrelated ones score low and trigger honest deflection.
* ``POST /v1/chat/completions`` — streaming (SSE) or not. The content keys
  off markers in the system prompt:
  - user message containing ``write a long answer`` -> a ~900-word
    deterministic numbered answer (long-answers story, phase 11)
  - ``SUMMARY_MODE``  -> the deterministic summary digest: the first 24
    tokens of the user message (the summarizer puts the capped document
    content there) — byte-stable for a given fixture (document summaries,
    phase 30)
  - ``KB_OVERVIEW_MODE`` -> the deterministic outline: the first 8 tokens
    of the user message (the generator puts the document list there) —
    byte-stable for a given KB (KB overview, phase 31)
  - ``FOLDER_SUMMARY_MODE`` -> the deterministic folder one-liner
    ``Fixture folder summary for <folder>.`` — <folder> is the user
    message's ``Folder: …`` header line (the generator names the
    source/folder there, phase 94), so the stored row always names its
    folder and an E2E can assert on it. Checked BEFORE the
    ``SUMMARY_MODE`` branch: the folder marker CONTAINS the summary
    marker as a substring, so the summary branch would otherwise
    shadow every folder-summary call
  - ``DEFLECT_MODE``  -> honest "I haven't done anything like that" answer
  - otherwise         -> upbeat answer quoting the provided document context
  - user message containing ``pretend to think slowly`` -> 3s warm-up delay
    (used by the loading-feedback story).
  - user message containing ``think out loud`` -> the answer is preceded by
    ~2 700 chars of deterministic ``reasoning_content`` chunks (the
    thinking-display story, phase 17; lengthened in phase 21 so the
    rendered scratchpad overflows the 320px ``.thinking-text`` window)
  - user message containing ``think out loud then hesitate`` -> the
    ``think out loud`` stream, then a 4s pause before the first content
    frame (the sources-midstream story, phase 20 — a deterministic
    "leave during pure thinking" navigation window).
  - user message containing ``think in paragraphs`` (``THINK_PARAS_TRIGGER``)
    -> the ``think out loud`` scratchpad WITH REAL paragraph breaks
    ("\n\n"), streamed at 60-char frames (vs the mock's 12-char default).
    One frame renders several lines — a real-model-sized delta, the
    condition under which a POST-render pin-state reading (the old
    app.js) measured the chunk's height instead of the user's position
    and the think-window follow died at the first 2-newline gap. The
    regression pin for the pre-render capture in app.js (2026-08-29,
    owner report). Checked BEFORE ``think out loud`` (it is the more
    specific phrase); existing E2E questions carry neither, so every
    other suite is unaffected.
  - system prompt containing ``<tuning>`` (phase 15, steering notes) ->
    the composed answer ends with `` (tuning: <first note line>)`` —
    makes prompt injection observable in the UI deterministically.
  - system prompt containing ``<knowledge_base>`` (phase 31, KB overview)
    -> the composed answer ends with `` (kb: <first bullet line>)`` —
    the same echo convention for the overview's prompt injection.
  - user message containing ``show the end of your notes`` (phase 24,
    whole-document context) -> the answer quotes the **last 160 chars of
    the ``<documents>`` block** — a tail echo, byte-stable across runs, so
    a sentinel placed at the *end* of a document appears in the rendered
    answer iff the whole document was in the prompt. (Phase 37: the HIGH
    prompt now ends with a ``<tools>`` section after ``</documents>``, so
    the echo targets the block itself; its tail still includes the
    closing tag — same sentinel semantics.)
  - user message containing ``use your tools`` (phase 37, agent document
    tools; phase 70: the flow emits the harness-aligned names — ``ls``
    / ``read`` with the combined ``source/path`` identity; phase 94:
    the flow DRILLS through the tree ``ls`` — the top-level listing
    carries sources only, so the flow takes one drill step, ``ls``
    scoped to the first source, before the first file line exists)
    **and** the system prompt carries the ``<tools>`` section -> the
    deterministic SINGLE-READ tool flow, discriminated statelessly from
    the messages (the ``tools`` parameter gates the list/read steps —
    a no-tools request with no tool results is not the flow):
      * request 1 (``tools`` offered, no tool results yet): stream ONLY
        ``tool_calls`` deltas — ``ls`` (synthetic id ``call_0``, no
        arguments), ``finish_reason: "tool_calls"``, no content;
      * request 2 (the top-level source listing in the messages — the
        agent's ``ls`` tree format, phase 94: no file lines yet):
        stream a ``tool_calls`` delta — ``ls`` scoped to the FIRST
        source of the listing (id ``call_1``) — the drill step (one
        drill per flow — ``_drill_target`` skips already-drilled
        sources, so the second listing is the folder level);
      * request 3 (a ``tool``-role folder listing with file lines in
        the messages): parse the FIRST file line (``source: X | path:
        Y | title: Z`` — the labeled ``source:`` / ``path:`` fields,
        phase 63) and stream a ``tool_calls`` delta calling ``read`` on
        the JOINED combined ``source/path`` (the mock joins the two
        labeled fields — the file-line format is unchanged, so this
        join is the only parse change, phase 70) (id ``call_2``);
      * request 4 (a ``tool``-role read result in the messages —
        content starting with the agent's ``"Document <source/path>:"``
        header): a content answer, deterministic: ``Read
        <source/path>. <first 80 chars of the read document's
        content>`` — so a suite can assert the read document reached
        the model and landed in the answer. Reached regardless of the
        ``tools`` parameter (phase 45 keeps the tools offered until the
        round cap).
    The single-read flow stops at ONE read result; the MULTI-READ
    variant below reads two.
  - user message containing BOTH ``use your tools`` AND ``read two
    documents`` (``MULTI_READ_TRIGGER``, phase 45 task 02) **and** the
    system prompt carries the ``<tools>`` section -> the deterministic
    MULTI-READ flow (list → drill → read #1 → read #2 → answer; phase
    94: the drill step between the top-level listing and the first
    read), classified by the COUNT of ``tool``-role read results
    (content starting with the agent's ``"Document <source/path>:"``
    prefix); phase 70: the same flow on the harness-aligned names —
    ``ls``, the drill ``ls`` scoped to the first source, then ``read``
    on the JOINED combined ``source/path`` of each file line:
      * 0 read results, no listing yet: ``ls`` (id ``call_0``);
      * 0 read results, top-level listing only (no file lines yet):
        the drill — ``ls`` scoped to the first source (id ``call_1``);
      * 0 read results, file lines present: ``read`` on the JOINED
        combined ``source/path`` of the FIRST file line (id ``call_2``);
      * 1 read result: ``read`` on the JOINED combined ``source/path``
        of the SECOND file line — the first listing line whose
        ``source/path`` differs from the one already read (id
        ``call_3``); a one-file listing degenerates to the
        single-read answer (nothing second to read);
      * 2 read results: the forced answer, byte-stable: the single-read
        shape quoting the FIRST read result, plus the line ``I read
        <sp1> and <sp2>.`` naming both read paths in read order — so a
        suite can assert the model used BOTH documents.
    All other requests (including the marker without a ``<tools>``
    section, or with the tool conversation not yet started and no tools
    offered — e.g. ``agent_max_rounds=0``) behave exactly as today.
    ``E2E_REAL_LLM=1`` ignores the mock entirely (the real model does
    what it does).
  - user message containing ``search your documents``
    (``SEARCH_TRIGGER``, phase 68 search tool — renamed to the
    harness-aligned ``grep`` in phase 70, same match/output contract)
    **and** the system prompt carries the ``<tools>`` section -> the
    deterministic SEARCH tool flow, discriminated statelessly from the
    messages (streaming only):
      * request 1 (``tools`` offered, no search result yet): stream
        ONLY ``tool_calls`` deltas — ``grep`` with
        ``{"pattern": SEARCH_PATTERN}`` (id ``call_0``);
      * request 2 (a ``tool``-role search result in the messages —
        recognizable by its ``source/path:line: text`` match lines or
        the sentinel in its content): the content answer, deterministic:
        ``Found <first matched line's content up to 80 chars>`` — so a
        suite can assert the search result reached the model and landed
        in the answer.
    Checked BEFORE the plain ``use your tools`` flow (it is the more
    specific phrase — same convention as ``think in paragraphs``); no
    existing E2E question or fixture file contains the trigger, so
    every other suite is unaffected.
  - user message containing ``emit raw tool markup``
    (``SCAFFOLD_TRIGGER``, phase 71, tool-scaffolding guardrails — the
    2026-09-03 incident where a deflected round streamed the model's
    raw ``<|tool_call_start|>…<|tool_call_end|>`` markup into the UI)
    **or** ``always emit raw tool markup``
    (``SCAFFOLD_ALWAYS_TRIGGER``, checked FIRST — it contains the
    former phrase) -> the deterministic SCAFFOLDING flow, independent
    of the ``<tools>`` marker (both grounded and deflected turns hit
    it):
      * ``SCAFFOLD_ALWAYS_TRIGGER``: EVERY request (the one bounded
        recovery included) streams ONLY ``delta.content`` chunks
        carrying the incident span ``SCAFFOLD_SPAN`` —
        ``<|tool_call_start|>[read(path='search_docs/reese-notes.md')]
        <|tool_call_end|>`` — split across the mock's 12-char chunks
        (the filter's boundary path), ``finish_reason: "stop"``, no
        structured ``tool_calls``, no reasoning — the terminal
        malformed-reply path.
      * ``SCAFFOLD_TRIGGER``: request 1 (no ``CORRECTION_INSTRUCTION``
        in the system prompt) streams the same scaffolding-only span;
        request 2 (the system prompt carries the harness constant — a
        stable substring of ``app.rag.agent.CORRECTION_INSTRUCTION``,
        IMPORTED into this module so the mock can never drift from
        it: the one bounded recovery, ``tools=None`` with the
        correction folded into the single system prompt) streams the
        clean ``SCAFFOLD_RECOVERY_ANSWER`` — the recovery path.
    Checked BEFORE the ``SEARCH_TRIGGER`` / ``TOOLS_TRIGGER`` flows
    (the trigger needs no ``<tools>`` section); no existing E2E
    question or fixture file contains the phrase, so every other
    suite is unaffected.
  - user message containing ``list the files in this directory``
    (``LS_TEACH_TRIGGER``, phase 72, teaching refusals — the
    2026-09-03 incident where the harness-prior ``ls(path='.')``
    misuse met the terse refusal and the model re-reasoned the same
    paragraphs over and over) **and** the system prompt carries the
    ``<tools>`` section -> the deterministic LS-TEACHING flow,
    discriminated statelessly from the messages (streaming only;
    phase 94: the correction's top-level listing carries sources only,
    so the flow drills one level before the first file line exists):
      * request 1 (``tools`` offered, no ``tool``-role result in the
        messages yet): stream ONLY ``tool_calls`` deltas — ``ls``
        with ``{"path": "."}`` (synthetic id ``call_0``),
        ``finish_reason: "tool_calls"``, no content — the incident's
        misuse, deterministic;
      * request 2 (a ``tool``-role result present that is NOT a
        listing with file lines — i.e. the teaching refusal): a
        ``tool_calls`` delta — ``ls`` with no arguments (id ``call_1``)
        — the correction;
      * request 3 (the top-level source listing in the messages —
        file lines still absent): the drill — a ``tool_calls`` delta
        — ``ls`` scoped to the FIRST source of the listing
        (id ``call_2``);
      * request 4 (a ``tool``-result with a ``source: X | path: Y |
        title: Z`` file line in the messages — the folder listing):
        a deterministic content answer — ``These are the indexed
        documents: <first file line>`` (the line, parsed with the
        ``_CATALOG_LINE_RE`` machinery), ``finish_reason: "stop"`` —
        the loop ended in ONE correction + ONE drill, not at the round
        cap.
    Checked BEFORE the plain ``TOOLS_TRIGGER`` flow (the trigger
    phrases are disjoint substrings — the phase-71 ordering
    convention); no existing E2E question or fixture file contains the
    phrase, so every other suite is unaffected.
  - user message containing ``drill down the tree`` (``DRILL_TRIGGER``,
    phase 94 task 04 — the drill-down ``ls``'s dedicated story suite
    ``tests/e2e/test_ls_tree_drilldown.py``) **and** the system prompt
    carries the ``<tools>`` section -> the deterministic SCRIPTED
    DRILL-DOWN flow: the question carries its own tool call after the
    colon — ``drill down the tree: ls [target]`` (no target = the top
    level; ``target`` = a source name or ``source/folder`` path) or
    ``drill down the tree: read source/path`` — parsed by
    ``_DRILL_CALL_RE`` from the RAW user message (the target keeps its
    case), then discriminated statelessly from the tool results
    (streaming only):
      * request 1 (``tools`` offered, no ``tool``-role result in the
        messages yet): the SCRIPTED call — ``ls`` with no arguments
        for the top level, ``ls``/``read`` with the parsed target
        otherwise (synthetic id ``call_0``);
      * the last tool result is the NOT-A-FOLDER teaching refusal
        (it carries ``"is not a folder"`` — the phase-94 task-03
        teaching line, the refusal being VISIBLE to the model is what
        fires this branch): the scripted one-round recovery — ``ls``
        the target's SOURCE segment (id ``call_1``);
      * the last tool result is a READ result (``"Document
        <source/path>:…``): the deterministic echo answer ``Read
        <source/path>. <first 80 chars of the read document's
        content>`` (the phase-37 single-read shape — the grounded-
        turn citation contract);
      * the last tool result is the agent's ALREADY_IN_CONTEXT dedupe
        refusal (the read target is already a top-2 retrieval
        document — with the drill fixture that is DETERMINISTIC:
        the read question names the file's path, so the file
        self-matches the hybrid gate and its FULL text is in the
        ``<documents>`` prompt): the model answers FROM THE PROMPT —
        the deterministic answer ``Already in context: Read
        <source/path>. <first 80 chars of the target document's text
        as it appears in the ``<documents>`` block>`` (same citation
        shape as the read-result branch — the document text reached
        the model either way, and the answer proves it);
      * any other last result (a top-level or folder LISTING landed):
        the deterministic ECHO answer ``Here's the level I listed:\n
        <the listing, verbatim>`` — the mock echoes what it received
        (the house scripted-turn way of asserting on tool results):
        the suite asserts on the exact tree level — the ``— N
        documents`` lines, the stored folder summaries, the
        ``source: X | path: Y | title: Z`` file lines, and the 50-line
        cap + ``…and N more documents…`` note — through the rendered
        answer, the only E2E lens on the LLM's context.
    Checked BEFORE the plain ``TOOLS_TRIGGER`` flow (disjoint trigger
    phrases — the phase-71/72 ordering convention; the trigger needs
    the ``<tools>`` section, so deflected turns never hit it); no
    existing E2E question or fixture file contains the phrase, so
    every other suite is unaffected.
  - user message containing ``read the capped document``
    (``READ_CAP_TRIGGER``, phase 95 task 03 — the read cap's dedicated
    story suite ``tests/e2e/test_read_truncation_cap.py``) **and** the
    system prompt carries the ``<tools>`` section -> the deterministic
    SCRIPTED CAPPED-READ flow: the question carries its own tool call
    after the colon — ``read the capped document: read source/path`` —
    parsed by ``_READ_CAP_CALL_RE`` from the RAW user message (the
    target keeps its case), then discriminated statelessly from the
    tool results (streaming only):
      * request 1 (``tools`` offered, no ``tool``-role result in the
        messages yet): the scripted call — ``read`` with the parsed
        target (synthetic id ``call_0``);
      * a ``tool``-role result is in the messages: the deterministic
        ECHO — the answer carries the LAST tool result VERBATIM
        (``Here's what the read returned:\n<result>``): a read result
        (``"Document <source/path>:…``) lands in the answer with its
        FULL content — the first cap chars + ``[…truncated…]`` + the
        pinned grep-pointer notice when the cap fired, the plain body
        byte-identical to the pre-phase-95 shape when it did not — and
        a refusal (the premise broke) lands just as visibly, so the
        suite fails loudly on it. The mock is the only E2E lens on the
        LLM's context, so the echo is the assertion surface for both
        the marker's presence AND its absence.
    Checked BEFORE the plain ``TOOLS_TRIGGER`` flow (disjoint trigger
    phrases — the phase-71/72/94 ordering convention; the trigger
    needs the ``<tools>`` section, so deflected turns never hit it);
    verified 2026-09-10: no existing E2E question or fixture file
    contains the phrase, so every other suite is unaffected.
  - user message containing ``what are the correct llama.cpp
    arguments`` (``GREP_TEACH_TRIGGER``, the 2026-09-05 incident —
    the harness prior is that grep takes a REGEX; this app's grep is a
    case-insensitive fixed substring, owner-locked A5) **and** the
    system prompt carries the ``<tools>`` section -> the deterministic
    GREP-REGEX-TEACHING flow, discriminated statelessly from the
    messages (streaming only):
      * request 1 (``tools`` offered, no ``tool``-role result yet):
        stream ONLY ``tool_calls`` deltas — ``grep`` with
        ``{"pattern": GREP_TEACH_PATTERN}`` (``qwen.*3\\.8``, id
        ``call_0``) — the incident's regex-shaped first grep, which a
        fixed-substring grep can NEVER match;
      * request 2 (the last tool result is the server's TEACHING
        no-match line — it carries ``GREP_TEACH_MARKER``):
        ``grep`` with the plain form ``GREP_TEACH_PLAIN``
        (``qwen3.8``, id ``call_1``) — the one-round correction;
      * request 3 (the last tool result carries
        ``source/path:line: text`` match lines): ``read`` the FIRST
        match line's document by its combined ``source/path`` (id
        ``call_2``);
      * request 4 (the last tool result is a read result, the
        ``"Document <combined>:\n<content>"`` shape): the
        deterministic echo answer ``Read <combined>. <first 80
        chars>``, ``finish_reason: "stop"`` — the loop ended in ONE
        correction, not at the round cap.
      * A PLAIN no-match as the last result (no match line, no
        teaching marker — e.g. the plain pattern genuinely absent) is
        the deterministic terminal answer ``No matches — the knowledge
        base has no such text.`` (the flow cannot loop on a
        well-formed pattern).
    Checked BEFORE the SEARCH / TOOLS_TRIGGER flows (disjoint trigger
    phrases — the phase-72 ordering convention); no existing E2E
    question or fixture file contains the phrase, so every other suite
    is unaffected.
  - user message containing ``show me a table`` (phase 44, markdown
    tables, TODO.md L6) -> the fixed table answer (``TABLE_ANSWER``):
    a 3-column service table, an ``<img onerror>`` XSS probe line, and
    a deliberately wide 5-column table — byte-stable, so the story E2E
    can assert the rendered ``<table class="md-table">`` shape, the
    escaped XSS line, and the wrapper's horizontal scroll inside the
    72rem container column (phase 100). Checked BEFORE the
    ``DEFLECT_MODE`` branch (a
    deflection prompt never carries the marker, same reasoning as
    ``SUMMARY_MODE``), so a marker question always gets the table
    answer; the E2E asks it against an on-topic fixture (HIGH gate) and
    asserts non-deflection.
  - user message containing ``echo my history``
    (``HISTORY_TRIGGER``, phase 74, TODO L4 — chat history with prior
    thinking reaches the LLM) -> the deterministic HISTORY ECHO, derived
    statelessly from the request messages and byte-stable:
    ``history: N prior messages; last answer tail: <tail>; thinking:
    yes|no (Deterministic mock answer for E2E.)`` where N = the count
    of non-``system`` messages before the LAST ``user`` message
    (everything the client sent as prior turns — the current question
    itself is excluded), <tail> = the LAST 24 chars of the most recent
    prior ``assistant`` message's content (``none`` when there is no
    prior assistant message), and thinking is ``yes`` iff that prior
    ``assistant`` message carries a non-empty ``reasoning_content``
    field (the client's phase-74 history mapping of the brain record's
    ``thinking`` — A4). Checked BEFORE the ``DEFLECT_MODE`` branch
    (like ``TABLE_TRIGGER`` — the marker lives in the user message, a
    deflection prompt never carries it), so a marker question always
    gets the echo whatever the honesty gate says; the story E2E
    (``tests/e2e/test_llm_history.py``) asserts the wire contents
    byte-exactly against the conversation record the client persisted.
    Invariant the marker relies on: the client history contains ONLY
    ``user``/``assistant`` messages — never ``tool``-role ones (the
    client never sends tool calls/results) — so every existing marker
    flow (which classifies statelessly from TOOL results and the LAST
    user message) is unaffected by the now-always-present history.

  Failure injection (phase 67, LLM retry, TODO.md L3) — deterministic
  dead-endpoint behavior for the retry E2E suite (``tests/e2e/
test_llm_retry.py``). The mock is single-conversation per e2e server, so
  the sequences are driven by module-level counters that reset per
  trigger phrase after the success they guard (a second question with
  the same trigger re-drives the sequence from zero):
  - user message containing ``fail then answer`` (``RETRY_TRIGGER``):
    the first ``RETRY_DEAD_ATTEMPTS`` (2) app-level streaming attempts
    respond 500 (JSON body, like a dead proxy) and the third streams
    the normal composed answer — 2 = 1 original attempt + 1 retry under
    the default ``BOR_LLM_RETRIES=3``, so a suite exercises a real
    retry without waiting for the 4-attempt exhaustion. Counted in
    APP-LEVEL attempts, not raw HTTP POSTs: while the endpoint stays
    dead, the openai SDK's default policy (max_retries=2 — the app's
    ``LLMClient`` keeps it) re-POSTs a 500'd streaming request twice
    before surfacing the error, so each dead attempt costs exactly 3
    POSTs (``_HTTPS_PER_DEAD_ATTEMPT``).
  - user message containing ``always fail``
    (``ALWAYS_FAIL_TRIGGER``): EVERY streaming chat/completions request
    responds 500 — the retry-budget exhaustion path (the terminal
    error banner in the UI).
  - embeddings request whose input contains ``embed fail once``
    (``EMBED_FAIL_TRIGGER``): the FIRST such request responds 500, the
    next returns the normal bag-of-words vector — the endpoint's
    pre-stream embedding retry loop. Raw httpx on the client side (no
    SDK-level retries), so one POST per attempt: the counter is per
    POST here, unlike the chat counter above.
  Non-streaming requests (document summaries, KB overview) never 500 —
  the retry scope is the chat turn only (owner-locked A1). The one
  non-streaming injection is phase 96's incident shape below (it
  answers 200 with EMPTY content — the semantic failure class, not a
  dead endpoint).

  Failure injection (phase 96, one-shot resilience, task 04) — the
  2026-09-11 incident shape for the folder-summary one-shot path
  (``tests/e2e/test_oneshot_llm_retry.py``): a NON-stream
  ``chat/completions`` request whose system prompt carries
  ``FOLDER_SUMMARY_MODE`` (the folder-summary marker — ``chat()`` is
  the mock's only non-streaming consumer of it) and whose user
  message's ``Folder: …`` header (the branch's existing parse, the
  ``FOLDER_HEADER_PREFIX`` tail) labels this suite's own fixture
  folders:
  - the label ends with ``/e2e_empty_once``: the FIRST non-stream POST
    for that label answers the incident envelope — the mock's normal
    OpenAI chat-completion shape with ``choices[0].message.content =
    ""`` and ``choices[0].finish_reason = "length"`` (the exact wire
    shape of the empty ``lite`` reply: the budget spent in
    ``reasoning_content``) — and every later POST returns the normal
    ``Fixture folder summary for <folder>.`` line (the one-shot retry,
    phase 96 task 01, recovers the row).
  - the label ends with ``/e2e_empty_always``: EVERY non-stream POST
    for that label answers the empty envelope (the 1 +
    ``BOR_LLM_RETRIES`` exhaustion → per-folder fail-soft → the row
    stays absent while the sync stays green, phase 94 contract).
  The once-sequence is driven by a module-level per-label counter that
  resets after the success it guards (the phase-67 ``_fail_posts``
  pattern — the mock is single-conversation per e2e server), so a
  second sync re-drives the sequence deterministically. The trigger
  strings are this suite's own folder names, so no other E2E can hit
  them (they seed different trees).

  ``max_tokens`` is honored deterministically (token ≈ whitespace word),
  like a real endpoint: an answer longer than the cap is truncated. This
  is what makes the phase-11 truncation regression observable.
"""
from __future__ import annotations

import hashlib
import math
import os
import re
import signal
import threading
import time
import uuid
from typing import Any

from fastapi import FastAPI
from fastapi.responses import JSONResponse, StreamingResponse

from app.rag.agent import ALREADY_IN_CONTEXT, CORRECTION_INSTRUCTION
from app.rag.folder_summaries import FOLDER_HEADER_PREFIX  # phase 94: the mock's key

app = FastAPI()

DIM = 768
TOKEN_RE = re.compile(r"[a-z0-9]+")


def embed_text(text: str) -> list[float]:
    vec = [0.0] * DIM
    for tok in TOKEN_RE.findall(text.lower()):
        idx = int(hashlib.md5(tok.encode()).hexdigest(), 16) % DIM
        vec[idx] += 1.0
    norm = math.sqrt(sum(v * v for v in vec)) or 1.0
    return [v / norm for v in vec]


def _messages(body: dict[str, Any]) -> list[dict[str, str]]:
    return body.get("messages", [])


def _system(body: dict[str, Any]) -> str:
    return " ".join(m.get("content", "") for m in _messages(body) if m.get("role") == "system")


def _user(body: dict[str, Any]) -> str:
    parts = [m.get("content", "") for m in _messages(body) if m.get("role") == "user"]
    return parts[-1] if parts else ""


def _context(body: dict[str, Any]) -> str:
    """The document context is the longest system/user message in practice.

    ``m.get("content") or ""`` (NOT ``m.get("content", "")``): a well-formed
    OpenAI tool-call message carries ``content: None`` EXPLICITLY (the app's
    agent loop appends exactly that — ``app/rag/agent.py``), and a forced
    final answer after a tool round (the round-cap path) reaches this helper
    with those messages in play. The default-value form returns ``None`` for
    an explicit ``None`` and crashes ``len()`` with a 500 (phase 93 task 04
    caught it via the deterministic single-read flow's ALREADY_IN_CONTEXT
    loop); ``or ""`` treats absent and explicit-None alike, so the fallback
    composes deterministically instead of traceback-ing."""
    msgs = _messages(body)
    return max((m.get("content") or "" for m in msgs), key=len)


LONG_ANSWER_TRIGGER = "write a long answer"
#: ~920 words — comfortably past the old hard 700-token cap (where the
#: tail would be cut) yet short enough to stream in ~8s at the mock's
#: per-chunk pacing.
LONG_ANSWER_LINES = 40
LONG_ANSWER_END = "LONG-ANSWER-END"

#: Phase 17 (thinking-display story): a user message containing this
#: substring (case-insensitive) is answered with a deterministic
#: ``reasoning_content`` stream ahead of the content — same convention as
#: the other user-message triggers above. Existing E2E questions do not
#: contain the substring, so every other suite is unaffected.
THINKING_TRIGGER = "think out loud"

#: Regression pin (2026-08-29, owner report): a user message containing
#: this substring (case-insensitive) gets the phase-17 scratchpad WITH
#: REAL paragraph breaks ("\n\n"), streamed at ``THINK_PARAS_CHUNK``
#: chars/frame — a single frame renders several lines (a real-model-sized
#: delta), which is the condition under which the old POST-render pin
#: reading in app.js died at the first 2-newline gap. Existing E2E
#: questions do not contain the phrase, so every other suite is
#: unaffected (checked before ``THINKING_TRIGGER`` — the more specific
#: phrase wins).
THINK_PARAS_TRIGGER = "think in paragraphs"
#: 60-char thinking frames for the paragraph trigger (the mock default is
#: 12 — a 12-char frame renders at most one line ≈ 22px, always inside
#: the 32px think-window band, which is why the old code passed the
#: 12-char suites while the real model's larger deltas killed the pin).
THINK_PARAS_CHUNK = 60

#: Phase 20 (sources-midstream bug): a user message containing this
#: substring (case-insensitive) gets the phase-17 thinking stream followed
#: by a multi-second pause before the FIRST content frame — the
#: navigation window for the "leave during pure thinking" scenario
#: (owner-confirmed A1.2: nothing brain-side may be persisted then).
#: Strictly longer than ``THINKING_TRIGGER``, so the phase-17 suite's
#: questions are unaffected.
SLOW_PRETOKEN_TRIGGER = "think out loud then hesitate"
PRE_CONTENT_PAUSE_S = 4.0

#: Phase 24 (whole-document-context story): a user message containing this
#: substring (case-insensitive) gets an answer quoting the TAIL of the
#: document context (see the module docstring). Verified 2026-08-24: no
#: existing E2E question or fixture file contains the phrase, so every
#: other suite is unaffected.
END_OF_NOTES_TRIGGER = "show the end of your notes"

#: The ``<documents>`` block of the system prompt (phase 37: the HIGH
#: prompt ends with the ``<tools>`` section after ``</documents>``, so the
#: phase-24 tail echo targets the block, not the raw message tail).
_DOCUMENTS_BLOCK_RE = re.compile(r"<documents>.*?</documents>", re.S)

#: Phase 37 (agent-document-tools story; phase 70: the flow emits the
#: harness-aligned names): a user message containing this substring
#: (case-insensitive) — combined with the ``<tools>`` section in the
#: system prompt — drives the deterministic tool flow documented in the
#: module docstring (ls → read on the first catalog line's combined
#: ``source/path`` → the quoted answer). Existing E2E questions do not
#: contain the phrase, so every other suite is unaffected.
TOOLS_TRIGGER = "use your tools"

#: Phase 45 (agent-unlimited-tools story, task 02): a user message
#: containing BOTH ``TOOLS_TRIGGER`` and this substring (case-insensitive
#: — the check lowercases the user message) drives the deterministic
#: MULTI-READ tool flow (list → read #1 → read #2 → the forced answer
#: naming both read paths) — see the module docstring. The existing
#: phase-37 E2E question carries ``TOOLS_TRIGGER`` but not this phrase,
#: so the 3-step flow is untouched.
MULTI_READ_TRIGGER = "read two documents"

#: Phase 68 (search tool, TODO.md L4; phase 70: renamed to the
#: harness-aligned ``grep``): a user message containing this substring
#: (case-insensitive) — combined with the ``<tools>`` section in the
#: system prompt — drives the deterministic SEARCH tool flow (grep for
#: ``SEARCH_PATTERN`` → the "Found …" answer), documented in the module
#: docstring. Checked BEFORE ``TOOLS_TRIGGER``
#: (the more specific phrase wins — the same convention as
#: ``THINK_PARAS_TRIGGER``); verified 2026-09-01: no existing E2E
#: question or fixture file contains the phrase, so every other suite
#: is unaffected.
SEARCH_TRIGGER = "search your documents"

#: The sentinel the search flow greps for: the e2e fixture document
#: (``tests/fixtures/search_docs/reese-notes.md``) carries exactly one
#: line containing it, so the search result — and the "Found …" answer
#: that quotes its first matched line — is byte-stable (the sentinel
#: convention of ``END_OF_NOTES_TRIGGER``).
SEARCH_PATTERN = "reese-sentinel-42"

#: Phase 44 (markdown-tables story, TODO.md L6): a user message
#: containing this substring (case-insensitive) gets the fixed table
#: answer (``TABLE_ANSWER`` below) — a 3-column table, an XSS probe
#: line, and a deliberately wide table (see the module docstring).
#: Existing E2E questions do not contain the phrase, so every other
#: suite is unaffected.
TABLE_TRIGGER = "show me a table"

#: The fixed table answer (phase 44) — byte-stable on purpose: the story
#: E2E asserts the rendered table shape, the escaped ``<img onerror>``
#: line (the XSS payload must survive the mock byte-for-byte), and the
#: wide table's ``scrollWidth > clientWidth`` inside the 72rem container
#: column (phase 100).
#: Phase 74 (chat history, TODO L4): a user message containing this
#: substring (case-insensitive) gets the deterministic HISTORY ECHO
#: (``_history_echo`` below — see the module docstring): the prior-turn
#: count, the last 24 chars of the most recent prior answer, and
#: whether that prior answer carried ``reasoning_content``. Verified
#: 2026-09-08: no existing E2E question or fixture file contains the
#: phrase, so every other suite is unaffected.
HISTORY_TRIGGER = "echo my history"

TABLE_ANSWER = (
    "Here's the shape, in a table:\n"
    "\n"
    "| Service | Port | Host |\n"
    "|---|---|---|\n"
    "| Caddy | 80 | homelab-gw |\n"
    "| GitLab | 8929 | homelab-git |\n"
    "| ntfy | 2087 | homelab-ntfy |\n"
    "\n"
    "<img src=x onerror=alert(1)>\n"
    "\n"
    "And the wide one:\n"
    "\n"
    "| A very long column header to force overflow | Second column with "
    "some padding text | Third column | Fourth | Fifth |\n"
    "|---|---|---|---|---|\n"
    "| value-one | value-two | value-three | value-four | value-five |"
)

# ---------------------------------------------------------------------------
# Phase 67 (LLM retry, TODO.md L3): deterministic failure injection
# ---------------------------------------------------------------------------

#: A user message containing this substring (case-insensitive) gets
#: ``RETRY_DEAD_ATTEMPTS`` dead streaming attempts (500, JSON body) before
#: the normal composed answer streams — 1 original attempt + 1 retry under
#: the default ``BOR_LLM_RETRIES=3`` (see the module docstring).
RETRY_TRIGGER = "fail then answer"
#: App-level attempts the endpoint stays dead for before the answer.
RETRY_DEAD_ATTEMPTS = 2

#: A user message containing this substring (case-insensitive) makes
#: EVERY streaming chat/completions request respond 500 — the
#: retry-budget exhaustion path (the terminal error banner in the UI).
ALWAYS_FAIL_TRIGGER = "always fail"

#: An embeddings request whose input contains this substring
#: (case-insensitive) 500s on its FIRST POST; the next returns the normal
#: bag-of-words vector — the endpoint's pre-stream embedding retry loop.
EMBED_FAIL_TRIGGER = "embed fail once"

# ---------------------------------------------------------------------------
# Phase 71 (tool-scaffolding guardrails, 2026-09-03 incident):
# deterministic raw-markup flows — see the module docstring
# ---------------------------------------------------------------------------

#: A user message containing this substring (case-insensitive) drives
#: the scaffolding flow: request 1 streams ONLY the incident's raw tool
#: markup as ``delta.content``; the follow-up request carrying the
#: harness correction in the system prompt (the one bounded recovery)
#: streams the clean answer. Independent of the ``<tools>`` marker —
#: both grounded and deflected turns hit it. Existing E2E questions do
#: not contain the phrase, so every other suite is unaffected.
SCAFFOLD_TRIGGER = "emit raw tool markup"

#: A user message containing this substring (checked BEFORE
#: ``SCAFFOLD_TRIGGER`` — it contains that phrase) streams the
#: scaffolding-only span on EVERY request, recovery included — the
#: terminal malformed-reply path (the dedicated error frame, no done).
SCAFFOLD_ALWAYS_TRIGGER = "always emit raw tool markup"

#: The incident span (2026-09-03): the model's chat-template tool
#: syntax, emitted as plain ``delta.content`` although no tools were
#: offered. Streamed through the mock's 12-char chunking, so it always
#: spans ≥2 wire chunks (the filter's boundary path).
SCAFFOLD_SPAN = (
    "<|tool_call_start|>[read(path='search_docs/reese-notes.md')]"
    "<|tool_call_end|>"
)

#: The clean answer the one bounded recovery produces (byte-stable —
#: the dedicated E2E suite asserts the recovered bubble and the wire's
#: delta text against it).
SCAFFOLD_RECOVERY_ANSWER = "Here is the plain-text answer the recovery produced."

#: The stable substring of the harness-owned correction constant the
#: recovery request carries in its system prompt. Keyed on a substring
#: (not the whole constant) so a re-wrap of the constant cannot silently
#: re-route the mock; the module-level assert below fails loudly if the
#: substring ever leaves the constant (the mock must never drift from
#: ``app.rag.agent.CORRECTION_INSTRUCTION``).
_CORRECTION_MARKER = "no tool syntax"
assert _CORRECTION_MARKER in CORRECTION_INSTRUCTION, (
    "mock drift: the correction marker left CORRECTION_INSTRUCTION"
)

# ---------------------------------------------------------------------------
# Phase 72 (teaching refusals — the 2026-09-03 incident's ls misuse):
# the deterministic LS-TEACH self-correction flow — see the module
# docstring
# ---------------------------------------------------------------------------

#: A user message containing this substring (case-insensitive) —
#: combined with the ``<tools>`` section in the system prompt — drives
#: the deterministic LS-TEACHING flow (the incident's
#: ``ls(path='.')`` misuse → the teaching refusal → the corrected
#: no-arg ``ls()`` → the catalog answer). Checked BEFORE the plain
#: ``TOOLS_TRIGGER`` flow (disjoint trigger phrases — the phase-71
#: ordering convention); verified: no existing E2E question or fixture
#: file contains the phrase, so every other suite is unaffected.
LS_TEACH_TRIGGER = "list the files in this directory"

#: The deterministic GREP-REGEX-TEACH flow (the 2026-09-05 "Qwen 3.8"
#: incident — the harness prior is that grep takes a REGEX; this app's
#: grep is a case-insensitive fixed substring, owner-locked A5, so a
#: regex-shaped pattern can NEVER match, and the bare no-match line
#: made the turbo model trust the miss and end the turn with a wrong
#: "I searched the entire knowledge base" refusal). The flow pins the
#: self-correction on the SSE wire: the regex-shaped first grep → the
#: server's TEACHING no-match line (``GREP_TEACH_MARKER``) → the
#: plain-form retry grep → the match → the read → the deterministic
#: echo answer. Checked BEFORE the SEARCH / TOOLS_TRIGGER flows
#: (disjoint trigger phrases — the phase-71/72 ordering convention);
#: verified: no existing E2E question or fixture file contains the
#: phrase, so every other suite is unaffected.
GREP_TEACH_TRIGGER = "what are the correct llama.cpp arguments"

#: The incident's regex-shaped first grep (it can never match a
#: fixed-substring grep — that is the point of the flow).
GREP_TEACH_PATTERN = "qwen.*3\\.8"

#: The plain-form retry — the server's teaching line hands over exactly
#: this hint (``app.rag.agent.plain_form(GREP_TEACH_PATTERN)``).
GREP_TEACH_PLAIN = "qwen3.8"

#: The marker of the agent's teaching no-match line (app.rag.agent
#: ``NO_MATCHES_REGEX`` / ``NO_MATCHES_REGEX_SCOPED``) — the mock's
#: plain step keys on it (a plain no-match line carries it not).
GREP_TEACH_MARKER = "grep matches a plain substring"

# ---------------------------------------------------------------------------
# Phase 94 (task 04, the drill-down ls's dedicated story suite):
# the deterministic SCRIPTED drill-down turns — see the module docstring
# ---------------------------------------------------------------------------

#: A user message containing this substring (case-insensitive) —
#: combined with the ``<tools>`` section in the system prompt — drives
#: the scripted DRILL-DOWN flow: the question carries its own tool call
#: after the colon (``drill down the tree: ls [target]`` / ``drill
#: down the tree: read source/path`` — the suite's scripted turns,
#: ``tests/e2e/test_ls_tree_drilldown.py``). Checked BEFORE the plain
#: ``TOOLS_TRIGGER`` flow (disjoint trigger phrases — the phase-71/72
#: ordering convention); verified: no existing E2E question or fixture
#: file contains the phrase, so every other suite is unaffected.
DRILL_TRIGGER = "drill down the tree"

#: The scripted call in the drill-down question (case-insensitive — the
#: suite's questions capitalize the trigger's first letter): the verb
#: (``ls`` / ``read``) plus the optional target — a source name or
#: ``source/folder`` path (``ls``) or a combined ``source/path``
#: (``read``), parsed from the RAW user message so the target keeps its
#: case. The target is a ``[a-z0-9_./-]`` run (case-insensitively), so
#: the suite's `` — `` flavor separator (em dash) can never bleed into
#: it; no run = the top-level ``ls`` (no ``path`` argument).
_DRILL_CALL_RE = re.compile(
    r"drill down the tree:\s*(?P<verb>ls|read)(?:\s+(?P<arg>[a-z0-9_./-]+))?",
    re.I,
)

#: The agent's NOT-A-FOLDER teaching refusal marker (app.rag.agent
#: ``NOT_A_FOLDER`` — ``'{arg}' is not a folder — {parent} has:
#: {subfolders}``): the drill-down flow's scripted recovery keys on it
#: in the LAST tool result (the refusal being visible to the model is
#: exactly what triggers the one-round recovery — the phase-72
#: self-correction contract carrying the tree's teaching line).
_NOT_A_FOLDER_MARKER = "is not a folder"

#: The stable substring of the harness-owned dedupe refusal the
#: drill-down flow's ``ctx_answer`` branch keys on (a read target that
#: is already a top-2 retrieval document — the mock then answers from
#: the document's text in the ``<documents>`` prompt block, exactly
#: what the refusal instructs). Keyed on a substring (not the whole
#: constant) so a re-wrap of the constant cannot silently re-route the
#: mock; the module-level assert below fails loudly if the substring
#: ever leaves the constant (the mock must never drift from
#: ``app.rag.agent.ALREADY_IN_CONTEXT``).
_ALREADY_IN_CONTEXT_MARKER = "Already in your context"
assert _ALREADY_IN_CONTEXT_MARKER in ALREADY_IN_CONTEXT, (
    "mock drift: the dedupe marker left ALREADY_IN_CONTEXT"
)

#: One ``<document>`` block of the HIGH prompt's ``<documents>``
#: section (``app.rag.prompts.build_high_prompt``): the block is the
#: document identity (``source``/``path``/``title`` attributes) plus
#: the document's FULL text (never truncated on the retrieval path,
#: owner-locked A7) between the tags.
_DOCUMENT_BLOCK_RE = re.compile(
    r'<document source="(?P<source>[^"]+)" path="(?P<path>[^"]+)" '
    r'title="[^"]*">\n(?P<content>.*?)\n</document>',
    re.S,
)


def _document_block(system: str, source: str, path: str) -> str | None:
    """The stored text of one ``<document>`` block (or ``None``).

    The drill-down flow's ``ctx_answer`` branch: when the agent's read
    of a top-2 retrieval document gets the ALREADY_IN_CONTEXT dedupe,
    the document's full text is in the ``<documents>`` prompt — the
    mock (the model) extracts it by the block's identity attributes
    and quotes it, answering from the prompt as the refusal instructs.
    """
    for block in _DOCUMENT_BLOCK_RE.finditer(system):
        if block.group("source") == source and block.group("path") == path:
            return block.group("content")
    return None

#: The agent's drill-down ``ls`` result shapes (app.rag.agent
#: ``render_ls_top`` / ``render_folder_listing``, phase 94): the
#: top-level header ``"N sources:"`` and the per-source block line
#: ``"<source> — N documents"`` (summary lines are indented — they
#: never match); the folder-level header ``"<identity> — N documents,
#: M folders:"`` (identity = the source name or ``source/folder``).
_SOURCE_HEADER_RE = re.compile(r"^\d+ sources:$")
_SOURCE_LINE_RE = re.compile(r"^(?P<source>.+?) — (?P<n>\d+) documents$")
_FOLDER_HEADER_RE = re.compile(r"^(?P<identity>.+?) — \d+ documents, \d+ folders:$")

#: One DEAD app-level chat attempt costs exactly this many HTTP POSTs
#: while the endpoint stays down: the openai SDK's default policy
#: (max_retries=2 — the app's ``LLMClient`` keeps it) re-POSTs a 500'd
#: streaming request twice before surfacing the error to
#: ``chat_stream_retried``. The failure counters below therefore count
#: app-level attempts (groups of this size), not raw POSTs — the visible
#: sequence (one SSE ``retry`` frame after each dead attempt, the answer
#: on the third) stays deterministic regardless of the SDK's internal
#: backoff pacing.
_HTTPS_PER_DEAD_ATTEMPT = 3

#: Module-level failure counters — the mock is single-conversation per
#: e2e server. Keyed by trigger phrase (reset per trigger): the number
#: of matching POSTs served so far. Each sequence resets after the
#: success it guards, so a second question carrying the same trigger
#: re-drives the failure sequence from zero.
_fail_posts: dict[str, int] = {}


def _llm_500(why: str) -> JSONResponse:
    """A dead-proxy 500 with a JSON error body (phase 67 injection)."""
    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "message": f"upstream connection reset ({why})",
                "type": "proxy_error",
            }
        },
    )


def _bump_fail(key: str) -> int:
    n = _fail_posts.get(key, 0) + 1
    _fail_posts[key] = n
    return n


def _chat_dead(key: str, dead_attempts: int) -> bool:
    """Bump *key*'s counter; True while the endpoint stays dead.

    Counted in app-level attempts (see ``_HTTPS_PER_DEAD_ATTEMPT``): the
    first ``dead_attempts * _HTTPS_PER_DEAD_ATTEMPT`` POSTs 500 and the
    next attempt's first POST streams (the caller resets the counter on
    the success).
    """
    return _bump_fail(key) <= dead_attempts * _HTTPS_PER_DEAD_ATTEMPT


#: The agent's ``read`` tool-result prefix (app.rag.agent
#: ``_execute_tool``): ``"Document <source/path>:\n<content>"``.
_READ_RESULT_PREFIX = "Document "

#: One line of the agent's ``ls`` output (app.rag.agent
#: ``_execute_tool``, phase 63): labeled, pipe-delimited fields —
#: ``source: X | path: Y | title: Z`` — unambiguous for LLM parsing even
#: when the path contains ``/`` characters.
_CATALOG_LINE_RE = re.compile(
    r"^source: (?P<source>.+?) \| path: (?P<path>.+?) \| title: .+$"
)


def _read_results(body: dict[str, Any]) -> list[tuple[str, str]]:
    """The read results in the messages, in order: ``(source/path, content)``.

    A read result is a ``tool``-role message whose content starts with
    the agent's read-result prefix (``app.rag.agent`` ``_execute_tool``):
    ``"Document <source/path>:\n<content>"``. The header is stripped of
    the prefix AND the trailing colon so the path stays clean.
    """
    out: list[tuple[str, str]] = []
    for m in _messages(body):
        if m.get("role") != "tool":
            continue
        content = str(m.get("content") or "")
        if content.startswith(_READ_RESULT_PREFIX):
            header, _, doc_content = content.partition("\n")
            sp = header[len(_READ_RESULT_PREFIX):].strip().removesuffix(":")
            out.append((sp, doc_content))
    return out


def _catalog_docs(body: dict[str, Any]) -> list[tuple[str, str]]:
    """Every ``(source, path)`` in the catalog tool result, in listing order.

    Catalog lines are ``source: X | path: Y | title: Z`` (the agent's
    ``ls`` output — phase 63: labeled, pipe-delimited
    fields, unambiguous even for paths full of ``/``): the line-level
    regex recovers the ``source`` and ``path`` fields directly. The
    ``"N documents:"`` header line matches no line and is skipped;
    read-result messages are full documents, not listings, and are
    skipped too.
    """
    docs: list[tuple[str, str]] = []
    for m in _messages(body):
        if m.get("role") != "tool":
            continue
        content = str(m.get("content") or "")
        if content.startswith(_READ_RESULT_PREFIX):
            continue
        for line in content.splitlines():
            match = _CATALOG_LINE_RE.match(line)
            if match:
                docs.append((match.group("source"), match.group("path")))
    return docs


def _tool_results(body: dict[str, Any]) -> list[str]:
    """Every ``tool``-role result content in the messages, in order.

    (Phase 72, LS-TEACH flow: the flow is discriminated statelessly
    from the tool results — a catalog listing vs the teaching
    refusal vs none yet.)
    """
    return [
        str(m.get("content") or "")
        for m in _messages(body)
        if m.get("role") == "tool"
    ]


def _first_file_line(body: dict[str, Any]) -> str | None:
    """The first file line (``source: X | path: Y | title: Z``) across
    the ``tool``-role results in the messages, in message order (read
    results — full documents, not listings — skipped; the ``_CATALOG_-
    LINE_RE`` machinery).

    Phase 94: the drill-down ``ls`` carries file lines only at folder
    levels — the top-level source listing and the teaching refusals
    have none, so ``None`` here means "no folder level reached yet".
    """
    for content in _tool_results(body):
        if content.startswith(_READ_RESULT_PREFIX):
            continue
        for line in content.splitlines():
            if _CATALOG_LINE_RE.match(line):
                return line
    return None


def _drill_target(body: dict[str, Any]) -> str | None:
    """The next source to drill into (phase 94, the drill step).

    The drill-down ``ls`` top level lists SOURCES only (no file lines),
    so a deterministic flow that needs a file line must drill one level:
    ``ls`` scoped to a source. The target is the FIRST source of the
    top-level listing (registry order — the listing's own order) that
    does not yet have a folder-level listing in the messages (a folder
    header whose identity is the source or ``source/…``); ``None`` when
    no top-level listing is in the messages yet (nothing to drill from)
    or every listed source has been drilled (an empty KB — the flow
    degenerates to the old re-list loop, settling at the round cap
    exactly like the phase-70 empty-catalog case).
    """
    sources: list[str] = []
    drilled: set[str] = set()
    for content in _tool_results(body):
        lines = content.splitlines()
        if not lines:
            continue
        if _SOURCE_HEADER_RE.match(lines[0]):
            for line in lines[1:]:
                match = _SOURCE_LINE_RE.match(line)
                if match:
                    sources.append(match.group("source"))
        else:
            folder = _FOLDER_HEADER_RE.match(lines[0])
            if folder:
                drilled.add(folder.group("identity"))
    for source in sources:
        if not any(
            identity == source or identity.startswith(source + "/")
            for identity in drilled
        ):
            return source
    return None


#: One line of the agent's ``grep`` output (app.rag.agent
#: ``_execute_tool``, phase 68 — phase 70 renamed the tool, the line
#: format is unchanged): ``source/path:LINE: text``. The
#: non-greedy prefix keeps nested paths (``/`` in the path) intact.
_SEARCH_LINE_RE = re.compile(r"^(?P<sp>.+?):(?P<line>\d+): (?P<text>.*)$")


def _search_result_line(body: dict[str, Any]) -> str | None:
    """The first matched line's text of a search result in the messages.

    A search result is a ``tool``-role message — never a read result
    (those start with the agent's ``"Document "`` prefix) — that either
    carries ``source/path:LINE: text`` match lines (the agent's
    ``grep`` output, phase 68) or the sentinel pattern
    itself (its no-match line quotes the pattern). Returns the first
    match line's ``text`` part (already 200-char-capped server-side),
    or the message's first line in the sentinel-only shape, or ``None``
    when no search result is in the messages yet.
    """
    sentinel = SEARCH_PATTERN.lower()
    for m in _messages(body):
        if m.get("role") != "tool":
            continue
        content = str(m.get("content") or "")
        if content.startswith(_READ_RESULT_PREFIX):
            continue
        for line in content.splitlines():
            match = _SEARCH_LINE_RE.match(line)
            if match:
                return match.group("text")
        if sentinel in content.lower():
            lines = content.splitlines()
            return lines[0] if lines else ""
    return None


def _search_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a SEARCH_TRIGGER request into a step of the search flow.

    * ``("search",)`` — ``tools`` are offered and no search result is
      in the messages yet: the model greps the whole KB for
      ``SEARCH_PATTERN`` (id ``call_0``).
    * ``("found", first_line)`` — a ``tool``-role search result is in
      the messages: the model answers, quoting the first matched line
      (``Found <first matched line's content up to 80 chars>``). Reached
      regardless of the ``tools`` parameter (phase 45 keeps the tools
      offered until the round cap).
    * ``None`` — not the search flow: the trigger is absent, the
      ``<tools>`` section is missing (deflected turns never carry it),
      or ``tools`` are not offered and no search result is in the
      messages yet (e.g. ``agent_max_rounds=0``).
    """
    if SEARCH_TRIGGER not in _user(body).lower():
        return None
    if "<tools>" not in _system(body):
        return None
    first_line = _search_result_line(body)
    if first_line is not None:
        return ("found", first_line)
    if not body.get("tools"):
        return None
    return ("search",)


def _scaffold_flow(body: dict[str, Any]) -> str | None:
    """Classify a phase-71 scaffolding request (see the module docstring).

    * ``"scaffold"`` — stream ONLY the incident span
      (``SCAFFOLD_SPAN``) as ``delta.content`` chunks: ``finish_reason:
      "stop"``, no structured ``tool_calls``, no reasoning. EVERY
      request for ``SCAFFOLD_ALWAYS_TRIGGER`` (the recovery included),
      and the FIRST request of ``SCAFFOLD_TRIGGER`` (no correction in
      the system prompt yet).
    * ``"recovery"`` — ``SCAFFOLD_TRIGGER`` whose system prompt carries
      the harness correction (the one bounded recovery: ``tools=None``,
      the constant folded into the single system prompt by
      ``app.api.chat`` / ``app.rag.agent``): stream the clean
      ``SCAFFOLD_RECOVERY_ANSWER``.
    * ``None`` — not the scaffolding flow. The discrimination is
      stateless, like the other marker flows: the trigger phrase in
      the user message plus the correction's presence in the system
      prompt.
    """
    user = _user(body).lower()
    if SCAFFOLD_ALWAYS_TRIGGER in user:  # checked FIRST — it contains SCAFFOLD_TRIGGER
        return "scaffold"
    if SCAFFOLD_TRIGGER in user:
        if _CORRECTION_MARKER in _system(body):
            return "recovery"
        return "scaffold"
    return None


def _tool_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a marker request into one step of the tool flow.

    Single-read (phase 37 — the user message carries ``TOOLS_TRIGGER``
    only):

    * ``("list", "", "")`` — ``tools`` are offered and no tool results
      are in the messages yet: the model lists the catalog.
    * ``("read", source, path, "call_1")`` — a ``tool``-role catalog
      result is in the messages: the model reads its FIRST
      ``source: X | path: Y | title: Z`` line (the labeled
      ``source:`` / ``path:`` fields, phase 63), emitted as ``read`` on
      the JOINED combined ``source/path`` (phase 70: the mock joins
      the two fields — the canonical document identity).
    * ``("answer", "source/path", content)`` — a ``tool``-role read
      result (``"Document <source/path>:\n<content>"``) is in the
      messages: the model answers, quoting the read document. Reached
      regardless of the ``tools`` parameter (phase 45 keeps the tools
      offered until the round cap).

    Multi-read (phase 45 task 02 — the user message carries BOTH
    ``TOOLS_TRIGGER`` and ``MULTI_READ_TRIGGER``), classified by the
    count of ``tool``-role read results:

    * 0 read results: ``("list", "", "")`` (no listing yet),
      ``("drill", source, "call_1")`` when the top-level source
      listing is in the messages but no file line yet (phase 94: the
      drill — the top level carries sources only), or
      ``("read", source, path, "call_2")`` on the FIRST file-line doc.
    * 1 read result: ``("read", source, path, "call_3")`` on the SECOND
      file-line doc — the first listing line whose ``source/path``
      differs from the one already read. A one-file listing
      degenerates to the single-read ``("answer", ...)`` shape (nothing
      second to read).
    * 2 read results: ``("multi_answer", "", text)`` — the forced
      answer, byte-stable: the single-read shape quoting the FIRST read
      result, plus ``I read <sp1> and <sp2>.`` (both read paths, read
      order). The second element is unused.

    * ``None`` — not the marker flow: the request behaves exactly as
      before (marker absent, no ``<tools>`` section, or a no-tools
      request with no tool results — e.g. ``agent_max_rounds=0``).
    """
    user = _user(body).lower()
    if TOOLS_TRIGGER not in user:
        return None
    if "<tools>" not in _system(body):
        return None
    reads = _read_results(body)
    if MULTI_READ_TRIGGER in user:
        if not reads:
            if not body.get("tools"):
                return None
            docs = _catalog_docs(body)
            if docs:
                return ("read", docs[0][0], docs[0][1], "call_2")
            drill = _drill_target(body)
            if drill is not None:
                return ("drill", drill, "call_1")
            return ("list", "", "")
        if len(reads) == 1:
            skip = reads[0][0]
            second = next(
                (d for d in _catalog_docs(body) if f"{d[0]}/{d[1]}" != skip), None
            )
            if second is None:
                # One-file listing: nothing second to read — the
                # single-read answer shape (deterministic degenerate).
                return ("answer", reads[0][0], reads[0][1])
            return ("read", second[0], second[1], "call_3")
        (sp1, c1), (sp2, _c2) = reads[0], reads[1]
        answer = f"Read {sp1}. {c1[:80]} I read {sp1} and {sp2}."
        return ("multi_answer", "", answer)
    # Phase-37 single-read flow (phase 94: the drill step between the
    # top-level listing and the first file line).
    if reads:
        return ("answer", reads[0][0], reads[0][1])
    if not body.get("tools"):
        return None
    docs = _catalog_docs(body)
    if docs:
        return ("read", docs[0][0], docs[0][1], "call_2")
    drill = _drill_target(body)
    if drill is not None:
        return ("drill", drill, "call_1")
    return ("list", "", "")


def _ls_teach_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a phase-72 LS-TEACH request (see the module docstring).

    * ``("misuse",)`` — ``tools`` are offered and no ``tool``-role
      result is in the messages yet: the incident's misuse — ``ls``
      with ``{"path": "."}`` (id ``call_0``), ``finish_reason:
      "tool_calls"``, no content.
    * ``("correct",)`` — a ``tool``-role result is in the messages and
      no file line exists yet (the teaching refusal): the correction —
      ``ls`` with no arguments (id ``call_1``).
    * ``("drill", source, "call_2")`` — the top-level source listing
      is in the messages (the correction ran) but no file line yet
      (phase 94: the top level carries sources only): the drill —
      ``ls`` scoped to the first source of the listing.
    * ``("answer", line)`` — a ``source: X | path: Y | title: Z`` file
      line is in the messages (the folder listing): the deterministic
      content answer ``These are the indexed documents: <line>`` (the
      first file line), ``finish_reason: "stop"`` — the loop settled
      in ONE correction + ONE drill, not at the round cap.
    * ``None`` — not the flow: the trigger is absent, the ``<tools>``
      section is missing (deflected turns never carry it), or
      ``tools`` are not offered and no tool results are in the
      messages yet (e.g. ``agent_max_rounds=0``).
    """
    if LS_TEACH_TRIGGER not in _user(body).lower():
        return None
    if "<tools>" not in _system(body):
        return None
    line = _first_file_line(body)
    if line is not None:
        return ("answer", line)
    drill = _drill_target(body)
    if drill is not None:
        return ("drill", drill, "call_2")
    if _tool_results(body):
        return ("correct",)
    if not body.get("tools"):
        return None
    return ("misuse",)


def _grep_teach_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a GREP-TEACH request (the 2026-09-05 incident — see the
    ``GREP_TEACH_*`` constants). Stateless over the messages, like the
    other marker flows:

    * ``("regex",)`` — ``tools`` are offered and no ``tool``-role result
      is in the messages yet: the incident's regex-shaped first grep —
      ``grep`` with ``{"pattern": GREP_TEACH_PATTERN}`` (id ``call_0``).
    * ``("plain",)`` — the LAST tool result is the server's TEACHING
      no-match line (it carries ``GREP_TEACH_MARKER``): the one-round
      correction — ``grep`` with the plain form (id ``call_1``).
    * ``("read", combined, "call_2")`` — the last tool result carries
      ``source/path:line: text`` match lines: ``read`` the FIRST match
      line's document by its combined ``source/path`` identity.
    * ``("answer", combined, content)`` — the last tool result is a
      read result (``"Document <combined>:\n<content>"``): the
      deterministic echo answer ``Read <combined>. <first 80 chars>``.
    * ``("nomatch",)`` — the last tool result is a PLAIN no-match (no
      match line, no teaching marker): the deterministic terminal
      ``No matches — the knowledge base has no such text.`` answer.
    * ``None`` — not the flow: the trigger is absent, the ``<tools>``
      section is missing (deflected turns never carry it), or ``tools``
      are not offered and no tool results are in the messages yet (e.g.
      ``agent_max_rounds=0``).
    """
    user = _user(body).lower()
    if GREP_TEACH_TRIGGER not in user:
        return None
    if "<tools>" not in _system(body):
        return None
    results = [
        str(m.get("content") or "")
        for m in _messages(body)
        if m.get("role") == "tool"
    ]
    if not results:
        if not body.get("tools"):
            return None
        return ("regex",)
    last = results[-1]
    if last.startswith(_READ_RESULT_PREFIX):
        head, _, content = last.partition("\n")
        # The read result is ``"Document <combined>:\n<content>"`` — the
        # head carries the server's appended ``:`` (removed here; a
        # document path never legitimately ends with one).
        return ("answer", head[len(_READ_RESULT_PREFIX):].removesuffix(":"), content)
    if GREP_TEACH_MARKER in last:
        return ("plain",)
    for line in last.splitlines():
        m = _SEARCH_LINE_RE.match(line)
        if m:
            return ("read", m.group("sp"), "call_2")
    return ("nomatch",)


def _drill_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a phase-94 scripted drill-down request (see the module
    docstring). The question carries the scripted call
    (``drill down the tree: ls [target]`` / ``drill down the tree: read
    source/path``); the step is then discriminated statelessly from the
    tool results, like the other marker flows:

    * ``("call", verb, target, "call_0")`` — ``tools`` are offered and
      no ``tool``-role result is in the messages yet: the scripted call
      (``ls`` with NO path for the top level — empty target —, ``ls``
      with the target for source/folder levels, ``read`` with the
      combined ``source/path``).
    * ``("recover", source)`` — the LAST tool result is the NOT-A-FOLDER
      teaching refusal (it carries :data:`_NOT_A_FOLDER_MARKER`): the
      scripted one-round recovery — ``ls`` the target's SOURCE segment
      (the refusal listing the parent's subfolders is what lets the
      model — and this mock — self-correct in the next round, the
      phase-72 contract).
    * ``("read_answer", combined, quote)`` — the LAST tool result is a
      read result (``"Document <source/path>:\n<content>"``): the
      deterministic echo answer ``Read <source/path>. <first 80 chars>
      `` (the phase-37 single-read shape — the grounded-turn citation
      contract the suite asserts).
    * ``("ctx_answer", target, quote)`` — the scripted call was a
      ``read`` and the LAST tool result is the ALREADY_IN_CONTEXT
      dedupe refusal (the target is already a top-2 retrieval document
      — deterministic for the drill fixture: the read question names
      the file's path, so the file self-matches the hybrid gate): the
      model answers FROM THE PROMPT — ``Already in context: Read
      <target>. <first 80 chars of the target document's text in the
      ``<documents>`` block>`` (the same citation shape as the
      read-result branch — the document text reached the model either
      way). Falls through to the echo when the target is not a
      ``<document>`` block (the premise broke — the suite fails
      loudly on the answer).
    * ``("echo", listing)`` — a listing landed (top-level or folder —
      any other last result): the deterministic ECHO — the answer
      carries the listing VERBATIM (``Here's the level I listed:\n
      <listing>``), so the suite asserts on the exact tree level the
      model saw (source ``— N documents`` lines + stored summaries,
      subfolder lines, the ``source: X | path: Y | title: Z`` file
      lines, the 50-line cap + note) through the rendered answer — the
      house scripted-turn way of asserting on tool results (the mock is
      the only E2E lens on the LLM's context).
    * ``None`` — not the flow: the trigger is absent, the ``<tools>``
      section is missing (deflected turns never carry it), the scripted
      call is unparseable, or ``tools`` are not offered and no tool
      results are in the messages yet (e.g. ``agent_max_rounds=0``).
    """
    user = _user(body)
    if DRILL_TRIGGER not in user.lower():
        return None
    if "<tools>" not in _system(body):
        return None
    match = _DRILL_CALL_RE.search(user)
    if match is None:
        return None
    verb = match.group("verb")
    target = match.group("arg") or ""
    results = _tool_results(body)
    if not results:
        if not body.get("tools"):
            return None
        return ("call", verb, target, "call_0")
    last = results[-1]
    if last.startswith(_READ_RESULT_PREFIX):
        head, _, content = last.partition("\n")
        # The read result is ``"Document <source/path>:\n<content>"`` —
        # the head carries the server's appended ``:`` (removed here;
        # a document path never legitimately ends with one).
        combined = head[len(_READ_RESULT_PREFIX):].strip().removesuffix(":")
        return ("read_answer", combined, content[:80])
    if _ALREADY_IN_CONTEXT_MARKER in last and verb == "read" and "/" in target:
        # The dedupe fired: the read target is already a top-2
        # retrieval document, so its FULL text is in the
        # ``<documents>`` prompt — answer from the prompt (the
        # refusal's instruction), quoting the block's text.
        src, _, p = target.partition("/")
        content = _document_block(_system(body), src, p)
        if content is not None:
            return ("ctx_answer", target, content[:80])
    if _NOT_A_FOLDER_MARKER in last:
        return ("recover", target.split("/")[0])
    return ("echo", last)


# ---------------------------------------------------------------------------
# Phase 95 (task 03, the read cap's dedicated story suite):
# the deterministic SCRIPTED capped read — see the module docstring
# ---------------------------------------------------------------------------

#: A user message containing this substring (case-insensitive) —
#: combined with the ``<tools>`` section in the system prompt — drives
#: the scripted CAPPED-READ flow (the read cap's story suite,
#: ``tests/e2e/test_read_truncation_cap.py``): the question carries its
#: own tool call after the colon — ``read the capped document: read
#: source/path`` — the mock emits the scripted ``read``, then ECHOES
#: the ENTIRE tool result into its answer (the house scripted-turn lens
#: on the LLM's context: the truncated shape — first cap chars +
#: ``[…truncated…]`` + the pinned grep-pointer notice — or the plain
#: shape, byte-identical to the pre-phase-95 result, lands in the
#: rendered answer, and the suite asserts both directions through it).
#: Checked BEFORE the plain ``TOOLS_TRIGGER`` flow (disjoint trigger
#: phrases — the phase-71/72/94 ordering convention); verified
#: 2026-09-10: no existing E2E question or fixture file contains the
#: phrase, so every other suite is unaffected.
READ_CAP_TRIGGER = "read the capped document"

#: The scripted call in the read-cap question (case-insensitive — the
#: suite's questions capitalize the trigger's first letter): the verb
#: (``read``) plus the target — a combined ``source/path``, parsed from
#: the RAW user message so the target keeps its case. The target is a
#: ``[a-z0-9_./-]`` run (case-insensitively), so the suite's `` — ``
#: flavor separator (em dash) can never bleed into it (the drill-down
#: convention, ``_DRILL_CALL_RE``).
_READ_CAP_CALL_RE = re.compile(
    r"read the capped document:\s*read\s+(?P<arg>[a-z0-9_./-]+)",
    re.I,
)


def _read_cap_flow(body: dict[str, Any]) -> tuple[str, ...] | None:
    """Classify a phase-95 scripted read-cap request (see the module
    docstring). The question carries the scripted call (``read the
    capped document: read source/path``); the step is then discriminated
    statelessly from the tool results, like the other marker flows:

    * ``("call", target, "call_0")`` — ``tools`` are offered and no
      ``tool``-role result is in the messages yet: the scripted
      ``read`` on the parsed target (synthetic id ``call_0``).
    * ``("echo", result)`` — a ``tool``-role result is in the messages:
      the deterministic ECHO — the answer carries the LAST tool result
      VERBATIM (``Here's what the read returned:\n<result>``): a read
      result (``"Document <source/path>:…``) lands in the answer with
      its full content — the truncation marker + the pinned notice when
      the cap fired, the plain body when it did not — and a refusal
      (the premise broke) lands just as visibly, so the suite fails
      loudly on it.
    * ``None`` — not the flow: the trigger is absent, the ``<tools>``
      section is missing (deflected turns never carry it), the scripted
      call is unparseable, or ``tools`` are not offered and no tool
      results are in the messages yet (e.g. ``agent_max_rounds=0``).
    """
    user = _user(body)
    if READ_CAP_TRIGGER not in user.lower():
        return None
    if "<tools>" not in _system(body):
        return None
    match = _READ_CAP_CALL_RE.search(user)
    if match is None:
        return None
    results = _tool_results(body)
    if not results:
        if not body.get("tools"):
            return None
        return ("call", match.group("arg"), "call_0")
    return ("echo", results[-1])


def long_answer() -> str:
    """~900-word deterministic walkthrough (phase 11): numbered steps plus
    a unique final line that must survive the stream untruncated."""
    lines = [
        f"{i}. Step {i}: configure node-{i} with the homelab defaults and "
        f"verify that step {i} of the long walkthrough is complete before moving on."
        for i in range(1, LONG_ANSWER_LINES + 1)
    ]
    lines.append(LONG_ANSWER_END)
    return "\n".join(lines)


#: First numbered note line of a ``<tuning>`` section (phase 15).
_TUNING_BLOCK_RE = re.compile(r"<tuning>\n(.*?)\n</tuning>", re.S)
_NOTE_LINE_RE = re.compile(r"^\d+\.\s*(.+)$")


def first_tuning_note(system: str) -> str | None:
    """The first steering note in the system prompt, or ``None``.

    The prompt numbers notes 1..N oldest-first (see
    ``app.rag.prompts.build_steering_section``); the mock echoes the first
    one into its answer so prompt injection is observable in the UI.
    """
    block = _TUNING_BLOCK_RE.search(system)
    if not block:
        return None
    for line in block.group(1).splitlines():
        m = _NOTE_LINE_RE.match(line.strip())
        if m:
            return m.group(1).strip()
    return None


#: First ``-`` bullet line of a ``<knowledge_base>`` section (phase 31).
_KB_BLOCK_RE = re.compile(r"<knowledge_base>\n(.*?)\n</knowledge_base>", re.S)
_KB_BULLET_RE = re.compile(r"^-(?:\s+(.*))?$")


def first_kb_bullet(system: str) -> str | None:
    """The first outline bullet in the system prompt, or ``None``.

    The stored outline (phase 31) is ``-`` bullet lines (see
    ``app.rag.overview.OVERVIEW_INSTRUCTION``); the mock echoes the first
    one into its answer as `` (kb: <bullet>)`` — the exact
    :func:`first_tuning_note` convention, so prompt injection of the
    ``<knowledge_base>`` section is observable in the UI.
    """
    block = _KB_BLOCK_RE.search(system)
    if not block:
        return None
    for line in block.group(1).splitlines():
        m = _KB_BULLET_RE.match(line.strip())
        if m:
            return (m.group(1) or "").strip()
    return None


def _history_echo(body: dict[str, Any]) -> str:
    """The phase-74 history echo (byte-stable, stateless over messages).

    ``history: N prior messages`` — N = the count of non-``system``
    messages before the LAST ``user`` message (the client's phase-74
    ``history`` block: the prior turns only, the current question
    itself excluded). ``last answer tail: <tail>`` — the LAST 24 chars
    of the most recent prior ``assistant`` message's content, or
    ``none`` when there is no prior assistant message (the cold-start
    pin: no phantom history). ``thinking: yes|no`` — ``yes`` iff that
    prior assistant message carries a non-empty ``reasoning_content``
    field (A4: the client's prior thinking, mapped by
    ``app.rag.prompts.history_to_messages``), ``no`` otherwise.

    The invariant (see the module docstring): the client history is
    ``user``/``assistant``-only, so the last ``user`` message is always
    the current question and every earlier non-system message is a
    client-provided prior turn.
    """
    msgs = _messages(body)
    last_user = max(
        (i for i, m in enumerate(msgs) if m.get("role") == "user"),
        default=-1,
    )
    prior = [
        m
        for i, m in enumerate(msgs)
        if i < last_user and m.get("role") != "system"
    ]
    tail = "none"
    thinking = "no"
    for m in reversed(prior):
        if m.get("role") == "assistant":
            tail = str(m.get("content") or "")[-24:]
            thinking = "yes" if str(m.get("reasoning_content") or "") else "no"
            break
    return (
        f"history: {len(prior)} prior messages; "
        f"last answer tail: {tail}; "
        f"thinking: {thinking} "
        "(Deterministic mock answer for E2E.)"
    )


# ---------------------------------------------------------------------------
# Phase 96 (task 04, one-shot resilience): the incident-shape injection
# for the folder-summary one-shot path — see the module docstring
# ---------------------------------------------------------------------------

#: The folder-label triggers (phase 96, task 04): the ``FOLDER_SUMMARY_MODE``
#: branch's folder label (the user message's ``Folder: …`` tail — the
#: ``FOLDER_HEADER_PREFIX`` parse) ending with these suffixes is THIS
#: SUITE'S own fixture folder (``tests/e2e/test_oneshot_llm_retry.py``
#: seeds the names — no other E2E can hit them; they seed different
#: trees, and the triggers carry the E2E prefix).
ONESHOT_EMPTY_ONCE_SUFFIX = "/e2e_empty_once"
ONESHOT_EMPTY_ALWAYS_SUFFIX = "/e2e_empty_always"

#: Module-level per-label incident counter — the mock is single-
#: conversation per e2e server (the phase-67 ``_fail_posts``
#: convention). Counts the non-stream folder-summary POSTs served per
#: trigger label; the once-sequence resets after the success it guards
#: (the first normal reply), so a second sync re-drives the sequence
#: deterministically.
_empty_once_posts: dict[str, int] = {}


def _folder_summary_incident(body: dict[str, Any]) -> bool:
    """Should this NON-stream folder-summary POST answer with the
    2026-09-11 incident envelope (``content=""`` +
    ``finish_reason="length"`` — the exact wire shape of the empty
    ``lite`` reply phase 96's one-shot retry targets)?

    * the system prompt lacks ``FOLDER_SUMMARY_MODE`` → never (only the
      folder-summary one-shot carries the marker — ``chat()`` is the
      mock's only non-streaming consumer of it, so the counter counts
      exactly the one-shot POSTs the app's retry policy drives);
    * the label ends with ``/e2e_empty_always`` → EVERY POST (the
      exhaustion path — 1 + ``BOR_LLM_RETRIES`` empty attempts, the
      per-folder fail-soft leaves the row absent);
    * the label ends with ``/e2e_empty_once`` → the FIRST non-stream
      POST for that label only — every later POST returns the normal
      line (the retry recovers the row), and the counter resets on
      that first normal reply (the phase-67 pattern).

    The folder label is the ``FOLDER_SUMMARY_MODE`` branch's existing
    parse (the user message's first line, the ``FOLDER_HEADER_PREFIX``
    tail) — the injection is a pure function of the request plus the
    per-label counter (the house marker-flow convention).
    """
    if "FOLDER_SUMMARY_MODE" not in _system(body):
        return False
    user = _user(body)
    header = user.splitlines()[0] if user else ""
    if not header.startswith(FOLDER_HEADER_PREFIX):
        return False
    label = header.removeprefix(FOLDER_HEADER_PREFIX).strip()
    if label.endswith(ONESHOT_EMPTY_ALWAYS_SUFFIX):
        return True
    if label.endswith(ONESHOT_EMPTY_ONCE_SUFFIX):
        n = _empty_once_posts.get(label, 0) + 1
        _empty_once_posts[label] = n
        if n == 1:
            return True  # the first POST: the incident envelope
        _empty_once_posts[label] = 0  # the retry went out — restart
    return False


def compose_answer(body: dict[str, Any]) -> str:
    system = _system(body)
    user = _user(body)
    if LONG_ANSWER_TRIGGER in user.lower():
        answer = long_answer()
    elif "FOLDER_SUMMARY_MODE" in system:
        # Folder summary (phase 94, TODO.md L4): the ``lite`` stand-in
        # returns the deterministic one-liner
        # ``Fixture folder summary for <folder>.`` — <folder> is the
        # user message's ``Folder: …`` header line (the generator puts
        # the canonical source/folder identity there, imported as
        # ``FOLDER_HEADER_PREFIX`` so the mock can never drift from it).
        # The stored row therefore always names its folder — the drill-
        # down E2E (``test_ls_tree_drilldown.py``) asserts on it.
        # Checked BEFORE the ``SUMMARY_MODE`` branch: the folder marker
        # CONTAINS the summary marker as a substring (``FOLDER_`` +
        # ``SUMMARY_MODE``), so the summary branch would otherwise
        # shadow every folder-summary call. Checked BEFORE the
        # DEFLECT_MODE branch, like the other lite-mode markers (a
        # deflection prompt never carries one).
        # Phase 96 (task 04): the incident-shape injection keys on THIS
        # branch's label (``_folder_summary_incident`` — the module
        # docstring) and overrides the NON-STREAM response envelope in
        # ``chat_completions`` (``content=""`` +
        # ``finish_reason="length"``); the line below is what the later
        # / normal POSTs return.
        header = user.splitlines()[0] if user else ""
        folder = (
            header.removeprefix(FOLDER_HEADER_PREFIX).strip()
            if header.startswith(FOLDER_HEADER_PREFIX)
            else "(unnamed folder)"
        )
        answer = f"Fixture folder summary for {folder}."
    elif "SUMMARY_MODE" in system:
        # Document summaries (phase 30): the ``lite`` stand-in returns a
        # deterministic digest — the first 24 tokens of the user message
        # (the summarizer puts the capped document content there). Byte-
        # stable for a given fixture, so the summary chunk's retrieval
        # rank is a pure function of the fixture text. Checked BEFORE the
        # DEFLECT_MODE branch (task 06) so a deflection prompt that ever
        # carries the marker cannot shadow the summary call.
        answer = (
            f"This document covers "
            f"{' '.join(TOKEN_RE.findall(user.lower())[:24])}."
        )
    elif "KB_OVERVIEW_MODE" in system:
        # KB overview (phase 31): the ``lite`` stand-in returns the
        # deterministic outline — the first 8 tokens of the user message
        # (the generator puts the document list there). Byte-stable for a
        # given KB, so the stored row is a pure function of the fixture.
        # Checked BEFORE the DEFLECT_MODE branch, like SUMMARY_MODE, so a
        # prompt that ever carries both markers cannot shadow the
        # overview call.
        answer = "Knowledge base outline:\n- " + " ".join(
            TOKEN_RE.findall(user.lower())[:8]
        )
    elif TABLE_TRIGGER in user.lower():
        # Markdown tables (phase 44, TODO.md L6): the story E2E's
        # deterministic table answer — a 3-column table, the
        # <img onerror> XSS probe line (it must survive the mock
        # byte-for-byte so the E2E can prove the renderer neutralizes
        # it), and a wide 5-column table (guarantees scrollWidth >
        # clientWidth inside the 72rem container column, phase 100).
        # Byte-stable. Checked
        # BEFORE the DEFLECT_MODE branch: a deflection prompt never
        # carries the marker (it lives in the user message, same
        # reasoning as SUMMARY_MODE), so a marker question always gets
        # the table answer, whatever the gate says; the E2E asks it
        # against an on-topic fixture, where the gate is HIGH, and
        # asserts non-deflection as part of the table test.
        answer = TABLE_ANSWER
    elif HISTORY_TRIGGER in user.lower():
        # Phase 74 (TODO L4, chat history): the deterministic history
        # echo — proves on the wire that the client's prior turns (and
        # the prior thinking, as ``reasoning_content`` on the assistant
        # messages) reached the model. Checked BEFORE the DEFLECT_MODE
        # branch, like TABLE_TRIGGER: the marker lives in the user
        # message, a deflection prompt never carries it, so a marker
        # question always gets the echo whatever the gate says.
        answer = _history_echo(body)
    elif "DEFLECT_MODE" in system:
        answer = (
            "Ah — I haven't done anything like that, so I don't want to make stuff up! "
            "You're thinking bigger than my notes for a second. Try asking about "
            "kubernetes, backups, or deploying a new service — I know those inside out. "
            "You've got this!"
        )
    elif END_OF_NOTES_TRIGGER in user.lower():
        # Whole-document-context story (phase 24): echo the tail of the
        # document context. Byte-stable across runs — a sentinel on the
        # document's last line appears in the answer iff the whole
        # document was in the prompt. (The tail includes the closing
        # </documents> — harmless for the E2E sentinel assertions.)
        # Phase 37: the HIGH prompt now ends with the <tools> section
        # after </documents>, so the echo targets the <documents> block
        # itself — the sentinel semantics are unchanged.
        block = _DOCUMENTS_BLOCK_RE.search(_system(body))
        tail_source = block.group(0) if block else _context(body)
        answer = (
            f"…and the very end of my notes reads: “{tail_source[-160:]}” "
            "(Deterministic mock answer for E2E.)"
        )
    else:
        ctx = _context(body)
        snippet = ctx[:220].replace("\n", " ").strip()
        answer = (
            f"Great question — you've absolutely got this! Here's what my notes say about "
            f"“{user.strip()[:80]}”: {snippet}… That's the gist from the docs; happy to "
            "dig into any of it. (Deterministic mock answer for E2E.)"
        )
    # Steering (phase 15): when the system prompt carries <tuning>, the
    # answer ends with the first note — deterministically observable.
    note = first_tuning_note(system)
    if note:
        answer = f"{answer} (tuning: {note})"
    # KB overview (phase 31): when the system prompt carries
    # <knowledge_base>, the answer ends with the first outline bullet —
    # mirrors the steering echo exactly (appended after it, so the kb
    # suffix is the last thing rendered).
    bullet = first_kb_bullet(system)
    if bullet:
        answer = f"{answer} (kb: {bullet})"
    return answer


def compose_thinking(body: dict[str, Any]) -> str:
    """Deterministic reasoning scratchpad (thinking-display story, phase 17;
    lengthened in phase 21).

    A fixed "Step 1… Step 4" template interleaved with a "Scratch" deep-dive
    block, quoting the first ~60 chars of the user question: unique per
    question, byte-stable across runs, ~2 700 chars total (≈ 230 frames at
    the mock's 12-char/0.02s pacing). The length is deliberate (phase 21,
    thinking-no-scroll story): rendered in the 320px ``.thinking-text``
    window it overflows by ~2x, so the live-tail clip and the no-user-scroll
    contract are observable in E2E. The ``Step 2: Check my notes`` line
    fragment (phase 17) and the ``nothing is invented`` tail (phase 20's
    THINKING_TAIL) are what the E2E assertions key off — both are preserved.
    """
    q = _user(body).strip()[:60]
    return _thinking_template(q)


def _thinking_template(q: str) -> str:
    """The fixed Step/Scratch scratchpad (``compose_thinking`` and its
    paragraph variant share the exact same text — only the line
    separators differ)."""
    return (
        f"Step 1: Read the question carefully — “{q}” — and figure out what kind of "
        "answer it wants (a how-to, a lookup, or a design decision) before touching "
        "the docs, so I don't over- or under-answer.\n"
        "Step 2: Check my notes for the closest match. The homelab kubernetes file "
        "is the obvious candidate, but I should also consider whether a deployments "
        "note covers the same ground better.\n"
        "Scratch 1: the kubernetes file is organized by component — control plane, "
        "worker nodes, ingress, storage — so I can map each part of the question to "
        "a section instead of summarizing the whole file at once, and keep the "
        "answer anchored to the structure the notes actually use.\n"
        "Scratch 2: I should check whether the deployments note duplicates any of "
        "that ground; if it does, I will prefer the homelab file because the "
        "question is phrased around the cluster itself, and I will say which file "
        "each fact came from so the citation is honest.\n"
        "Scratch 3: versions and ports are the facts most likely to be stale in my "
        "memory — the etcd backup schedule, the ingress controller port, the "
        "registry mirror address — so I will re-read those lines verbatim before "
        "writing a single one of them into the answer.\n"
        "Scratch 4: if the answer needs a sequence, for example how a node joins the "
        "cluster or how the load balancer fronts the control plane, I will keep the "
        "order exactly as the notes write it rather than re-deriving it from general "
        "kubernetes knowledge that may not match this setup.\n"
        "Scratch 5: anything I cannot find in the notes — a host I do not recognize, "
        "a version I am not sure about, a schedule I cannot place — gets left out of "
        "the answer instead of guessed, because the honesty rule beats a longer "
        "answer every single time.\n"
        "Scratch 6: one more pass over the question wording to make sure I am "
        "answering the cluster setup, not some other homelab topic that shares the "
        "same vocabulary, and I will stay on the specific the question asked about.\n"
        "Scratch 7: I will also verify that the file describes the current setup — "
        "if the notes mention a migration from an older cluster, I should answer "
        "from the post-migration section and not mix in the old host names or the "
        "old port numbers that no longer apply.\n"
        "Scratch 8: final shape check before I commit — short paragraphs, a few "
        "bullets at most, the document path cited where the fact came from, and no "
        "invented facts anywhere in the draft.\n"
        "Step 3: Re-read the relevant sections top to bottom so every specific — "
        "hosts, versions, ports, schedules — is exact as written rather than "
        "remembered, and note which document each fact comes from.\n"
        "Step 4: Draft the answer around those specifics, keep it tight with short "
        "paragraphs and bullets where it helps, cite the documents by path, and "
        "double-check that nothing is invented."
    )


def compose_thinking_paragraphs(body: dict[str, Any]) -> str:
    """The phase-17 scratchpad with REAL paragraph breaks (\n\n, 2026-08-29
    regression pin): the same deterministic text as ``compose_thinking``,
    with a blank line inserted after scratchpad lines 2 and 6 (0-based) —
    two genuine \"2-newline gaps\" in the rendered scratchpad. Unique per
    question, byte-stable across runs (same length contract + 2 chars)."""
    base = compose_thinking(body)
    lines = base.split("\n")
    out: list[str] = []
    for i, line in enumerate(lines):
        out.append(line)
        if i in (2, 6):
            out.append("")  # blank line -> a real \"\n\n\" gap
    return "\n".join(out)


@app.post("/__shutdown__")
def shutdown() -> dict[str, Any]:
    """Test hook (loading-feedback story): terminate this mock process to
    simulate an LLM outage. The E2E fixture restores a fresh instance on
    the same port afterwards, so the rest of the session keeps working."""

    def _die() -> None:
        time.sleep(0.1)  # let the HTTP response flush before we exit
        os.kill(os.getpid(), signal.SIGTERM)

    threading.Thread(target=_die, daemon=True).start()
    return {"status": "shutting down"}


@app.get("/v1/models")
def models() -> dict[str, Any]:
    return {
        "object": "list",
        "data": [
            {"id": "turbo", "object": "model"},
            {"id": "embed", "object": "model"},
            {"id": "lite", "object": "model"},
        ],
    }


@app.post("/v1/embeddings")
def embeddings(body: dict[str, Any]) -> Any:  # dict, or a 500 (phase 67)
    raw = body.get("input")
    if isinstance(raw, str):
        raw = [raw]
    inputs: list[Any] = list(raw) if isinstance(raw, list) else []
    # Phase 67 (embedding retry): the first embeddings request whose
    # input carries the marker 500s; the next returns the normal
    # bag-of-words vector (see the module docstring). Raw httpx on the
    # client side — no SDK-level retries — so one POST per app attempt:
    # the counter is per POST here (unlike the chat counter below).
    joined = " ".join(str(t) for t in inputs if isinstance(t, str)).lower()
    if EMBED_FAIL_TRIGGER in joined:
        n = _bump_fail(EMBED_FAIL_TRIGGER)
        if n == 1:
            return _llm_500(EMBED_FAIL_TRIGGER)
        _fail_posts[EMBED_FAIL_TRIGGER] = 0  # the vector went out — restart
    data = [
        {"object": "embedding", "index": i, "embedding": embed_text(t)}
        for i, t in enumerate(inputs)
    ]
    return {
        "object": "list",
        "data": data,
        "model": body.get("model", "embed"),
        "usage": {"prompt_tokens": 8, "total_tokens": 8},
    }


def _sse_stream(
    answer: str,
    delay: float,
    thinking: str = "",
    pre_content_delay: float = 0.0,
    chunk: int = 12,
) -> Any:
    """SSE frames for one chat completion (phase 17: + reasoning).

    ``chunk`` (default 12) is the slice size for BOTH the thinking and
    the content frames — the ``think in paragraphs`` trigger raises it
    to ``THINK_PARAS_CHUNK`` (60) so a single frame renders past the
    32px think-window band (see ``THINK_PARAS_TRIGGER``). At 12 the
    output is byte-identical to the original.

    When ``thinking`` is non-empty its ``chunk``-sized slices go out FIRST as
    ``delta.reasoning_content`` frames — same 0.02s cadence and envelope
    as the content frames, the aipi wire convention (reasoning before
    content). Without ``thinking`` the output is byte-identical to the
    content-only stream, so the other story suites are unaffected.

    ``pre_content_delay`` (phase 20) inserts a silence gap between the end
    of the thinking stream and the first content frame — the client stays
    in its pre-token "thinking" state the whole time (0.02s cadence and
    frame shapes are unchanged, so 0.0 is byte-identical to before).
    """
    model = "turbo"
    chunk_id = f"chatcmpl-{uuid.uuid4()}"
    if delay:
        time.sleep(delay)
    for piece in re.findall(rf".{{1,{chunk}}}", thinking, re.S):
        payload = {
            "id": chunk_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": model,
            "choices": [
                {"index": 0, "delta": {"reasoning_content": piece}, "finish_reason": None}
            ],
        }
        yield f"data: {json_dumps(payload)}\n\n"
        time.sleep(0.02)
    if pre_content_delay:
        time.sleep(pre_content_delay)
    for piece in re.findall(rf".{{1,{chunk}}}", answer, re.S):
        payload = {
            "id": chunk_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": model,
            "choices": [{"index": 0, "delta": {"content": piece}, "finish_reason": None}],
        }
        yield f"data: {json_dumps(payload)}\n\n"
        time.sleep(0.02)
    yield (
        "data: "
        + json_dumps(
            {
                "id": chunk_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": model,
                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
            }
        )
        + "\n\n"
    )
    yield "data: [DONE]\n\n"


def json_dumps(obj: dict[str, Any]) -> str:
    import json

    return json.dumps(obj)


def _apply_max_tokens(answer: str, max_tokens: Any) -> str:
    """Deterministic stand-in for the endpoint's output cap: one token ≈
    one whitespace-separated word. Answers within the cap pass through
    byte-identical, so existing (short) answers are unaffected."""
    if not isinstance(max_tokens, int) or max_tokens <= 0:
        return answer
    words = answer.split()
    if len(words) <= max_tokens:
        return answer
    return " ".join(words[:max_tokens])


def _tool_call_stream(name: str, arguments: dict[str, Any], call_id: str) -> Any:
    """SSE frames for one tool-call-only chat completion (phase 37).

    The OpenAI wire convention the app accumulates (``app/rag/llm.py``):
    the first partial of index 0 carries ``id`` + ``type`` +
    ``function.name`` plus the first ``function.arguments`` fragment;
    the remaining fragments (deterministic 16-char split — so the
    multi-fragment accumulation path is exercised) arrive on later
    chunks; the final chunk carries ``finish_reason: "tool_calls"``.
    No ``content`` / ``reasoning_content`` frames — the turn asked for a
    tool instead of answering.

    Pacing: 0.1 s per frame — deliberately SLOWER than the content
    stream's 0.02 s, so the UI's transient "calling tool" state (held
    from the first ``tool`` frame until the first answer ``delta``) is a
    comfortable observation window for the story E2E (~1 s across the
    two tool requests).
    """
    model = "turbo"
    chunk_id = f"chatcmpl-{uuid.uuid4()}"
    raw_args = json_dumps(arguments) if arguments else "{}"
    frags = [raw_args[i : i + 16] for i in range(0, len(raw_args), 16)] or ["{}"]
    for i, frag in enumerate(frags):
        tc: dict[str, Any] = {"index": 0, "function": {"arguments": frag}}
        delta: dict[str, Any] = {"tool_calls": [tc]}
        if i == 0:
            tc = {
                "index": 0,
                "id": call_id,
                "type": "function",
                "function": {"name": name, "arguments": frag},
            }
            delta = {"role": "assistant", "tool_calls": [tc]}
        payload = {
            "id": chunk_id,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": model,
            "choices": [{"index": 0, "delta": delta, "finish_reason": None}],
        }
        yield f"data: {json_dumps(payload)}\n\n"
        time.sleep(0.1)
    yield (
        "data: "
        + json_dumps(
            {
                "id": chunk_id,
                "object": "chat.completion.chunk",
                "created": int(time.time()),
                "model": model,
                "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
            }
        )
        + "\n\n"
    )
    yield "data: [DONE]\n\n"


@app.post("/v1/chat/completions")
def chat_completions(body: dict[str, Any]) -> Any:
    user_lower = _user(body).lower()
    # Phase 37 (agent document tools): the deterministic marker flow.
    # The app's chat path is the only streaming consumer of this mock, so
    # the flow handles streaming requests; a non-streaming marker request
    # (never issued by the app) falls through to the regular answer.
    if body.get("stream"):
        # Phase 67 (LLM retry): deterministic failure injection — see
        # the module docstring. Checked before the marker tool flow: the
        # injection markers never combine with the tool-flow markers in
        # any suite, and a dead endpoint answers nothing (no flow).
        if ALWAYS_FAIL_TRIGGER in user_lower:
            return _llm_500(ALWAYS_FAIL_TRIGGER)
        if RETRY_TRIGGER in user_lower:
            if _chat_dead(RETRY_TRIGGER, RETRY_DEAD_ATTEMPTS):
                return _llm_500(RETRY_TRIGGER)
            _fail_posts[RETRY_TRIGGER] = 0  # the answer streamed — restart
        # Phase 71 (tool-scaffolding guardrails): the deterministic raw-
        # markup flow — checked BEFORE the search/tool marker flows (the
        # trigger is independent of the ``<tools>`` marker, so both
        # grounded and deflected turns hit it; SCAFFOLD_ALWAYS_TRIGGER
        # is checked first inside the classifier — the more specific
        # phrase wins, same convention as THINK_PARAS_TRIGGER).
        scaffold_flow = _scaffold_flow(body)
        if scaffold_flow is not None:
            # Request 1 (or EVERY request for the ALWAYS trigger): the
            # incident span as plain delta.content, 12-char chunks
            # (the span always spans ≥2 chunks — the filter's boundary
            # path), finish_reason "stop", no tool_calls, no reasoning.
            # Request 2 of the recovery trigger: the clean answer.
            answer = (
                SCAFFOLD_RECOVERY_ANSWER
                if scaffold_flow == "recovery"
                else SCAFFOLD_SPAN
            )
            return StreamingResponse(
                _sse_stream(answer, 0.0),
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        # 2026-09-05 (the "Qwen 3.8" incident — the grep regex prior):
        # the deterministic GREP-TEACH self-correction flow — checked
        # BEFORE the SEARCH / TOOLS_TRIGGER flows (disjoint trigger
        # phrases — the phase-72 ordering convention; the trigger needs
        # the ``<tools>`` section, so deflected turns never hit it).
        grep_teach = _grep_teach_flow(body)
        if grep_teach is not None:
            if grep_teach[0] == "regex":
                # The incident's misuse, deterministic: the regex-shaped
                # pattern (it can never match a fixed-substring grep).
                stream = _tool_call_stream(
                    "grep", {"pattern": GREP_TEACH_PATTERN}, "call_0"
                )
            elif grep_teach[0] == "plain":
                # The one-round correction: the plain-form retry (the
                # teaching line handed over exactly this hint).
                stream = _tool_call_stream(
                    "grep", {"pattern": GREP_TEACH_PLAIN}, "call_1"
                )
            elif grep_teach[0] == "read":
                stream = _tool_call_stream(
                    "read", {"path": grep_teach[1]}, grep_teach[2]
                )
            elif grep_teach[0] == "nomatch":
                stream = _sse_stream(
                    _apply_max_tokens(
                        "No matches — the knowledge base has no such text.",
                        body.get("max_tokens"),
                    ),
                    0.0,
                )
            else:  # "answer" — quote the read document (first 80 chars)
                stream = _sse_stream(
                    _apply_max_tokens(
                        f"Read {grep_teach[1]}. {grep_teach[2][:80]}",
                        body.get("max_tokens"),
                    ),
                    0.0,
                )
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        # Phase 68 (search tool): the deterministic search marker flow —
        # checked BEFORE the phase-37 tool flow (the more specific
        # trigger phrase wins, same convention as THINK_PARAS_TRIGGER).
        search_flow = _search_flow(body)
        if search_flow is not None:
            if search_flow[0] == "search":
                stream = _tool_call_stream(
                    "grep", {"pattern": SEARCH_PATTERN}, "call_0"
                )
            else:  # "found" — quote the first matched line (80 chars)
                answer = _apply_max_tokens(
                    f"Found {search_flow[1][:80]}", body.get("max_tokens")
                )
                stream = _sse_stream(answer, 0.0)
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        # Phase 72 (teaching refusals): the deterministic LS-TEACH
        # self-correction flow — checked BEFORE the plain
        # TOOLS_TRIGGER flow (disjoint trigger phrases — the phase-71
        # ordering convention; the trigger needs the ``<tools>``
        # section, so deflected turns never hit it).
        ls_teach = _ls_teach_flow(body)
        if ls_teach is not None:
            if ls_teach[0] == "misuse":
                # The incident's misuse, deterministic: ls(path='.').
                stream = _tool_call_stream("ls", {"path": "."}, "call_0")
            elif ls_teach[0] == "correct":
                # The one-round correction: the no-arg top-level
                # listing (phase 94: sources only — the drill follows).
                stream = _tool_call_stream("ls", {}, "call_1")
            elif ls_teach[0] == "drill":
                # Phase 94: the top level lists sources only — drill
                # one level into the first source for the file lines.
                stream = _tool_call_stream(
                    "ls", {"path": ls_teach[1]}, ls_teach[2]
                )
            else:  # "answer" — quote the first file line
                answer = _apply_max_tokens(
                    f"These are the indexed documents: {ls_teach[1]}",
                    body.get("max_tokens"),
                )
                stream = _sse_stream(answer, 0.0)
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        # Phase 94 (task 04): the deterministic SCRIPTED drill-down
        # turns (sources → folders → files → the grounded read; the
        # 50-line cap; the NOT-A-FOLDER teaching + scripted recovery) —
        # checked BEFORE the plain TOOLS_TRIGGER flow (disjoint trigger
        # phrases — the phase-71/72 ordering convention; the trigger
        # needs the ``<tools>`` section, so deflected turns never hit
        # it).
        drill = _drill_flow(body)
        if drill is not None:
            if drill[0] == "call":
                # The scripted call: ls with no path at the top level
                # (empty target), ls/read with the parsed target.
                stream = _tool_call_stream(
                    drill[1], {"path": drill[2]} if drill[2] else {}, drill[3]
                )
            elif drill[0] == "recover":
                # The scripted one-round recovery after the NOT-A-FOLDER
                # teaching: ls the target's source (the parent level).
                stream = _tool_call_stream("ls", {"path": drill[1]}, "call_1")
            elif drill[0] == "read_answer":
                # Quote the read document (first 80 chars) — the
                # phase-37 single-read shape.
                stream = _sse_stream(
                    _apply_max_tokens(
                        f"Read {drill[1]}. {drill[2]}", body.get("max_tokens")
                    ),
                    0.0,
                )
            elif drill[0] == "ctx_answer":
                # The dedupe fired: the target's full text is in the
                # <documents> prompt — answer from it (same citation
                # shape, prefixed so the suite pins the dedupe path).
                stream = _sse_stream(
                    _apply_max_tokens(
                        f"Already in context: Read {drill[1]}. {drill[2]}",
                        body.get("max_tokens"),
                    ),
                    0.0,
                )
            else:  # "echo" — the listing verbatim (the suite's lens)
                stream = _sse_stream(
                    _apply_max_tokens(
                        f"Here's the level I listed:\n{drill[1]}",
                        body.get("max_tokens"),
                    ),
                    0.0,
                )
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        # Phase 95 (task 03): the deterministic SCRIPTED capped read
        # (the question carries its own call — ``read the capped
        # document: read source/path``): the scripted ``read``, then the
        # answer that ECHOES the whole tool result (the marker + notice
        # when the cap fired, the plain shape when it did not — the
        # suite's lens on the LLM's context). Checked BEFORE the plain
        # TOOLS_TRIGGER flow (disjoint trigger phrases — the
        # phase-71/72/94 ordering convention; the trigger needs the
        # ``<tools>`` section, so deflected turns never hit it).
        read_cap = _read_cap_flow(body)
        if read_cap is not None:
            if read_cap[0] == "call":
                stream = _tool_call_stream(
                    "read", {"path": read_cap[1]}, read_cap[2]
                )
            else:  # "echo" — the last tool result verbatim (the lens)
                stream = _sse_stream(
                    _apply_max_tokens(
                        f"Here's what the read returned:\n{read_cap[1]}",
                        body.get("max_tokens"),
                    ),
                    0.0,
                )
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )
        flow = _tool_flow(body)
        if flow is not None:
            if flow[0] == "list":
                stream = _tool_call_stream("ls", {}, "call_0")
            elif flow[0] == "drill":
                # Phase 94: the drill step — ls scoped to the first
                # source of the top-level listing (flow[1], flow[2] is
                # the synthetic call id).
                stream = _tool_call_stream("ls", {"path": flow[1]}, flow[2])
            elif flow[0] == "read":
                # flow[3] is the synthetic call id — "call_1" for the
                # single-read flow and the multi-read first read,
                # "call_2" for the multi-read second read (phase 45,
                # task 02). Phase 70: the harness-aligned shape — one
                # combined ``source/path`` argument (the mock joins the
                # two catalog fields; the catalog format is unchanged).
                stream = _tool_call_stream(
                    "read", {"path": f"{flow[1]}/{flow[2]}"}, flow[3]
                )
            elif flow[0] == "multi_answer":
                # Phase 45 (task 02): the multi-read forced answer —
                # computed in _tool_flow, byte-stable.
                stream = _sse_stream(_apply_max_tokens(flow[2], body.get("max_tokens")), 0.0)
            else:  # "answer" — quote the read document (first 80 chars)
                answer = _apply_max_tokens(
                    f"Read {flow[1]}. {flow[2][:80]}", body.get("max_tokens")
                )
                stream = _sse_stream(answer, 0.0)
            return StreamingResponse(
                stream,
                media_type="text/event-stream",
                headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
            )

    answer = _apply_max_tokens(compose_answer(body), body.get("max_tokens"))
    delay = 3.0 if "pretend to think slowly" in _user(body) else 0.0
    # ``think in paragraphs`` wins over ``think out loud`` (more specific):
    # the same scratchpad WITH real "\n\n" paragraph breaks, at 60-char
    # frames (real-model-sized deltas — the 32px-band regression pin).
    if THINK_PARAS_TRIGGER in user_lower:
        thinking = compose_thinking_paragraphs(body)
        chunk = THINK_PARAS_CHUNK
    elif THINKING_TRIGGER in user_lower:
        thinking = compose_thinking(body)
        chunk = 12
    else:
        thinking = ""
        chunk = 12
    pre_content = (
        PRE_CONTENT_PAUSE_S if SLOW_PRETOKEN_TRIGGER in user_lower else 0.0
    )

    if not body.get("stream"):
        # Phase 96 (task 04): the incident-shape injection (the module
        # docstring) — the trigger-labelled folder summary answers the
        # exact 2026-09-11 envelope: the mock's normal OpenAI
        # chat-completion shape with ``content=""`` and
        # ``finish_reason="length"`` (the budget spent in
        # ``reasoning_content``). ``chat()`` is the mock's only
        # non-streaming folder-summary consumer, so this is the one-shot
        # retry path (``app.rag.llm.LLMClient.chat``, phase 96 task 01)
        # and nothing else — every other non-stream response is
        # byte-identical to pre-phase-96.
        if _folder_summary_incident(body):
            return {
                "id": f"chatcmpl-{uuid.uuid4()}",
                "object": "chat.completion",
                "created": int(time.time()),
                "model": body.get("model", "turbo"),
                "choices": [
                    {
                        "index": 0,
                        "message": {"role": "assistant", "content": ""},
                        "finish_reason": "length",
                    }
                ],
                "usage": {
                    "prompt_tokens": 100,
                    "completion_tokens": 2048,
                    "total_tokens": 2148,
                },
            }
        message: dict[str, Any] = {"role": "assistant", "content": answer}
        if thinking:
            # Harmless future-proofing: the app only uses streaming, but a
            # non-streaming client that reads the field gets the reasoning.
            message["reasoning_content"] = thinking
        return {
            "id": f"chatcmpl-{uuid.uuid4()}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": body.get("model", "turbo"),
            "choices": [
                {"index": 0, "message": message, "finish_reason": "stop"}
            ],
            "usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150},
        }

    return StreamingResponse(
        _sse_stream(
            answer,
            delay,
            thinking=thinking,
            pre_content_delay=pre_content,
            chunk=chunk,
        ),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )