webapp/backend.py

"""Pure, typed backend for the Agenda Parser.

All the PDF parsing, packet slicing, and the chunk -> ChromaDB -> Gemma-4
summarizer live here as plain functions with full type hints. The type hints
matter: :func:`gradio.api` derives each endpoint's JSON schema from them (see
``server.py``), and the React frontend consumes that schema.

The app takes an **uploaded agenda-packet PDF** (no external API): the user uploads
the packet and marks the agenda (table-of-contents) page range; we parse it into
agenda items mapped to their backup-packet page ranges (bookmarks first, text
fallback), then summarize / report / answer / run the agent against that packet.

Nothing in this module touches Gradio UI components -- it is the API surface,
not the presentation.
"""

from __future__ import annotations

import os
import re
import secrets
import tempfile
import threading
from collections import OrderedDict
from pathlib import Path
from typing import Iterator, TypedDict

from chroma import (
    AgendaStore,
    chunk_text,
    generate_report,
    summarize_text,
)
from chroma.llm import chat_complete, load_llm_config
from pdf_utils import extract_pdf_outline

REPO = Path(__file__).resolve().parent.parent

# Let the LLM client pick up the deployed key even if pi_models.json isn't found.
_key_file = REPO / "model" / ".gemma4_api_key"
if _key_file.exists() and not os.getenv("GEMMA_API_KEY"):
    os.environ["GEMMA_API_KEY"] = _key_file.read_text().strip()

# One persistent Chroma store + one resolved LLM config, reused across requests.
STORE = AgendaStore(str(REPO / ".chroma"))
LLM_CFG = load_llm_config()

# Which LLM path to use:
#   "remote" (default) -> the OpenAI-compatible endpoint in LLM_CFG (Modal vLLM).
#   "local"            -> an in-process llama.cpp GGUF run inside @spaces.GPU
#                         (webapp/local_llm.py); set this on the ZeroGPU Space.
LLM_BACKEND = os.getenv("LLM_BACKEND", "remote").strip().lower()


def _model_label(model: str = "") -> str:
    """Human label for the model that produced an output, for the 'done' frame."""
    if LLM_BACKEND == "local":
        from webapp import local_llm

        return local_llm.model_label(model)
    return str(LLM_CFG["model"])


def available_models() -> list[str]:
    """Which model-picker keys are usable here. On the Space (local backend) these are
    the in-process GGUFs whose file can be downloaded (the private 26B/full need
    HF_TOKEN). All three run locally — no remote endpoint."""
    if LLM_BACKEND == "local":
        from webapp import local_llm

        keys = local_llm.available_models()
        return keys or ["e4b"]
    return ["e4b", "26b", "full"]


def _complete(prompt: str, system: str | None = None) -> str:
    """LLM completer with a generous timeout (the Modal GPU can cold-start)."""
    from openai import OpenAI

    client = OpenAI(
        base_url=LLM_CFG["base_url"], api_key=LLM_CFG["api_key"],
        timeout=900.0, max_retries=1,
    )
    return chat_complete(prompt, system=system, client=client, model=LLM_CFG["model"])


# --------------------------------------------------------------------------- #
# Bundled sample agenda packets — a one-click "try it" alternative to uploading.
# The PDFs ship in samples/ (served from disk, parsed exactly like an upload), so
# there's no client-side CORS / external fetch at runtime.
# --------------------------------------------------------------------------- #
_SAMPLES_DIR = REPO / "samples"
SAMPLES: list[dict] = [
    {
        "id": "oakland-1570",
        "label": "Oakland County — Full Board",
        "description": "773-page board packet · 50 items, bookmark-mapped",
        "file": "oakland-1570.pdf",
        "agenda_pages": "1-3",
        "source_url": "https://oaklandcomi.portal.civicclerk.com/event/1570/files/agenda/13735",
    },
    {
        "id": "baycity-928",
        "label": "Bay City — City Commission",
        "description": "228-page packet · 20 items incl. a budget presentation",
        "file": "baycity-928.pdf",
        "agenda_pages": "1-3",
        "source_url": "https://baycitymi.portal.civicclerk.com/event/928/files/agenda/3731",
    },
    {
        "id": "ionia-2026-02",
        "label": "Ionia — City Council",
        "description": "144-page council packet · 30 items, bookmark-mapped",
        "file": "ionia-2026-02.pdf",
        "agenda_pages": "1-3",
        "source_url": "",
    },
]
_SAMPLE_BY_ID = {s["id"]: s for s in SAMPLES}
_SAMPLE_UID_PREFIX = "sample-"


def _sample_uid(sample_id: str) -> str:
    """Stable upload_id for a bundled sample (so its PDF is always re-derivable)."""
    return f"{_SAMPLE_UID_PREFIX}{sample_id}"


def list_samples() -> list[dict]:
    """The bundled sample agendas available to try (only those present on disk)."""
    return [
        {"id": s["id"], "label": s["label"], "description": s["description"],
         "agenda_pages": s["agenda_pages"], "source_url": s["source_url"]}
        for s in SAMPLES
        if (_SAMPLES_DIR / s["file"]).exists()
    ]


def ingest_sample(sample_id: str) -> dict:
    """Parse a bundled sample agenda packet (same return shape as upload_packet).

    Truncated bookmark titles are restored from the agenda text at parse time (see
    :func:`_untruncate_item_names`), so samples show full titles like any upload.
    """
    sid = (sample_id or "").strip()
    s = _SAMPLE_BY_ID.get(sid)
    if s is None:
        return {"error": f"Unknown sample '{sample_id}'."}
    path = _SAMPLES_DIR / s["file"]
    if not path.exists():
        return {"error": f"Sample '{sample_id}' is not bundled on this server."}
    # Deterministic upload_id (vs a random one) so the cached PDF is re-derivable from
    # the bundle: cached_packet rehydrates a sample on a miss (Space restart / LRU
    # eviction / a saved sample from a prior session), so sample downloads & page views
    # never 404 the way an expired user upload would.
    return upload_packet(path.read_bytes(), s["file"], s["agenda_pages"],
                         upload_id=_sample_uid(sid))


# --------------------------------------------------------------------------- #
# Typed shapes (gradio.api derives each endpoint's JSON schema from these)
# --------------------------------------------------------------------------- #
class AgendaItem(TypedDict):
    id: str
    number: str          # leading outline enumerator, e.g. "11." / "a." / ""
    name: str            # item title with the enumerator stripped
    summary: str
    is_section: bool     # a grouping header (no backup pages of its own)
    level: int           # nesting depth (0 = top-level item, 1 = sub-item)
    status: str
    attachments: int
    has_pages: bool      # whether this item has backup pages in the packet
    pages: str           # "24-35" 1-indexed human range, "" when none
    start: int           # 0-indexed start page (inclusive), 0 when none
    end: int             # 0-indexed end page (exclusive), 0 when none


class AgendaOutline(TypedDict):
    items: list[AgendaItem]
    source: str  # "outline" | "text" | "none"
    confidence: str  # "good" | "poor" | "empty"
    message: str
    model: str
    title: str  # suggested agenda name (from the agenda header), "" if none


class ItemPage(TypedDict):
    page: int  # 1-indexed page number in the packet
    image: str  # "data:image/jpeg;base64,…" — renderable directly in an <img>


class ItemPages(TypedDict):
    stage: str  # "working" | "done" | "error" — streamed so the browser keeps the
                # connection alive while the packet renders
    sliced: bool  # whether the item's pages were confidently located
    pages: str  # "26-48" page range, or "" when not sliced
    note: str  # human explanation (range, or why it couldn't be isolated)
    code: str  # "" normally; "no_packet" when the upload is gone (re-upload)
    images: list[ItemPage]


class AgentMessage(TypedDict):
    role: str  # "user" | "assistant"
    content: str


class AgentFrame(TypedDict):
    # One streamed step of an Agent Mode turn. Stable shape (every key always present)
    # so gradio.api derives a clean schema and the React client can switch on `stage`.
    stage: str  # "thinking" | "tool_call" | "tool_result" | "notice" | "answer" | "error"
    text: str   # thinking / notice / answer / error message (empty for tool_* frames)
    tool: str   # tool name (tool_call / tool_result)
    args: dict  # tool_call arguments
    result: str  # tool_result compact JSON (empty otherwise)
    summary: str  # tool_result one-line human gist (empty otherwise)
    step: int   # 1-based step index (0 for answer/error)
    model: str  # model label, set on the final answer frame


# --------------------------------------------------------------------------- #
# Agenda outline from the PDF text (fallback when the packet has no bookmarks).
# Numbered lines ("10. ...") become top-level items; lettered lines ("a. ...")
# nest one level under the preceding numbered item — that is where the substance
# lives. Recognized headers ("Consent Agenda", ...) become sections.
# --------------------------------------------------------------------------- #
_TAG_RE = re.compile(r"<[^>]+>")
_OUTLINE_START_RE = re.compile(r"(?i)^a\s*g\s*e\s*n\s*d\s*a\s*:?\s*$")
_OUTLINE_FOOTER_RE = re.compile(
    r"(?i)^(public access information|if you require special accommodation|"
    r"view meeting|updated agenda|watch |how to (view|participate|join)|"
    r"persons with disabilities|americans with disabilities act|"
    r"closed captioning|to (view|watch|join|participate)|"
    r"in accordance with|note:|the public may)\b"
)
_OUTLINE_NUM_RE = re.compile(r"^\s*(\d{1,3})[\.\)]\s+(.+)$")
_OUTLINE_ALPHA_RE = re.compile(r"^\s*([a-zA-Z]|[ivxIVX]{1,4})[\.\)]\s+(.+)$")
_OUTLINE_BULLET_RE = re.compile(r"^\s*[•\-\*]\s+")
# Per-item metadata lines (e.g. "Item Category: Appointment", "Presenter: ...").
_OUTLINE_META_RE = re.compile(
    r"(?i)^(item category|presenter|action required|action|fiscal impact|"
    r"department|sponsor|recommended action)\s*:\s*(.+)$"
)

_OUTLINE_SECTION_WORDS = {
    "consent agenda", "consent calendar", "regular agenda",
    "reports of standing committees", "reports of special committees",
    "presentations", "communications", "public comment", "old business",
    "new business", "unfinished business", "special order of business",
    "items for discussion", "action items", "information items",
}


def _looks_like_section(line: str) -> bool:
    """True if a non-numbered agenda line reads like a grouping header.

    Deliberately conservative: a known header phrase, or a short ALL-CAPS line.
    We do *not* treat arbitrary Title-Case lines as headers — in practice those
    are wrapped continuations of an item's title, and misreading them as sections
    shreds the outline (and orphans the sub-items that follow).
    """
    s = line.strip()
    if s.rstrip(":").lower() in _OUTLINE_SECTION_WORDS:
        return True
    if len(s) > 60 or ":" in s or s.endswith((".", "?", ",")):
        return False
    words = [w for w in re.split(r"\s+", s) if w]
    if not (1 <= len(words) <= 6) or not any(c.isalpha() for c in s):
        return False
    return s == s.upper()  # ALL-CAPS header only


def _parse_agenda_outline(text: str) -> list[AgendaItem]:
    """Heuristically turn an agenda PDF's text into an ordered item outline.

    Numbered lines (``10. ...``) become top-level items; lettered lines
    (``a. ...``) nest one level under the preceding numbered item — that is where
    the substance lives (a committee's individual resolutions, appointments,
    etc.). Recognized headers (``Consent Agenda``, ``Regular Agenda``, ...) become
    sections, and per-item metadata (``Item Category:`` -> status; ``Presenter:``,
    ``Action Required:`` -> summary) folds into the item it follows. Best-effort:
    a readable rendering of the published agenda, not a parser of record.
    """
    lines = text.splitlines()
    start = 0
    for i, ln in enumerate(lines):
        if _OUTLINE_START_RE.match(ln.strip()):
            start = i + 1
            break

    out: list[AgendaItem] = []
    last: AgendaItem | None = None
    n = 0

    def add(number: str, name: str, is_section: bool, level: int) -> AgendaItem:
        nonlocal n
        n += 1
        it = AgendaItem(
            id=f"pdf:{n}", number=number, name=name.strip(), summary="",
            is_section=is_section, level=level, status="", attachments=0,
            has_pages=False, pages="", start=0, end=0,
        )
        out.append(it)
        return it

    def add_summary(it: AgendaItem, piece: str) -> None:
        it["summary"] = f"{it['summary']} · {piece}" if it["summary"] else piece

    for raw in lines[start:]:
        s = raw.strip()
        if not s:
            continue
        if _OUTLINE_FOOTER_RE.match(s):
            break

        m = _OUTLINE_NUM_RE.match(raw)
        if m:
            last = add(m.group(1) + ".", m.group(2), False, 0)
            continue
        m = _OUTLINE_ALPHA_RE.match(raw)
        if m and last is not None and not last["is_section"]:
            base = last["level"] if last["number"].rstrip(".").isalpha() else last["level"] + 1
            last = add(m.group(1) + ".", m.group(2), False, base)
            continue
        # Per-item metadata — must precede section detection ("Item Category:
        # Appointment" would otherwise read as a Title-Case header).
        m = _OUTLINE_META_RE.match(s)
        if m and last is not None:
            key, val = m.group(1).strip().lower(), m.group(2).strip()
            if key == "item category" and not last["status"]:
                last["status"] = val
            else:
                add_summary(last, f"{m.group(1).strip().title()}: {val}")
            continue
        if _OUTLINE_BULLET_RE.match(s) and last is not None:
            add_summary(last, re.sub(r"^[•\-\*]\s+", "", s))
            continue
        if _looks_like_section(s):
            last = None
            add("", s.rstrip(":"), True, 0)
            continue
        # Otherwise: a wrapped continuation of the previous item's title.
        if last is not None and not last["is_section"]:
            last["name"] = (last["name"] + " " + s).strip()

    return out


# --------------------------------------------------------------------------- #
# Item -> packet page anchoring (shared by the report slicer and the page viewer).
#
# A title is only distinctive enough to anchor in the packet if it's reasonably
# long; short generic lines ("Roll Call", "Invocation") never carry their own
# packet section. When an item's slice has no downstream item to bound it, cap how
# far it can run so we don't sweep in the rest of a long packet.
# --------------------------------------------------------------------------- #
_ITEM_ANCHOR_MIN_LEN = 18
_ITEM_ANCHOR_KEY_LEN = 50
_ITEM_SLICE_MAX_PAGES = 25
# A page carrying this many *other* item titles (besides the one being anchored) is
# an agenda outline / table-of-contents page, not an item's content section, so it
# never anchors an item. A content page carries its item's title and few others.
_ITEM_OUTLINE_OTHER_TITLES = 2

# Leading outline enumerator on a title ("1.", "a.", "iv.", "V.") — the packet body
# repeats the item's wording but not its outline number, so we strip it for matching.
_ENUM_RE = re.compile(r"^\s*(?:\d{1,3}|[a-z]|[ivxlcdm]{1,6})[.)]\s+", re.IGNORECASE)


def _norm_ws(s: str) -> str:
    return re.sub(r"\s+", " ", (s or "")).strip().lower()


def _title_key(title: str) -> str:
    """Whitespace-normalized title with any leading outline enumerator stripped —
    the form used to find the item's wording in the packet body."""
    return _norm_ws(_ENUM_RE.sub("", (title or "").strip()))


def _split_enumerator(title: str) -> tuple[str, str]:
    """Split a bookmark/agenda title into ``(number, name)``.

    ``"11. Economic Development…"`` -> ``("11.", "Economic Development…")``;
    a title with no leading enumerator (a section header) -> ``("", title)``.
    """
    t = re.sub(r"\s+", " ", (title or "").strip())
    m = _ENUM_RE.match(t)
    if m:
        return t[: m.end()].strip(), t[m.end():].strip()
    return "", t


# --------------------------------------------------------------------------- #
# Outline-based anchoring: match each agenda item to the packet's PDF bookmark.
#
# Government agenda packets are compiled with a bookmark outline that mirrors the
# agenda — a bookmark per item (titled "<number> <name>") pointing at the page where
# that item's content begins, with the item's attachments/reports nested one level
# deeper. That is the packet author's own item→page map, so we prefer it over
# re-deriving page ranges from body text. We match an item to its bookmark on an
# alphanumeric-squeezed title prefix (robust to HTML tags, "<br>"-joined words, and
# outline-number/punctuation differences), scanning bookmarks in document order so
# repeated titles stay in sync. An item only anchors when its bookmark has a deeper
# child bookmark — i.e. the item actually carries supporting documents in the packet
# — which keeps procedural items (whose bookmark merely points at the agenda front
# matter) from claiming a section.
# --------------------------------------------------------------------------- #
_OUTLINE_MATCH_MIN_LEN = 6  # squeezed item title shorter than this is too generic
_OUTLINE_MATCH_PREFIX = 28  # compare this many squeezed leading chars


def _squeeze_title(title: str) -> str:
    """Lowercased alphanumeric-only form of a title (tags + enumerator stripped).

    Collapsing to ``[a-z0-9]`` makes the item title and the packet bookmark title
    line up despite HTML markup, ``<br>``-joined words, non-breaking/zero-width
    spaces, and outline-number/punctuation differences."""
    t = _TAG_RE.sub(" ", title or "")
    t = _ENUM_RE.sub("", t.strip())
    return re.sub(r"[^a-z0-9]+", "", t.lower())


def _outline_anchors(outline: list[dict], titles: list[str]) -> list[int | None]:
    """For each agenda item, the 0-indexed packet page from its bookmark (or None).

    ``outline`` is :func:`pdf_utils.extract_pdf_outline`'s flat bookmark list.
    Anchors only content-bearing items (whose bookmark has a deeper child) and never
    the front-matter page; everything else is ``None`` so the caller can fall back to
    text anchoring.
    """
    anchors: list[int | None] = [None] * len(titles)
    if not outline:
        return anchors
    # A bookmark is content-bearing iff the next bookmark nests under it (its own
    # attachment/report), so its page is where the item's documents actually start.
    has_child = [
        i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"]
        for i in range(len(outline))
    ]
    cand = [
        (outline[i]["page"], _squeeze_title(outline[i]["title"]), has_child[i])
        for i in range(len(outline))
        if outline[i]["page"] is not None
    ]
    pos = 0  # bookmarks are consumed in document order to disambiguate repeats
    for i, title in enumerate(titles):
        key = _squeeze_title(title)
        if len(key) < _OUTLINE_MATCH_MIN_LEN:
            continue
        for j in range(pos, len(cand)):
            page, bkey, child = cand[j]
            if not bkey:
                continue
            if bkey.startswith(key[:_OUTLINE_MATCH_PREFIX]) or key.startswith(
                bkey[:_OUTLINE_MATCH_PREFIX]
            ):
                pos = j + 1
                if child and page > 0:
                    anchors[i] = page
                break
    return anchors


def _slice_packet_for_item(
    pages: list[str], titles: list[str], index: int,
    outline: list[dict] | None = None,
) -> tuple[str, dict]:
    """Slice an agenda packet down to the selected item's section.

    The packet is one compiled PDF. We locate each item's section by **two anchoring
    strategies, outline first**:

    1. *Outline* — match the item to its bookmark in the packet's PDF outline (the
       packet author's own item→page map). Deterministic and exact when present.
    2. *Text* — fallback when an item has no bookmark: anchor the item's title to a
       packet *content* page by whitespace-normalized substring match, skipping
       outline/TOC pages (which list several item titles at once).

    The selected item's slice runs from its anchor page to the next later-anchored
    item's page (text anchors are capped; outline anchors aren't, since they're
    exact). When the item can't be confidently located by either strategy, we don't
    guess — the caller falls back to the whole packet.

    Returns ``(sliced_text, info)`` where ``info`` is ``{"sliced", "pages", "start",
    "end", "method"}`` (``start``/``end`` are 0-indexed, end-exclusive page bounds, or
    ``None`` when not sliced; ``method`` is ``"outline"``/``"text"``/``""``).
    ``sliced_text`` is empty when no confident anchor was found.
    """
    norm_pages = [_norm_ws(p) for p in pages]
    keys = [_title_key(t) for t in titles]
    probes = [k[:_ITEM_ANCHOR_KEY_LEN] for k in keys]
    out_anchors = _outline_anchors(outline or [], titles)

    def distinctive(k: str) -> bool:
        return len(k) >= _ITEM_ANCHOR_MIN_LEN

    def text_anchor(idx: int) -> int | None:
        """First content page whose text contains item ``idx``'s title and at most
        one *other* item title (i.e. not an outline/TOC page)."""
        if not (0 <= idx < len(keys)) or not distinctive(keys[idx]):
            return None
        probe = probes[idx]
        for pi, npg in enumerate(norm_pages):
            if probe not in npg:
                continue
            others = sum(1 for j, p in enumerate(probes)
                         if j != idx and distinctive(keys[j]) and p and p in npg)
            if others < _ITEM_OUTLINE_OTHER_TITLES:
                return pi
        return None

    def anchor(idx: int) -> int | None:
        """Item ``idx``'s packet start page — its bookmark if any, else its title."""
        if 0 <= idx < len(out_anchors) and out_anchors[idx] is not None:
            return out_anchors[idx]
        return text_anchor(idx)

    not_sliced = {"sliced": False, "pages": "", "start": None, "end": None,
                  "method": ""}
    start = anchor(index)
    if start is None:
        return "", dict(not_sliced)

    method = "outline" if (0 <= index < len(out_anchors)
                           and out_anchors[index] is not None) else "text"
    later = [a for a in (anchor(j) for j in range(index + 1, len(keys)))
             if a is not None and a > start]
    if later:
        end = min(later)
    elif method == "outline":
        end = len(pages)  # exact map — run to the packet end
    else:
        end = min(len(pages), start + _ITEM_SLICE_MAX_PAGES)
    text = "\n\n".join(p for p in pages[start:end] if p).strip()
    if not text:
        return "", dict(not_sliced)
    return text, {"sliced": True, "pages": f"{start + 1}-{end}",
                  "start": start, "end": end, "method": method}


def _outline_slice_for_item(
    outline: list[dict], titles: list[str], index: int, page_count: int,
) -> dict:
    """Outline-anchored slice info for one item — **without** per-page text.

    The page viewer renders by page *index*, so when the packet's PDF outline anchors
    the item we never need to extract the whole packet's text. We match the item to its
    bookmark exactly as :func:`_outline_anchors` does (squeezed-title prefix, in
    document order, only content-bearing bookmarks) so the viewer's start page agrees
    with the report slicer — then bound the section at the **next bookmark of the same
    or a higher level** (the item's next sibling or the next section), i.e. the
    bookmark's natural extent. That is the item's true page span; we fall back to the
    packet end only when no such bookmark follows. Returns the same ``info`` dict shape
    (``method`` always ``"outline"``), or ``{"sliced": False, ...}`` when the item has
    no content-bearing bookmark — the caller then falls back to the text-based slice.
    """
    not_sliced = {"sliced": False, "pages": "", "start": None, "end": None,
                  "method": ""}
    if page_count <= 0 or not outline or not (0 <= index < len(titles)):
        return dict(not_sliced)
    # Candidate bookmarks in document order, each carrying its level and whether it is
    # content-bearing (has a deeper child = the item's own attachments/reports).
    has_child = [
        i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"]
        for i in range(len(outline))
    ]
    cand = [
        (outline[i]["page"], outline[i]["level"], _squeeze_title(outline[i]["title"]),
         has_child[i])
        for i in range(len(outline))
        if outline[i]["page"] is not None
    ]

    pos = 0  # bookmarks consumed in document order to disambiguate repeated titles
    for ti, title in enumerate(titles):
        key = _squeeze_title(title)
        if len(key) < _OUTLINE_MATCH_MIN_LEN:
            if ti == index:
                return dict(not_sliced)
            continue
        match_j = None
        for j in range(pos, len(cand)):
            page, level, bkey, child = cand[j]
            if bkey and (bkey.startswith(key[:_OUTLINE_MATCH_PREFIX])
                         or key.startswith(bkey[:_OUTLINE_MATCH_PREFIX])):
                match_j, pos = j, j + 1
                break
        if ti != index:
            continue
        if match_j is None:
            return dict(not_sliced)
        page, level, _, child = cand[match_j]
        if not (child and page > 0):
            return dict(not_sliced)
        end = page_count
        for k in range(match_j + 1, len(cand)):
            npage, nlevel = cand[k][0], cand[k][1]
            if nlevel <= level and npage > page:
                end = npage
                break
        if end <= page:
            return dict(not_sliced)
        return {"sliced": True, "pages": f"{page + 1}-{end}",
                "start": page, "end": end, "method": "outline"}
    return dict(not_sliced)


# --------------------------------------------------------------------------- #
# Build the agenda item tree from a parsed packet — bookmarks first, text fallback.
# --------------------------------------------------------------------------- #
def _items_from_outline(outline: list[dict], page_count: int) -> list[AgendaItem]:
    """Build the agenda item tree from the packet's PDF bookmarks.

    Surfaces level-0 (top-level items + section headers) and level-1 (sub-items)
    bookmarks as agenda items; deeper bookmarks are an item's own attachments and are
    not surfaced. An item is content-bearing (``has_pages``) iff its bookmark has a
    deeper child AND points beyond the agenda front matter (page index > 0); its page
    span runs to the next bookmark of the same-or-higher level (its next sibling /
    section). 0-indexed ``start``/``end`` (end-exclusive) plus a 1-indexed ``pages``
    string, matching the slicer.
    """
    if not outline:
        return []
    has_child = [
        i + 1 < len(outline) and outline[i + 1]["level"] > outline[i]["level"]
        for i in range(len(outline))
    ]
    items: list[AgendaItem] = []
    n = 0
    for i, bm in enumerate(outline):
        level = int(bm.get("level", 0) or 0)
        if level > 1:  # attachments / reports — not surfaced as agenda items
            continue
        page = bm.get("page")
        number, name = _split_enumerator(bm.get("title", ""))
        if not name:
            continue
        content = bool(has_child[i] and page is not None and page > 0)
        start = end = 0
        pages_str = ""
        if content:
            sec_end = page_count
            for k in range(i + 1, len(outline)):
                npage, nlevel = outline[k].get("page"), int(outline[k].get("level", 0) or 0)
                if npage is not None and nlevel <= level and npage > page:
                    sec_end = npage
                    break
            if sec_end > page:
                start, end, pages_str = page, sec_end, f"{page + 1}-{sec_end}"
            else:
                content = False
        n += 1
        items.append(AgendaItem(
            id=f"outline:{n}",
            number=number,
            name=name,
            summary="",
            is_section=(level == 0 and not number),
            level=level,
            status="",
            attachments=0,
            has_pages=content,
            pages=pages_str,
            start=start,
            end=end,
        ))
    return items


def _items_from_text(pages: list[str], agenda_pages: str) -> list[AgendaItem]:
    """Text fallback when the packet has no bookmarks.

    Parses the user-marked agenda page range into an item list, then text-anchors each
    item into the later packet pages to fill ``start``/``end``/``has_pages``/``pages``.
    """
    if not pages:
        return []
    a_start, a_end = _parse_page_range(agenda_pages, len(pages))
    agenda_text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip()
    items = _parse_agenda_outline(agenda_text)
    if not items:
        return []
    titles = [f"{it['number']} {it['name']}".strip() for it in items]
    for idx, it in enumerate(items):
        _, info = _slice_packet_for_item(pages, titles, idx, outline=[])
        if info.get("sliced"):
            it["has_pages"] = True
            it["pages"] = info["pages"]
            it["start"] = int(info["start"])
            it["end"] = int(info["end"])
    return items


def _parse_page_range(agenda_pages: str, page_count: int) -> tuple[int, int]:
    """Parse a 1-indexed inclusive page range (``"1-3"`` / ``"5"``) to a 0-indexed,
    end-exclusive ``(start, end)`` slice, clamped to ``[0, page_count]``. Empty input
    defaults to the first few pages (a typical agenda front section)."""
    s = (agenda_pages or "").strip()
    if not s:
        return 0, min(page_count, 3) if page_count else 0
    m = re.match(r"^\s*(\d+)\s*(?:[-–to]+\s*(\d+))?\s*$", s)
    if not m:
        return 0, min(page_count, 3) if page_count else 0
    lo = max(1, int(m.group(1)))
    hi = int(m.group(2)) if m.group(2) else lo
    if hi < lo:
        lo, hi = hi, lo
    start = max(0, lo - 1)
    end = hi if page_count <= 0 else min(page_count, hi)
    if end <= start:
        end = min(page_count or hi, start + 1)
    return start, end


def parse_agenda_outline_from_packet(
    upload_id: str, agenda_pages: str = ""
) -> AgendaOutline:
    """Parse an uploaded packet into agenda items — bookmarks first, text fallback.

    Returns the item tree (each item carrying ``has_pages`` + 0-indexed
    ``start``/``end`` + a 1-indexed ``pages`` string) plus ``source``
    (``outline`` | ``text`` | ``none``), a ``confidence`` hint, a message, and a
    suggested agenda ``title``. Item names truncated by the PDF bookmarks are restored
    deterministically from the agenda (table-of-contents) page text — no LLM.
    """
    def result(items, source, confidence, message, title="") -> AgendaOutline:
        return {"items": items, "source": source, "confidence": confidence,
                "message": message, "model": "", "title": title}

    if not upload_id:
        return result([], "none", "empty", "Upload an agenda packet first.")
    if not agenda_pages:
        with _PKT_LOCK:
            meta = _PKT_META.get(upload_id)
        agenda_pages = (meta or {}).get("agenda_pages", "")

    outline = cached_packet_outline(upload_id)
    page_count = cached_packet_page_count(upload_id)
    if page_count <= 0:
        return result([], "none", "empty",
                      "The packet is no longer on the server — re-upload it.")

    if outline:
        items = _items_from_outline(outline, page_count)
        if items:
            # Restore truncated bookmark titles + a default agenda name from the TOC text.
            agenda_text = _agenda_text_for(upload_id, agenda_pages, page_count)
            if agenda_text:
                _untruncate_item_names(items, agenda_text)
            return result(items, "outline", "good",
                          "Parsed from the packet's PDF bookmarks.",
                          title=_agenda_title_from_text(agenda_text))

    pages = cached_packet_pages(upload_id)
    items = _items_from_text(pages, agenda_pages)
    if not items:
        return result([], "none", "empty",
                      "No agenda items could be parsed — set the agenda page range, "
                      "or add items/ranges manually.")
    a_start, a_end = _parse_page_range(agenda_pages, len(pages))
    agenda_text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip()
    n_numbered = sum(1 for it in items if it["number"] and not it["is_section"])
    confidence = "good" if n_numbered >= 3 else "poor"
    return result(items, "text", confidence,
                  "Parsed from the agenda text (no PDF bookmarks) — verify the page "
                  "ranges.", title=_agenda_title_from_text(agenda_text))


# --------------------------------------------------------------------------- #
# Local packet cache (keyed by upload_id).
#
# The uploaded Agenda Packet is one multi-MB PDF used by every feature: the page
# viewer renders pages from it, the item report reads its text, the agent searches
# it. We save it once to a local file at upload time and reuse it everywhere. Per-page
# text and the bookmark outline (the next-biggest costs) are extracted from the saved
# PDF once and memoized alongside. The cache is keyed by the server-minted upload_id.
# --------------------------------------------------------------------------- #
_PKT_DIR = Path(
    os.getenv("PACKET_CACHE_DIR") or (Path(tempfile.gettempdir()) / "agenda_packets")
)
_PKT_META: "OrderedDict[str, dict]" = OrderedDict()  # uid -> {path, name, agenda_pages}
_PKT_PAGES: "OrderedDict[str, list]" = OrderedDict()  # uid -> [page text]
_PKT_OUTLINE: "OrderedDict[str, list]" = OrderedDict()  # uid -> [bookmark]
_PKT_PAGECOUNT: "OrderedDict[str, int]" = OrderedDict()  # uid -> page count
_PKT_CACHE_MAX = 24

# Gradio runs API handlers in a threadpool, so the caches above and the in-flight
# map below are shared across threads. One lock guards every mutation/read of them.
# The expensive work (text extraction) runs OUTSIDE the lock via the single-flight
# helper, so the lock is only ever held briefly.
_PKT_LOCK = threading.Lock()

# pypdfium2 (PDFium) is NOT thread-safe — it keeps global state, so two threads doing
# any PDFium work at once (rendering pages, counting pages, extracting text/outline)
# corrupt the native heap and abort the process ("corrupted double-linked list").
# Gradio runs API handlers in a threadpool, so EVERY PDFium call in this module must be
# serialized behind this one global lock.
_PDFIUM_LOCK = threading.Lock()


class _Flight:
    """One in-flight expensive op for a cache key — followers share its result."""

    __slots__ = ("event", "result", "error")

    def __init__(self) -> None:
        self.event = threading.Event()
        self.result = None
        self.error: BaseException | None = None


_PKT_PAGES_FLIGHTS: "dict[str, _Flight]" = {}  # text extractions


def _compute_once(flights: "dict[str, _Flight]", key: str, producer):
    """Run ``producer()`` once across threads for ``key``; share its result/error.

    The first caller becomes the leader and runs ``producer()`` **without holding**
    ``_PKT_LOCK`` (it may parse for many seconds); concurrent callers become followers
    that wait on the leader's event and return the same result (or re-raise the same
    exception). This collapses a burst of rapid item clicks into a single extraction
    instead of N duplicate, mutually-starving ones.
    """
    with _PKT_LOCK:
        flight = flights.get(key)
        leader = flight is None
        if leader:
            flight = _Flight()
            flights[key] = flight
    if not leader:
        flight.event.wait()
        if flight.error is not None:
            raise flight.error
        return flight.result

    try:
        flight.result = producer()
    except BaseException as e:  # noqa: BLE001 - shared with followers, then re-raised
        flight.error = e
    finally:
        with _PKT_LOCK:
            flights.pop(key, None)
        flight.event.set()
    if flight.error is not None:
        raise flight.error
    return flight.result


def _evict_packet(uid: str) -> None:
    """Drop a packet's file + every derived cache entry. Caller holds ``_PKT_LOCK``."""
    meta = _PKT_META.pop(uid, None)
    _PKT_PAGES.pop(uid, None)
    _PKT_OUTLINE.pop(uid, None)
    _PKT_PAGECOUNT.pop(uid, None)
    if meta:
        try:
            Path(meta["path"]).unlink()
        except OSError:
            pass


def upload_packet(
    file_bytes: bytes, filename: str, agenda_pages: str = "", upload_id: str = ""
) -> dict:
    """Persist an uploaded agenda packet and parse its agenda outline.

    Writes the PDF under ``_PKT_DIR`` keyed by a server-minted ``upload_id`` (or the
    one passed in, for a re-upload — which overwrites in place and refreshes the parse),
    registers the cache meta, evicting the least-recently-used packet past the cap.
    Returns ``{upload_id, name, size, page_count, agenda_pages, items, source,
    confidence, message, model}``.
    """
    uid = upload_id or secrets.token_urlsafe(12)
    _PKT_DIR.mkdir(parents=True, exist_ok=True)
    safe = re.sub(r"[^A-Za-z0-9_-]", "_", uid)
    path = _PKT_DIR / f"{safe}.pdf"
    path.write_bytes(file_bytes)
    name = (filename or "agenda-packet.pdf").strip()
    if not name.lower().endswith(".pdf"):
        name += ".pdf"

    with _PKT_LOCK:
        # A re-upload reuses the uid — clear any stale derived caches for it.
        _PKT_PAGES.pop(uid, None)
        _PKT_OUTLINE.pop(uid, None)
        _PKT_PAGECOUNT.pop(uid, None)
        _PKT_META[uid] = {"path": str(path), "name": name, "agenda_pages": agenda_pages}
        _PKT_META.move_to_end(uid)
        while len(_PKT_META) > _PKT_CACHE_MAX:
            old_uid = next(iter(_PKT_META))
            if old_uid == uid:
                break
            _evict_packet(old_uid)

    page_count = cached_packet_page_count(uid)
    outline_result = parse_agenda_outline_from_packet(uid, agenda_pages)
    return {
        "upload_id": uid,
        "name": name,
        "size": len(file_bytes),
        "page_count": page_count,
        "agenda_pages": agenda_pages,
        **outline_result,
    }


def cached_packet(upload_id: str) -> tuple[Path, str] | None:
    """Local path + filename of an uploaded packet, or ``None`` if unknown/evicted.

    The packet is written at upload time, so this is a pure cache lookup. ``None``
    means the upload is gone (tmp wiped on a Space restart, or LRU-evicted) — the
    caller surfaces a "re-upload" prompt.
    """
    if not upload_id:
        return None
    with _PKT_LOCK:
        meta = _PKT_META.get(upload_id)
        if meta is not None and Path(meta["path"]).exists():
            _PKT_META.move_to_end(upload_id)
            return Path(meta["path"]), meta["name"]
    # Bundled samples are always re-derivable from disk: on a cache miss (Space restart,
    # LRU eviction, or a saved sample from a prior session), rehydrate from samples/
    # rather than 404. User uploads have no such source, so they still surface no_packet.
    if upload_id.startswith(_SAMPLE_UID_PREFIX):
        s = _SAMPLE_BY_ID.get(upload_id[len(_SAMPLE_UID_PREFIX):])
        if s is not None and (_SAMPLES_DIR / s["file"]).exists():
            return _rehydrate_sample(upload_id, s)
    return None


def _rehydrate_sample(upload_id: str, s: dict) -> tuple[Path, str] | None:
    """Re-cache a bundled sample's PDF on disk + register its meta (no re-parse), so the
    /packet route can serve it after the in-memory cache lost it."""
    _PKT_DIR.mkdir(parents=True, exist_ok=True)
    safe = re.sub(r"[^A-Za-z0-9_-]", "_", upload_id)
    path = _PKT_DIR / f"{safe}.pdf"
    try:
        if not path.exists():
            path.write_bytes((_SAMPLES_DIR / s["file"]).read_bytes())
    except OSError:
        return None
    with _PKT_LOCK:
        _PKT_META[upload_id] = {"path": str(path), "name": s["file"],
                                "agenda_pages": s["agenda_pages"]}
        _PKT_META.move_to_end(upload_id)
    return path, s["file"]


def _extract_pages_text(src: bytes | str) -> list[str]:
    """Per-page text via pypdfium2 (PDFium native).

    ~20-25x faster than pdfplumber's pdfminer backend on a 150-page packet (≈0.4s
    vs ≈9s), and the slicer's whitespace-normalized title anchoring lands on exactly
    the same pages — we only need the page *text*, not pdfplumber's layout analysis.
    Accepts PDF bytes or a path (renders from the locally-cached file).
    """
    import pypdfium2 as pdfium

    out: list[str] = []
    with _PDFIUM_LOCK:  # PDFium is not thread-safe — serialize all access
        pdf = pdfium.PdfDocument(src)
        try:
            for i in range(len(pdf)):
                textpage = pdf[i].get_textpage()
                try:
                    out.append(textpage.get_text_range())
                finally:
                    textpage.close()
        finally:
            pdf.close()
    return out


def cached_packet_pages(upload_id: str) -> list[str]:
    """Per-page text of the cached packet PDF — extracted once and memoized.

    Drives the slicer's text fallback (and the item report). Reads the locally-saved
    PDF. Returns ``[]`` when the upload is unknown/evicted. The page viewer avoids this
    whole-packet text pass when the PDF outline anchors the item (see
    :func:`agenda_item_pages`). Concurrent callers share one extraction.
    """
    if not upload_id:
        return []
    with _PKT_LOCK:
        cached = _PKT_PAGES.get(upload_id)
        if cached is not None:
            _PKT_PAGES.move_to_end(upload_id)
            return cached
    cp = cached_packet(upload_id)
    if cp is None:
        return []

    def _extract() -> list[str]:
        with _PKT_LOCK:
            cached = _PKT_PAGES.get(upload_id)
            if cached is not None:
                _PKT_PAGES.move_to_end(upload_id)
                return cached
        pages = _extract_pages_text(str(cp[0]))  # CPU — no lock held
        with _PKT_LOCK:
            _PKT_PAGES[upload_id] = pages
            _PKT_PAGES.move_to_end(upload_id)
            while len(_PKT_PAGES) > _PKT_CACHE_MAX:
                _PKT_PAGES.popitem(last=False)
        return pages

    return _compute_once(_PKT_PAGES_FLIGHTS, upload_id, _extract)


def cached_packet_page_count(upload_id: str) -> int:
    """Number of pages in the cached packet PDF — read once and memoized.

    The outline-only slice (page viewer) needs just the page count for its end bound,
    not the per-page text. Opens the locally-saved PDF (no text extraction). Returns 0
    when the upload is unknown/evicted.
    """
    if not upload_id:
        return 0
    with _PKT_LOCK:
        cached = _PKT_PAGECOUNT.get(upload_id)
        if cached is not None:
            _PKT_PAGECOUNT.move_to_end(upload_id)
            return cached
    cp = cached_packet(upload_id)
    if cp is None:
        return 0

    import pypdfium2 as pdfium

    with _PDFIUM_LOCK:  # PDFium is not thread-safe — serialize all access
        pdf = pdfium.PdfDocument(str(cp[0]))
        try:
            count = len(pdf)
        finally:
            pdf.close()
    with _PKT_LOCK:
        _PKT_PAGECOUNT[upload_id] = count
        _PKT_PAGECOUNT.move_to_end(upload_id)
        while len(_PKT_PAGECOUNT) > _PKT_CACHE_MAX:
            _PKT_PAGECOUNT.popitem(last=False)
    return count


def cached_packet_outline(upload_id: str) -> list[dict]:
    """The cached packet PDF's bookmark outline — read once and memoized.

    Feeds the bookmarks-first parse and the slicer's outline anchoring (see
    :func:`_outline_anchors`). Reads the locally-saved PDF; returns ``[]`` when the
    upload is unknown/evicted or the packet carries no outline.
    """
    if not upload_id:
        return []
    with _PKT_LOCK:
        cached = _PKT_OUTLINE.get(upload_id)
        if cached is not None:
            _PKT_OUTLINE.move_to_end(upload_id)
            return cached
    cp = cached_packet(upload_id)
    if cp is None:
        return []

    try:
        with _PDFIUM_LOCK:  # extract_pdf_outline uses PDFium — serialize all access
            outline = extract_pdf_outline(Path(cp[0]).read_bytes())
    except Exception:  # noqa: BLE001 - a packet with no/garbled outline just falls back
        outline = []
    with _PKT_LOCK:
        _PKT_OUTLINE[upload_id] = outline
        _PKT_OUTLINE.move_to_end(upload_id)
        while len(_PKT_OUTLINE) > _PKT_CACHE_MAX:
            _PKT_OUTLINE.popitem(last=False)
    return outline


def _agenda_pages_for(upload_id: str) -> str:
    with _PKT_LOCK:
        meta = _PKT_META.get(upload_id)
    return (meta or {}).get("agenda_pages", "")


# --------------------------------------------------------------------------- #
# Deterministic title untruncation — recover full item titles (and a default agenda
# name) by searching the agenda (table-of-contents) page text. PDF bookmark titles are
# commonly truncated (~50 chars) and may drop a department prefix; the agenda text holds
# each item's full title, so we find the truncated substring there and lift the whole
# enclosing title block. No LLM.
# --------------------------------------------------------------------------- #
_UNTRUNC_MIN_KEY = 24  # squeezed bookmark shorter than this isn't worth searching
_UNTRUNC_MAX_LEN = 320  # never adopt a runaway "title" longer than this

# A line that ends an item's title block (its metadata / a page footer / the AGENDA
# banner). Continuation lines (wrapped title text) match none of these.
_TITLE_META_RE = re.compile(
    r"(?i)^\s*(item category|presenter|action required|action|fiscal impact|department|"
    r"sponsor|recommended action|recommendation|background|summary|whereas|now therefore)\b"
)
_PAGE_FOOTER_RE = re.compile(r"(?i)(page \d+ of \d+|agenda page \d+)")
# Lenient enumerator at the START of a line ("11.", "a)", "iv.", or even "1 " / "2 ").
_ENUM_LINE_RE = re.compile(r"^\s*(?:\d{1,3}|[a-zA-Z]|[ivxlcdmIVXLCDM]{1,6})[.):]?\s+\S")
# Lenient enumerator to strip from a recovered full title (allows the bare "1 " form).
_ENUM_LEAD_RE = re.compile(r"^\s*(?:\d{1,3}|[a-z]|[ivxlcdm]{1,6})[.):]?\s+", re.IGNORECASE)


def _is_title_boundary(line: str) -> bool:
    s = line.strip()
    if not s:
        return True
    return bool(_TITLE_META_RE.match(s) or _PAGE_FOOTER_RE.search(s)
                or _OUTLINE_START_RE.match(s))


def _title_block(lines: list[str], hit: int) -> str:
    """The full title block of agenda ``lines`` that the matched line ``hit`` belongs to.

    Walks up to the block's enumerator line (or a boundary above) and down across wrapped
    continuation lines, stopping at the next enumerator / a metadata line / a page footer.
    """
    top = hit
    while top > 0:
        if _ENUM_LINE_RE.match(lines[top]):
            break
        if _is_title_boundary(lines[top - 1]):
            break
        top -= 1
    bot = hit
    while bot + 1 < len(lines):
        nxt = lines[bot + 1]
        if _is_title_boundary(nxt) or _ENUM_LINE_RE.match(nxt):
            break
        bot += 1
    return " ".join(ln.strip() for ln in lines[top:bot + 1] if ln.strip())


def _untruncate_item_names(items: list[dict], agenda_text: str) -> int:
    """Replace truncated bookmark item names with their full title from the agenda text.

    For each item, find its (enumerator-stripped, alphanumeric-squeezed) bookmark title as
    a substring of the squeezed agenda text — in document order to disambiguate repeats —
    then lift the enclosing title block. Adopts it only when it's genuinely longer. Returns
    the number of names improved.
    """
    lines = agenda_text.splitlines()
    squeezed_chars: list[str] = []
    line_of: list[int] = []  # squeezed char index -> agenda line index
    for li, ln in enumerate(lines):
        for ch in ln:
            if ch.isalnum():
                squeezed_chars.append(ch.lower())
                line_of.append(li)
    squeezed = "".join(squeezed_chars)
    if not squeezed:
        return 0

    cursor = 0
    changed = 0
    for it in items:
        name = it.get("name", "")
        bkey = _squeeze_title(name)
        if len(bkey) < _UNTRUNC_MIN_KEY:
            continue
        q = squeezed.find(bkey, cursor)
        if q == -1:  # fall back to a global search (order drift)
            q = squeezed.find(bkey)
        if q == -1:
            continue
        cursor = q + len(bkey)
        full = _title_block(lines, line_of[q])
        full = _ENUM_LEAD_RE.sub("", full).strip()
        if full and len(name) < len(full) <= _UNTRUNC_MAX_LEN:
            it["name"] = full
            changed += 1
    return changed


# Header lines (before the AGENDA banner) that name the meeting + its date.
_AGENDA_DATE_RE = re.compile(
    r"(?i)(?:(?:mon|tue|wed|thu|fri|sat|sun)[a-z]*,?\s+)?"
    r"(?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4}"
)
_AGENDA_BODY_RE = re.compile(
    r"(?i)\b(board|council|commission|committee|authority|district|trustees|supervisors|"
    r"township|village|city|county|district|department|agency)\b"
)


def _agenda_title_from_text(agenda_text: str) -> str:
    """A concise default agenda name (governing body + date) from the header lines.

    Deterministic, best-effort: scans the lines before the AGENDA banner for a governing-
    body line and a date. Returns ``""`` when nothing usable is found (caller falls back
    to the file name)."""
    lines = [re.sub(r"\s+", " ", ln).strip() for ln in agenda_text.splitlines()]
    lines = [ln for ln in lines if ln]
    end = len(lines)
    for i, ln in enumerate(lines):
        if _OUTLINE_START_RE.match(ln):
            end = i
            break
    header = lines[:end] or lines[:15]

    date = ""
    for ln in header:
        m = _AGENDA_DATE_RE.search(ln)
        if m:
            date = m.group(0).strip(" ,")
            break
    body = ""
    for ln in header:
        if _AGENDA_DATE_RE.search(ln) or any(c.isdigit() for c in ln):
            continue
        if 4 <= len(ln) <= 70 and _AGENDA_BODY_RE.search(ln):
            body = ln.title() if ln.isupper() else ln
            break
    return " — ".join(p for p in (body, date) if p)[:90]


def _agenda_text_for(upload_id: str, agenda_pages: str, page_count: int) -> str:
    """Text of just the marked agenda (table-of-contents) pages — a targeted PDFium read
    so big packets don't pay a full text extraction at parse time."""
    cp = cached_packet(upload_id)
    if cp is None or page_count <= 0:
        return ""
    a_start, a_end = _parse_page_range(agenda_pages, page_count)
    import pypdfium2 as pdfium

    out: list[str] = []
    with _PDFIUM_LOCK:  # PDFium is not thread-safe — serialize all access
        pdf = pdfium.PdfDocument(str(cp[0]))
        try:
            for i in range(max(0, a_start), min(len(pdf), a_end)):
                textpage = pdf[i].get_textpage()
                try:
                    out.append(textpage.get_text_range())
                finally:
                    textpage.close()
        finally:
            pdf.close()
    return "\n\n".join(p for p in out if p).strip()


def _packet_documents(upload_id: str) -> list[dict]:
    """Chroma documents for the whole uploaded packet (one doc; chunking splits it).

    Replaces the old per-event document loader. Used by the report, the agent's
    semantic search, and the summarize/report tools.
    """
    cp = cached_packet(upload_id)
    if cp is None:
        return []
    name = cp[1]
    pages = cached_packet_pages(upload_id)
    text = "\n\n".join(p for p in pages if p).strip()
    if not text:
        return []
    return [{
        "doc_id": f"{upload_id}:packet",
        "text": text,
        "metadata": {"source": "packet", "upload_id": upload_id, "file_name": name},
    }]


# --------------------------------------------------------------------------- #
# Summarize (chunk the agenda portion -> Gemma-4)
# --------------------------------------------------------------------------- #
def summarize_agenda(upload_id: str, model: str = "") -> Iterator[dict]:
    """Stream a summary of the uploaded agenda's table-of-contents portion.

    Summarizes the text of the page range the user marked as the agenda (not the whole
    multi-hundred-page packet). Yields progress dicts ``{"stage", "message", "summary",
    "chunks", "model"}`` where ``stage`` is ``working`` | ``done`` | ``error``.

    Args:
        upload_id: the uploaded packet handle.
        model: which GGUF to run (local backend) — ``"e4b"`` (default) or ``"26b"``.
    """
    def msg(stage: str, message: str = "", summary: str = "",
            chunks: int = 0, model: str = "") -> dict:
        return {"stage": stage, "message": message, "summary": summary,
                "chunks": chunks, "model": model}

    if not upload_id:
        yield msg("error", "Upload an agenda packet first.")
        return

    yield msg(
        "working",
        "Reading the agenda and summarizing. The model may cold-start on the first "
        "call — this can take a minute.",
    )

    pages = cached_packet_pages(upload_id)
    if not pages:
        yield msg("error", "The packet is no longer on the server — re-upload it.")
        return
    a_start, a_end = _parse_page_range(_agenda_pages_for(upload_id), len(pages))
    text = "\n\n".join(p for p in pages[a_start:a_end] if p).strip()
    if not text:
        yield msg(
            "error",
            "The agenda pages have no extractable text — they may be scanned / "
            "image-only. Try the Report tab, which reads the whole packet.",
        )
        return

    n_chunks = len(chunk_text(text))
    try:
        if LLM_BACKEND == "local":
            from webapp import local_llm

            summary = local_llm.gpu_summarize(text, model)
        else:
            summary = summarize_text(text, complete=_complete)
    except Exception as e:  # noqa: BLE001
        yield msg("error", f"Summarization failed: {type(e).__name__}: {e}")
        return

    yield msg(
        "done",
        f"Summarized the agenda pages {a_start + 1}-{a_end} ({n_chunks} block(s)).",
        summary=summary or "_(empty summary)_",
        chunks=n_chunks,
        model=_model_label(model),
    )


# --------------------------------------------------------------------------- #
# Report streaming (shared by the whole-packet and per-item reports)
# --------------------------------------------------------------------------- #
def _report_msg(stage: str, message: str = "", report: str = "",
                frac: float = 0.0, docs: int = 0, model: str = "",
                code: str = "") -> dict:
    return {"stage": stage, "message": message, "report": report,
            "frac": frac, "docs": docs, "model": model, "code": code}


def _stream_report(docs: list[dict], question: str, max_sections: int,
                   built_from: str, model: str = "") -> Iterator[dict]:
    """Run a query-framed report over pre-fetched ``docs`` and stream progress.

    Bridges :func:`chroma.report.generate_report` (remote) or
    :func:`webapp.local_llm.gpu_report` (in-process ZeroGPU) to ``working`` /
    ``done`` / ``error`` frames. ``built_from`` is surfaced in the done frame;
    ``model`` selects the GGUF (local backend).
    """
    import queue
    import threading

    def done_frame(report: str) -> dict:
        return _report_msg(
            "done", built_from, report=report or "_(empty report)_",
            frac=1.0, docs=len(docs), model=_model_label(model),
        )

    # Local backend: stream straight from the in-process GPU generator.
    if LLM_BACKEND == "local":
        from webapp import local_llm

        try:
            for frame in local_llm.gpu_report(docs, question, max_sections, model):
                if "error" in frame:
                    yield _report_msg("error", f"Report generation failed: {frame['error']}")
                    return
                if frame.get("done"):
                    yield done_frame(frame.get("report", ""))
                    return
                yield _report_msg("working", frame.get("message", "Working…"),
                                  frac=float(frame.get("frac", 0.0)))
        except Exception as e:  # noqa: BLE001
            yield _report_msg("error", f"Report generation failed: {type(e).__name__}: {e}")
        return

    # Remote backend: bridge generate_report's progress callback to this generator.
    q: "queue.Queue[tuple]" = queue.Queue()

    def cb(frac: float, message: str) -> None:
        q.put(("progress", frac, message))

    def worker() -> None:
        try:
            report = generate_report(
                docs, question, complete=_complete,
                max_chunks=int(max_sections), progress=cb,
            )
            q.put(("done", report))
        except Exception as e:  # noqa: BLE001
            q.put(("error", f"{type(e).__name__}: {e}"))

    threading.Thread(target=worker, daemon=True).start()

    while True:
        kind, *rest = q.get()
        if kind == "progress":
            frac, message = rest
            yield _report_msg("working", message, frac=float(frac))
        elif kind == "done":
            yield done_frame(rest[0])
            return
        else:  # error
            yield _report_msg("error", f"Report generation failed: {rest[0]}")
            return


def _stream_item_answer(docs: list[dict], question: str, max_sections: int,
                        built_from: str, engine: str = "auto",
                        model: str = "") -> Iterator[dict]:
    """Answer one agenda item in a single pass and stream progress.

    Bridges :func:`webapp.local_llm.gpu_answer` (in-process ZeroGPU, token-streamed)
    or :func:`chroma.answer.answer_question` (remote) to ``working`` / ``done`` /
    ``error`` frames. ``engine`` is the user's choice — ``"single"`` (semantic search /
    one-shot answer) or ``"mapreduce"`` (read every section) — defaulting to ``"auto"``
    (size-based heuristic). ``model`` selects the GGUF (local backend only). ``working``
    frames carry the partial ``report`` so the UI renders the answer as it streams.
    """
    import queue
    import threading

    def done_frame(report: str) -> dict:
        return _report_msg(
            "done", built_from, report=report or "_(empty answer)_",
            frac=1.0, docs=len(docs), model=_model_label(model),
        )

    # Local backend: stream tokens straight from the in-process GPU generator.
    if LLM_BACKEND == "local":
        from webapp import local_llm

        try:
            for frame in local_llm.gpu_answer(docs, question, max_sections, engine, model):
                if "error" in frame:
                    yield _report_msg("error", f"Answer generation failed: {frame['error']}")
                    return
                if frame.get("done"):
                    yield done_frame(frame.get("report", ""))
                    return
                yield _report_msg("working", frame.get("message", "Working…"),
                                  report=frame.get("report", ""),
                                  frac=float(frame.get("frac", 0.0)))
        except Exception as e:  # noqa: BLE001
            yield _report_msg("error", f"Answer generation failed: {type(e).__name__}: {e}")
        return

    # Remote backend: single-pass answer (or map-reduce fallback) in a worker thread.
    from chroma.answer import answer_question, budget_for, resolve_engine

    char_budget, max_tokens = budget_for(max_sections)
    q: "queue.Queue[tuple]" = queue.Queue()

    def cb(frac: float, message: str) -> None:
        q.put(("progress", frac, message))

    def worker() -> None:
        try:
            if resolve_engine(docs, engine, char_budget=char_budget) == "mapreduce":
                report = generate_report(docs, question, complete=_complete,
                                         max_chunks=int(max_sections), progress=cb)
            else:
                report = answer_question(docs, question, complete=_complete,
                                         char_budget=char_budget, max_tokens=max_tokens,
                                         progress=cb)
            q.put(("done", report))
        except Exception as e:  # noqa: BLE001
            q.put(("error", f"{type(e).__name__}: {e}"))

    threading.Thread(target=worker, daemon=True).start()

    while True:
        kind, *rest = q.get()
        if kind == "progress":
            frac, message = rest
            yield _report_msg("working", message, frac=float(frac))
        elif kind == "done":
            yield done_frame(rest[0])
            return
        else:  # error
            yield _report_msg("error", f"Answer generation failed: {rest[0]}")
            return


# --------------------------------------------------------------------------- #
# Agenda Report (query-framed over the whole uploaded packet)
# --------------------------------------------------------------------------- #
def generate_agenda_report(
    upload_id: str,
    question: str,
    max_sections: int = 60,
    model: str = "",
) -> Iterator[dict]:
    """Stream a query-framed markdown report over the whole uploaded packet.

    Chunks the packet into an ephemeral in-memory ChromaDB collection, mines every
    chunk for facts relevant to ``question`` (map) and writes the findings up as
    markdown (reduce) — see :func:`chroma.report.generate_report`.

    Yields progress dicts ``{"stage", "frac", "message", "report", "docs", "model",
    "code"}`` where ``stage`` is ``working`` | ``done`` | ``error``.

    Args:
        upload_id: the uploaded packet handle.
        question: the free-form question framing the report.
        max_sections: cap on how many chunks are mined (higher = more thorough, slower).
        model: which GGUF to run (local backend) — ``"e4b"`` (default) or ``"26b"``.
    """
    question = (question or "").strip()
    if not upload_id:
        yield _report_msg("error", "Upload an agenda packet, then ask a question.")
        return
    if not question:
        yield _report_msg("error", "Enter a question to frame the report.")
        return

    yield _report_msg("working", "Reading the agenda packet…", frac=0.02)
    docs = _packet_documents(upload_id)
    if not docs:
        yield _report_msg(
            "error", "The packet is no longer on the server — re-upload it.",
            code="no_packet",
        )
        return

    built_from = (f"Built from the agenda packet, chunked into an ephemeral ChromaDB "
                  "collection.")
    yield from _stream_report(docs, question, max_sections, built_from, model)


# --------------------------------------------------------------------------- #
# Agenda Item Report (slice the packet to one item's section, report over it)
# --------------------------------------------------------------------------- #
def _item_documents(
    upload_id: str, item_titles: list[str], item_index: int,
    start: int = 0, end: int = 0,
) -> tuple[list[dict], dict]:
    """Build report documents scoped to one agenda item.

    Reads the packet from the local cache and scopes it to the item's pages. When an
    explicit page range is given (``end > start``, the user-adjusted range from the
    client), that range is used directly (clamped); otherwise the item is located by
    anchoring (outline first, text fallback — see :func:`_slice_packet_for_item`). The
    item's own agenda line is always included as a focusing document. Returns
    ``(documents, info)`` with ``info`` = ``{"sliced", "pages", "note"}``.
    """
    cp = cached_packet(upload_id)
    packet_name = cp[1] if cp is not None else "Agenda Packet"
    pages = cached_packet_pages(upload_id) if cp is not None else []
    outline = cached_packet_outline(upload_id) if cp is not None else []

    title = item_titles[item_index] if 0 <= item_index < len(item_titles) else ""
    base_meta = {"source": "packet", "upload_id": str(upload_id)}
    docs: list[dict] = []
    if title:
        docs.append({
            "doc_id": f"{upload_id}:item",
            "text": f"Selected agenda item: {title}",
            "metadata": {**base_meta, "file_name": "Selected agenda item"},
        })

    if not pages:
        return docs, {"sliced": False, "pages": "", "note": "",
                      "code": "no_packet"}

    # Explicit (possibly user-adjusted) range wins — clamp to the packet.
    s = max(0, int(start or 0))
    e = min(len(pages), int(end or 0))
    if e > s:
        sliced_text = "\n\n".join(p for p in pages[s:e] if p).strip()
        if sliced_text:
            info = {"sliced": True, "pages": f"{s + 1}-{e}", "start": s, "end": e,
                    "method": "explicit"}
            docs.append({
                "doc_id": f"{upload_id}:packet-slice",
                "text": sliced_text,
                "metadata": {**base_meta, "file_name": packet_name, "pages": info["pages"]},
            })
            info["note"] = (f"Reporting on pages {info['pages']} of {len(pages)} "
                            "(the item's backup pages).")
            return docs, info

    sliced_text, info = _slice_packet_for_item(
        pages, item_titles, item_index, outline=outline)
    if info["sliced"]:
        docs.append({
            "doc_id": f"{upload_id}:packet-slice",
            "text": sliced_text,
            "metadata": {**base_meta, "file_name": packet_name, "pages": info["pages"]},
        })
        how = ("from the packet's bookmarks" if info.get("method") == "outline"
               else "by matching the item's title in the packet")
        info["note"] = (f"Sliced the agenda packet to this item's section "
                        f"(pages {info['pages']} of {len(pages)}, {how}).")
    else:
        docs.append({
            "doc_id": f"{upload_id}:packet",
            "text": "\n\n".join(p for p in pages if p),
            "metadata": {**base_meta, "file_name": packet_name},
        })
        info["note"] = ("Couldn't locate this item's pages in the packet, so this "
                        "report covers the full packet, framed by the item. Set the "
                        "item's page range to scope it.")
    return docs, info


def generate_agenda_item_report(
    upload_id: str,
    item_titles: list[str],
    item_index: int,
    question: str,
    start: int = 0,
    end: int = 0,
    max_sections: int = 60,
    engine: str = "auto",
    model: str = "",
) -> Iterator[dict]:
    """Stream a query-framed report scoped to a single agenda item.

    Scopes the packet to the item's backup pages — using the explicit (user-adjusted)
    ``start``/``end`` range when given, else anchoring by bookmark/title (see
    :func:`_item_documents`) — and answers the question over that slice in a single
    streamed pass. When the pages can't be located the answer falls back to the full
    packet, with a note saying so.

    Args:
        upload_id: the uploaded packet handle.
        item_titles: every agenda item's ``"<number> <name>"`` line, in agenda order —
            used to anchor the item when no explicit range is given.
        item_index: index into ``item_titles`` of the item to report on.
        question: the free-form question framing the answer.
        start: 0-indexed start page of the item's slice (inclusive), 0 to auto-locate.
        end: 0-indexed end page of the item's slice (exclusive), 0 to auto-locate.
        max_sections: thoroughness knob — packet context fed + answer length.
        engine: ``"single"`` (semantic search) | ``"mapreduce"`` (read every section) |
            ``"auto"`` (by slice size).
        model: which GGUF to run (local backend) — ``"e4b"`` (default) or ``"26b"``.
    """
    question = (question or "").strip()
    if not upload_id:
        yield _report_msg("error", "Upload an agenda packet, then pick an item.")
        return
    if not item_titles or not (0 <= item_index < len(item_titles)):
        yield _report_msg("error", "Pick an agenda item to report on.")
        return
    if not question:
        yield _report_msg("error", "Enter a question to frame the report.")
        return

    yield _report_msg("working", "Reading the agenda packet and locating this item's "
                                 "section…", frac=0.02)
    docs, info = _item_documents(upload_id, item_titles, item_index, start, end)
    if info.get("code") == "no_packet":
        yield _report_msg(
            "error", "The packet is no longer on the server — re-upload it.",
            code="no_packet",
        )
        return
    if not any((d.get("text") or "").strip() for d in docs):
        yield _report_msg("error", "No agenda text was available for this item — the "
                                   "packet may be empty or image-only.")
        return

    note = info.get("note", "")
    built_from = f"{note} Answered by the model below."
    for frame in _stream_item_answer(docs, question, max_sections, built_from, engine, model):
        if note and frame["stage"] in ("working", "done") and frame.get("report"):
            frame["report"] = f"> _{note}_\n\n{frame['report']}"
        yield frame


# --------------------------------------------------------------------------- #
# Agenda item packet pages (render the item's PDF pages to images).
# --------------------------------------------------------------------------- #
_ITEM_PAGE_RENDER_SCALE = 1.6  # ~115 dpi — legible, keeps the JPEG payload small


def _render_pages_b64_iter(src: bytes | str, start: int, end: int) -> Iterator[ItemPage]:
    """Rasterize packet pages ``[start, end)`` to base64 JPEG data URLs, one at a time.

    Yields each page's :class:`ItemPage` as soon as it rasterizes (the PDF is opened
    once). ``src`` is the PDF bytes or a path to the locally-cached packet (pypdfium2
    accepts either).
    """
    import base64
    import io

    import pypdfium2 as pdfium

    pdf = pdfium.PdfDocument(src)
    try:
        for i in range(start, end):
            pil = pdf[i].render(scale=_ITEM_PAGE_RENDER_SCALE).to_pil().convert("RGB")
            buf = io.BytesIO()
            pil.save(buf, format="JPEG", quality=80)
            b64 = base64.b64encode(buf.getvalue()).decode("ascii")
            yield ItemPage(page=i + 1, image=f"data:image/jpeg;base64,{b64}")
    finally:
        pdf.close()


def agenda_item_pages(
    upload_id: str,
    item_titles: list[str],
    item_index: int,
    start: int = 0,
    end: int = 0,
    max_pages: int = 0,
) -> Iterator[ItemPages]:
    """Stream the packet PDF pages that back a single agenda item.

    Uses the explicit (user-adjusted) ``start``/``end`` range when given; otherwise
    locates the item with the same confidence-gated slicer as
    :func:`generate_agenda_item_report` (outline first, text fallback). Rasterizes the
    page range to base64 JPEG data URLs the frontend drops straight into ``<img>``
    tags. When the item's pages can't be located, ``sliced`` is ``False`` and
    ``images`` is empty (with ``note`` explaining why).

    Yields ``ItemPages`` frames with ``stage`` ``working`` → ``done`` | ``error``. It
    streams so the browser keeps the connection alive while pages render.

    Args:
        upload_id: the uploaded packet handle.
        item_titles: every agenda item's ``"<number> <name>"`` line, in agenda order.
        item_index: index into ``item_titles`` of the item to show pages for.
        start: 0-indexed start page (inclusive), 0 to auto-locate.
        end: 0-indexed end page (exclusive), 0 to auto-locate.
        max_pages: optional cap on how many pages to rasterize; ``0`` = the whole slice.
    """
    def frame(stage: str, note: str = "", sliced: bool = False,
              pages: str = "", images: list[ItemPage] | None = None,
              code: str = "") -> ItemPages:
        return {"stage": stage, "sliced": sliced, "pages": pages,
                "note": note, "code": code, "images": images or []}

    if not upload_id or not (0 <= item_index < len(item_titles or [])):
        yield frame("error", "Pick an agenda item.")
        return

    limit = int(max_pages) if max_pages and int(max_pages) > 0 else 0

    cp = cached_packet(upload_id)
    if cp is None:
        yield frame("error", "The packet is no longer on the server — re-upload it.",
                    code="no_packet")
        return

    yield frame("working", "Locating this item's pages in the packet…")
    try:
        page_count = cached_packet_page_count(upload_id)
        n_pages = page_count
        # Explicit (user-adjusted) range wins.
        s = max(0, int(start or 0))
        e = min(page_count, int(end or 0))
        if e > s:
            info = {"sliced": True, "pages": f"{s + 1}-{e}", "start": s, "end": e,
                    "method": "explicit"}
        else:
            # Fast path: outline anchors the item from bookmarks + page count alone.
            outline = cached_packet_outline(upload_id)
            info = _outline_slice_for_item(outline, item_titles, item_index, page_count)
            if not info["sliced"]:
                pages = cached_packet_pages(upload_id)
                if not pages:
                    yield frame("error", "The packet is no longer on the server — "
                                         "re-upload it.", code="no_packet")
                    return
                n_pages = len(pages)
                _, info = _slice_packet_for_item(
                    pages, item_titles, item_index, outline=outline
                )
    except Exception as e:  # noqa: BLE001
        yield frame("error", f"Couldn't load the agenda packet: {type(e).__name__}: {e}")
        return

    if not info["sliced"]:
        yield frame(
            "done", sliced=False,
            note=(f"No matching pages were found for this item in the {n_pages}-page "
                  "agenda packet. Set the item's page range, or open the full packet "
                  "PDF below. (The ask box still searches the whole packet.)"),
        )
        return

    packet_path = cp[0]
    pstart, pend = info["start"], info["end"]
    total = pend - pstart
    n = total if limit <= 0 else min(total, limit)
    render_end = pstart + n
    located = {"outline": "via the packet's PDF bookmarks",
               "explicit": "from the item's page range",
               "text": "by matching the item title in the packet text"}.get(
        info.get("method", ""), "in the packet")
    note = f"Packet pages {info['pages']} — {total} page(s) for this item, located {located}"
    note += "." if n >= total else f"; showing the first {n}."

    yield frame("working", note=f"Rendering {n} packet page(s) {info['pages']}…",
                sliced=True, pages=info["pages"])
    # Rasterize the whole slice under the PDFium lock (PDFium is not thread-safe; the
    # lock can't be held across a yield, so render to memory, release, then return).
    try:
        with _PDFIUM_LOCK:
            rendered = list(_render_pages_b64_iter(str(packet_path), pstart, render_end))
    except Exception as e:  # noqa: BLE001
        yield frame("error", f"Couldn't render the packet pages: {type(e).__name__}: {e}")
        return

    yield frame("done", sliced=True, pages=info["pages"], note=note, images=rendered)


# --------------------------------------------------------------------------- #
# Agent Mode (a ReAct loop drives the packet tools — see webapp/agent_loop.py
# and webapp/agent_tools.py). The model picks tools, reads results, and answers;
# we stream its thinking / tool calls / results / final answer to the React panel.
# --------------------------------------------------------------------------- #
def _agent_frame(fr: dict, *, model: str = "") -> AgentFrame:
    """Normalize a loop frame to the stable :class:`AgentFrame` shape."""
    stage = fr.get("stage", "")
    return {
        "stage": stage,
        "text": str(fr.get("text", "")),
        "tool": str(fr.get("tool", "")),
        "args": fr.get("args") or {},
        "result": str(fr.get("result", "")),
        "summary": str(fr.get("summary", "")),
        "step": int(fr.get("step", 0) or 0),
        "model": model if stage in ("answer", "error") else "",
    }


def agent_chat(
    upload_id: str,
    messages: list[AgentMessage],
    model: str = "",
) -> Iterator[AgentFrame]:
    """Stream one turn of Agent Mode over the uploaded agenda packet.

    A ReAct loop lets the model call the packet tools (:mod:`webapp.agent_tools`) —
    listing agenda items, reading an item's pages, semantic-searching the packet, and
    (heavier) summarizing or reporting — then answer the user's latest message.

    On the ZeroGPU Space (``LLM_BACKEND=local``) the **entire turn** runs inside one
    ``@spaces.GPU`` window (the GGUF is loaded once and reused for every reasoning step
    and any LLM-backed tool); otherwise it runs against the remote vLLM endpoint.

    Args:
        upload_id: the uploaded packet handle the tools operate on.
        messages: the conversation so far, each ``{"role", "content"}``, ending with
            the user's current message.
        model: which in-process GGUF to use on the local backend (``"e4b"`` / ``"26b"``);
            ignored by the remote backend, which uses its configured model.

    Yields :class:`AgentFrame` steps: ``thinking`` → ``tool_call`` → ``tool_result``
    (repeated) → ``answer``, or ``error``.
    """
    if not upload_id:
        yield _agent_frame({"stage": "error", "text": "Upload an agenda packet first."})
        return
    if not messages or not (messages[-1].get("content") or "").strip():
        yield _agent_frame({"stage": "error", "text": "Ask a question to begin."})
        return

    model_lbl = _model_label(model)

    try:
        if LLM_BACKEND == "local":
            # Whole loop runs in one GPU window (model loaded there, reused per step).
            from webapp import local_llm

            frames = local_llm.gpu_agent_turn(upload_id, list(messages), model)
        else:
            from webapp.agent_loop import agent_turn
            from webapp.agent_tools import ToolContext

            ctx = ToolContext(upload_id=upload_id, complete=_complete)
            frames = agent_turn(list(messages), ctx, _complete)

        for fr in frames:
            yield _agent_frame(fr, model=model_lbl)
    except Exception as e:  # noqa: BLE001
        yield _agent_frame(
            {"stage": "error", "text": f"Agent failed: {type(e).__name__}: {e}"},
            model=model_lbl,
        )


def lii_agent_chat(
    messages: list[AgentMessage],
    model: str = "",
) -> Iterator[AgentFrame]:
    """Stream one turn of the Cornell LII legal-research agent.

    A ReAct loop over the legal tools (:mod:`webapp.lii_tools`) — searching the federal
    eCFR, resolving and verifying CFR/USC citations, and linking state materials on
    Cornell LII — then answering the user's latest message. Unlike :func:`agent_chat`
    this is **packet-independent** (no ``upload_id``); it researches public US law and
    its tools make live calls to the eCFR API / ``law.cornell.edu``.

    On the ZeroGPU Space (``LLM_BACKEND=local``) the whole turn runs in one
    ``@spaces.GPU`` window; otherwise it runs against the remote vLLM endpoint.

    Args:
        messages: the conversation so far, each ``{"role", "content"}``, ending with the
            user's current message.
        model: which in-process GGUF to use on the local backend; ignored remotely.

    Yields :class:`AgentFrame` steps: ``thinking`` → ``tool_call`` → ``tool_result``
    (repeated) → ``answer``, or ``error``.
    """
    if not messages or not (messages[-1].get("content") or "").strip():
        yield _agent_frame({"stage": "error", "text": "Ask a question to begin."})
        return

    model_lbl = _model_label(model)

    try:
        if LLM_BACKEND == "local":
            from webapp import local_llm

            frames = local_llm.gpu_lii_agent_turn(list(messages), model)
        else:
            from webapp.agent_loop import agent_turn
            from webapp.lii_tools import TOOLKIT, LiiContext

            ctx = LiiContext()
            frames = agent_turn(list(messages), ctx, _complete, toolkit=TOOLKIT)

        for fr in frames:
            yield _agent_frame(fr, model=model_lbl)
    except Exception as e:  # noqa: BLE001
        yield _agent_frame(
            {"stage": "error", "text": f"Agent failed: {type(e).__name__}: {e}"},
            model=model_lbl,
        )