booth/booth/app.py

"""The Booth — a standing web server that renders drop-folders as ephemeral media booths.

Model (deliberately dead-simple, no database):
  * The data dir holds one subfolder per "booth". A booth is created by a CC
    session simply making a folder and dropping files in — there is no upload API.
  * GET /            -> index: a card per booth (scan of the data dir).
  * GET /b/<name>/   -> if <name>/index.html exists, serve it verbatim; otherwise
                        auto-render a gallery of the images / webm-videos / audio in it.
  * GET /b/<name>/<file> -> serve a file out of the booth (also feeds a custom index.html's assets).
  * LIFETIME IS DERIVED, not set by a boolean (U4). Three states, and `sweep_once`
    is the only thing that honours the first two:
      KEPT      `.forever` present. Never swept, own lane at the top of the index.
                Durable operator-facing boards — chiefly the standing link board,
                whose whole purpose is to outlive the scrollback it replaces.
      HELD      an open pick in `.marks.json`, or marks that cannot be read at all.
                A booth the operator still owes an answer to is not the sweeper's
                to take, and one whose judgment we failed to READ is certainly not.
      EPHEMERAL everything else: wiped TTL hours after the last activity. Age is the
                *newest* mtime in the tree, so a booth lives while it is being
                worked on and self-destructs once it stops.
  * VIEWING IS ACTIVITY. A deliberate GET of a booth's own page writes VIEW_MARKER,
    which the age rule already counts — if the operator is still looking at it, it
    is still alive. Browsing the index is not a view, and neither is a session
    polling `marks.json`: an agent must not be able to hold its own booth open.
  * WHY DERIVED. `.forever` was the ONLY way to say three different things, and the
    measurement showed it carrying all of them — 17 of 24 live booths on 2026-09-22
    (70%, up from 54%), with three of the four booths awaiting an answer ALSO pinned
    by hand. Only "this is durable" is what keep means. The other two are facts the
    service already held and did not consult.

State is the filesystem — `ls ~/booth-data` tells you everything. That is the whole point.
"""

from __future__ import annotations

import asyncio
import fcntl
import hashlib
import io
import os
import re
import secrets
import shutil
import time
import zipfile
from contextlib import asynccontextmanager
from dataclasses import replace
from pathlib import Path
from typing import Sequence
from urllib.parse import quote, unquote

from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import (
    FileResponse,
    HTMLResponse,
    JSONResponse,
    RedirectResponse,
    Response,
)
from fastapi.templating import Jinja2Templates
from starlette.concurrency import run_in_threadpool
from jinja2 import Environment, FileSystemLoader, select_autoescape

try:
    import markdown as _markdown
except ImportError:  # optional dep — .md then degrades to a plain-text view
    _markdown = None

TEMPLATES_DIR = Path(__file__).parent / "templates"

# The item record lives in booth/items.py — ONE resolver every surface reads.
# These names are RE-EXPORTED rather than merely moved: 22 existing test sites
# import them from booth.app by name, and a silent drop would be found by a
# consumer instead of by us. `test_app_still_exports_the_moved_names` asserts it.
from booth.items import (  # noqa: E402,F401
    AUDIO_EXTS,
    BLUR_FILE,
    CAPTION_MAX,
    DOC_MAX_BYTES,
    IMAGE_EXTS,
    MARKDOWN_EXTS,
    TEXT_EXTS,
    VIDEO_EXTS,
    Item,
    booth_items,
    classify,
    doc_kind,
    find_item,
    image_chain,
    read_blurred,
    render_doc,
    render_doc_body,
)

# Sentinel dotfile that exempts a booth from the TTL sweep. A dotfile because
# the existing listing code already skips dotfiles, so it costs nothing in item
# counts or galleries, and because `touch`/`rm` is the entire user interface: no
# flag to remember, no state anywhere but the filesystem.
KEEP_MARKER = ".forever"

# Records the last deliberate look at a booth (U4). A dotfile for the same two
# reasons KEEP_MARKER is one — `booth_items` and `zip_booth` skip it, so it
# costs nothing in counts, galleries or zips — and NOT a `.lock` dotfile, so
# `_newest_mtime` COUNTS it and the existing age rule picks the view up with no
# new arithmetic. That is the whole integration: a view is one more thing in
# the tree, not a second term in the formula.
VIEW_MARKER = ".viewed"

# ⚠⚠ BLUR IS COSMETIC, NOT ACCESS CONTROL. The file is still served at its own
# URL, still in the zip, still on disk. This hides an item from a glance — a
# shoulder, a screen-share, a scroll past something you did not want to see
# full-size — and nothing more. The Booth has no auth by design; if a thing
# must not be seen by whoever can reach port 8090, it must not be in a booth.
# Anyone who reads this marker as protection has misread it.


def set_blurred(booth: Path, rel: str, on: bool) -> set[str]:
    """Add or remove one item from the blur set. Atomic replace, so a crash
    mid-write cannot leave a half-file that read_blurred would parse as a
    shorter — and therefore more revealing — set. Returns the new set."""
    current = read_blurred(booth)
    if on:
        current.add(rel)
    else:
        current.discard(rel)
    path = booth / BLUR_FILE
    if not current:
        path.unlink(missing_ok=True)
        return current
    tmp = path.with_suffix(".tmp")
    tmp.write_text("".join(f"{r}\n" for r in sorted(current)))
    tmp.replace(path)
    return current

# The link-board logic lives in booth/links.py (stdlib only) so the `booth` CLI
# can use it without pulling FastAPI in. Re-exported here because call sites and
# tests already reference these names through app.
from booth.asks import (  # noqa: E402
    ANSWER_SUFFIX,
    ASK_SUFFIX,
    AskError,
    is_answer_file,
    is_ask_file,
    valid_stem,
)
from booth.marks import (  # noqa: E402
    MARKS_FILE,
    Mark,
    MarksCorrupt,
    answer_pick,
    as_dict,
    declare_pick,
    delete_mark,
    import_legacy_asks,
    hold_read,
    marks_for,
    marks_for_target,
    open_marks,
    set_flag,
    write_note,
)
from booth.manifest import (  # noqa: E402
    MANIFEST_FILE,
    SERVICE_HANDLE,
    read_manifest,
    write_manifest,
)
from booth.benches import (  # noqa: E402
    BENCH_STATES,
    normalize_bench_url,
    read_benches,
    remove_bench,
    set_bench_state,
    upsert_bench,
)
from booth.links import (  # noqa: E402
    LINK_LOCK,
    LINKS_FILE,
    PINS_FILE,
    booth_target,
    link_entry_id,
    order_for_display,
    parse_link_entries,
    read_pins,
    remove_link_entry,
    toggle_pin,
)


def human_dur(seconds: float) -> str:
    s = int(seconds)
    if s <= 0:
        return "expired"
    h, rem = divmod(s, 3600)
    m, _ = divmod(rem, 60)
    if h and m:
        return f"{h}h {m}m"
    if h:
        return f"{h}h"
    if m:
        return f"{m}m"
    return "<1m"


def _newest_mtime(path: Path) -> float:
    """Newest mtime among a folder and everything under it — OUR LOCKS EXCEPT.

    A booth's age is how long since somebody touched it, and a lock sidecar is
    machinery: `marks.py` and `links.py` each create one on the way into a
    read-modify-write, including one that turns out to change nothing. Counting
    it made reading-through-a-write-path look like activity, and a no-op mark
    POST on a dead booth reset its clock.

    The exclusion is `.<something>.lock` — a DOTfile, which is the Booth's own
    namespace. An agent that posts a real artifact called `build.lock` still
    gets its clock counted. Everything else counts too, dotfiles included,
    because `.marks.json`, `.blurred` and `.pins` are the operator doing
    something.

    ⚠ A STAT WE CANNOT DO READS AS *FRESH*, NEVER AS EPOCH-OLD. This function
    feeds `is_expired`, which feeds `rmtree`. Returning 0.0 for a booth whose
    own stat fails made it maximally ancient and therefore the FIRST thing the
    sweeper takes — a permissions or ELOOP problem resolving to a deletion. The
    bug-hunt panel found this as one of four paths into the same shape. Not
    knowing a booth's age is a reason to leave it alone.

    `FileNotFoundError` on an entry is the exception, and it stays a skip: a
    dangling symlink and a file removed mid-scan both raise it, and neither is
    a thing with an mtime worth counting. Any OTHER per-entry OSError means we
    could not read something that IS there, so the age is unknowable and the
    booth reads as fresh.
    """
    now = time.time()
    try:
        newest = path.stat().st_mtime
    except OSError:
        return now
    for p in path.rglob("*"):
        if p.name.startswith(".") and p.name.endswith(".lock"):
            continue
        try:
            m = p.stat().st_mtime
        except FileNotFoundError:
            continue                      # dangling symlink, or gone mid-scan
        except OSError:
            return now                    # cannot read it — cannot judge the age
        if m > newest:
            newest = m
    return newest


def booth_age_seconds(path: Path, now: float | None = None) -> float:
    now = time.time() if now is None else now
    return now - _newest_mtime(path)


def is_expired(path: Path, ttl_seconds: float, now: float | None = None) -> bool:
    """Pure age question. Deliberately does NOT consider the keep sentinel.

    Expiry arithmetic (what `expires_in` renders) and reaper policy (what
    actually gets deleted) are kept apart so they cannot drift into each other.
    Only `sweep_once` honours the pin.
    """
    return booth_age_seconds(path, now) > ttl_seconds


def is_kept(path: Path) -> bool:
    """True if this booth carries the keep sentinel and must never be swept.

    `lstat`, not `Path.exists()`, and an unreadable answer counts as KEPT. The
    old form collapsed ELOOP and EACCES into False, so a kept booth whose
    sentinel could not be stat'd became eligible for the sweep — a failed read
    authorizing a delete, which is the shape the bug-hunt panel found four ways
    into. `lstat` also means a `.forever` SYMLINK counts, dangling or not:
    somebody put it there to mean keep.
    """
    try:
        (path / KEEP_MARKER).lstat()
        return True
    except FileNotFoundError:
        return False
    except OSError:
        return True


def record_view(booth: Path) -> None:
    """Note that somebody deliberately looked at this booth (U4).

    Touches VIEW_MARKER and lets `_newest_mtime` do the rest — a view enters
    the age rule as a file in the tree, not as a new term in the arithmetic.

    NEVER RAISES. A read-only mount, a booth owned by another uid, a full disk,
    a booth deleted between the route's resolve and this call: every one of
    those costs the timestamp, not the page. The same trade `_Locked.__enter__`
    makes on its `os.utime`, and for the same stated reason — not recording the
    look is a cost this service can absorb, not answering the request is not.
    A booth whose view cannot be recorded simply ages on its content mtime,
    which is what every booth did before this existed.
    """
    # O_NOFOLLOW, not `Path.touch()`. `touch` on an existing symlink follows it,
    # so a booth carrying a planted `.viewed -> /anywhere` turned EVERY page
    # view into an mtime write at an arbitrary path under the service uid — and
    # any fleet session can write into a booth, because making a folder is the
    # whole API. Three of four bug-hunt arms found it independently. A symlink
    # here now raises ELOOP into the swallow below: view-recording quietly stops
    # for that booth, which is the right way to lose this argument.
    #
    # O_CREAT alone does not move the mtime of a file that already exists, so
    # the utime is not decoration: the marker must read as NOW or the whole
    # mechanism is a file nobody's clock looks at.
    try:
        fd = os.open(booth / VIEW_MARKER,
                     os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o644)
        try:
            os.utime(fd)
        finally:
            os.close(fd)
    except OSError:
        pass


HOLD_UNREADABLE = "unreadable"
HOLD_OPEN = "open"


def hold_reason(marks: Sequence[Mark], error: str | None) -> str | None:
    """WHY this booth must not be swept, or None if it may be. THE hold predicate.

    Returns a reason rather than a bool so the surface that has to say why can
    read it off the same value the sweeper acts on. A boolean plus a separate
    error string is two representations of one state, and they drift.

    PURE — it takes the result of a read and does none of its own, so the index
    card and the sweeper cannot answer differently about the same booth. That is
    U1's rule (one resolver, every surface reads the record) applied to lifetime.

    FAIL-SAFE ON BOTH LEVELS OF DAMAGE, which is the correction the bug-hunt
    panel forced (2026-09-22). `marks_for` is lenient because a review page that
    will not load is worse than one missing an annotation — the right trade for
    a RENDER and the wrong one for a DELETE, where the same leniency wipes the
    booth whose judgment we had just failed to read, artifacts and all. The
    first cut of this caught FILE-level damage only:

      * file-level — `.marks.json` will not parse at all. `hold_read` reports it.
      * ENTRY-level — the document parses, but one mark fails normalization and
        `_hydrate_safe` hands back a `Mark` carrying `error`. `_is_open` returns
        False for an errored pick, ON PURPOSE (a broken pick can never be
        answered; the CLI spells that exit code 4) — so such a booth read as
        `not held` and SWEPT, while the panel beside it rendered the broken mark
        in full. Four arms found four ways into that shape; this was the worst.

    A mark that cannot be read is judgment we cannot see. Deleting the booth it
    belongs to is the one thing we must not do with it.

    Openness itself is `open_marks` and nothing else (U2 INV-2): a partially
    answered pick is STILL open and still holds, which is the reading that
    makes this rule correct rather than one that sweeps a review in flight.
    """
    if error is not None or any(m.error is not None for m in marks):
        return HOLD_UNREADABLE
    if open_marks(marks):
        return HOLD_OPEN
    return None


def sweep_once(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[str]:
    """Wipe every direct-child booth older than the TTL. Returns names wiped.

    Only ever removes direct children of data_dir (never data_dir itself), and
    skips dotfolders so a stray control dir can opt out.

    TWO exemptions, and this is the only function that honours either.

    A booth carrying KEEP_MARKER is exempt no matter how stale it is. That is
    the explicit escape hatch, opt-in per booth: the default stays ephemeral, so
    nobody inherits a cleanup chore they did not ask for. Removing the sentinel
    hands the booth straight back to the sweeper.

    A booth that is HELD — an open pick, or marks we cannot read — is exempt for
    as long as that holds (U4). This is the derived half: the operator was
    pressing `.forever` to mean "not yet" because nothing else could say it, and
    the service already knew. 17 of 24 live booths carried the sentinel on
    2026-09-22, and three of the four booths in the fleet awaiting an answer
    carried it too — the "not yet" case, caught in the act.

    Reading the marks costs ONE strict read per booth per tick — `hold_read`,
    which answers both halves of the hold question at once. It is deliberately
    not two calls: two reads of one file are not one read of one state, and the
    pair that loses that race is the pair that deletes. Do not "optimize" this
    back into `marks_for` plus `read_error`.
    """
    wiped: list[str] = []
    if not data_dir.is_dir():
        return wiped
    for child in data_dir.iterdir():
        if not child.is_dir() or child.name.startswith("."):
            continue
        try:
            if is_kept(child):
                continue
            if hold_reason(*hold_read(child)):   # ONE read — see hold_read
                continue
            if is_expired(child, ttl_seconds, now):
                shutil.rmtree(child)
                wiped.append(child.name)
        except OSError:
            pass
    return wiped


def list_booths(data_dir: Path, ttl_seconds: float, now: float | None = None) -> list[dict]:
    """One card per booth for the index.

    Counts and the cover come from `booth_items`, so the index agrees with the
    booth page about what an item IS. It did not before: this function counted
    every non-dot, non-ask file, which meant a caption sidecar was counted as an
    item here and (correctly) not counted there — an A/B pair with two captions
    read as "4 items" on the index and showed two tiles when you opened it.

    Doc bodies are deliberately NOT rendered — see `booth_items`. The index
    touches every booth on every page load.
    """
    now = time.time() if now is None else now
    booths: list[dict] = []
    if not data_dir.is_dir():
        return booths
    for child in data_dir.iterdir():
        if not child.is_dir() or child.name.startswith("."):
            continue
        items = booth_items(child)
        # Marks are judgment, not items: counted separately so the index can
        # flag a booth that is waiting on the operator. ONE file read per booth
        # — which is why marks live in one file per booth rather than a sidecar
        # per mark. This loop runs on every index page load.
        # ONE read for BOTH the badge and the lifetime decision. It has to be
        # one: `marks_for` is lenient, so an unreadable `.marks.json` reads as
        # no marks — fine for a card, wrong for the reaper, which would then
        # delete the booth whose judgment it had just failed to read. And
        # asking the two questions with two reads is not one read of one state:
        # a write landing between them yields `([], None)`, the pair that
        # deletes. `hold_read` answers both from one read; the lenient reader
        # comes back only on the error path, where leniency is the point.
        held_marks, read_err = hold_read(child)
        # The DECISION comes from that one read and nothing else. The lenient
        # re-read below is for DISPLAY only — feeding it back into the predicate
        # would rebuild the two-read seam this call exists to close.
        hold = hold_reason(held_marks, read_err)
        marks = held_marks if read_err is None else marks_for(child)
        # The booth's own announcement — who posted it and why. One more small
        # read per booth, beside the marks read already here, and `read_manifest`
        # cannot raise for the same reason `marks_for` must not: this loop runs
        # over EVERY booth on every index page load.
        manifest = read_manifest(child)
        kinds = {"image": 0, "video": 0, "audio": 0, "other": 0}
        thumb_url = None
        thumb_blurred = False
        for it in items:
            kinds[it.kind] += 1
            if it.kind == "image" and thumb_url is None:
                thumb_url = it.url
                # If the cover image is blurred inside the booth, blur it on the
                # index too — otherwise the front page cheerfully displays the
                # exact thing someone asked to hide. Read off the record now,
                # rather than re-opening .blurred here.
                thumb_blurred = it.blurred
        mtime = _newest_mtime(child)
        booths.append(
            {
                "name": child.name,
                "name_url": quote(child.name, safe=""),
                "manifest": manifest,
                "count": len(items),
                "kinds": kinds,
                "thumb_url": thumb_url,
                "thumb_blurred": thumb_blurred,
                "has_index": (child / "index.html").is_file(),
                "uploaded": (child / UPLOAD_MARKER).exists(),
                "kept": is_kept(child),
                "marks_total": len(marks),
                # `open_marks` and nothing else (INV-2). The count this replaced
                # tested `answer is None`, so a half-answered pick read as closed
                # here while the panel beside it rendered `◐ partial`.
                "marks_open": len(open_marks(marks)),
                # U4: WHY this booth is or is not counting down. The card must
                # never just stop the clock silently — `.forever` was at least
                # visible as a lane, and an invisible rule would be worse than
                # the boolean it replaces.
                # WHY it is or is not counting down — the reason, not a bool
                # beside a string that can disagree with it.
                "hold": hold,
                "expires_in": max(0.0, ttl_seconds - (now - mtime)),
                "mtime": mtime,
            }
        )
    # Newest first, NAME as the tie-break. Sorting on mtime alone left equal-mtime
    # booths ordered by whatever `iterdir()` yielded, which is not a rule — and
    # invariant 6 is not "usually stable", it is a sentence you can write down.
    # Two booths created by one `rsync` batch share an mtime exactly, and the
    # operator refers to cards positionally.
    booths.sort(key=lambda b: (b["mtime"], b["name"]), reverse=True)
    return booths


def build_gallery(child: Path) -> list[dict]:
    """The gallery's render dicts — a thin adapter over `booth_items`.

    The resolver owns every fact about an item; this only shapes them for the
    template and pulls doc bodies for the one surface that inlines them. Kept as
    a function (rather than inlined at the call site) because the existing test
    suite reaches for it by name in nine places.
    """
    out = []
    for it in booth_items(child):
        body = render_doc_body(child, it)
        rendered, rendered_html = body if body is not None else (None, False)
        out.append(
            {
                "name": it.rel,
                "kind": it.kind,
                "doc": it.doc,
                "url": it.url,
                "section": it.section,
                # U7. Derived in the resolver (INV-1); this only carries it.
                "group": it.group,
                "caption": it.caption,
                "rendered": rendered,
                "rendered_html": rendered_html,
                "blurred": it.blurred,
            }
        )
    return out


def zip_booth(booth: Path) -> bytes:
    """Zip a booth's whole tree (dotfiles excluded) into an in-memory archive.

    Lets a booth be downloaded as one artifact regardless of shape — the case a
    verbatim `index.html` booth (e.g. a rendered brief + its assets) has no
    per-file download affordance for, since the page is served raw.
    """
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in sorted(booth.rglob("*")):
            if p.is_file() and not p.name.startswith("."):
                zf.write(p, p.relative_to(booth).as_posix())
    return buf.getvalue()


def _zip_filename(name: str) -> str:
    """A Content-Disposition-safe `<booth>.zip` (strip quotes/control chars)."""
    safe = "".join(c for c in name if c.isprintable() and c != '"')
    return f"{safe or 'booth'}.zip"


# ---- the declared embed seam (U3) ------------------------------------------

# Mirror of base.html's favicon. The app templates set it there; this copy is
# what `/b/<name>/embed.json` hands to a VERBATIM report, so a raw page inherits
# the same icon. Keep the two in sync if the Booth's icon ever changes.
FAVICON_HREF = (
    "data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 32 32'"
    "%3E%3Crect width='32' height='32' rx='7' fill='%23171a23'/%3E%3Ccircle cx='16' "
    "cy='16' r='6' fill='none' stroke='%2342dcd1' stroke-width='2.5'/%3E%3Ccircle "
    "cx='16' cy='16' r='2.2' fill='%2342dcd1'/%3E%3C/svg%3E"
)

# The seam a verbatim report declares to get the Booth's chrome. ONE line, and
# the Booth appends it only when the page has not declared it itself.
EMBED_SRC = "/_booth/embed.js"
EMBED_SCRIPT_TAG = f'<script src="{EMBED_SRC}" defer></script>'
EMBED_JS_PATH = Path(__file__).parent / "static" / "embed.js"

# What counts as DECLARING the seam. Two substring tests, one per quote style,
# and each requires `src=` immediately before the path.
#
# The bare path was the first draft and it was wrong in the dangerous
# direction. A report that merely MENTIONS `/_booth/embed.js` — in a code
# sample, a comment, a sentence about this very feature — would have been read
# as declaring it, served untouched, and silently shown no chrome at all. The
# Booth's own design reports are exactly the pages that would quote it.
#
# These tests fail in the harmless direction instead. An unusual spelling
# (`src = "…"` with spaces, an unquoted attribute, a `?v=2` suffix) is read as
# NOT declared, so a second tag is appended — and embed.js mounts once
# regardless, because it guards on `window.__boothEmbed`. A missed declaration
# costs a duplicate tag; a false one costs the operator his chrome.
_EMBED_DECLARATIONS = (f'src="{EMBED_SRC}"', f"src='{EMBED_SRC}'")

WRAP_MAX_BYTES = 8 * 1024 * 1024  # above this, serve the verbatim page raw


def declares_embed(html: str) -> bool:
    """Whether a verbatim page already asks for the Booth's chrome.

    TWO SUBSTRING TESTS. This is the entire detection half of what used to be
    six regular expressions run against arbitrary author HTML — and the other
    half, the insertion, is a `+`. See `_EMBED_DECLARATIONS` for why it matches
    `src="…"` rather than the bare path: both spellings fail toward appending a
    harmless duplicate rather than toward silently withholding the chrome.
    """
    return any(d in html for d in _EMBED_DECLARATIONS)


def embed_verbatim(raw: bytes) -> bytes:
    """The ONLY thing the Booth does to a verbatim report. BYTES IN, BYTES OUT.

    Appended, never inserted, and never prepended. That is what retires both of
    the old wrapper's hard constraints rather than satisfying them more
    carefully: nothing can displace a leading doctype into quirks mode and
    nothing can push the charset <meta> out of its first-1024-byte detection
    window, because nothing in front of them moves. Content after `</html>` is
    parsed into the body by every browser, so there is no seam to find.

    ⚠ IT TAKES BYTES BECAUSE TEXT WAS QUIETLY EDITING THE DOCUMENT. The first
    version read the file with `read_text()` and returned a str. That opens in
    UNIVERSAL-NEWLINE mode, so a report written with CRLF came back with LF —
    and `errors="replace"` turned any byte that was not valid UTF-8 into U+FFFD.
    A declaring page was therefore NOT served as its author wrote it, which is
    this unit's headline promise, and the test could not see it because its
    fixture was LF-only ASCII. Found by a cross-frontier bug-hunt panel.

    Decoding still happens — `declares_embed` needs a string to look in — but
    the decoded copy is used ONLY to answer that question. What goes on the wire
    is the original bytes, plus the tag's bytes when it is appended, so the
    source is a byte-exact prefix of the response.
    """
    text = raw.decode("utf-8", errors="replace")
    return raw if declares_embed(text) else raw + EMBED_SCRIPT_TAG.encode("utf-8")


def ask_form_id(stem: str) -> str:
    """The shared `<form>` id a pick's scattered question groups bind to with
    the HTML5 `form=` attribute.

    Moved here from `booth/inline.py` when U3 deleted that module: it is not
    placement machinery, it is what makes four radio groups spread down a report
    submit as ONE POST, which is what a multi-question ask requires.
    """
    return f"bk-ask-form-{re.sub(r'[^A-Za-z0-9_-]', '-', stem)}"


# ---- uploads (browser drop-off for pickup) ---------------------------------

UPLOAD_MARKER = ".uploaded"  # dotfile stamped into upload booths (excluded from listings)

# Friendly, unambiguous words for human-readable pickup ids (4-wombat / star-84).
PICKUP_WORDS = (
    "wombat otter panda koala tiger walrus gecko heron badger beaver falcon marmot "
    "lemur narwhal ocelot puffin quokka raccoon tapir urchin vulture weasel yak zebra "
    "alpaca bison cobra dingo egret ferret gibbon hare ibis jaguar llama moose newt "
    "osprey possum quail robin seal toad viper wren lynx mole swan crane finch sloth "
    "shrew stoat skunk heronry orca walnut sparrow "
    "star comet moon cloud river maple cedar birch fern moss reef dune mesa cove glade "
    "brook pine cedarwood kelp coral amber opal jade onyx slate flint ember spark frost "
    "storm tide wave ridge peak vale marsh delta atoll canyon fjord geyser lagoon prairie "
    "anchor beacon lantern kettle copper brass velvet cobalt indigo crimson violet olive "
    "hazel cocoa mango guava papaya plum kiwi lime pear quince radish turnip acorn clover "
    "thistle poppy aster dahlia iris lily sage thyme basil clove nutmeg ginger honey"
).split()


def _form_text(form, key: str) -> str:
    """One form field as text, or "" for anything that is not text.

    A multipart FILE part named `notes` parses to an UploadFile, not a str, and
    every downstream cleaner calls `.replace` on what it is handed. Coercing
    here keeps that decision in one place instead of one `isinstance` per call
    site — which is how `/note` came to have the guard and `/answer` not to.
    """
    value = form.get(key)
    return value if isinstance(value, str) else ""


def safe_upload_name(name: str, fallback: str) -> str:
    """Reduce a client-supplied filename to a safe basename (no path, no hidden)."""
    base = (name or "").replace("\\", "/").split("/")[-1].strip()
    base = base.lstrip(".")  # a leading dot would hide the file from every listing
    return base[:200] or fallback


def _dedupe_name(name: str, used: set) -> str:
    if name not in used:
        return name
    stem, dot, ext = name.partition(".")
    i = 1
    while f"{stem}-{i}{dot}{ext}" in used:
        i += 1
    return f"{stem}-{i}{dot}{ext}"


def generate_pickup_id(exists) -> str:
    """A human-readable id like '4-wombat' or 'star-84'. `exists(name)->bool` gates collisions."""
    for _ in range(400):
        word = secrets.choice(PICKUP_WORDS)
        num = secrets.randbelow(99) + 1
        name = f"{num}-{word}" if secrets.randbelow(2) else f"{word}-{num}"
        if not exists(name):
            return name
    # astronomically unlikely fallback: two words keep it human-readable
    while True:
        name = f"{secrets.choice(PICKUP_WORDS)}-{secrets.choice(PICKUP_WORDS)}-{secrets.randbelow(999) + 1}"
        if not exists(name):
            return name


def create_app(
    data_dir,
    ttl_hours: float = 24.0,
    host_label: str = "",
    start_sweeper: bool = True,
    sweep_interval_s: int = 900,
    max_upload_mb: float = 1024.0,
    max_files: int = 50,
) -> FastAPI:
    data_dir = Path(data_dir).expanduser().resolve()
    data_dir.mkdir(parents=True, exist_ok=True)
    ttl_seconds = ttl_hours * 3600.0
    max_upload_bytes = int(max_upload_mb * 1024 * 1024)

    # TEMPLATES ARE CACHED AT STARTUP, DELIBERATELY — `auto_reload=False`.
    #
    # `booth.service` runs uvicorn with WorkingDirectory set to this repo, so the
    # repo IS the deployment root: there is no build step and no staging copy.
    # Jinja's default FileSystemLoader re-reads a template from disk on every
    # render, while the Python stays as it was when the process started. That
    # gives the two halves of the service different staleness rules, and editing
    # a template deploys it INSTANTLY against Python that may know nothing about
    # the context it wants.
    #
    # It cost an outage on 2026-09-21: 19 of 25 live booths returned 500 with
    # `UndefinedError: 'item_marks' is undefined` — new markup, old context, both
    # running at once, and neither version broken on its own. The Python had
    # started at 22:03 and the templates were from 23:40.
    #
    # With reload off there is ONE rule — nothing takes effect until you restart
    # — so the running process is always a coherent snapshot of one commit. The
    # price is that template work needs a `systemctl --user restart
    # booth.service` to see; that price is the whole point.
    env = Environment(
        loader=FileSystemLoader(str(TEMPLATES_DIR)),
        autoescape=select_autoescape(["html", "xml"]),
        auto_reload=False,
    )
    env.filters["dur"] = human_dur
    templates = Jinja2Templates(env=env)

    # embed.js IS READ ONCE, HERE, for exactly the reason above. It is the third
    # kind of thing this repo serves, and the only one that would otherwise be
    # free to hot-reload from the deployment root — which is the skew that put
    # 19 of 25 booths at 500. One rule: nothing takes effect until you restart.
    embed_js = EMBED_JS_PATH.read_text(encoding="utf-8")
    embed_etag = '"%s"' % hashlib.sha256(embed_js.encode("utf-8")).hexdigest()[:16]

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        task = None
        if start_sweeper:

            async def loop():
                while True:
                    try:
                        wiped = sweep_once(data_dir, ttl_seconds)
                        if wiped:
                            print(f"[booth] swept {len(wiped)} expired: {', '.join(wiped)}", flush=True)
                    except Exception as exc:  # never let the sweeper die
                        print(f"[booth] sweep error: {exc}", flush=True)
                    await asyncio.sleep(sweep_interval_s)

            task = asyncio.create_task(loop())
        try:
            yield
        finally:
            if task is not None:
                task.cancel()

    app = FastAPI(title="The Booth", lifespan=lifespan)
    # The template environment, reachable for assertion: the snapshot property
    # above is a promise about the DEPLOYED service, so it needs a test, and a
    # test needs a handle on the env that the app actually renders with.
    app.state.templates = templates

    @app.exception_handler(MarksCorrupt)
    async def _marks_corrupt(request: Request, exc: MarksCorrupt):
        """A write was refused because the booth's mark file is damaged.

        409, not 500: the service is fine and the request was well-formed — the
        state on disk is not, and the refusal is deliberate. Says what to do,
        because the alternative the operator will otherwise reach for is
        deleting the file, which is the thing being protected.
        """
        return JSONResponse(
            status_code=409,
            content={
                "error": "this booth's .marks.json cannot be read, so nothing was written",
                "detail": str(exc),
                "why": "writing would replace every mark in the booth with just this one",
                "fix": "repair or move the file by hand; the marks panel still renders as empty",
            },
        )

    ttl_display = int(ttl_hours) if float(ttl_hours).is_integer() else ttl_hours
    base_ctx = {
        "ttl_hours": ttl_display,
        "host": host_label,
        "data_dir": str(data_dir),
        "keep_marker": KEEP_MARKER,  # shown in the kept lane so the mechanism is discoverable
    }

    def resolve_booth(name: str) -> Path:
        if not name or name.startswith(".") or "/" in name or "\\" in name or ".." in name:
            raise HTTPException(status_code=404, detail="no such booth")
        candidate = data_dir / name
        try:
            resolved = candidate.resolve()
        except OSError:
            raise HTTPException(status_code=404, detail="no such booth")
        # resolved.parent must be the data dir itself — blocks symlink escape + nesting.
        if resolved.parent != data_dir or not resolved.is_dir():
            raise HTTPException(status_code=404, detail="no such booth")
        return resolved

    @app.get("/", response_class=HTMLResponse)
    def index(request: Request):
        # Two lanes, split here rather than in the template: kept boards are a
        # different KIND of thing from the ephemeral churn — durable, deliberate,
        # operator-facing — and burying them in a feed that turns over daily is
        # exactly how they would get lost, which is the problem they exist to
        # solve. Kept renders first.
        everything = list_booths(data_dir, ttl_seconds)
        return templates.TemplateResponse(
            request,
            "index.html",
            {
                **base_ctx,
                "kept": [b for b in everything if b["kept"]],
                "booths": [b for b in everything if not b["kept"]],
            },
        )

    @app.get("/healthz")
    def healthz():
        return {"ok": True, "ttl_hours": ttl_hours, "booths": len(list_booths(data_dir, ttl_seconds))}

    @app.get(EMBED_SRC)
    def embed_script():
        """The declared seam's one static asset.

        Served from the startup read, with an ETag over its content so a
        browser revalidates instead of holding a stale copy across a restart —
        `no-cache` here means "ask me", not "do not store".
        """
        return Response(
            content=embed_js,
            media_type="text/javascript; charset=utf-8",
            headers={"ETag": embed_etag, "Cache-Control": "no-cache"},
        )

    @app.get("/b/{name}", include_in_schema=False)
    def booth_redirect(name: str):
        resolve_booth(name)
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=307)

    @app.get("/b/{name}/", response_class=HTMLResponse)
    def booth_view(request: Request, name: str, download: int = 0, filter: str = "all"):
        booth = resolve_booth(name)
        # U4: viewing is activity. ABOVE both early returns — the zip download
        # and the verbatim-index.html branch are looks at this booth too, and a
        # verbatim report is the shape the operator stares at longest.
        record_view(booth)
        if download:
            # whole-booth zip — the download path for a verbatim index.html booth
            # (which has no gallery/per-file chrome), and a "download all" for any.
            return Response(
                content=zip_booth(booth),
                media_type="application/zip",
                headers={"Content-Disposition": f'attachment; filename="{_zip_filename(name)}"'},
            )
        own_index = booth / "index.html"
        if own_index.is_file():
            # The operator's verbatim report. U3: the page declares the seam and
            # the Booth mounts into it — so a page carrying the script tag is
            # served exactly as written, and one that is not gets that single
            # line appended. Nothing is parsed, matched or inserted.
            #
            # The read is still bounded: a pathological file falls back to
            # serving raw, which costs it the chrome exactly as it did before.
            try:
                if own_index.stat().st_size <= WRAP_MAX_BYTES:
                    # ONE read, and it is a byte read: see embed_verbatim.
                    return Response(
                        content=embed_verbatim(own_index.read_bytes()),
                        media_type="text/html; charset=utf-8",
                    )
            except OSError:
                pass
            return FileResponse(str(own_index), media_type="text/html")
        # links.md is rendered AS the board below, so it must not also appear as
        # a markdown doc tile — that would show the same content twice, once
        # interactive and once not.
        gallery = [
            it for it in build_gallery(booth)
            if not ((booth / LINKS_FILE).is_file() and it["name"] == LINKS_FILE)
        ]
        held_marks, read_err = hold_read(booth)  # ONE read; see list_booths
        hold = hold_reason(held_marks, read_err)
        marks = held_marks if read_err is None else marks_for(booth)
        rail, shown = _rail(gallery, marks, filter)
        return templates.TemplateResponse(
            request,
            "booth.html",
            {
                **base_ctx,
                "name": name,
                "name_url": quote(name, safe=""),
                # The page could not previously tell keep from release, so it
                # offered neither and you had to go back to the index.
                "kept": is_kept(booth),
                # THE GRID RENDERS `shown`; everything else reads `gallery`.
                # Filtering is a VIEW: `shown` is `gallery` with non-matching
                # items removed and NOTHING re-sorted, so "the third one" means
                # the same thing with a filter on as with it off. Sorting by
                # anything filter-derived would look right and silently misfile
                # the operator's judgment — CLAUDE.md invariant 6.
                "items": shown,
                "all_items": gallery,
                "rail": rail,
                "filter": rail["active"],
                # A booth carrying links.md is the standing link board: render
                # its rows as real UI (link, provenance, pin, per-row + bulk
                # remove) instead of a markdown blob you can only edit by hand.
                # Ordered pinned-first then newest-first, each row stamped with a
                # `pinned` flag. Empty list for every other booth, so the template
                # branch simply does not fire.
                # `is_file()` then an UNGUARDED read was a 500 waiting on a
                # mode change or an EIO: the board is one tile on this page, and
                # a page that will not load is worse than one missing a tile —
                # the same posture `read_blurred`, `marks_for` and
                # `read_manifest` already take. A booth whose `links.md` cannot
                # be read renders as a booth with no board.
                "board": _board_rows(booth),
                # The bench registry, rendered on the STANDING BOARD's page and
                # nowhere else: it belongs to exactly one booth, and a read per
                # gallery page view would buy noise. `_board_rows` is empty for
                # every other booth, so this pair is read only when it renders.
                # `read_benches` never raises; a damaged registry costs its own
                # panel and says so, which is the v0.2.2 lesson.
                # `is_board` is PAGE IDENTITY, not page content. Gating the
                # panel on `board or benches` hid it — and its registration
                # form — exactly when the board was empty and the registry
                # absent, which is the state a new deployment starts in and the
                # one where "no benches registered yet" is most worth saying.
                # A panel that disappears when it has nothing to show is the
                # same defect as a damaged panel rendering as an absent one.
                "is_board": (booth / LINKS_FILE).is_file(),
                **dict(zip(("benches", "benches_error"),
                           read_benches(data_dir) if (booth / LINKS_FILE).is_file()
                           else ([], None))),
                # Marks: operator judgment attached to this booth or to one of
                # its items — a session's question (`pick`), the operator's own
                # remark (`note`), the operator's selection (`flag`). Rendered
                # as the panel above the gallery, and per item on each tile.
                # See booth/marks.py.
                "marks": marks,
                "marks_open": len(open_marks(marks)),
                # Per-item marks, keyed by rel, so a tile reads its own judgment
                # without every tile re-filtering the whole list. Keyed off the
                # FULL gallery, not the filtered one, so a tile hidden by the
                # current filter still has its marks if the filter changes.
                "item_marks": {
                    it["name"]: marks_for_target(marks, it["name"]) for it in gallery
                },
                "booth_marks": marks_for_target(marks, None),
                "uploaded": (booth / UPLOAD_MARKER).exists(),
                # The same provenance line the index card carries. Deliberate:
                # a booth URL handed to the operator lands HERE, never on the
                # index, and job 5 is "operator, look at this".
                "manifest": read_manifest(booth),
                # The lifetime line, same three states as the index card: a
                # booth URL handed to the operator lands HERE, not on the index,
                # so "why is this not counting down" has to be answerable here.
                "hold": hold,
                "expires_in": max(0.0, ttl_seconds - booth_age_seconds(booth)),
            },
        )

    FILTERS = ("all", "flagged", "annotated", "unanswered")

    def _rail(gallery: list[dict], marks, requested: str) -> tuple[dict, list[dict]]:
        """Per-filter counts, and the items the grid should render.

        `requested` ARRIVES FROM A URL, which is operator-editable and
        link-shared, so an unknown value falls back to `all` rather than
        indexing a dict by it. A filter nobody can mistype into a 500.

        `unanswered` means HAS AN OPEN PICK — the U4 hold predicate, which
        already exists and already has a home. The other reading ("has no mark
        at all") is a genuinely different question and is an open question on
        the U7 contract, not something to guess at here.
        """
        active = requested if requested in FILTERS else "all"
        open_ids = {m.id for m in open_marks(marks)}
        buckets: dict[str, list[dict]] = {f: [] for f in FILTERS}
        for it in gallery:
            mine = marks_for_target(marks, it["name"])
            buckets["all"].append(it)
            if any(m.shape == "flag" and m.flagged for m in mine):
                buckets["flagged"].append(it)
            if any(m.shape == "note" for m in mine):
                buckets["annotated"].append(it)
            if any(m.id in open_ids for m in mine):
                buckets["unanswered"].append(it)
        shown = buckets[active]
        rail = {
            "active": active,
            # ORDER: the declaration order of FILTERS. Stated because a rail is
            # an ordered collection and invariant 6 binds to it like any other.
            "counts": [{"key": f, "n": len(buckets[f])} for f in FILTERS],
            "total": len(gallery),
            "groups": _groups(shown),
        }
        return rail, shown

    def _groups(shown: list[dict]) -> list[dict]:
        """The jump-to-group rows, or [] when grouping would not help.

        DERIVED FROM `shown`, NOT FROM THE FULL GALLERY, so every anchor lands
        on a tile the page actually rendered. A row pointing at an item the
        current filter has hidden scrolls nowhere, which is the same defect as
        a wrong id arriving by a different route.

        ORDER: the position of each group's FIRST member in the rendered
        sequence — which is `sorted(rel)` narrowed by the filter and never
        re-sorted. So the rail reads in the direction the grid does, and adding
        a file reshuffles nothing unless it lands first in its group. Settled by
        the operator 2026-09-22; ROADMAP carries the row. `dict` preserves
        insertion order, so walking `shown` once IS the rule.

        ⚠ THE RAIL IS ABSENT UNLESS GROUPING IS INFORMATIVE: two or more
        groups, and the middle group holding more than one item. TWO
        degeneracies, not one. The contract named only the first --
        `sindra` and `sc-iso-spread` put every file in ONE group, and a rail
        with a single row cannot navigate. The second is the one the live set
        actually exhibits: `pewpew-ui-brief` yields 23 groups for 34 items and
        `dfa-concepts` 13 for 20, a rail that is a second copy of the grid.
        Both render as no rail, because a navigation affordance that cannot
        navigate is worse than none -- it occupies the space where the real one
        would be.
        """
        by_group: dict[str, list[dict]] = {}
        for it in shown:
            if it["group"] is not None:
                by_group.setdefault(it["group"], []).append(it)
        sizes = sorted(len(v) for v in by_group.values())
        if len(sizes) < 2 or sizes[len(sizes) // 2] <= 1:
            return []
        return [
            # The anchor is the FIRST member's existing tile id. The template
            # already stamps `id="item-<rel>"` on every figure; minting a
            # parallel `#group-<key>` would be a second identity for one tile.
            {"key": k, "n": len(v), "anchor": f"item-{v[0]['name']}"}
            for k, v in by_group.items()
        ]

    def _board_rows(booth: Path) -> list[dict]:
        """The link board's rows, or [] for a board that cannot be read.

        NEVER RAISES, for the reason every other read on this page does not:
        one damaged file must cost its own tile, not the booth page."""
        try:
            if not (booth / LINKS_FILE).is_file():
                return []
            rows = order_for_display(
                parse_link_entries((booth / LINKS_FILE).read_text()),
                read_pins(booth),
            )
            # DEAD = the row points at a booth that no longer exists. 156 of the
            # board's 221 rows are exactly that, and nothing on the page could
            # tell them apart, so the bulk-delete control that has existed since
            # before this unit was unusable at that scale. Marking is all this
            # does: removal stays the operator's two deliberate clicks, because
            # "a migration that deletes anything" is not in v1.
            for row in rows:
                target = booth_target(row["url"])
                row["dead"] = target is not None and not _booth_exists(target)
            return rows
        except (OSError, ValueError, UnicodeDecodeError):
            return []

    def _booth_exists(name: str) -> bool:
        """Whether a booth name is a live directory. NEVER RAISES.

        SEAM REVIEW SR-2: this deliberately does NOT call `resolve_booth`, which
        raises HTTPException(404) — called once per board row, one swept booth
        would 404 the whole page, which is the opposite of the marker's purpose.
        `booth_target` has already applied the same addressability rules
        `resolve_booth` enforces, so the two cannot disagree about what is
        reachable; all that is left is the existence check itself.

        Cost: one stat per booth-shaped row per render of the standing board —
        178 of 221 rows today, on the ONE booth that carries a links.md.
        """
        try:
            candidate = (data_dir / name).resolve()
            # THE SAME CONTAINMENT `resolve_booth` ENFORCES. Without it the two
            # disagree on a symlink: the marker would call a booth pointing
            # outside the data root ALIVE while the page 404s it, so the row
            # renders healthy and the link is dead — the worst of both, and
            # invisible. 3-of-4 cold bug-hunt arms found the disagreement.
            return candidate.parent == data_dir and candidate.is_dir()
        except (OSError, ValueError):
            # ValueError, not only OSError: an embedded NUL raises it rather
            # than an OSError, and this predicate runs once per board row — one
            # bad row must never cost the other 220.
            return False

    def _mark_redirect(name: str, form, anchor: str) -> RedirectResponse:
        """Land where the form was: the standalone marks page for a verbatim
        booth (its own index.html cannot show the recorded judgment), else the
        booth page, scrolled to the mark that was just written."""
        base = f"/b/{quote(name, safe='')}/"
        if form.get("back") == "marks":
            base = f"/b/{quote(name, safe='')}/marks"
        return RedirectResponse(url=f"{base}#{anchor}", status_code=303)

    @app.post("/b/{name}/answer")
    async def booth_answer(request: Request, name: str):
        """Record the operator's pick — one of N options a session declared in
        advance. Validates every choice against the declaration and rewrites
        `.marks.json` atomically. Re-submitting overwrites: the mark is the
        CURRENT judgment, not a log.

        Form fields: `ask` (the mark id); single-question → `choice` + `notes`;
        multi-question → `choice.<key>` per question, optional `notes.<key>`,
        plus the form-level `notes`. 404 for an unknown id, 400 for a missing
        choice or one the declaration does not offer.

        Kept at `/answer` with an `ask` field rather than renamed: the inline
        fragments a report author has already marked up POST here, and breaking
        every landed verbatim report to tidy a URL is not a trade worth making.
        """
        booth = resolve_booth(name)
        form = await request.form()
        mark_id = form.get("ask")
        if not isinstance(mark_id, str) or not valid_stem(mark_id):
            raise HTTPException(status_code=404, detail="no such pick")
        spec = next((m for m in marks_for(booth) if m.id == mark_id and m.shape == "pick"), None)
        if spec is None:
            raise HTTPException(status_code=404, detail="no such pick")
        if spec.error is not None:
            raise HTTPException(status_code=400, detail=spec.error)
        who = request.client.host if request.client else ""
        # `notes` is whatever the form parser yielded. A multipart FILE part
        # named `notes` is an UploadFile, and `_clean_notes` calls `.replace` on
        # it — a 500 on hostile-but-legal input, where the sibling `/note` route
        # returns 400 for exactly the same class of value. Same parser, same
        # question, one answer.
        notes = _form_text(form, "notes")
        try:
            if spec.multi:
                choice = {q["key"]: _form_text(form, f"choice.{q['key']}")
                          for q in spec.questions}
                qnotes = {q["key"]: _form_text(form, f"notes.{q['key']}")
                          for q in spec.questions}
                await run_in_threadpool(answer_pick, booth, mark_id, choice, notes,
                                        who=who, qnotes=qnotes)
            else:
                # `choice` through the same reader as `notes`. It was raw, so a
                # multipart FILE part named `choice` reached the answer builder
                # as an UploadFile — the asymmetry that had already been fixed
                # once on the field beside it.
                await run_in_threadpool(answer_pick, booth, mark_id,
                                        _form_text(form, "choice"), notes, who=who)
        except AskError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        return _mark_redirect(name, form, f"mark-{quote(mark_id, safe='')}")

    @app.post("/b/{name}/note")
    async def booth_note(request: Request, name: str):
        """Attach free text to one item, or to the booth itself.

        The operator telling the session — a direction that had no mechanism at
        all before marks, which is exactly why it was running through chat.
        `target` empty or absent means the booth. 400 on empty text.
        """
        booth = resolve_booth(name)
        form = await request.form()
        raw_target = form.get("target")
        target = raw_target if isinstance(raw_target, str) and raw_target else None
        text = form.get("text")
        try:
            mark = await run_in_threadpool(
                write_note, booth, target, text if isinstance(text, str) else "",
                who=request.client.host if request.client else "")
        except AskError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        return _mark_redirect(name, form, f"mark-{quote(mark.id, safe='')}")

    @app.post("/b/{name}/flag")
    async def booth_flag(request: Request, name: str):
        """Flag or unflag one item — the operator pointing at the good ones.

        The shape that makes a 270-image booth tractable, and the one that
        closes the loop `golden-candidates` / `sindra-finalists` / the
        `pancake-*` ladders were running through conversation.
        """
        booth = resolve_booth(name)
        form = await request.form()
        target = form.get("target")
        if not isinstance(target, str) or not target:
            raise HTTPException(status_code=400, detail="a flag needs a target")
        on = str(form.get("on", "1")) not in ("0", "", "false", "off")
        try:
            await run_in_threadpool(
                set_flag, booth, target, on,
                who=request.client.host if request.client else "")
        except AskError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        return _mark_redirect(name, form, f"item-{quote(target, safe='')}")

    @app.post("/b/{name}/unmark")
    async def booth_unmark(request: Request, name: str):
        """Withdraw one mark — the operator's undo. Withdrawing a judgment is
        his to do; nothing else here removes a mark."""
        booth = resolve_booth(name)
        form = await request.form()
        mark_id = form.get("mark")
        if not isinstance(mark_id, str) or not mark_id:
            raise HTTPException(status_code=400, detail="which mark?")
        await run_in_threadpool(delete_mark, booth, mark_id)
        return _mark_redirect(name, form, "marks")

    @app.post("/b/{name}/import-asks")
    async def booth_import_asks(request: Request, name: str):
        """Import this booth's legacy `*.ask.json` sidecars into `.marks.json`.

        Idempotent, and it deletes nothing — the sidecars stay on disk. Exposed
        as a route as well as a CLI verb so a booth that predates marks can be
        migrated from the page you are already looking at.
        """
        booth = resolve_booth(name)
        await run_in_threadpool(import_legacy_asks, booth)
        form = await request.form()
        return _mark_redirect(name, form, "marks")

    _frag = templates.env.get_template("_ask_inline.html").module

    def _pick_fragments(name: str, mark) -> dict:
        """One pick, rendered into the pieces a page can mount independently.

        Rendered HERE, by the same Jinja macros the gallery page uses, so there
        is exactly ONE renderer of an ask. embed.js places these; it never
        builds one. A second renderer in JavaScript is the shape INV-1 was
        written to stop after the zoom view re-derived an item and lost its
        captions doing it.
        """
        url = quote(name, safe="")
        fid = ask_form_id(mark.id)
        if mark.error:
            # `whole` renders the broken-ask box. A question the session
            # believes it posted has to be visible; the pieces of a pick that
            # could not be read do not exist to offer.
            return {"id": mark.id, "error": mark.error,
                    "whole": str(_frag.whole(mark, fid, url)), "submit": "",
                    "questions": []}
        return {
            "id": mark.id,
            "error": None,
            "whole": str(_frag.whole(mark, fid, url)),
            "submit": str(_frag.submit(mark, fid, url)),
            # A LIST, not an object keyed by question key: a single-question
            # pick normalizes to one question whose key is None, which JSON
            # would write as the string "null" and so invent a name. The list
            # also carries declaration order in the format itself.
            "questions": [
                {"key": q.get("key"), "html": str(_frag.question(mark, q, fid, url))}
                for q in mark.questions
            ],
        }

    def _safe_fragments(name: str, mark) -> dict:
        """`_pick_fragments`, with the promise that it cannot raise.

        `marks_for` hydrates an entry whose JSON is well-formed but whose SHAPE
        is wrong — `{"answer": {"answers": []}}` survives `_hydrate` with no
        error and then raises `UndefinedError` in the template, because the
        macro asks a list for `.get`. Verified, not assumed.

        This endpoint renders every pick in the booth on every page load of the
        operator's report, so one such entry would 500 the whole seam and the
        report would show no chrome at all — while `hold_read` reported the file
        as perfectly readable. Same leniency `_hydrate_safe` already applies one
        layer down, at the layer that actually renders: one unreadable pick
        costs that pick, never the page.
        """
        try:
            return _pick_fragments(name, mark)
        except Exception as exc:                     # noqa: BLE001 - deliberate
            broken = replace(mark, error=f"this question could not be rendered: {exc}")
            try:
                whole = str(_frag.whole(broken, ask_form_id(mark.id),
                                        quote(name, safe="")))
            except Exception:                        # noqa: BLE001 - deliberate
                # THE HANDLER MUST SURVIVE THE FAILURE IT IS HANDLING. The
                # fallback re-rendered through the SAME macro module that had
                # just raised, so when `whole` itself was the broken thing this
                # guard re-raised and took the report anyway — a guard that
                # only works when the failure is somewhere else. Found while
                # building a falsifier for the guard: the falsifier tripped it.
                # Plain text, escaped by the caller, no macro involved.
                whole = ""
            return {"id": mark.id, "error": broken.error,
                    "whole": whole, "submit": "", "questions": []}

    @app.get("/b/{name}/embed.json")
    def booth_embed_json(name: str):
        """Everything a verbatim report needs to mount the Booth's chrome.

        The READ half of the declared seam. `embed.js` fetches this and places
        what comes back; every decision — what a mark says, whether it is still
        open, what order the marks come in — is made here and never re-derived
        on the page.

        Marks are ordered `(created, id)`, which is what both readers below
        sort by. Questions are in declaration order. `open` is `open_marks`,
        the ONE openness predicate, so a half-answered multi-question pick
        counts as open here exactly as it does on the index badge.

        DOES NOT RECORD A VIEW. `booth_view` already did, above both of its
        early returns; counting a script's fetch of the page it is already on
        would reset the TTL on machinery rather than on the operator.

        The read is LENIENT and the status stays 200, copied from
        `/marks.json`: a damaged `.marks.json` must cost the chrome, never the
        operator's report. That is the v0.2.2 lesson.
        """
        booth = resolve_booth(name)
        marks, read_err = hold_read(booth)      # ONE read; see list_booths
        if read_err is not None:
            marks = marks_for(booth)
        picks = [m for m in marks if m.shape == "pick"]
        body = {
            "booth": name,
            # No `home`: the way-home chip mounts from a constant BEFORE this
            # fetch, so that a failed one still leaves the operator a way out.
            # Carrying the value anyway would put a second representation of it
            # on the wire for nothing to read.
            "favicon": FAVICON_HREF,
            # Picks only. It is also what keeps a flag's `flag:<target>` id —
            # the one mark id containing the separator an anchor spec splits
            # on — out of a payload whose specs split on the first colon.
            "marks": [_safe_fragments(name, m) for m in picks],
            "open": [m.id for m in open_marks(picks)],
        }
        if read_err is not None:
            body["marks"] = []
            body["error"] = "this booth's .marks.json cannot be read"
            body["detail"] = read_err
        return JSONResponse(body)

    @app.get("/b/{name}/marks", response_class=HTMLResponse)
    def booth_marks_page(request: Request, name: str):
        """The marks panel on its own page. Reachable from any booth, and the ONLY
        place a verbatim-index.html booth can show its marks — that page is served
        untouched by design, so the inline panel never renders there."""
        booth = resolve_booth(name)
        # U4: for a verbatim booth this IS the booth page. `/b/<n>/asks` is a
        # 308 into here, so the legacy URL records through this call and must
        # not get one of its own.
        record_view(booth)
        held_marks, read_err = hold_read(booth)  # ONE read; see list_booths
        hold = hold_reason(held_marks, read_err)
        marks = held_marks if read_err is None else marks_for(booth)
        return templates.TemplateResponse(
            request,
            "marks.html",
            {**base_ctx, "name": name, "name_url": quote(name, safe=""),
             "marks": marks, "marks_open": len(open_marks(marks)),
             "booth_marks": marks_for_target(marks, None), "marks_page": True,
             # U4 INV-4, and this page is WHY the invariant needs a third home.
             # A verbatim booth's own index.html is served untouched, so it has
             # no Booth-rendered header to carry the lifetime line — this page
             # is the only surface besides the index card where the Booth owns
             # the chrome. Without it, the booths most likely to be held (a
             # report that ASKS something is the archetype) would be the ones
             # that never say they are.
             "kept": is_kept(booth),
             "hold": hold,
             "expires_in": max(0.0, ttl_seconds - booth_age_seconds(booth))},
        )

    @app.get("/b/{name}/asks", include_in_schema=False)
    def booth_asks_redirect(name: str):
        """`/asks` moved to `/marks` when asks became one shape of mark. A
        redirect rather than a 404: the URL is in the operator's history and in
        landed reports, and a dead link teaches nothing."""
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/marks", status_code=308)

    @app.get("/b/{name}/marks.json")
    def booth_marks_json(name: str):
        """Every mark in the booth, as JSON — the READ path for a session that
        is not on this host.

        `booth marks` covers a session with filesystem access; a session on
        another box rsyncs its work in and has only HTTP. Before marks it polled
        `<stem>.answer.json` and waited for a 404 to become a 200, which is why
        this endpoint has to exist: without it, moving picks out of per-question
        sidecars would take that capability away. One request now answers for
        the whole booth instead of one question at a time.
        """
        booth = resolve_booth(name)
        marks, read_err = hold_read(booth)
        body = {
            "booth": name,
            "marks": [as_dict(m) for m in marks],
            "open": [m.id for m in open_marks(marks)],
        }
        # A DAMAGED file used to come back as an empty list and nothing else,
        # which is indistinguishable from "you were never asked anything" — and
        # this endpoint is the ONLY reader a remote session has. Its filesystem
        # sibling has told the truth since U2: `booth marks` exits 3 on an
        # unreadable file precisely so a caller can tell "not yet" from
        # "broken". One question, two surfaces, two answers.
        #
        # The STATUS stays 200 and that is deliberate. Reads are lenient here —
        # the same rule that keeps a poisoned booth from 500ing the index — and
        # a pinned status code is a promise to remote clients this fix has no
        # business breaking. The information goes in the body instead: a client
        # that wants the CLI's exit-3 parity reads `error`, and one that does
        # not behaves exactly as it does today.
        if read_err is not None:
            body["error"] = "this booth's .marks.json cannot be read"
            body["detail"] = read_err
        return JSONResponse(body)

    @app.get("/b/{name}/view", response_class=HTMLResponse)
    def booth_view_file(request: Request, name: str, f: str):
        """Full-size view of ONE item — image zoom, or a doc as a readable page.

        Reads the item's RECORD rather than re-deriving it. That is the whole
        point of U1: this route used to call classify/doc_kind/render_doc and
        booth_image_names itself, deriving a strictly smaller set of facts than
        the gallery did, and the fact it lacked was the caption. An annotated
        image lost its annotation at exactly the size where it is most readable.
        """
        booth = resolve_booth(name)
        try:
            target = (booth / f).resolve()
        except OSError:
            raise HTTPException(status_code=404, detail="no such file")
        if not str(target).startswith(str(booth) + os.sep) or not target.is_file():
            raise HTTPException(status_code=404, detail="no such file")

        items = booth_items(booth)
        item = find_item(items, f)
        # U4: a bookmarked zoom URL is somebody looking — but only once we know
        # there is an ITEM to look at. Below the 404s, and gated on the record,
        # because `f` is any path that stats inside the booth: the bug-hunt
        # panel pointed `?f=.marks.lock` at this and held a booth open with a
        # file the service created itself. A dotfile is not an item, and a view
        # of a thing that is not an item is not a view of the booth.
        if item is not None:
            record_view(booth)
        marks = marks_for(booth)
        item_marks = marks_for_target(marks, f)
        common = {
            **base_ctx,
            "name": name,
            "name_url": quote(name, safe=""),
            "file": f,
            "file_url": quote(f, safe="/"),
            # The facts this route never used to carry.
            "caption": item.caption if item else None,
            "section": item.section if item else None,
            "blurred": item.blurred if item else False,
            # INV-3, U1's rule extended from the caption to the judgment: the
            # notes and the flag state travel to full size, which is the size at
            # which the judgment is actually being made.
            "marks": item_marks,
            "flagged": any(m.shape == "flag" for m in item_marks),
        }

        if item is not None and item.kind == "image":
            # prev/next ring (wraps; only when there is more than one image)
            names = image_chain(items)
            prev_url = next_url = None
            if f in names and len(names) > 1:
                i = names.index(f)
                prev_url = quote(names[(i - 1) % len(names)], safe="/")
                next_url = quote(names[(i + 1) % len(names)], safe="/")
            return templates.TemplateResponse(
                request, "view.html", {**common, "prev_url": prev_url, "next_url": next_url}
            )

        # .md renders, .txt/.log show as text — viewable in-booth, no download
        if item is not None:
            body = render_doc_body(booth, item)
            if body is not None:
                rendered, is_html = body
                return templates.TemplateResponse(
                    request,
                    "doc.html",
                    {**common, "kind": item.doc, "body": rendered, "is_html": is_html},
                )

        # nothing to render — hand back the raw file
        return RedirectResponse(
            url=f"/b/{quote(name, safe='')}/{quote(f, safe='/')}", status_code=307
        )

    @app.get("/b/{name}/{filepath:path}")
    def booth_file(name: str, filepath: str, dl: int = 0):
        booth = resolve_booth(name)
        try:
            target = (booth / filepath).resolve()
        except OSError:
            raise HTTPException(status_code=404, detail="no such file")
        if not str(target).startswith(str(booth) + os.sep) or not target.is_file():
            raise HTTPException(status_code=404, detail="no such file")
        # ?dl=1 forces a download (Content-Disposition: attachment) instead of the
        # browser rendering inline — the fix for html/md/text that otherwise opens
        # in-page with no easy "save".
        if dl:
            return FileResponse(str(target), filename=target.name)
        return FileResponse(str(target))

    @app.post("/upload")
    async def upload(files: list[UploadFile] = File(...)):
        """Browser/curl drop-off: files land in a new booth with a human-readable
        pickup id (e.g. 4-wombat), sweep-expiring in the usual TTL. Redirects (303)
        to the pickup page; curl clients read the Location header for the id."""
        files = [f for f in files if f and f.filename]
        if not files:
            raise HTTPException(status_code=400, detail="no files uploaded")
        if len(files) > max_files:
            raise HTTPException(status_code=413, detail=f"too many files (max {max_files})")

        booth_id = generate_pickup_id(lambda n: (data_dir / n).exists())
        dest = data_dir / booth_id
        dest.mkdir(parents=True)

        total = 0
        # Both markers are belt-and-braces: `safe_upload_name` strips leading
        # dots, so an uploaded file can never be named either of them. Listed
        # anyway so the set says what the directory already contains.
        used: set = {UPLOAD_MARKER, MANIFEST_FILE}
        try:
            (dest / UPLOAD_MARKER).write_text("")   # dotfile, not listed
            # A booth the SERVICE made says so, rather than being exempted from
            # the unannounced marker. INSIDE the guard, with the marker: both
            # sat above it, so a failure here left a half-booth on disk with no
            # files in it — and the manifest's unique temp name meant a leaked
            # `.booth.json.<hex>.tmp` was never overwritten, was not a `.lock`,
            # and so kept that empty booth alive past every sweep. Found 4/4.
            write_manifest(dest, SERVICE_HANDLE, title=booth_id,
                           why="browser upload, for pickup")
            for i, f in enumerate(files):
                name = _dedupe_name(safe_upload_name(f.filename, f"file-{i + 1}"), used)
                used.add(name)
                with (dest / name).open("wb") as out:
                    while chunk := await f.read(1024 * 1024):
                        total += len(chunk)
                        if total > max_upload_bytes:
                            raise HTTPException(
                                status_code=413,
                                detail=f"upload too large (max {max_upload_mb:g} MB)",
                            )
                        out.write(chunk)
                await f.close()
        except Exception:
            shutil.rmtree(dest, ignore_errors=True)  # never leave a half-written booth
            raise

        return RedirectResponse(url=f"/b/{quote(booth_id, safe='')}/", status_code=303)

    # Releasing a kept board. The kept lane has no wipe control on purpose —
    # destroying a durable board should not be one misclick — but "deliberate"
    # had been built as "impossible from the UI": the only ways out were ssh or
    # a hand-written API call. These two routes make the release step reachable
    # while keeping deletion two deliberate acts (release, then wipe).
    #
    # NOTE ON THE TTL, which is not intuitive: removing the sentinel BUMPS the
    # booth directory's mtime, and booth_age_seconds reads the newest mtime in
    # the tree — so a released board's clock resets to zero and it survives
    # another full TTL. "Unkeep and let the sweeper take it" therefore does NOT
    # delete promptly. Release is the step that makes the × available; the ×
    # is what deletes. Anything relying on release-then-sweep is relying on a
    # 24h delay it probably did not intend.

    @app.post("/b/{name}/unlink")
    def board_unlink(name: str, entry: str = Form(...)):
        """Remove ONE row from a link board, by content id.

        Deliberately not by index: the board is append-only and multi-writer,
        so between rendering the page and clicking × another session may have
        posted. A content id either matches the row the operator saw or matches
        nothing — it can never resolve to a neighbour.
        """
        removed = remove_link_entry(resolve_booth(name), entry)
        if removed is None:
            # Already gone (double-click, stale tab, someone else pruned it).
            # Not an error worth a 404 page — the desired end state holds.
            pass
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/unlink-many")
    def board_unlink_many(name: str, sel: list[str] = Form(default=[])):
        """Remove SEVERAL rows in one go — the multi-select delete.

        Each `sel` is a content id (same identity the per-row × uses), so the same
        race-safety holds: an id either matches the row the operator selected or
        matches nothing, never a neighbour that another session appended in the
        meantime. An empty selection is a no-op, not an error.
        """
        board = resolve_booth(name)
        for entry_id in sel:
            remove_link_entry(board, entry_id)
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/pin")
    def board_pin(name: str, entry: str = Form(...)):
        """Toggle a row's pinned (favorite) state, by content id. Pinned rows
        float to the top of the board; toggling again unpins. Reversible, so no
        confirmation — unlike removal."""
        toggle_pin(resolve_booth(name), entry)
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/bench-add")
    def bench_add(name: str, url: str = Form(...), bname: str = Form("", alias="name")):
        """Register or update a bench by normalized URL.

        A rejected URL must not 500 the page it was posted from. THE REJECTION
        IS SILENT HERE, and that is stated rather than dressed up: the form's
        `type="url"` catches the ordinary typo in the browser before the post,
        and this `except` is the last resort for what slips past it — the bench
        simply does not appear. Surfacing the reason would need a flash message,
        which this service has no mechanism for; inventing one for a path the
        browser already guards is not worth a unit's scope.

        `resolve_booth` is called for its 404: a POST at a booth that does not
        exist is not a silent no-op.
        """
        resolve_booth(name)
        try:
            upsert_bench(data_dir, url, bname, "operator")
        except (ValueError, OSError):
            pass
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/bench-state")
    def bench_state(name: str, bench: str = Form(...), state: str = Form(...)):
        resolve_booth(name)
        if state in BENCH_STATES:
            try:
                set_bench_state(data_dir, bench, state)
            except (ValueError, OSError):
                pass
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/bench-remove")
    def bench_remove(name: str, bench: str = Form(...)):
        resolve_booth(name)
        try:
            remove_bench(data_dir, bench)
        except (ValueError, OSError):
            pass
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    def _safe_next(nxt: str) -> str:
        """Where to land after keep/unkeep. Defaults to the index; a booth page
        can ask to stay put. Only same-site absolute paths are honoured — `//`
        and any scheme are refused, because a redirect target taken from a form
        field is an open redirect if you do not check it."""
        if nxt.startswith("/") and not nxt.startswith("//") and "\\" not in nxt:
            return nxt
        return "/"

    @app.post("/b/{name}/keep")
    def booth_keep(name: str, next: str = Form("/")):
        (resolve_booth(name) / KEEP_MARKER).touch()
        return RedirectResponse(url=_safe_next(next), status_code=303)

    @app.post("/b/{name}/unkeep")
    def booth_unkeep(name: str, next: str = Form("/")):
        # missing_ok: releasing an already-released board is a no-op, not a 500.
        booth = resolve_booth(name)
        marker = booth / KEEP_MARKER
        try:
            marker.unlink()
            released = True
        except FileNotFoundError:
            released = False              # already released: a no-op, not a 500
        except OSError:
            # A `.forever` that is a DIRECTORY raised IsADirectoryError straight
            # through this route and 500'd it, which made the card's release
            # button permanently dead for that booth. Pre-existing; the panel
            # re-exposed it. Removing it is still best-effort, and failing to is
            # not worth refusing the request over.
            released = False
        if not released:
            return RedirectResponse(url=_safe_next(next), status_code=303)
        # U4: RELEASE IS ACTIVITY, and now it is a rule rather than an accident.
        # A released board already survived another full TTL, because unlinking
        # a file bumps the directory's mtime — behaviour the note above calls
        # "not intuitive" precisely because nothing declared it. The behaviour
        # is unchanged; its reason is now stated. Releasing a board is somebody
        # touching it, so it gets one full TTL, the same as any other look.
        #
        # ONLY when something was actually released, which is the correction the
        # bug-hunt panel forced: an unconditional call made POSTing release at
        # an already-released booth an endless TTL refresh, contradicting this
        # route's own no-op promise and diverging from the CLI, which `rm`s the
        # sentinel without recording anything.
        record_view(booth)
        return RedirectResponse(url=_safe_next(next), status_code=303)

    @app.post("/b/{name}/blur")
    def booth_blur(name: str, f: str = Form(...), on: str = Form("1")):
        """Toggle one item's blur. Reversible and cosmetic, so no confirmation.
        See BLUR_FILE: this hides an item from a glance, it does not protect it."""
        booth = resolve_booth(name)
        # Guard the path the same way the file route must: a blur entry is only
        # ever a booth-relative path, never an escape.
        rel = f.strip().lstrip("/")
        if ".." in Path(rel).parts:
            raise HTTPException(status_code=400, detail="bad item path")
        set_blurred(booth, rel, on not in ("0", "false", ""))
        return RedirectResponse(url=f"/b/{quote(name, safe='')}/", status_code=303)

    @app.post("/b/{name}/delete")
    def booth_delete_form(name: str):
        shutil.rmtree(resolve_booth(name))
        return RedirectResponse(url="/", status_code=303)

    @app.delete("/b/{name}")
    def booth_delete_api(name: str):
        shutil.rmtree(resolve_booth(name))
        return JSONResponse({"wiped": name})

    return app


def _from_env() -> FastAPI:
    data = os.environ.get("BOOTH_DATA_DIR", str(Path.home() / "booth-data"))
    ttl = float(os.environ.get("BOOTH_TTL_HOURS", "24"))
    host = os.environ.get("BOOTH_HOST_LABEL", "")
    interval = int(float(os.environ.get("BOOTH_SWEEP_INTERVAL_MIN", "15")) * 60)
    max_mb = float(os.environ.get("BOOTH_MAX_UPLOAD_MB", "1024"))
    max_n = int(os.environ.get("BOOTH_MAX_FILES", "50"))
    return create_app(
        data,
        ttl_hours=ttl,
        host_label=host,
        sweep_interval_s=interval,
        max_upload_mb=max_mb,
        max_files=max_n,
    )


app = _from_env()