app.py

"""

PDF Injection Detector - MiMo-7B.



Upload a PDF, and this reads it the way the corpus was read, cuts it into batches of regions that

fit one run of the model, and asks MiMo-7B-RL whether a payload is hidden in the batch you choose.



`app.py` holds the interface, the batching and the aggregation. It contains no detection logic of

its own: the text extraction lives in `corpus_text.py`, the prompt and parser in `mimo.py`, the

embedding lookup in `neighbours.py`, and each is quoted from the notebook that measured it.

"""

import re
import traceback
from pathlib import Path

import gradio as gr

import corpus_text
import mimo
import neighbours

# What to do about each family, shown under a payload verdict.
#
# One entry per family in `mimo.FAMILIES`, and the keys are asserted against that list at import so
# a renamed family cannot silently lose its advice. Each is written against the generator's own
# object shapes rather than the general security literature, so a reader who opens a flagged file
# finds the thing the text describes.
#
# **This is advice conditional on a guess.** MiMo names the family correctly 43% of the time, so
# more often than not the paragraph shown here is the remedy for a different attack than the one
# present. It is worth showing anyway - the first line of every entry is a containment step that
# holds whatever the family turns out to be - but the interface says so rather than implying the
# app knows what it is looking at.
TREATMENT = {
    "javascript_injection": (
        "Quarantine the file immediately. Do not open in Adobe Reader or any JS-enabled viewer. "
        "Strip the /JS and /JavaScript PDF objects using a PDF sanitiser (e.g. qpdf or mutool). "
        "Disable JavaScript in PDF readers organisation-wide via Group Policy."),
    "cross_site_scripting": (
        "Do not open in a browser-based PDF viewer. Use a sandboxed desktop reader only. "
        "Submit to your security team for content stripping before redistribution."),
    "ssrf": (
        "Block outbound HTTP requests from any server that processes this file. "
        "Do not open on cloud infrastructure without egress filtering - the payload targets "
        "the AWS/GCP metadata endpoint (169.254.169.254). Flag for security review."),
    "object_action_injection": (
        "Open only in read-only sandboxed mode. Strip /Launch, /OpenAction and /AA objects "
        "using a PDF sanitiser. Alert your SOC team - this payload attempts to execute a "
        "program outside the PDF viewer, with the opening user's own permissions."),
    "llm_prompt_injection": (
        "Do not feed this document to any AI/LLM pipeline without human review first. "
        "The payload attempts to override AI instructions. Sanitise or reject the file "
        "before any automated processing."),
    "shellcode_embedded_exe": (
        "Quarantine immediately and isolate the machine. Run through antivirus before any "
        "attachment is extracted, and do not let anyone double-click what comes out. "
        "Report the binary stream hash to your threat intelligence team."),
    "dde_template_injection": (
        "Do not copy any field contents into Excel or Word - the payload activates as a DDE "
        "formula once it reaches Office, not while the PDF is open. Strip /AcroForm field "
        "values and any remote template reference, and block outbound requests to the linked "
        "template host."),
    "polyglot_file": (
        "Treat this as two files, not one. Scan it again as an archive as well as a PDF: "
        "whichever format your scanner did not choose was never inspected at all. Do not let "
        "it through a filter that decides file type from the extension or the leading bytes "
        "alone."),
    "ransomware_simulation": (
        "Isolate the host before anything else, and do not open the file. Confirm backups are "
        "offline and restorable, then hand the sample to incident response. In this corpus the "
        "payload is a harmless RANSIM/EICAR simulation - an unknown file of this shape should "
        "be treated as the real thing until proven otherwise."),
    "steganographic_payload": (
        "Do not forward the file. Extract and inspect the embedded image streams separately: "
        "the payload hides in pixel data and survives filters that look for code or links. "
        "Re-encode or drop the images before any redistribution."),
    "uri_redirect_phishing": (
        "Do not click anywhere in the document - the link annotation covers the entire page, "
        "so any click opens it. Verify the destination host out of band, strip /URI actions "
        "with a PDF sanitiser, and report the URL to your phishing intake."),
    "xfa_acroform_injection": (
        "Strip the XFA form definition before the file is processed further. The payload lives "
        "in an XML document inside the PDF, so scanners that parse only PDF objects never see "
        "it. Do not open in a reader with XFA support enabled."),
}

# A verdict can be `injected=true` with `injection_type="none"` - the model is sure something is
# wrong and will not say what. That still deserves an answer.
TREATMENT_UNNAMED = (
    "MiMo flagged this region but would not name a family, so no specific remediation applies. "
    "Treat the file as untrusted: do not open it on a machine that matters, do not forward it, "
    "and pass it to whoever handles security for you.")

assert set(TREATMENT) == set(mimo.FAMILIES), (
    f"TREATMENT does not cover mimo.FAMILIES: "
    f"missing {sorted(set(mimo.FAMILIES) - set(TREATMENT))}, "
    f"unknown {sorted(set(TREATMENT) - set(mimo.FAMILIES))}")

# Measured in Part B on a T4, on the project's own 1,100-document corpus.
PART_B = {"f1": 0.945, "precision": 0.988, "recall": 0.906, "family_acc": 0.433,
          "false_alarm_rate": 0.05, "unparsable": 155, "n": 1100}

GPU = mimo.BACKEND == "gpu"

# How many regions fit in ONE run is not a taste decision. On ZeroGPU a single grant is capped at
# 300 seconds and the whole scan plus the 4-bit load must fit inside it; on 2 vCPUs a region costs
# two minutes and a browser will not wait for many. That ceiling is what a batch is.
MAX_PER_BATCH = mimo.MAX_WINDOWS_GPU if GPU else 6
DEFAULT_PER_BATCH = 8 if GPU else 3

WINDOW_COLUMNS = ["#", "where in the skeleton", "kind", "markers", "verdict", "family", "evidence"]

# One example per injection family, plus one clean control, taken from the generation repo.
#
# The files on disk are named Example_N and nothing else, and that stays true: the uploader shows
# a neutral filename, and nothing about the file announces its own answer. The *buttons* are
# labelled by family so the demo can be driven deliberately ("show me ransomware"), which costs
# nothing on the model side - a filename never reaches the prompt, only extracted text does.
#
# Labels come from `examples/manifest.json`, written by the same script that copies the PDFs, so
# a button cannot end up pointing at the wrong family. `examples/Example_Key.txt` is the same
# mapping in prose, for whoever is marking this.
EXAMPLE_DIR = Path(__file__).resolve().parent / "examples"


def load_examples():
    """[(label, path)], families first in the model's own order, clean last. Empty if absent."""
    manifest = EXAMPLE_DIR / "manifest.json"
    if not manifest.exists():
        return []
    import json
    mapping = json.loads(manifest.read_text(encoding="utf-8"))
    order = {f: i for i, f in enumerate(mimo.FAMILIES)}
    rows = [(fam, EXAMPLE_DIR / name) for name, fam in mapping.items()
            if (EXAMPLE_DIR / name).exists()]
    # "clean" is not in FAMILIES, so it sorts last - which is where it belongs: it is the control,
    # read after you have seen what a hit looks like.
    return sorted(rows, key=lambda r: order.get(r[0], len(order)))


EXAMPLES = load_examples()


def fmt_eta(n: int, runtime: str = None) -> str:
    seconds = n * mimo.SECONDS[runtime or mimo.BACKEND]
    if seconds < 90:
        return f"about {max(20, int(seconds))}s"
    return f"{seconds * 0.6 / 60:.0f}-{seconds * 1.5 / 60:.0f} min"


def extract(path, cover_all=True):
    """PDF bytes to ranked candidate regions. No model touched, so this runs on upload."""
    with open(path, "rb") as fh:
        data = fh.read()
    skeleton, truncated, dropped = corpus_text.build_skeleton(data)
    return (data, skeleton, truncated, dropped,
            corpus_text.candidate_windows(skeleton, cover_all=bool(cover_all)))


def batches_of(candidates, per_batch):
    """The ranked regions cut into runnable chunks. Batch 1 is the most marker-dense."""
    per_batch = max(1, int(per_batch))
    return [candidates[i:i + per_batch] for i in range(0, len(candidates), per_batch)]


def batch_label(batch, i, n_batches, runtime=None):
    """What one batch is, in a line, so the choice is informed rather than a number."""
    first, last = batch[0], batch[-1]
    if first["is_head"]:
        where = "head of document"
    else:
        where = f"chars {min(w['start'] for w in batch):,}-{max(w['end'] for w in batch):,}"
    fams = sorted({f for w in batch for f in w["families"]})
    n_marker = sum(1 for w in batch if w["source"] == "marker")
    kind = ("marker regions" if n_marker == len(batch) else
            "sweep of the document" if n_marker == 0 else
            f"{n_marker} marker + {len(batch) - n_marker} sweep")
    # "signature:" and not the bare family name. The regex has only seen a token like `/JS (` in
    # the text; plenty of harmless PDFs contain one. Printing "javascript_injection" on its own
    # before the model has read anything reads as a verdict the app has not made.
    tail = f" · signature: {', '.join(fams)}" if fams else ""
    return (f"Batch {i + 1} of {n_batches} — {len(batch)} region(s), {kind} · {where}{tail} · "
            f"~{fmt_eta(len(batch), runtime)}")


def on_upload(path, per_batch, cover_all, runtime):
    """Extract, rank and batch. Fast enough to run on every upload and every slider move."""
    if not path:
        return (None, "Upload a PDF to see what will be read.",
                gr.update(choices=[], value=None, interactive=False))

    try:
        data, skeleton, truncated, dropped, candidates = extract(path, cover_all)
    except Exception as e:
        return (None, f"Could not read that file: `{type(e).__name__}: {e}`",
                gr.update(choices=[], value=None, interactive=False))

    groups = batches_of(candidates, per_batch)

    # Plain label strings, NOT (label, index) pairs. The dropdown allows custom values so the API
    # can name a batch before any PDF has been uploaded, and that turns it into a free-text
    # combobox: a programmatic integer value of 0 is falsy, so the displayed text refused to
    # refresh and the box kept showing the previous split after the slider moved. Strings are
    # never falsy here, and `resolve_batch` reads the index straight back out of "Batch N of M".
    choices = [batch_label(b, i, len(groups), runtime) for i, b in enumerate(groups)]

    lines = [
        f"**{len(data):,} bytes** on disk, rendered to a **{len(skeleton):,}-character skeleton**"
        + (f" (truncated to the {corpus_text.SKELETON_CHAR_BUDGET:,}-character budget)"
           if truncated else "")
        + (f", {dropped} binary stream(s) dropped." if dropped else "."),
    ]
    n_marker = sum(1 for w in candidates if w["source"] == "marker")
    fams = corpus_text.detect_markers(data)

    if n_marker:
        lines.append(
            f"**{n_marker} region(s) carry a marker**, and the remaining "
            f"{len(candidates) - n_marker} cover the rest of the document — "
            f"**{len(candidates)} in total, cut into {len(groups)} batch(es)** of at most "
            f"{int(per_batch)}. Structural signatures in the raw file: "
            f"`{'`, `'.join(fams) or 'none'}`.")
        lines.append(
            "_Ranking decides **reading order only** — batch 1 is the most marker-dense, not the "
            "guilty one. The verdict is MiMo's alone._")
    elif cover_all:
        lines.append(
            f"**No structural marker anywhere in the file.** The triage only recognises the twelve "
            f"families this project generated, so this means either a clean file or a payload "
            f"shaped like none of them — which is why the batches below sweep the **whole** "
            f"skeleton rather than stopping here: **{len(candidates)} region(s) in "
            f"{len(groups)} batch(es)**.")
    else:
        lines.append(
            "**No structural marker anywhere in the file**, and the sweep is switched off — so "
            "only the head of the document will be read, which is exactly what the corpus builder "
            "produced for a *clean* file. Switch the sweep on to look at the rest of it.")

    lines.append(f"Pick a batch and press **Check this batch**. One batch is one run of the model, "
                 f"sized to fit the **{(runtime or mimo.BACKEND).upper()}** runtime's limit; run as many "
                 f"batches as you like, one at a time.")

    return ((skeleton, candidates), "\n\n".join(lines),
            gr.update(choices=choices, value=choices[0], interactive=True))


def resolve_batch(value, n_batches: int) -> int:
    """

    Whatever the dropdown handed back, as a batch index that exists.



    Because the dropdown allows custom values it can arrive as the integer index, as `None` before

    anything was picked, or as the label string itself. All three mean something, and none of them

    should be an exception in front of a user who just pressed a button.

    """
    if isinstance(value, (int, float)):
        idx = int(value)
    else:
        digits = re.search(r"\d+", str(value or ""))
        idx = int(digits.group()) - 1 if digits else 0     # labels are 1-based, indices are not
    return idx if 0 <= idx < n_batches else 0


def run(path, per_batch, batch_index, want_neighbours, cover_all, runtime,

        progress=gr.Progress()):
    """Score one batch."""
    if not path:
        return "Upload a PDF first.", [], [], ""

    progress(0.05, desc="reading the PDF")
    try:
        _, _, _, _, candidates = extract(path, cover_all)
    except Exception as e:
        return f"Could not read that file: `{type(e).__name__}: {e}`", [], [], ""

    groups = batches_of(candidates, per_batch)
    idx = resolve_batch(batch_index, len(groups))
    batch = groups[idx]

    runtime = runtime or mimo.BACKEND
    progress(0.15, desc=f"loading MiMo ({runtime})")
    try:
        answers = mimo.judge_all([w["text"] for w in batch], backend=runtime,
                                 progress=lambda m: progress(0.4, desc=m))
    except Exception as e:
        hint = ""
        if "quota" in str(e).lower():
            hint = ("\n\nA free account gets about **five minutes of ZeroGPU per day**, and the "
                    "scheduler reserves a whole run up front. The quota is per visitor, so this "
                    "affects your account rather than the Space.")
            hint += ("\n\nSwitch the runtime to **cpu** — much slower, but no quota — or come back "
                     "tomorrow." if mimo.CPU_AVAILABLE else
                     " Fewer regions per batch costs less of it; otherwise it resets in 24 hours.")
        return (f"## MiMo could not run\n\n`{type(e).__name__}: {e}`{hint}\n\nRuntime selected: "
                f"**{runtime}**."), [], [], traceback.format_exc()

    offset = idx * max(1, int(per_batch))
    rows, log, results = [], [], list(zip(batch, answers))
    for i, (win, r) in enumerate(results):
        where = "head of document" if win["is_head"] else f"chars {win['start']:,}-{win['end']:,}"
        verdict = ("PAYLOAD" if r["pred_injected"] else "clean") + (
            "" if r["parse_ok"] else " (unreadable answer)")
        rows.append([offset + i + 1, where, win["source"], ", ".join(win["families"]) or "-",
                     verdict, r["pred_family"], (r["evidence"] or "-")[:160]])
        log.append(f"--- region {offset + i + 1} ({where}, prompt via {r['prompt_route']}) ---\n"
                   f"{r['raw']}")

    report = build_report(results, idx, groups, len(candidates), runtime)

    nb_rows = []
    if want_neighbours and results:
        progress(0.95, desc="embedding and looking up the corpus")
        flagged = next((w for w, r in results if r["pred_injected"]), None)
        query = flagged or batch[0]
        try:
            neighbours.check_provenance()
            nb_rows = neighbours.neighbour_rows(query["text"], k=5)
            basis = ("the first flagged region" if flagged else
                     "the first region in this batch (nothing was flagged)")
            report += (f"\n\n### Nearest files in the corpus\n\nEmbedded from **{basis}** with "
                       f"Part A's winning configuration. Precision@5 on this index is **35.6%** "
                       f"against a 6.8% random baseline: fewer than 2 of the 5 listed are the same "
                       f"kind of attack. Read it as *resemblance*, not identification.")
        except Exception as e:
            report += (f"\n\n### Nearest files in the corpus\n\nUnavailable: "
                       f"`{type(e).__name__}: {e}`")
            log.append(traceback.format_exc())

    return report, rows, nb_rows, "\n\n".join(log)


def begin(per_batch, runtime):
    """

    Lock the button and say so, before the slow part starts.



    A second press while MiMo is mid-scan is worse here than in most apps: on ZeroGPU it queues a

    second grant against a daily quota that only affords one or two, so the cost of a stray click

    is the rest of the day. The button is disabled for the duration and `mimo._lock` serialises

    the model itself, so neither the UI nor the server can be made to run two scans at once.



    Stale results are cleared at the same time - leaving the previous batch's verdict on screen

    under a "loading" banner is how someone reads the wrong answer for the wrong file.

    """
    runtime = runtime or mimo.BACKEND
    n = max(1, int(per_batch or 1))
    return (gr.update(interactive=False, value="Checking…"),
            f"### Loading…\n\nMiMo is reading up to **{n} region(s)** on the **{runtime}** "
            f"runtime — roughly {fmt_eta(n, runtime)} once the model is in memory. The first scan "
            f"after a restart also downloads the weights, which takes several minutes longer.\n\n"
            f"_Leave this tab open; the report replaces this message when it is done._",
            [], [], "")


def finish():
    """Give the button back. Chained with `.then()`, so it runs even if the scan raised."""
    return gr.update(interactive=True, value="Check this batch")


def build_report(results, idx, groups, n_candidates, runtime) -> str:
    """The verdict for this batch, and an explicit account of what is still unread."""
    if not results:
        return "Nothing was read."

    hits = [(w, r) for w, r in results if r["pred_injected"]]
    unparsed = sum(1 for _, r in results if not r["parse_ok"])
    read = len(results)
    unread = n_candidates - read
    others = [i for i in range(len(groups)) if i != idx]

    if hits:
        fams = sorted({r["pred_family"] for _, r in hits if r["pred_family"] != "none"})
        head = (f"## Payload found in batch {idx + 1}\n\n"
                f"> ### ⚠️ Instructions: do not open the file. Erase immediately!\n\n"
                f"MiMo flagged **{len(hits)} of the {read} region(s)** in this batch.")
        head += (f" It named the family as **{', '.join(fams)}** — correct 43% of the time in "
                 f"Part B, so treat it as a suggestion." if fams
                 else " It did not commit to a family.")

        # Treatment goes directly under the verdict, because it is the only part of this report
        # anyone acts on. One block per named family; a flagged region with no family still gets
        # the generic containment advice rather than silence.
        head += "\n\n### What to do about it\n"
        if fams:
            for family in fams:
                head += f"\n**{family}**\n\n{TREATMENT[family]}\n"
            named = "family" if len(fams) == 1 else "families"
            head += (f"\n_The {named} named above {'is' if len(fams) == 1 else 'are'} MiMo's "
                     f"guess, right about 43% of the time — so this advice may be the remedy for a "
                     f"different attack. The containment step in each first sentence holds either "
                     f"way._")
        else:
            head += f"\n{TREATMENT_UNNAMED}\n"
    else:
        head = (f"## Nothing found in batch {idx + 1}\n\nMiMo read **{read} region(s)** in this "
                f"batch and flagged none of them.")

    caveats = []
    if unread > 0:
        caveats.append(
            f"**This is 1 of {len(groups)} batches.** {unread} region(s) across "
            f"{len(others)} other batch(es) have not been read. Whatever this batch says, it says "
            f"it about {read} of the file's {n_candidates} candidate regions — nothing more.")
    if unparsed:
        sweep_bad = sum(1 for w, r in results if not r["parse_ok"] and w["source"] == "sweep")
        note = (f"{unparsed} answer(s) could not be parsed and count as *not injected*, exactly as "
                f"Part B scored them (155 of 1,100 there).")
        if sweep_bad:
            note += (
                f" **{sweep_bad} of those were sweep regions**, and that is expected rather than "
                f"surprising: Part B only ever showed MiMo marker-centred windows or the head of a "
                f"document, never arbitrary mid-file content streams. Given a page of font "
                f"positioning operators it tends to carry on copying the input instead of "
                f"answering. Sweep regions buy coverage of text that would otherwise never be "
                f"looked at; they do not inherit Part B's accuracy, and a *clean* verdict on one "
                f"is close to no evidence at all.")
        caveats.append(note)

    body = head
    if caveats:
        body += "\n\n" + "\n\n".join("- " + c for c in caveats)
    body += (f"\n\n---\n\n**On the corpus Part B measured**, MiMo scored F1 {PART_B['f1']}, "
             f"precision {PART_B['precision']}, recall {PART_B['recall']} on {PART_B['n']:,} "
             f"documents that were 82% injected — where a detector that flags everything without "
             f"reading it scores F1 0.900. Read 0.945 against 0.900, not against zero.\n\n"
             f"_Runtime: **{runtime}**. {mimo.CAVEATS[runtime]}_")
    return body


INTRO = f"""

# PDF Injection Detector — MiMo-7B



Upload a PDF. It is rendered to text with the extractor that built the project corpus, the regions

carrying structural markers are ranked and cut into **batches that fit one run of the model**, and

**MiMo-7B-RL** reads the batch you choose — reporting whether a payload is hidden there, and the

substring that convinced it.



Batching is what keeps a long document inside the runtime's limit: one batch is one run, and you

decide how many runs to spend. The report always states how much of the file is still unread.



**None of what it tells you is guaranteed correct — not the verdict, not the family, not the

treatment.** On the corpus it was measured against, MiMo got the injected/clean call right often

enough to score F1 0.945, but it named the attack family correctly only **43%** of the time — so

more often than not the family shown, and therefore the remediation advice attached to it, belongs

to a different attack. A clean verdict is not proof of a clean file either. Read every output as a

prompt to look closer yourself, never as a decision that has already been made.



This is a coursework artefact built on a synthetic corpus of 1,100 PDFs carrying harmless

EICAR/AMTSO/WICAR/RANSIM test markers. **It is not a general malware scanner**, and real malware

does not announce itself the way these samples do.



**Why MiMo and not Gemma?** Part B's winner was Gemma-2-9B at F1 0.969, against MiMo's 0.945. But

Gemma is gated behind a licence and a token, and it is 2.6× slower per window (10.95 s vs 4.18 s).

On free ZeroGPU — one grant capped at 300 s, and roughly five minutes of GPU per day — that is the

difference between a working demo and one that refuses strangers at the door and then runs out of

quota. The cost of the swap is 0.024 F1 and family-naming dropping from 63% to 43%.



Running on the **{mimo.BACKEND.upper()}** runtime

({'4-bit NF4 — Part B’s own configuration' if GPU else 'Q4_K_M GGUF via llama.cpp'}),

about {mimo.SECONDS_PER_WINDOW:g}s per region.

"""


with gr.Blocks(title="PDF Injection Detector") as demo:
    gr.Markdown(INTRO)
    state = gr.State()

    with gr.Row():
        # The example rail, down the left edge. Plain buttons rather than `gr.Examples`: the
        # built-in renders a horizontal table of filenames, and what is wanted here is one
        # labelled tab per attack type that loads its document on click. Buttons also sidestep
        # `gr.Examples`' caching, which would run a full scan of all thirteen files at startup
        # and spend the whole day's ZeroGPU quota before anyone opened the page.
        with gr.Column(scale=1, min_width=170):
            if EXAMPLES:
                gr.Markdown("### Examples\nOne document per attack type.")
                example_buttons = [(gr.Button(label.replace("_", " "), size="sm"), path)
                                   for label, path in EXAMPLES]
            else:
                example_buttons = []

        with gr.Column(scale=2):
            pdf = gr.File(label="PDF", file_types=[".pdf"], type="filepath")
            # Offered rather than decided, because the GPU here is the scarce resource: a free
            # account gets ~5 minutes of ZeroGPU a day and one batch reserves most of a run. The
            # CPU path is ~30x slower and has no quota at all, which makes it the right answer
            # once the day's GPU is gone - the Space should not go dark until midnight.
            runtime_pick = gr.Radio(
                choices=mimo.BACKENDS, value=mimo.BACKEND, label="Runtime",
                visible=len(mimo.BACKENDS) > 1,
                info=("gpu = 4-bit NF4, Part B's own configuration, ~4s/region, limited by your "
                      "daily ZeroGPU quota. cpu = Q4_K_M GGUF via llama.cpp, ~2min/region, "
                      "unlimited."))
            per_batch = gr.Slider(1, MAX_PER_BATCH, value=DEFAULT_PER_BATCH, step=1,
                                  label="Regions per batch",
                                  info=(f"How many regions one run of the model reads — it just "
                                        f"re-cuts the same list, so fewer per batch means more "
                                        f"batches. A run reserves the same GPU time whatever this "
                                        f"is set to, so lowering it inspects less of the file for "
                                        f"the same quota. Leave it at {MAX_PER_BATCH} unless you "
                                        f"want a faster single run."))
            # allow_custom_value: the choices are empty until a PDF is uploaded, and without this
            # Gradio validates any incoming value against that empty list and rejects it - which
            # makes the batch un-selectable over the API even though the UI had populated it.
            # `resolve_batch` below is what actually decides the index, from the file itself.
            batch_pick = gr.Dropdown(label="Batch to check", choices=[], interactive=False,
                                     allow_custom_value=True,
                                     info="Each batch is a separate run — spend as many as you "
                                          "like.")
            sweep = gr.Checkbox(
                value=True, label="Sweep the rest of the document too",
                info="Off = marker regions only, which is the shape Part B measured. On = the "
                     "batches cover the whole file, at the cost of regions MiMo often will not "
                     "answer about.")
            want_nb = gr.Checkbox(value=True, label="Also show the nearest files in the corpus",
                                  info="Adds a one-off 550 MB embedding-model download.")
            go = gr.Button("Check this batch", variant="primary")
            plan = gr.Markdown("Upload a PDF to see what will be read.")

        with gr.Column(scale=4):
            with gr.Tab("Report"):
                report = gr.Markdown()
            with gr.Tab("Regions read"):
                window_table = gr.Dataframe(headers=WINDOW_COLUMNS, wrap=True, interactive=False)
            with gr.Tab("Nearest corpus files"):
                nb_table = gr.Dataframe(headers=neighbours.NEIGHBOUR_COLUMNS, interactive=False)
            with gr.Tab("What MiMo actually said"):
                # No `show_copy_button`: gradio 6 removed it, and this Space should survive an
                # sdk_version bump rather than crash at startup on a cosmetic argument.
                raw = gr.Textbox(lines=22, interactive=False,
                                 label="The prompt route and untouched generation per region")

    # Each example button just drops its path into the file component. That fires `pdf.change`
    # below, so an example goes through exactly the same triage as a real upload - there is no
    # second code path for demo files, and nothing about an example is pre-computed.
    for button, path in example_buttons:
        button.click(lambda p=str(path): p, None, pdf)

    inputs = [pdf, per_batch, sweep, runtime_pick]
    for ev in (pdf.change, per_batch.change, sweep.change, runtime_pick.change):
        ev(on_upload, inputs, [state, plan, batch_pick])
    # Three chained steps: lock and show "Loading…", scan, unlock. `.then()` rather than
    # `.success()` for the last one, because the button must come back even when the scan raised -
    # a quota refusal that left the app permanently disabled would look like a crash.
    # concurrency_limit=1 is the server-side half of the same guarantee.
    scan = go.click(begin, [per_batch, runtime_pick],
                    [go, report, window_table, nb_table, raw], queue=False)
    scan = scan.then(run, [pdf, per_batch, batch_pick, want_nb, sweep, runtime_pick],
                     [report, window_table, nb_table, raw], concurrency_limit=1)
    scan.then(finish, None, go, queue=False)

if __name__ == "__main__":
    demo.queue(max_size=8).launch()