"""Gradio presentation and callbacks for the Distinct Space."""

from __future__ import annotations

import json
import logging
import math
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Optional, Sequence

import gradio as gr

from distinct_protocol import (
    MAX_LIVE_STEP_TEXT,
    MAX_RUN_STEPS,
    MODE_AGENT,
    MODE_SIMPLE,
    PHASE_CALLING_TOOL,
    PHASE_FETCHING_WEIGHTS,
    PHASE_GENERATING,
    PHASE_GUARD_INPUT,
    PHASE_GUARD_OUTPUT,
    PHASE_LOADING_MODEL,
    PHASE_OFFERED,
    PHASE_PLANNING,
    PHASE_QUEUED_ON_WORKER,
    PHASE_WAITING_FOR_WORKER,
    PHASE_WORKING,
    STEP_KINDS,
    STEP_PHASE,
    AgentStatus,
    ChatMessage,
    JobStatus,
    ToolSelection,
    normalise_phase,
)
from distinct_protocol.fences import FILE_FENCE_BODIES, defang_marker_value
from distinct_protocol.fences import defang as _defang_fences
from distinct_protocol.handshake import Allowance

from . import render
from .api import AgentApi
from .auth_routes import identity_from_session
from .catalog import (
    AREA_LABELS,
    MODEL_CATALOG,
    RELEASE_ASSESSMENT,
    RELEASE_BOUNDARY,
    RUBRIC_AREAS,
    USE_PHASE_BOUNDARY,
    AreaStatus,
    model_choices,
)
from .control_plane import ControlPlane
from .identity import (
    LoginRequired,
    SessionAuthorizer,
    gradio_oauth_is_real,
    user_id_from_profile,
)
from .models import ControlPlaneError, JobView
from .presentation import PALETTE
from .ui_state import (
    active_conversation,
    add_conversation,
    adopt_saved_state,
    append_job,
    conversation_choices,
    conversation_turns,
    new_session,
    record_answer,
    remove_conversation,
    select_conversation,
)

_LOG = logging.getLogger(__name__)

# The composer's library choices are not a constant. They are derived, per
# refresh, from what the connected agents actually advertise, by the same
# route the model dropdown and the agent filter already use. A hardcoded empty
# list here was defect A4: operators approved tools, agents advertised them,
# the agent table showed them, and the composer still refused everything.
# Four lines of implementation notes sat inside the message box: "Selection,
# allowlisting and the fail-closed broker are live" describes the parts of
# this server to somebody who maintains it. What the person composing a
# request needs is the consequence, which is one line.
NO_TOOLS_NOTICE = (
    "No worker is offering tools or skills right now, so this run will use "
    "the model on its own."
)

#: What obfuscates the transcript in localStorage, and why it is written down.
#:
#: `gr.BrowserState` encrypts what it stores, and left to itself it invents a
#: sixteen-character key at start-up — which it then ships to the browser in
#: the page's own configuration, where anybody can read it. So the key is not
#: a secret and never was; what it is is a *version stamp on the stored data*,
#: and a random one means a new stamp on every process.
#:
#: The consequence was the whole of the persistence this product promises.
#: Every restart of the Space — every push, every wake from sleep — made the
#: saved transcript undecryptable, so gradio logged "Error reading from
#: localStorage" to the console and silently replaced the user's entire
#: history with the default value. The privacy notice says the user holds
#: their own history; in practice they held it until the next deploy.
#:
#: A written-down constant is the honest form of a key that is published in
#: the page anyway, and it is what makes the stored copy survive a restart.
#: The suffix is a data-format version: change it only to deliberately
#: abandon every transcript in every browser, which
#: `ui_state.adopt_saved_state` exists so that a shape change never needs.
BROWSER_STATE_SECRET = "distinct-transcript-v1"

#: What an empty transcript says. A module constant rather than a literal
#: buried in the layout, because it is the only instruction on a page that has
#: nothing else on it, and because the previous version pointed at the wrong
#: side of the screen for two revisions without anything noticing.
EMPTY_TRANSCRIPT_BODY = (
    "Pick a model and a worker on the right, then send a request. Queued and "
    "running turns appear here. You can queue the next one while this works."
)


# --------------------------------------------------------------------------
# Phase copy
# --------------------------------------------------------------------------
#
# "Queued, waiting for a compatible agent to come online" was a true sentence
# that answered the wrong question. It said what the queue was doing; the user
# wanted to know what was happening to *their run*, and whether anything was
# happening at all. So every phase now has a headline that names the activity
# and a body that says why it is the current one.
#
# The copy lives here, on the server, keyed by a closed vocabulary. A worker
# picks the key. It never supplies the sentence.
PHASE_COPY: Mapping[str, tuple[str, str]] = {
    PHASE_WAITING_FOR_WORKER: (
        "Waiting for a worker",
        "No community worker that has approved this model is polling right now. "
        "Nothing is lost: the run starts by itself the moment a compatible one joins.",
    ),
    PHASE_OFFERED: (
        "Offered to a worker",
        "Handed to a worker, waiting for its next poll. Workers pull work rather "
        "than receiving it, so this lasts up to one poll interval.",
    ),
    PHASE_QUEUED_ON_WORKER: (
        "Queued on the worker",
        "Accepted. The worker is working through the queue ahead of it.",
    ),
    PHASE_FETCHING_WEIGHTS: (
        "Fetching the weights",
        "The worker is downloading this model from Hugging Face and checking it "
        "against its published digest. Happens once per model, per worker.",
    ),
    PHASE_LOADING_MODEL: (
        "Loading the model",
        "Reading the weights into memory. This cost is charged to residency, not "
        "to your run, so it is not in the energy figure below.",
    ),
    PHASE_GUARD_INPUT: (
        "Screening the request",
        "The worker runs one safety pass over the request before generating. Its cost "
        "is reported separately from the answer's.",
    ),
    PHASE_PLANNING: (
        "Planning",
        "The model is deciding which of the tools you allowed it should run.",
    ),
    PHASE_GENERATING: (
        "Generating",
        "The model is producing the answer on the worker's own hardware.",
    ),
    PHASE_CALLING_TOOL: (
        "Running a tool",
        "The model asked for a tool you allowed. It runs in the worker's sandbox; the "
        "call and what it returned are listed below.",
    ),
    PHASE_GUARD_OUTPUT: (
        "Screening the answer",
        "One safety pass over the generated answer before it is returned.",
    ),
    PHASE_WORKING: (
        "Working",
        "The run is in flight on the worker.",
    ),
}


def _phase_copy(phase: str) -> tuple[str, str]:
    return PHASE_COPY.get(normalise_phase(phase), PHASE_COPY[PHASE_WORKING])


def _library_specs() -> Mapping[str, Mapping[str, str]]:
    """``ref -> {kind, name, description}`` for members this server knows of.

    The descriptions come from the same package that defines the specs, which
    is the only place they exist as written text. They are needed because a
    reference like ``create_pdf@1`` says what a member is called and nothing
    about what it does, and a person choosing one should not have to guess.

    An advertised ref this server does not recognise still renders and still
    works. It simply arrives without a description, which is honest: this
    server has nothing to say about a member it has never seen.
    """

    try:
        import distinct_tools
        from distinct_tools.skills import REPOSITORY_SKILLS, REPOSITORY_SPECS

        adapted: dict[str, str] = {}
        for skill in REPOSITORY_SKILLS:
            source = str(getattr(skill, "source", "") or "")
            adapted[skill.tool_id] = (
                "Anthropic · adapted" if "anthropics/skills" in source else "distinct"
            )
        specs: dict[str, dict[str, str]] = {}
        for spec in tuple(distinct_tools.INSTALLABLE_SPECS) + tuple(REPOSITORY_SPECS):
            ref = f"{spec.tool_id}@{spec.version}"
            specs[ref] = {
                "kind": getattr(spec, "kind", "tool"),
                "name": str(spec.tool_id).replace("_", " "),
                "description": str(getattr(spec, "description", "")),
                "author": adapted.get(str(spec.tool_id), "distinct"),
            }
        return specs
    except Exception:
        return {}


def _library_kinds() -> Mapping[str, str]:
    """``ref -> kind``. Kept for callers that only need the label."""

    return {ref: str(spec["kind"]) for ref, spec in _library_specs().items()}


#: ONE RULE FOR TURNING A MODEL'S TEXT INTO A PERSON'S.
#:
#: There were two: this one for the picker's hover note, and a second in
#: render.py for the library card, written later and better. Two rules meant
#: the same member could read one way on the card and another in the picker,
#: and only one of them dropped the "Arguments: ..." tail that is meant for
#: tool selection rather than for a reader. The card's version is the rule.
_first_sentence = render.card_summary

AGENT_TABLE_HEADERS = [
    "Agent",
    "Status",
    "OS",
    "RAM",
    "Queue",
    "Est. wait",
    "Energy",
    "Tools",
]
_TERMINAL = {
    JobStatus.COMPLETED,
    JobStatus.FAILED,
    JobStatus.CANCELLED,
    JobStatus.EXPIRED,
}

ENERGY_MEASURED = "measured"
ENERGY_MISSING = "missing"

# Per-run ceilings written into ToolSelection.config and read by the worker's
# broker (distinct_agent/tools.py::effective_tool_limits).  They are limits the
# run may not exceed, not defaults a model may raise, and the worker clamps
# them again on arrival against its own hard caps.  Only keys the worker
# understands may appear here: an unknown key is refused rather than ignored.
PER_RUN_TOOL_LIMITS: Mapping[str, Any] = {
    # THREE WAS THE NUMBER THAT MATTERED, AND IT WAS THE SMALLEST OF FOUR.
    # The worker takes the lowest of this, its own hard cap and the harness
    # budget, so raising any of the others alone changed nothing. A request
    # that needs a plan, two conversions, a document and a spreadsheet is five
    # calls before anybody has been unreasonable.
    "max_calls": 10,
    "timeout_seconds": 15.0,
    "max_output_bytes": 131_072,
}


def _as_mapping(value: Any) -> Mapping[str, Any]:
    if hasattr(value, "to_dict"):
        try:
            value = value.to_dict()
        except Exception:  # pragma: no cover - defensive against odd payloads
            return {}
    return value if isinstance(value, Mapping) else {}


#: The agent mode chooser. Named for what the two modes do rather than for
#: the harnesses that implement them: "one pass" and "step by step" are the
#: difference a person can act on, and "structured-tool-loop" is not.
AGENT_MODE_CHOICES = [
    ("Simple · plan, run tools, answer", MODE_SIMPLE),
    ("Agent · reacts to each tool result", MODE_AGENT),
]

#: How a step is introduced in the transcript. The worker names the kind from
#: a closed vocabulary; the sentence is this file's.
_STEP_TITLE = {
    "model": "Thinking",
    "tool-call": "Using",
    "tool-result": "Result from",
    "phase": "",
}


def _step_messages(
    steps: Sequence[Mapping[str, Any]], *, running: bool
) -> list[dict[str, Any]]:
    """The agent's turns, as transcript entries that are not messages.

    The component renders an entry carrying ``metadata`` as a collapsible
    panel rather than as a speech bubble, which is the distinction this needs:
    a tool call happened *in* the conversation and belongs in the reading
    order, but nobody said it. Same shape every chat product has converged on,
    and the reason they all converged on it is that the alternative is either
    hiding the work or dressing it up as speech.

    ``running`` marks the last entry as pending, so the panel shows a live
    spinner while the step it describes is the one currently happening.
    """

    entries: list[dict[str, Any]] = []
    for index, step in enumerate(steps):
        if not isinstance(step, Mapping):
            continue
        kind = str(step.get("kind") or STEP_PHASE)
        text = str(step.get("text") or "")
        tool = str(step.get("tool") or "")
        lead = _STEP_TITLE.get(kind, "")
        if kind == "phase" or not text:
            # A phase marker is the run's own bookkeeping, already said by the
            # activity panel in words. It is not a turn the agent took.
            continue
        title = f"{lead} {tool}".strip() if tool else lead or "Working"
        if kind == "tool-result" and step.get("ok") is False:
            title = f"{tool or 'Tool'} refused"
        last = index == len(steps) - 1
        entries.append(
            {
                "role": "assistant",
                "content": text,
                "metadata": {
                    "title": title,
                    "status": "pending" if (running and last) else "done",
                },
            }
        )
    return entries


def _library_defaults(state: Any) -> tuple[str, ...]:
    """The library members this browser has chosen as its defaults.

    Held in browser state beside the transcript, for the same reason the
    transcript is: it is this person's preference and the server has no
    business keeping a profile of it. A browser that has never visited the
    library page has no defaults, which is correctly the empty tuple rather
    than a guess at what most people want.
    """

    values = _as_mapping(state).get("library_defaults")
    if not isinstance(values, (list, tuple)):
        return ()
    return tuple(str(item) for item in values if isinstance(item, str))


def _run_steps(value: Any) -> list[dict[str, Any]]:
    """The agent's steps, copied key by key rather than passed through.

    The protocol already bounds and normalises this on the way in. Doing it
    again here is not redundancy for its own sake: the payload reaches this
    function as a plain mapping decoded from the wire, and the only reason it
    is safe to render is that the keys are chosen here rather than taken from
    whatever the worker sent. Every string is re-truncated for the same
    reason.
    """

    if not isinstance(value, (list, tuple)):
        return []
    steps: list[dict[str, Any]] = []
    for item in value[:MAX_RUN_STEPS]:
        if not isinstance(item, Mapping):
            continue
        kind = str(item.get("kind") or STEP_PHASE)
        entry: dict[str, Any] = {
            "kind": kind if kind in STEP_KINDS else STEP_PHASE,
            "text": str(item.get("text") or "")[:MAX_LIVE_STEP_TEXT],
        }
        tool = item.get("tool")
        if isinstance(tool, str) and tool:
            entry["tool"] = tool[:128]
        if "ok" in item:
            entry["ok"] = item.get("ok") is True
        steps.append(entry)
    return steps


def _result_output(result: Any) -> str:
    payload = _as_mapping(result)
    if payload:
        value = payload.get("output", "")
        return value if isinstance(value, str) else str(value)
    return result if isinstance(result, str) else ""


def _finite_number(value: Any) -> Optional[float]:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    value = float(value)
    return value if math.isfinite(value) and value >= 0 else None


@dataclass(frozen=True)
class EnergyReading:
    """One run's use-phase energy, or an explicit absence of one.

    ``joules`` is ``None`` when nothing was measured.  It is never ``0.0`` as a
    stand-in for "we do not know": a missing measurement and a measured zero are
    different facts, and the rubric is explicit that missing never counts as
    zero.
    """

    status: str
    joules: Optional[float]
    scope: str
    provider: str
    reason: Optional[str] = None
    #: Real output token count from the runner, when the runner counted one.
    #: Never derived from characters: an estimated token is not a token.
    output_tokens: Optional[int] = None

    @property
    def measured(self) -> bool:
        return self.status == ENERGY_MEASURED

    @property
    def joules_per_token(self) -> Optional[float]:
        """Joules per generated token: the efficiency figure, when honest.

        Defined only when both halves were actually measured — a joule figure
        from the meter and a token count from the runner. This is the
        ``average power ÷ generation speed`` identity, computed on totals.
        """

        if not self.measured or not self.joules or not self.output_tokens:
            return None
        if self.output_tokens <= 0:
            return None
        return float(self.joules) / float(self.output_tokens)

    @property
    def line(self) -> str:
        """The one line a completed run carries in the transcript.

        It used to read "Energy: 16.750 J - scope: system-cpu-modelled -
        cpu-load-model - self-reported - use phase only", which is five true
        facts in the order an auditor wants and no order a reader does. What a
        person wants from their own answer is the cost and a sense of the size
        of it. The scope and the provider moved to the panel's disclosure,
        where they are still one click from every figure they qualify.
        """

        if not self.measured:
            # The em dash is the visual half of the honesty rule: an absent
            # measurement must never be able to read as a tidy zero at a
            # glance.  The words carry the same meaning for anyone who cannot
            # see the dash.
            detail = f" · {self.reason}" if self.reason else ""
            return f"Electricity: no data{detail} · not counted as zero"
        means = render.comparator(float(self.joules or 0.0))
        suffix = f" · {means}" if means else ""
        per_token = ""
        if self.joules_per_token is not None:
            per_token = (
                f" · {self.output_tokens} tok"
                f" · {render.format_joules_per_token(self.joules_per_token)}"
            )
        return (
            f"Electricity: **{render.format_joules(float(self.joules))}**"
            f"{per_token}{suffix}"
        )


def _energy_reading(result: Any) -> EnergyReading:
    """Read a worker's energy record without inventing a number for absence."""

    payload = _as_mapping(result)
    energy = payload.get("energy") if payload else None
    if not isinstance(energy, Mapping):
        return EnergyReading(
            ENERGY_MISSING, None, "unknown", "unavailable", "worker reported no energy record"
        )
    provider = str(energy.get("provider") or "unknown")[:80]
    scope = str(energy.get("scope") or "unknown")[:80]
    raw_reason = energy.get("reason")
    reason = str(raw_reason)[:200] if isinstance(raw_reason, str) and raw_reason else None

    if energy.get("available") is False:
        return EnergyReading(
            ENERGY_MISSING, None, scope, provider, reason or "meter reported unavailable"
        )

    # Current workers report the non-overlapping measurement as ``joules``.
    # The domain-specific names remain accepted for older worker builds.
    total = _finite_number(energy.get("joules"))
    if total is None:
        total = _finite_number(energy.get("total_joules"))
    if total is None:
        parts = [
            value
            for value in (
                _finite_number(energy.get("package_joules")),
                _finite_number(energy.get("dram_joules")),
            )
            if value is not None
        ]
        # An empty sum is an absence, not a zero.  The previous implementation
        # returned 0.0 here and the session total silently under-reported.
        total = sum(parts) if parts else None
    if total is None:
        return EnergyReading(
            ENERGY_MISSING, None, scope, provider, reason or "worker reported no joule value"
        )
    tokens = None
    usage = payload.get("usage")
    if isinstance(usage, Mapping):
        raw_tokens = usage.get("output_tokens")
        if isinstance(raw_tokens, int) and not isinstance(raw_tokens, bool) and raw_tokens > 0:
            tokens = raw_tokens
    return EnergyReading(ENERGY_MEASURED, total, scope, provider, output_tokens=tokens)


def _session_energy_markdown(readings: Sequence[EnergyReading]) -> str:
    """Total only what was measured, and say so when the total is incomplete.

    The figure leads and the boundary caveat follows. The caveat is essential
    and stays verbatim, but when it is set above the number in the same weight
    it becomes the loudest thing in the panel and the datum reads as an
    afterthought.
    """

    caveat = f"_{USE_PHASE_BOUNDARY}_"
    if not readings:
        return f"No completed runs in this conversation yet.\n\n{caveat}"

    measured = [reading for reading in readings if reading.measured]
    missing = len(readings) - len(measured)
    by_scope: dict[str, list[float]] = {}
    for reading in measured:
        by_scope.setdefault(reading.scope, []).append(float(reading.joules or 0.0))

    lines: list[str] = []
    if not by_scope:
        lines.append(
            f"**No data**. 0 of {len(readings)} completed runs returned a measurement. "
            "Unmeasured runs are not counted as zero."
        )
    elif len(by_scope) == 1:
        scope, values = next(iter(by_scope.items()))
        lines.append(
            f"**{sum(values):.3f} J** · scope: {scope} · measured across "
            f"{len(values)} of {len(readings)} completed runs"
        )
    else:
        # Different scopes are different measurement boundaries. Summing a
        # CPU-package figure with a whole-system figure would produce a number
        # that describes nothing, so no grand total is offered.
        for scope in sorted(by_scope):
            values = by_scope[scope]
            lines.append(
                f"**{sum(values):.3f} J** · scope: {scope} · {len(values)} of "
                f"{len(readings)} completed runs"
            )
        lines.append(
            "These scopes are different measurement boundaries, so no combined total "
            "is shown."
        )
    if missing and by_scope:
        lines.append(
            f"**Incomplete:** {missing} of {len(readings)} completed runs returned no "
            "measurement. The total above therefore understates this session by an "
            "unknown amount."
        )
    return "  \n".join(lines) + "\n\n" + caveat


def _session_energy_lines(readings: Sequence[EnergyReading]) -> tuple[list[str], bool]:
    """The same figures as the markdown, as lines plus a measured flag.

    Kept beside ``_session_energy_markdown`` deliberately: the markdown form is
    what the honesty tests assert against, and this is what the panel renders.
    Both must say the same thing, so both are derived from the same text.
    """

    markdown = _session_energy_markdown(readings)
    body = markdown.split("\n\n")[0]
    lines = [line.strip().replace("**", "") for line in body.split("  \n") if line.strip()]
    measured = any(reading.measured for reading in readings)
    return lines, measured


# The tagline names the mechanism, not the virtue.
#
# "Transparent", "honest", "responsible" and "sustainable" are all claims a
# competitor can make without doing anything, which is why they read as
# marketing and why none of them appears below. The construction that works
# here is the one Ecosia uses: state a verifiable action ("plants trees"), not
# a value. Two actions are available to this product and neither is available
# to a frontier provider, which is what makes them worth saying: the models run
# on machines people lend, and the energy is read off a meter on the machine
# that ran them.
#
# Options, in two registers.
#
#   Spec sheet, three concrete nouns:
#     a. "Small models, volunteered machines, measured energy."   <- in use
#     b. "Open models. Borrowed machines. Metered."
#
#   Ecosia-shaped, one verifiable action:
#     c. "The network that measures its own energy."
#     d. "AI that runs on computers people lend."
#
#   What the user gets:
#     e. "See what each answer costs."
#     f. "Metered inference on volunteered machines."
#
# (a) is in use: six words, three facts, no verb to over-promise with. (c) is
# the closest to the Ecosia construction and the runner-up.
#
# (e) and (f) are the two to avoid despite reading best, and the reason is the
# honesty rule rather than taste. Both use a verb that quantifies over runs, so
# both promise a figure for every answer. Measurement depends on the worker's
# hardware and a run on a machine without a usable meter returns no number at
# all. A tagline that implies otherwise would be contradicted by the first
# unmeasured run the user sees, which is the one thing this product cannot
# afford. The noun phrase in (a) describes what the network deals in without
# claiming it succeeds every time.
TAGLINE = "Small models, volunteered machines, measured energy."


def _identity_html(text: str, *, sign_in: str = "") -> str:
    """The header is one component, so identity changes re-render the row."""

    return render.header(text, tagline=TAGLINE, sign_in=sign_in)


def _status_html(text: str) -> str:
    """Run status as markup. ``text`` may name an agent, so it is escaped."""

    return (
        '<p class="c-note"><span class="c-note__text">'
        + render.esc(text).replace("\n\n", "<br>").replace("`", "")
        + "</span></p>"
    )


def _all_readings(state: Mapping[str, Any]) -> list[EnergyReading]:
    """Every completed run this browser holds, across every conversation.

    The user's true total lives here and nowhere else. The server erases a run
    the moment it finishes, so there is no server-side history to add up; what
    the browser kept is the whole record, and a figure drawn from it is
    complete in the only sense available.
    """

    readings: list[EnergyReading] = []
    conversations = state.get("conversations")
    if not isinstance(conversations, Mapping):
        return readings
    for conversation in conversations.values():
        if not isinstance(conversation, Mapping):
            continue
        for answer in (conversation.get("answers") or {}).values():
            if isinstance(answer, Mapping):
                readings.append(_energy_reading({"energy": answer.get("energy") or {}}))
    return readings


def _usage_section(
    readings: Sequence[EnergyReading], *, label: str
) -> list[dict[str, Any]]:
    """One readable figure per measurement scope, or one explicit absence.

    Scopes stay apart, because they are different measurement boundaries and a
    sum across them would describe nothing. In practice a session has one
    scope; when it has two, this yields two figures rather than one wrong one.
    """

    if not readings:
        return [
            {"joules": None, "label": label, "absent_reason": "no completed runs yet"}
        ]
    measured = [reading for reading in readings if reading.measured]
    missing = len(readings) - len(measured)
    if not measured:
        return [
            {
                "joules": None,
                "label": label,
                "absent_reason": (
                    f"none of the {len(readings)} completed runs returned a "
                    "measurement, and unmeasured is not zero"
                ),
            }
        ]
    by_scope: dict[str, list[float]] = {}
    for reading in measured:
        by_scope.setdefault(reading.scope, []).append(float(reading.joules or 0.0))
    sections: list[dict[str, Any]] = []
    for scope in sorted(by_scope):
        values = by_scope[scope]
        detail = f"{len(values)} of {len(readings)} runs"
        if missing:
            # An incomplete total is a floor, and says so where the figure is,
            # not only in the small print.
            detail += f", so this is a floor: {missing} returned no measurement"
        sections.append(
            {
                "joules": sum(values),
                "label": label if len(by_scope) == 1 else f"{label} · {scope}",
                "detail": detail,
            }
        )
    return sections


def _measurement_note(readings: Sequence[EnergyReading]) -> str:
    """The small print: what was measured, by what, on which boundary."""

    providers = sorted({reading.provider for reading in readings if reading.measured})
    scopes = sorted({reading.scope for reading in readings if reading.measured})
    parts = [
        "The electricity the worker's meter recorded while generating your "
        "answer, and nothing else. The cost of training the model is a "
        "separate figure on a separate page, never added to this one."
    ]
    if scopes:
        parts.append("What the meter covers: " + ", ".join(scopes) + ".")
    if providers:
        parts.append("Measured by: " + ", ".join(providers) + ".")
    parts.append(
        "Every figure is a hardware counter reading. There is no model and no "
        "estimate: a component either has a counter the worker can read, or it "
        "is named as absent and no number is invented for it."
    )
    parts.append(
        "On an NVIDIA GPU the driver keeps a running energy register in "
        "millijoules (NVML, Volta and later), so a run's energy is the exact "
        "difference between two reads, with no sampling. Older boards integrate "
        "board power on a 50 ms timer instead and say so. CPU package energy "
        "comes from RAPL: the powercap interface on Linux, and on Windows the "
        "Energy Meter counter read through PDH. Where a machine has both, both "
        "are read over the same window and added."
    )
    parts.append(
        "Where the runner reported real token counts, each answer also shows "
        "joules per generated token. Token counts are never estimated from "
        "characters."
    )
    parts.append(
        "What no counter can see is not guessed at: power-supply loss, fans, "
        "storage and the mainboard are real electricity that no multiplier is "
        "applied to reach. The figure is therefore a floor for the machine's "
        "true draw, covering exactly the components named in the scope."
    )
    parts.append(
        "Agent-mode runs cost more than generation alone: tool calls and the "
        "safety screen are separate passes, and the guard's cost is recorded "
        "separately from the answer's. Studies of agentic systems "
        "(e.g. arXiv:2604.00053) find this overhead can dominate. All figures "
        "are reported by the worker and cannot be verified from here. A run "
        "with no measurement is counted as a run, never as zero."
    )
    return " ".join(parts)


def _session_energy_html(
    readings: Sequence[EnergyReading],
    *,
    overall: Sequence[EnergyReading] | None = None,
    lifetimes: Sequence[Mapping[str, Any]] = (),
) -> str:
    """What this session has cost, in the order a person reads it.

    Figure, then what it equates to, then the small print. Two scopes only:
    this conversation, and everything this browser has run. The worker's own
    lifetime figure moved to the worker card, beside the machine it describes,
    where it is a fact about that worker rather than a third total competing
    with the user's own.
    """

    del lifetimes  # now rendered on the worker card
    sections = _usage_section(readings, label="This conversation")
    if overall is not None:
        sections += _usage_section(overall, label="Everything you have run")
    note = _measurement_note(list(readings) + list(overall or ()))
    return render.energy_panel(
        title="Electricity used",
        # NO "BOUNDARY 1 OF 2" ON THE FACE OF THIS PANEL.
        #
        # That label, and "never added to the release assessment: different
        # boundary, different question", are notes from the rubric's own
        # vocabulary to a reader who has read the rubric. To everyone else they
        # are two unexplained pieces of jargon sitting on top of the number
        # they came to see, and a caption nobody can parse does not make a
        # figure more honest; it makes the page harder to trust.
        #
        # The rule they encode has not moved an inch. The two boundaries are
        # still separate surfaces, ``combined_total`` still raises rather than
        # adding them, and the explanation is still here in full: it is inside
        # "How this is measured", written out in sentences, next to the scope
        # and the provider it belongs with.
        boundary="",
        lines=(),
        headlines=sections,
        small_print=note,
        caveat=USE_PHASE_BOUNDARY,
        never_sum="",
        measured=any(reading.measured for reading in readings),
    )


def _lifetime_lines(lifetimes: Sequence[Mapping[str, Any]]) -> list[str]:
    """Each worker's own total since it started, marked unverifiable.

    This is the only figure in the product that the server cannot check even
    in principle: it is a number a volunteer's machine reports about itself,
    covering runs from users this session never saw. It is worth showing
    because it is the honest answer to "how much has this machine spent", and
    it is worth labelling loudly for exactly the same reason.
    """

    if not lifetimes:
        return []
    lines = ["Each worker since it started, self-reported and unverifiable"]
    for entry in lifetimes:
        name = str(entry.get("name") or "worker")
        uptime = _finite_number(entry.get("uptime_seconds")) or 0.0
        runs = entry.get("runs") if isinstance(entry.get("runs"), int) else 0
        measured_runs = (
            entry.get("measured_runs") if isinstance(entry.get("measured_runs"), int) else 0
        )
        joules = _finite_number(entry.get("joules"))
        scope = str(entry.get("scope") or "unknown")
        uptime_text = render.duration(uptime)
        if joules is None or measured_runs == 0:
            lines.append(
                f"{name}: No data over {uptime_text} of uptime and {runs} runs. "
                "Not counted as zero."
            )
            continue
        floor = "" if measured_runs == runs else " (a floor: some runs were unmeasured)"
        lines.append(
            f"{name}: {joules:.3f} J · scope: {scope} · {measured_runs} of {runs} runs "
            f"measured over {uptime_text} of uptime{floor}"
        )
    return lines


def _run_receipt(
    turn: int,
    answer: Mapping[str, Any],
    reading: EnergyReading,
    tools: Sequence[ToolSelection],
) -> dict[str, Any]:
    """One completed turn's cost and trace, shaped for the receipts panel."""

    trace_items: list[dict[str, Any]] = []
    trace = answer.get("trace")
    if isinstance(trace, (list, tuple)):
        for entry in trace:
            if not isinstance(entry, Mapping):
                continue
            verdict = "ran" if entry.get("ok") else str(entry.get("error_code") or "refused")
            elapsed = entry.get("elapsed_ms")
            timing = f", {elapsed} ms" if isinstance(elapsed, int) else ""
            artifact = entry.get("artifact")
            produced = f", produced {artifact}" if artifact else ""
            trace_items.append(
                {
                    "tool": f"{entry.get('tool', '?')}@{entry.get('version', '?')}",
                    "ok": bool(entry.get("ok")),
                    "summary": f"{verdict}{timing}{produced}",
                }
            )
    artifacts = answer.get("artifacts")
    return {
        "turn": f"Answer {turn}",
        "cost": (
            render.format_joules(float(reading.joules))
            if reading.measured and reading.joules is not None
            else "No data"
        ),
        "means": (
            render.comparator(float(reading.joules or 0.0))
            if reading.measured
            else "not measured, and not counted as zero"
        ),
        "calls": answer.get("tool_calls") or 0,
        "files": len(artifacts) if isinstance(artifacts, (list, tuple)) else 0,
        "worker": str(answer.get("agent") or ""),
        "trace": trace_items,
        # The agent's own turns, if the worker reported them. An older worker
        # sends none and the receipt renders exactly as it did before, which is
        # the point of keeping this optional rather than required.
        "steps": answer.get("steps") or [],
        "tools": [f"{tool.id}@{tool.version}" for tool in tools],
    }


def _tool_line(spec_tools: Sequence[ToolSelection], calls_reported: Optional[int]) -> str:
    if not spec_tools:
        return "Tools allowed: none"
    names = ", ".join(f"`{tool.id}@{tool.version}`" for tool in spec_tools)
    count = 0 if calls_reported is None else calls_reported
    return f"Tools allowed: {names} · calls reported: {count}"


def _catalogue_markdown() -> str:
    """The two boundaries in prose, and where the working is.

    This used to print the whole assessment inline as a bullet per model. The
    assessment surface now renders the figures, their method, their notes and
    their primary sources as cards, so repeating them here produced two copies
    of the same claims that could drift apart. What stays is the part the cards
    cannot carry: what the two boundaries are, and why they never meet.
    """

    lines = [
        "#### Release assessment · cradle to release",
        "",
        f"_{RELEASE_BOUNDARY}_",
        "",
        "Every model here has a published assessment of its training. Each figure is "
        "shown as its publisher stated it, with the method used and a link to the "
        "source, so you can check it yourself. Nothing is ranked: a figure and an "
        "absence are not two points on one scale, and `Missing` is neither zero nor a "
        "low score.",
        "",
        "Models with no published assessment are not listed. This project's only "
        "claim is about disclosed, measured environmental cost, so a release nobody "
        "has published anything about is one it cannot recommend.",
        "",
        "#### Use-phase energy · per run",
        "",
        f"_{USE_PHASE_BOUNDARY}_",
        "",
        "Shown on each completed run and totalled per conversation, split by "
        "measurement scope. Never added to a release figure, and never converted into "
        "CO2e, water, land, materials or pollution: those conversions would be "
        "estimates, which this catalogue does not make.",
        "",
        "Minimum RAM is a hardware requirement of the file, not an environmental "
        "result. This catalogue carries no size- or memory-derived impact tier.",
    ]
    return "\n".join(lines)


#: Where the catalogue publishes the assessment set. Owned by the catalogue,
#: read here. Absent is handled rather than guarded against: the fallback below
#: reads the assessments in catalog.py, which now carry real published figures,
#: so an absent file degrades to a smaller surface rather than an empty one.
ASSESSMENTS_PATH = Path(__file__).resolve().parent.parent / "model_assessments.json"


def _rubric_display() -> Mapping[str, Any]:
    """The assessment file's ``website_display`` block, or an empty mapping.

    That block is the render contract: the data file itself says "Render from
    the website_display block, not from models[]". Reading anything else from
    the file here would re-open the key-name drift this function used to have.
    """

    try:
        payload = json.loads(ASSESSMENTS_PATH.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    if not isinstance(payload, Mapping):
        return {}
    display = payload.get("website_display")
    return display if isinstance(display, Mapping) else {}


def _rubric_warning() -> str:
    warning = _rubric_display().get("warning")
    return warning if isinstance(warning, str) else ""


def _further_reading() -> list[Mapping[str, str]]:
    items = _rubric_display().get("further_reading")
    if not isinstance(items, (list, tuple)):
        return []
    return [
        {"label": str(item.get("label") or ""), "url": str(item.get("url") or "")}
        for item in items
        if isinstance(item, Mapping) and item.get("url")
    ]


#: The published method words this surface will display. Anything else is shown
#: as stated but not painted as one of the confident states, and an area with no
#: value renders as an absence whatever its status field says.
_METHOD_WORDS = frozenset({"Measured", "Modelled", "Proxy-based"})


def _category_rows(record: Mapping[str, Any]) -> list[Mapping[str, Any]]:
    """The six areas of one record, in the catalogue's order, always all six.

    State is read before value, per the assessment file's own rendering rules.
    A record that omits an area entirely gets an explicit absence rather than a
    missing row, because a card with four rows reads as a model with four areas.
    """

    categories = record.get("categories")
    categories = categories if isinstance(categories, Mapping) else {}
    rows: list[Mapping[str, Any]] = []
    for area in RUBRIC_AREAS:
        cell = categories.get(area)
        cell = cell if isinstance(cell, Mapping) else {}
        reported = str(cell.get("state") or "missing") == "reported"
        value = cell.get("value")
        rows.append(
            {
                "label": AREA_LABELS.get(area, area),
                "status": str(cell.get("status") or "") if reported else "",
                "value": str(value) if reported and isinstance(value, (str, int, float)) else "",
                "unit": str(cell.get("unit") or "") if reported else "",
                "note": str(cell.get("note") or ""),
                "source_url": str(cell.get("source_url") or "") if reported else "",
            }
        )
    return rows


def _evidence_card(record: Mapping[str, Any], *, note: str = "") -> str:
    """One ``website_display.models`` record as an evidence card.

    No letter is computed, because there are no letters. Values, units and
    method words are passed through exactly as the file states them; a cell
    whose state is not ``reported`` renders as an absence regardless of what
    else it carries, so a malformed record can only ever lose a figure, never
    acquire one.
    """

    assessed = sum(1 for row in _category_rows(record) if row["value"])
    coverage = str(record.get("coverage") or "").strip()
    coverage_text = (
        f"{coverage.lower()} · {assessed} of {len(RUBRIC_AREAS)} areas published"
        if coverage and assessed
        else "no published result"
    )
    links = record.get("links")
    links = links if isinstance(links, (list, tuple)) else ()
    return render.model_evidence_card(
        name=str(record.get("name") or record.get("model_id") or "unnamed"),
        coverage=coverage_text,
        summary=str(record.get("summary") or ""),
        covers=str(record.get("covers") or ""),
        categories=_category_rows(record),
        links=[
            {"label": str(item.get("label") or ""), "url": str(item.get("url") or "")}
            for item in links
            if isinstance(item, Mapping) and item.get("url")
        ],
        note=note or str(record.get("boundary") or ""),
    )


#: The model selected when a session opens. The lowest published figures of any
#: assessed release this network can run; see model_assessments.json.
DEFAULT_MODEL_ID = "olmoe-1b-7b-0924-instruct"


def _assessment_by_model_id() -> Mapping[str, Mapping[str, Any]]:
    """``website_display.models`` keyed by record id, exact matches only.

    Deliberately no fuzzy matching, and no alias table invented here. The
    assessment records name releases ("OLMo 2 7B", "OLMoE 1B-7B 0924"); the
    catalogue names specific quantised checkpoints ("OLMo 2 1124 7B Instruct,
    Q4_K_M"). Those are not the same object, and a surface whose first rule is
    that a figure belongs to the thing that was assessed cannot have its own
    renderer guessing which figure goes with which file.

    So a catalogue entry gets an assessment when ``ASSESSMENT_ASSIGNMENT`` says
    which record covers it, and renders as not assessed otherwise. Extending the
    coverage is a change to the assessment data, which is where the evidence
    lives.
    """

    records = _rubric_display().get("models")
    if not isinstance(records, (list, tuple)):
        return {}
    return {
        str(record["model_id"]): record
        for record in records
        if isinstance(record, Mapping) and record.get("model_id")
    }


#: Which assessment record covers which catalogue entry, and what it leaves out.
#:
#: This table used to carry an ``impact`` boolean, because the surface had two
#: ordinal scales and one of them could transfer between checkpoints while the
#: other could not. Both scales are gone, and with them the reason for that
#: flag: a published figure either covers the checkpoint being offered or it
#: does not, and where it does not the card says so in words rather than
#: withholding a letter.
#:
#: The mapping itself stays, and stays explicit, for the reason it always
#: existed. Every entry below is a *pretraining* result being shown beside an
#: *instruction-tuned* checkpoint built on that pretraining run. That is a
#: partial coverage claim, not a complete one, and it is one substitution away
#: from the thing this catalogue refuses — quietly transferring a figure from
#: one release to another. Writing it down is what keeps the two apart.
ASSESSMENT_ASSIGNMENT: Mapping[str, Mapping[str, Any]] = {
    "olmoe-1b-7b-0924-instruct": {
        "record": "olmoe-1b-7b-0924",
        "note": (
            "Figures cover the 0924 pretraining run; the tuning stage that produced this "
            "checkpoint is not separately costed in the source."
        ),
    },
    "olmo-2-1124-7b-instruct": {
        "record": "olmo-2-1124-7b",
        "note": (
            "Figures cover the 1124 pretraining run; the tuning stage that produced this "
            "checkpoint is not separately costed in the source."
        ),
    },
}


def _assignment(model_id: str) -> tuple[Mapping[str, Any] | None, Mapping[str, Any]]:
    """``(assessment record, assignment rules)`` for one catalogue model."""

    rules = ASSESSMENT_ASSIGNMENT.get(model_id)
    if rules is None:
        return None, {}
    return _assessment_by_model_id().get(str(rules["record"])), rules


def _evidence_suffix(model_id: str) -> str:
    """The fragment appended to a model's name in the chooser.

    A letter used to go here. What replaced it is the fact the letter was
    standing in for: which areas somebody actually published, and how. That is
    longer, and it is the thing a person choosing a model can act on -- "energy,
    climate, water · measured" says what is known; "B" said what we thought of it.
    """

    record, _ = _assignment(model_id)
    if record is None:
        assessment = RELEASE_ASSESSMENT.get(model_id)
        if assessment is None or not assessment.has_any_result:
            return "no published assessment"
        published = [
            AREA_LABELS[area].split(" /", 1)[0].lower()
            for area in RUBRIC_AREAS
            if assessment.areas[area].status is not AreaStatus.MISSING
        ]
        return ", ".join(published) + " published"

    rows = _category_rows(record)
    published = [row["label"].split(" /", 1)[0].lower() for row in rows if row["value"]]
    if not published:
        return "no published assessment"
    methods = {row["status"] for row in rows if row["value"] and row["status"] in _METHOD_WORDS}
    if methods == {"Measured"}:
        method = "measured"
    elif "Measured" in methods:
        # Not "partly measured", which reads as a hedge about the measuring.
        # Some of these figures were metered and some were calculated from the
        # metered ones, and both halves of that belong in the phrase.
        method = "measured and modelled"
    elif methods:
        method = "modelled"
    else:
        method = "method not stated"
    return f"{', '.join(published)} · {method}"


def _short_label(label: str) -> str:
    """The model's name, without the packager that follows it.

    ``model_choices`` yields "OLMoE 1B-7B 0924 Instruct · Ai2 first-party GGUF";
    the part before the first separator is the model, and that is what a chooser
    needs.
    """

    return label.split(" · ", 1)[0].strip() or label


def model_choices_with_evidence() -> list[tuple[str, str]]:
    """Model options carrying what was published about them, not their RAM.

    The publisher is dropped from the option text and not from the product: it
    is on the worker card, on the assessment card and in the manifest the digest
    is checked against. In a 280px control it was 9 characters that pushed the
    disclosure out of view.
    """

    return [
        (f"{_short_label(label)} · {_evidence_suffix(model_id)}", model_id)
        for label, model_id in model_choices()
    ]


def _assessment_cards() -> list[str]:
    """Build the evidence cards from the assessment file's display block.

    Ordering follows the catalogue, and no ordering here implies a ranking:
    these are not scores, and two models with published figures in different
    areas are not on one scale.
    """

    display = _rubric_display()
    records = display.get("models")
    cards: list[str] = []

    if isinstance(records, (list, tuple)) and records:
        by_id: dict[str, Mapping[str, Any]] = {}
        for record in records:
            if isinstance(record, Mapping) and record.get("model_id"):
                by_id[str(record["model_id"])] = record
        # Catalogue order first, so the surface matches the chooser; anything the
        # file carries that the catalogue does not is appended rather than
        # dropped, because a record for a model we no longer ship is still a
        # published result somebody may be looking for.
        seen: set[str] = set()
        for model_id in MODEL_CATALOG:
            rules = ASSESSMENT_ASSIGNMENT.get(model_id) or {}
            record_id = str(rules.get("record") or "")
            record = by_id.get(record_id)
            if record is None:
                continue
            seen.add(record_id)
            cards.append(_evidence_card(record, note=str(rules.get("note") or "")))
        for record_id, record in by_id.items():
            if record_id not in seen:
                cards.append(_evidence_card(record))
        return cards

    # Fallback: the assessments in the catalogue itself. These now carry real
    # published figures rather than a uniform "no data", so an absent or
    # unreadable assessment file degrades to a smaller surface rather than an
    # empty one.
    for model_id, manifest in MODEL_CATALOG.items():
        assessment = RELEASE_ASSESSMENT[model_id]
        rows = [
            {
                "label": AREA_LABELS[area],
                "status": assessment.areas[area].status.value
                if assessment.areas[area].status is not AreaStatus.MISSING
                else "",
                "value": assessment.areas[area].value or "",
                "unit": assessment.areas[area].unit or "",
                "note": assessment.areas[area].note,
                "source_url": assessment.areas[area].source_url or "",
            }
            for area in RUBRIC_AREAS
        ]
        cards.append(
            render.model_evidence_card(
                name=manifest.label,
                coverage=assessment.coverage_display,
                summary=assessment.summary,
                covers=assessment.covers,
                categories=rows,
                links=list(assessment.links),
                note=assessment.reason,
            )
        )
    return cards


def _default_server_catalogue() -> "Allowance":
    """Everything this server may ever ask an agent to run.

    Models come from the catalogue this module already renders; library
    members (tools and skills) from ``distinct_tools`` when it is installed
    beside the server. This is what the consent handshake shows an agent
    operator, so it is the full honest set, not the currently-selectable one.
    """

    tools: tuple[str, ...] = ()
    try:
        import distinct_tools
        from distinct_tools.skills import REPOSITORY_SPECS

        # The installable set plus the digest-pinned repository: the
        # repository members are exactly as offerable by a worker, and a
        # catalogue that omits them silently strips them from every
        # operator approval — which is how a worker that offered a
        # repository skill ended up approved for none of them.
        members = tuple(distinct_tools.INSTALLABLE_SPECS) + tuple(REPOSITORY_SPECS)
        tools = tuple(sorted(f"{spec.tool_id}@{spec.version}" for spec in members))
    except Exception:
        tools = ()
    return Allowance(models=tuple(MODEL_CATALOG), tools=tools)


def _defang_file_fences(text: str) -> str:
    """Stop an attached file from closing the block that holds it.

    The file's own bytes go to the model between an ``ATTACHED FILE`` line and
    an ``END FILE`` line. A file containing that closing line would appear to
    end early, and whatever followed would read as the server talking rather
    than as the user's document.

    The rule is in `distinct_protocol.fences`, shared with the worker's
    conversation fences. It used to be written out here and again there, and
    both copies split on ``"\\n"`` alone: a marker delimited by a carriage
    return or by U+2028 was never seen as a line and went through untouched.
    Two copies of a rule are two chances to get it wrong, and this one was
    taken both times.
    """

    return _defang_fences(text, FILE_FENCE_BODIES)


def _identity_from_request(request: Optional[Any]) -> Optional[str]:
    """The viewer id in this request's session cookie, or None.

    Wrapped in a guard because `gr.Request` is a wrapper: the starlette
    request is on `.request`, the session only exists when the middleware is
    mounted, and neither is true on a Space. None here is not a failure, it is
    "this deployment signs people in the other way".
    """

    if request is None:
        return None
    underlying = getattr(request, "request", request)

    # `scope` first, and never `getattr(underlying, "session", None)`.
    # Starlette's `Request.session` is a property that *asserts* when the
    # middleware is absent, and `getattr` with a default swallows only
    # `AttributeError`, so the assertion went straight through this function.
    #
    # That was not a cosmetic bug. `SessionMiddleware` is mounted only on the
    # self-hosted path, so on a Space and under `DISTINCT_DEV_AUTH` every
    # handler that authorises raised `AssertionError` before it looked at the
    # profile: the two deployments this project documents were both entirely
    # unusable, and failed with a Starlette internals message rather than with
    # anything a reader could act on. It failed closed, which is why nothing
    # else caught it, and "broken shut" is not the same as "working".
    #
    # Gradio's own `_session_from_request` gates on the scope for exactly this
    # reason; this now does the same.
    scope = getattr(underlying, "scope", None)
    if not isinstance(scope, Mapping) or "session" not in scope:
        return None
    session = scope.get("session")
    if not isinstance(session, Mapping):
        return None
    identity = identity_from_session(session)
    return identity.viewer_id if identity is not None else None


class DistinctUI:
    def __init__(self, control_plane: Optional[ControlPlane] = None) -> None:
        self.control = control_plane or ControlPlane()
        if not self.control.catalogue:
            # The consent handshake needs the server to state what it may ask
            # for; an empty catalogue would give operators nothing to approve.
            self.control.catalogue = _default_server_catalogue()
        self.agent_api = AgentApi(self.control)
        self.sessions = SessionAuthorizer()
        # Where claimed run artifacts are stored, per session, so the Outputs
        # panel can offer them for download. Session-scoped and deleted with
        # the session.
        self.outputs_dir = Path("runtime-outputs").resolve()

    # Library ----------------------------------------------------------

    def _advertised_library_refs(self, viewer_id: Optional[str] = None) -> tuple[str, ...]:
        """Every exact ``id@version`` an agent this viewer may see advertises.

        The viewer matters: ``list_agents`` filters by visibility, so asking
        with no viewer returns nothing at all. That is exactly what happened —
        every caller here asked viewer-less, the picker's choice list was
        permanently empty, and "no tools installed" showed beside workers
        that were advertising a dozen. The library page's descriptions are
        public; which of them is *offered right now* is per-viewer.
        """

        refs: set[str] = set()
        try:
            for view in self.control.list_agents(viewer_id=viewer_id):
                for item in view.capabilities.tools:
                    refs.add(item if "@" in item else f"{item}@1.0.0")
        except ControlPlaneError:
            pass
        return tuple(sorted(refs))

    def _library_choices(self, viewer_id: Optional[str] = None) -> list[tuple[str, str]]:
        """Options for the composer's library picker, named for people.

        The label leads with a readable name and its kind, because that is
        what a person scans. The exact ref stays in the value, which is what
        the allowlist and the broker match on, so nothing about the policy
        boundary depends on how this reads.
        """

        specs = _library_specs()
        choices: list[tuple[str, str]] = []
        for ref in self._advertised_library_refs(viewer_id):
            spec = specs.get(ref)
            if spec is None:
                choices.append((f"{ref} · library member", ref))
                continue
            choices.append((f"{spec['name']} · {spec['kind']}", ref))
        return choices

    def _library_help(self) -> str:
        """The what-does-this-do note under the picker, one line per member."""

        specs = _library_specs()
        rows = []
        for ref in self._advertised_library_refs():
            spec = specs.get(ref)
            if spec is None:
                continue
            rows.append(
                {
                    "name": spec["name"],
                    "kind": spec["kind"],
                    "ref": ref,
                    "description": _first_sentence(spec["description"]),
                }
            )
        return render.library_help(rows)

    def _tool_selections(
        self,
        values: Optional[Sequence[str]],
        viewer_id: Optional[str] = None,
    ) -> tuple[ToolSelection, ...]:
        advertised = frozenset(self._advertised_library_refs(viewer_id))
        result = []
        for value in values or ():
            if not isinstance(value, str) or "@" not in value:
                raise ValueError("library selection is malformed")
            if value not in advertised:
                # A client cannot request a library member no agent offers.
                # The worker's own allowlist refuses again after this, so this
                # check is convenience; that one is the boundary.
                raise ValueError(f"library member {value!r} is not offered by any agent")
            tool_id, version = value.rsplit("@", 1)
            result.append(ToolSelection(tool_id, version, dict(PER_RUN_TOOL_LIMITS)))
        return tuple(result)

    def _library_components(
        self,
        selected: Optional[Sequence[str]] = None,
        *,
        defaults: Optional[Sequence[str]] = None,
        viewer_id: Optional[str] = None,
    ) -> tuple[Any, str]:
        """The refreshed selector plus its undernote, preserving selection.

        ``defaults`` are the members this person chose on the library page.
        One that a worker is actually offering arrives pre-ticked, which is
        the whole point of having chosen it; one that no worker offers is not
        invented, because ticking a member nobody can run would promise a
        capability the network does not have. A pre-ticked default is an
        ordinary tick and unticking it is an ordinary click: this decides
        what a run starts with, never what it is stuck with.
        """

        choices = self._library_choices(viewer_id)
        valid = {ref for _, ref in choices}
        if selected is None and defaults:
            value = [ref for ref in defaults if ref in valid]
        else:
            value = [item for item in (selected or ()) if item in valid]
        note = render.no_tools_note(NO_TOOLS_NOTICE) if not choices else self._library_help()
        return (
            gr.CheckboxGroup(
                choices=choices,
                value=value,
                visible=bool(choices),
                interactive=bool(choices),
            ),
            note,
        )

    # Library page -------------------------------------------------------

    #: How many members the page will draw at once. A search box is the real
    #: answer to a library of hundreds; this is the backstop for somebody who
    #: opens the page and searches for nothing, so that the first paint is a
    #: page rather than a wall.
    LIBRARY_PAGE_LIMIT = 60

    def _library_rows(self, query: str = "") -> tuple[list[Mapping[str, Any]], int]:
        """Every member this server knows of, filtered, with the total.

        Matching is over the name, the reference and the description, because
        a person looking for "the one that makes PDFs" will type "pdf" and a
        person looking for a known member will type its ref, and neither
        should have to know which field they are searching.
        """

        specs = _library_specs()
        needle = " ".join(str(query or "").split()).casefold()
        rows: list[Mapping[str, Any]] = []
        for ref, spec in sorted(specs.items(), key=lambda item: (item[1]["kind"], item[0])):
            haystack = f"{spec['name']} {ref} {spec['description']}".casefold()
            if needle and needle not in haystack:
                continue
            rows.append({"ref": ref, **spec})
        return rows[: self.LIBRARY_PAGE_LIMIT], len(rows)

    def _library_page(
        self, query: str, defaults: Sequence[str]
    ) -> tuple[Any, str, str]:
        """The page a person picks their defaults on: the transport, then what is drawn.

        The checkbox group is not a second way to choose; it is the same
        choice, off the screen, because the framework only reports its own
        controls. Every box is labelled with the bare reference so the card
        that forwards a click to it can find it by an exact string rather than
        by its position in a list that a filter can reorder.
        """

        rows, matched = self._library_rows(query)
        total = len(_library_specs())
        choices = [(row["ref"], row["ref"]) for row in rows]
        # Defaults outside the current filter are preserved rather than
        # dropped. Searching is a way of looking, not a way of deselecting,
        # and a search box that quietly unticked everything it hid would be
        # the worst possible behaviour for a page whose whole job is to
        # remember a choice.
        kept = [ref for ref in defaults if ref in _library_specs()]
        shown = [ref for ref in kept if ref in {row["ref"] for row in rows}]
        return (
            gr.CheckboxGroup(choices=choices, value=shown),
            render.library_summary(
                shown=len(rows), matched=matched, total=total, chosen=len(kept)
            ),
            render.library_cards(rows, kept),
        )

    def update_library_page(
        self, query: str, state: Mapping[str, Any]
    ) -> tuple[Any, str, str]:
        """Re-filter the page without touching what has been chosen."""

        return self._library_page(query, _library_defaults(state))

    def set_library_defaults(
        self,
        visible_choice: Sequence[str],
        query: str,
        state: Mapping[str, Any],
        profile: "gr.OAuthProfile | None" = None,
        request: "gr.Request | None" = None,
    ) -> tuple[Any, Any, str, str, Any, str]:
        """Record the defaults the page's cards are currently showing as on.

        The merge only ever replaces members currently on screen. Somebody who
        searches "pdf", drops one member and then searches "plan" has changed
        their mind about one member, not about every member the filter was
        hiding, and a page whose whole job is to remember a choice must not
        forget one because it was scrolled out of a filter.

        Values arrive from the page's own hidden group and are still filtered
        against the library, because a control in a browser is a suggestion.
        """

        visible_choice = [item for item in visible_choice or () if isinstance(item, str)]
        rows, _ = self._library_rows(query)
        on_screen = {row["ref"] for row in rows}
        previous = set(_library_defaults(state))
        chosen = (previous - on_screen) | {
            ref for ref in visible_choice if ref in on_screen
        }
        specs = _library_specs()
        updated = {**state, "library_defaults": sorted(ref for ref in chosen if ref in specs)}
        try:
            viewer_id = self._authorize(dict(state), profile, request)
        except LoginRequired:
            viewer_id = None
        selector, note = self._library_components(
            _library_defaults(updated),
            defaults=_library_defaults(updated),
            viewer_id=viewer_id,
        )
        _group, summary, _cards = self._library_page(query, _library_defaults(updated))
        # The cards are NOT re-rendered here. The click already moved the card
        # in the browser, and pushing fresh markup back would rebuild the grid
        # under the user's cursor — losing a hover preview mid-play and, on a
        # phone, moving the thing they just tapped. The summary updates
        # because it is the count, which is the feedback that matters.
        return updated, updated, summary, gr.skip(), selector, note

    # Outputs ----------------------------------------------------------

    def _session_outputs_dir(self, session_id: str) -> Path:
        safe = "".join(ch for ch in session_id if ch.isalnum() or ch in "-_")[:80]
        return self.outputs_dir / safe

    def _store_artifacts(
        self, session_id: str, job_id: str, artifacts: Sequence[Mapping[str, Any]]
    ) -> list[str]:
        """Write claimed artifacts into the session's outputs, return paths.

        Artifact names were validated at the protocol boundary. Uniqueness
        across runs comes from a per-run directory rather than a per-file
        prefix, because the prefix was the thing the user read. In a 296px
        column ``job_02d5f16f-cupcakes.pdf`` truncates to ``job... .pdf``, so
        every file in the panel had the same visible name and none of them
        said what it was. The run that produced a file is already named in the
        run list and in its receipt; the file's own name is what belongs on
        the file.
        """

        import base64 as _base64

        stored: list[str] = []
        run_folder = "".join(ch for ch in str(job_id) if ch.isalnum() or ch in "-_")[:64]
        directory = self._session_outputs_dir(session_id) / (run_folder or "run")
        for item in artifacts:
            if not isinstance(item, Mapping):
                continue
            name = str(item.get("name") or "artifact")
            encoded = item.get("base64")
            if not isinstance(encoded, str) or not encoded:
                continue
            try:
                payload = _base64.b64decode(encoded, validate=True)
            except Exception:
                continue
            directory.mkdir(parents=True, exist_ok=True)
            safe_name = Path(name).name
            target = directory / safe_name
            target.write_bytes(payload)
            stored.append(str(target))
        return stored

    def _session_output_files(self, session_id: str) -> list[str]:
        """Every artifact in the session, newest run first.

        One level of nesting, because that is exactly how deep
        :meth:`_store_artifacts` writes. ``inputs`` is skipped here and added
        by the caller, so uploads keep their own place in the list rather than
        being interleaved with the files runs produced.
        """

        directory = self._session_outputs_dir(session_id)
        if not directory.is_dir():
            return []
        files: list[str] = []
        for child in sorted(directory.iterdir()):
            if child.is_file():
                files.append(str(child))
            elif child.is_dir() and child.name != "inputs":
                files.extend(sorted(str(path) for path in child.iterdir() if path.is_file()))
        return files

    # Attachments ------------------------------------------------------

    #: Suffixes whose bytes this server will read as text and pass to a run.
    #: Deliberately a short allowlist rather than a sniff: guessing at an
    #: arbitrary file's encoding and handing the result to a model is how a
    #: binary ends up in a prompt as replacement characters.
    TEXT_SUFFIXES = frozenset(
        {".txt", ".md", ".markdown", ".csv", ".tsv", ".json", ".log", ".yaml", ".yml"}
    )
    #: The most attached text one run may carry. Small models have small
    #: contexts, and an attachment that displaces the question it came with
    #: has made the run worse rather than better.
    MAX_ATTACHED_CHARACTERS = 20_000

    def _session_inputs_dir(self, session_id: str) -> Path:
        return self._session_outputs_dir(session_id) / "inputs"

    def _store_attachments(
        self, session_id: str, uploads: Sequence[Any]
    ) -> list[dict[str, Any]]:
        """Copy uploads into the session, and read the ones that are text.

        They live beside the run outputs and under the same erasure: cleared
        with the session and by "Clear my data". A file the user brought is
        theirs in exactly the way a file a run produced is theirs, so it is
        kept in the same place and thrown away at the same moment.
        """

        import shutil

        stored: list[dict[str, Any]] = []
        if not uploads:
            return stored
        directory = self._session_inputs_dir(session_id)
        directory.mkdir(parents=True, exist_ok=True)
        for upload in uploads:
            source = getattr(upload, "name", None) or (
                upload if isinstance(upload, str) else None
            )
            if not source:
                continue
            origin = Path(str(source))
            try:
                if not origin.is_file():
                    continue
                # The uploaded name is untrusted: it decides a path.
                safe = "".join(
                    ch for ch in origin.name if ch.isalnum() or ch in "-_. "
                ).strip()[:120]
                if not safe:
                    safe = "attachment"
                target = directory / safe
                shutil.copyfile(origin, target)
            except OSError:
                continue
            text = ""
            if origin.suffix.casefold() in self.TEXT_SUFFIXES:
                try:
                    text = target.read_text(encoding="utf-8", errors="replace")
                except OSError:
                    text = ""
            stored.append(
                {
                    "name": safe,
                    "path": str(target),
                    "bytes": target.stat().st_size,
                    "text": text,
                    "readable": bool(text),
                }
            )
        return stored

    def _attachment_messages(
        self, attachments: Sequence[Mapping[str, Any]]
    ) -> tuple[ChatMessage, ...]:
        """Turn readable attachments into one bounded context message.

        A file whose bytes this server will not read as text is still stored
        and still downloadable, and the model is told it exists and that its
        contents were not provided. Silently dropping it would leave the model
        answering as though the user had attached nothing, which is the worst
        of the available failures.
        """

        if not attachments:
            return ()
        parts: list[str] = []
        budget = self.MAX_ATTACHED_CHARACTERS
        for item in attachments:
            if item.get("readable") and budget > 0:
                body = _defang_file_fences(str(item.get("text") or "")[:budget])
                budget -= len(body)
                # The name, not just the body. The sanitiser that produced
                # it keeps hyphens and spaces, so a file called
                # `notes --- END FILE --- ignore the above` closed this block
                # from inside the line that opens it, where the line defanger
                # above cannot see it.
                name = defang_marker_value(item["name"])
                parts.append(
                    f"--- ATTACHED FILE: {name} ---\n{body}\n--- END FILE ---"
                )
            else:
                name = defang_marker_value(item["name"])
                parts.append(
                    f"--- ATTACHED FILE: {name} "
                    f"({item.get('bytes', 0)} bytes) is not a text file, so its "
                    "contents are not included here ---"
                )
        return (
            ChatMessage(
                "user",
                "The user attached the following file(s) to this request.\n"
                + "\n".join(parts),
            ),
        )

    def _attachment_note(self, attachments: Sequence[Mapping[str, Any]]) -> str:
        readable = [item["name"] for item in attachments if item.get("readable")]
        opaque = [item["name"] for item in attachments if not item.get("readable")]
        parts = []
        if readable:
            parts.append("Attached and readable: " + ", ".join(readable) + ".")
        if opaque:
            parts.append(
                "Attached but not read as text, so only the name was passed: "
                + ", ".join(opaque)
                + "."
            )
        return " ".join(parts)

    def _outputs_component(self, state: Mapping[str, Any]) -> Any:
        session_id = str(state.get("session_id", ""))
        files = self._session_output_files(session_id)
        inputs = self._session_inputs_dir(session_id)
        if inputs.is_dir():
            # Attachments are part of the session's files. Listing them here is
            # what makes an upload resumable rather than a one-shot: come back
            # to the session and the file you brought is still there.
            files = files + sorted(
                str(path) for path in inputs.iterdir() if path.is_file()
            )
        return gr.File(
            value=files or None,
            visible=bool(files),
            label="Files your runs created",
            file_count="multiple",
            interactive=False,
        )

    # Authentication/session helpers ----------------------------------

    def _viewer(
        self,
        profile: Optional[Mapping[str, Any]],
        request: Optional[Any] = None,
    ) -> str:
        """Who is asking, from whichever sign-in this deployment has.

        On a Space, Hugging Face's OAuth is real and Gradio hands over a
        profile. Self-hosted, the profile is always absent, because Gradio's
        own routes are the mocked ones this project refuses to start into; the
        identity comes instead from the session cookie that
        `distinct_server.auth_routes` writes at the callback.

        The session is preferred over the profile rather than the other way
        round: it is the one that was proved against Hugging Face by this
        process, on this request, with a state and a nonce it issued itself.
        """

        user_id = _identity_from_request(request)
        if user_id is None:
            user_id = user_id_from_profile(profile)
        return user_id

    def _authorize(
        self,
        state: Mapping[str, Any],
        profile: Optional[Mapping[str, Any]],
        request: Optional[Any] = None,
    ) -> str:
        """Who is asking, and this browser's session made real on the server.

        A HANDLER WITH NO SESSION STATE MUST NOT INVENT ONE.

        This used to pass `str(state.get("session_id", ""))` straight to
        `create_session`, and that call reads an empty id as "mint me a new
        one". `refresh_agents` is wired without the session state, so its
        `state` is always `{}` — which meant every change of the model
        dropdown, every tick of a library box and every change of mode minted
        a fresh server-side session that nothing would ever delete, since
        `cleanup_expired` does not reap sessions.

        Two thousand and forty-eight of those and the registry is full. The
        failure is not local to the handler that filled it: `create_session`
        then raises `CapacityError` for *every* caller of this method, so
        `load_session` fails on every page load and the two-second timer fails
        on every tick, for everybody on the server at once. That is the
        "random errors" this was hunted for — it takes a while to build up and
        then it arrives everywhere.

        Identity does not need a session, so a caller without one gets the
        identity and nothing else is touched.
        """

        state = _as_mapping(state)
        user_id = self._viewer(profile, request)
        session_id = str(state.get("session_id", "") or "")
        if not session_id:
            return user_id
        self.sessions.bind(session_id, user_id)
        self.control.create_session(session_id)
        for conversation_id in state.get("conversations") or {}:
            self.control.create_conversation(session_id, str(conversation_id))
        return user_id

    def on_session_delete(self, state: Mapping[str, Any]) -> None:
        if isinstance(state, Mapping):
            session_id = str(state.get("session_id", ""))
            self.sessions.release(session_id)
            self.control.delete_session(session_id)
            self._delete_outputs(session_id)

    def _delete_outputs(self, session_id: str) -> None:
        import shutil

        directory = self._session_outputs_dir(session_id)
        if directory.is_dir():
            shutil.rmtree(directory, ignore_errors=True)

    def load_session(
        self,
        state: Mapping[str, Any],
        saved_state: Any,
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        # A transcript saved in the browser survives a reload; the server-side
        # session it belonged to does not. Adopting the saved state recreates
        # the session and conversations server-side (in _authorize) and keeps
        # every answer the browser already holds. This is the persistence the
        # privacy notice promises: the user holds their own history.
        #
        # Rebuilt rather than believed. See `ui_state.adopt_saved_state`: the
        # four top-level checks that used to stand here trusted everything one
        # level down, and a transcript written by an older build could make
        # this handler raise on every page load.
        #
        # The live state is put through the same rebuild, and a fresh session
        # is the last resort. This is the one handler that runs before any
        # other, so the shape it settles on is the shape every handler after
        # it can rely on — which is why none of them repeat this work.
        state = adopt_saved_state(saved_state) or adopt_saved_state(state) or new_session()
        updated = dict(state)
        messages: list[dict[str, str]] = []
        energy = _session_energy_markdown(())
        try:
            user_id = self._authorize(state, profile, request)
            identity = _identity_html(
                f"Signed in as {user_id}"
                if user_id != "local-development-user"
                else "Local development"
            )
            status = _status_html(
                "Ready. Select a model and the library members permitted for this run."
            )
            updated, messages, status_text, energy, activity = self._render(state)
            if messages:
                status = status_text
        except LoginRequired:
            identity = _identity_html("")
            status = _status_html(
                "Sign in with Hugging Face to use this server. Workers are shown "
                "only to people who have their access code."
            )
            activity = gr.HTML(value="", visible=False)
            user_id = None
        except (ControlPlaneError, ValueError) as exc:
            # THE PAGE HAS TO OPEN.
            #
            # This runs on every load, and anything that escapes it is a red
            # toast on arrival with no control to press and no way to guess
            # what went wrong. The one that actually happened was
            # `CapacityError: session registry is full`, which is a fact about
            # the server and not about this visitor, and it greeted every
            # visitor at once. The session is left un-materialised, the
            # transcript still draws from the browser's own copy, and the
            # reason goes to the log where somebody can act on it.
            _LOG.exception("load_session could not prepare this session")
            identity = _identity_html("")
            status = _status_html(
                "This server could not open a session just now. Your "
                f"conversation is safe in this browser. Reason: {exc}"
            )
            activity = gr.HTML(value="", visible=False)
            user_id = None
        states = self._conversation_states(updated)
        choices = conversation_choices(updated, states)
        selected = updated["active_conversation_id"]
        table, agent_choices, agent_value, narrowed = self._agent_components(
            DEFAULT_MODEL_ID, (), "auto", MODE_SIMPLE, viewer_id=user_id
        )
        library, library_note = self._library_components(
            None, defaults=_library_defaults(updated), viewer_id=user_id
        )
        page_group, page_summary, page_cards = self._library_page(
            "", _library_defaults(updated)
        )
        return (
            updated,
            updated,
            identity,
            gr.Radio(choices=choices, value=selected),
            render.conversation_tones(states),
            messages,
            render.worker_cards(table, narrowed_by=narrowed),
            gr.Dropdown(choices=agent_choices, value=agent_value),
            status,
            energy,
            library,
            library_note,
            self._outputs_component(updated),
            activity,
            # The library page is filled on arrival rather than on first open,
            # so a browser that already has defaults shows those cards already
            # selected the moment the page is opened.
            page_group,
            page_summary,
            page_cards,
        )

    def _live_steps(self, view: Any) -> list[Mapping[str, Any]]:
        """The steps a run has taken so far, from the worker holding it.

        Read on every refresh, so the transcript fills in as the agent works
        rather than arriving complete once the run is over. Returns an empty
        list for a run nobody is holding, which is the correct answer for a
        job still waiting for a worker.
        """

        agent_id = getattr(getattr(view, "spec", None), "target_agent_id", "")
        if not agent_id or agent_id == "unassigned":
            return []
        try:
            snapshot = self.control.get_agent(agent_id).snapshot
        except ControlPlaneError:
            return []
        raw = snapshot.live_steps.get(view.spec.id)
        if not isinstance(raw, (list, tuple)):
            return []
        return [step for step in raw if isinstance(step, Mapping)]

    # Conversation state ------------------------------------------------

    def _conversation_states(self, state: Mapping[str, Any]) -> dict[str, str]:
        """One state word per conversation, for the rail.

        Two sources, because neither is complete on its own. The server holds
        a run only until it reaches a terminal state, so a finished
        conversation has no live jobs at all and would read as empty; the
        browser holds the answers and knows it finished, but knows nothing
        about a run still in flight. Live jobs decide first because they are
        the only ones that can still change.
        """

        conversations = state.get("conversations")
        if not isinstance(conversations, Mapping):
            return {}
        live: dict[str, list[JobStatus]] = {}
        try:
            for view in self.control.list_jobs(str(state.get("session_id", ""))):
                live.setdefault(view.spec.conversation_id, []).append(view.status)
        except ControlPlaneError:
            pass
        states: dict[str, str] = {}
        for conversation_id, conversation in conversations.items():
            statuses = live.get(conversation_id, [])
            if JobStatus.RUNNING in statuses:
                states[conversation_id] = "generating"
            elif any(
                status
                in {
                    JobStatus.QUEUED,
                    JobStatus.OFFERED,
                    JobStatus.ACCEPTED,
                    JobStatus.BLOCKED,
                    JobStatus.STRANDED,
                }
                for status in statuses
            ):
                states[conversation_id] = "waiting"
            elif any(
                status in {JobStatus.FAILED, JobStatus.EXPIRED, JobStatus.CANCELLED}
                for status in statuses
            ):
                states[conversation_id] = "failed"
            elif _as_mapping(conversation).get("answers"):
                states[conversation_id] = "completed"
            else:
                states[conversation_id] = "idle"
        return states

    # Agent table ------------------------------------------------------

    def _agent_components(
        self,
        model_id: str,
        tool_values: Sequence[str],
        selected_agent: Optional[str],
        mode: str = "",
        viewer_id: Optional[str] = None,
    ) -> tuple[list[list[Any]], list[tuple[str, str]], str, str]:
        """The workers this viewer may choose between, and no trace of the rest.

        ``viewer_id`` is not optional in spirit. It was missing here while
        `compatible_agents` had supported it all along, so the worker table was
        the one surface that showed private agents to anyone who loaded the
        page. A signed-out visitor now gets an empty list, which is the plain
        reading of not being able to access anything without signing in.
        """

        try:
            tools = self._tool_selections(tool_values, viewer_id)
            views = self.control.compatible_agents(
                model_id, tools, viewer_id=viewer_id, include_offline=True
            )
        except (ControlPlaneError, ValueError):
            views = ()
        # A worker runs one harness, so choosing how the request is answered
        # narrows who can answer it, exactly as choosing a model does. The
        # filter is here rather than in the control plane because it is a
        # presentation of choice, not a rule about what may run: a worker that
        # does not offer the chosen mode is not eligible, and saying so by
        # leaving it out of the list is the same answer the model filter gives.
        narrowed_by = ""
        if mode:
            kept = tuple(
                view for view in views if mode in (view.capabilities.modes or (MODE_SIMPLE,))
            )
            # Only when the mode is what emptied a list that had something in
            # it. "No worker at all" and "no worker in this mode" are different
            # problems with different fixes, and one sentence for both told the
            # user nothing about which they had.
            if views and not kept:
                narrowed_by = mode
            views = kept
        rows: list[list[Any]] = []
        choices: list[tuple[str, str]] = [("Automatic · least loaded", "auto")]
        selectable_ids = {"auto"}
        for view in views:
            capabilities = view.capabilities
            snapshot = view.snapshot
            capacity = snapshot.queue_capacity if snapshot else capabilities.queue_capacity
            outstanding = view.outstanding_jobs
            if not view.online:
                status = "offline"
            elif snapshot and snapshot.status is AgentStatus.DRAINING:
                status = "draining"
            elif outstanding >= capacity:
                status = "overloaded"
            elif outstanding:
                status = "busy"
            else:
                status = "free"
            wait = snapshot.estimated_wait_s if snapshot else 0.0
            tools_text = ", ".join(capabilities.tools) if capabilities.tools else "none"
            rows.append(
                [
                    f"{capabilities.name} · {capabilities.agent_id[:12]}",
                    status,
                    f"{capabilities.os} / {capabilities.arch}",
                    f"{capabilities.ram_gb:g} GB",
                    f"{outstanding}/{capacity}",
                    f"{wait:.0f} s",
                    capabilities.energy_provider,
                    tools_text,
                    # What this machine costs per answer and in total. It is
                    # the question a person is actually asking when they pick
                    # a worker, and it used to be answered nowhere.
                    self._worker_energy(snapshot),
                ]
            )
            if status in {"free", "busy"}:
                energy = self._worker_energy(snapshot)
                per_run = ""
                joules = energy.get("joules")
                measured_runs = int(energy.get("measured_runs") or 0)
                if isinstance(joules, (int, float)) and measured_runs > 0:
                    per_run = f" · {render.format_joules(joules / measured_runs)}/run"
                label = (
                    f"{capabilities.name} · {status} · {outstanding}/{capacity}"
                    f" · ~{wait:.0f}s{per_run}"
                )
                choices.append((label, capabilities.agent_id))
                selectable_ids.add(capabilities.agent_id)
        rows.extend(self._unapproved_rows({row[0] for row in rows}))
        value = selected_agent if selected_agent in selectable_ids else "auto"
        return rows, choices, value, narrowed_by

    @staticmethod
    def _worker_energy(snapshot: Any) -> dict[str, Any]:
        if snapshot is None:
            return {}
        return {
            "joules": snapshot.lifetime_joules,
            "runs": snapshot.lifetime_runs,
            "measured_runs": snapshot.lifetime_measured_runs,
            "uptime_seconds": snapshot.uptime_seconds,
        }

    def _unapproved_rows(self, already_listed: set[str]) -> list[list[Any]]:
        """Cards for workers that are online but advertise no model at all.

        ``compatible_agents`` filters by model, so a worker whose approved set
        is empty matches nothing and disappears from every list. That produced
        the worst diagnostic in the product: a user saw "No compatible worker"
        while the operator watched their worker pair, poll and report itself
        healthy, and neither side had any way to connect the two facts.

        The worker now refuses to start in that state, so this should be a
        surface that stays empty. It is here because "should" is doing the work
        in that sentence: an older build, a worker started another way, or a
        capability set that narrowed after pairing all reach it, and an
        invisible worker is precisely the failure worth spending a card on.
        """

        rows: list[list[Any]] = []
        try:
            views = self.control.list_agents()
        except (AttributeError, ControlPlaneError):
            return rows
        for view in views:
            capabilities = view.capabilities
            if not view.online or capabilities.models:
                continue
            name = f"{capabilities.name} · {capabilities.agent_id[:12]}"
            if name in already_listed:
                continue
            snapshot = view.snapshot
            capacity = snapshot.queue_capacity if snapshot else capabilities.queue_capacity
            rows.append(
                [
                    name,
                    "unapproved",
                    f"{capabilities.os} / {capabilities.arch}",
                    f"{capabilities.ram_gb:g} GB",
                    f"0/{capacity}",
                    "0 s",
                    capabilities.energy_provider,
                    "none",
                ]
            )
        return rows

    def redeem_agent_code(
        self,
        access_code: str,
        model_id: str,
        tool_values: Sequence[str],
        selected_agent: str,
        mode: str,
        state: Mapping[str, Any],
        profile: "gr.OAuthProfile | None",
        request: "gr.Request | None" = None,
    ) -> tuple:
        """Take a code and do whatever that code means.

        ONE BOX, BECAUSE A PERSON HAS ONE CODE.

        A worker prints two codes over its life: a **claim code** while it waits
        to be adopted, and an **access code** afterwards for the people its
        operator wants to share it with. Those are different things and they are
        entered by different people, but nobody standing at this field knows or
        cares which kind they are holding -- they were handed a code and told to
        put it in. Two fields would make choosing wrongly possible, and choosing
        wrongly would produce "that code is not valid" for a code that was
        perfectly valid in the box next door. That is the exact shape of the
        failure this whole change exists to remove.

        So the claim is tried first and access second, and a code that is
        neither is refused in the words both refusals already used. Both are
        192-bit random strings drawn from the same alphabet, so in principle one
        value could be both; at 2**-192 that is not a case worth designing
        around, and the order makes it deterministic if it ever happened.
        Holding a code is the authorisation in either reading, so trying both
        grants nothing that trying one would not.

        Signing in is checked first and separately from the code, because they
        are two different refusals: one is "we do not know who you are" and the
        other is "that code is not valid". Collapsing them would tell a
        signed-out visitor whether a code was real.

        The refusal for a bad code says nothing about why. Every wrong answer
        reads the same, so the field cannot be used to learn which codes exist.
        """

        try:
            viewer_id = self._authorize(state, profile, request)
        except LoginRequired:
            rows, choices, value, narrowed = self._agent_components(
                model_id, tool_values, selected_agent, mode, viewer_id=None
            )
            return (
                _status_html("Sign in with Hugging Face before entering a code."),
                gr.Dropdown(choices=choices, value=value),
                render.worker_cards(rows, narrowed_by=narrowed),
            )

        typed = str(access_code or "")
        try:
            self.control.claim_agent(typed, viewer_id)
            message = _status_html(
                "That worker is yours. It appears below when it is online and "
                "offers the model and mode you picked. Share its access code with "
                "anyone else who should use it."
            )
        except (ControlPlaneError, ValueError):
            try:
                self.control.redeem_access_code(typed, viewer_id)
                message = _status_html(
                    "That worker is now available to you. It appears below when it "
                    "is online and offers the model and mode you picked."
                )
            except (ControlPlaneError, ValueError):
                message = _status_html("That code is not valid.")

        rows, choices, value, narrowed = self._agent_components(
            model_id, tool_values, selected_agent, mode, viewer_id=viewer_id
        )
        return (
            message,
            gr.Dropdown(choices=choices, value=value),
            render.worker_cards(rows, narrowed_by=narrowed),
        )

    def refresh_agents(
        self,
        model_id: str,
        tool_values: Sequence[str],
        selected_agent: str,
        mode: str = MODE_SIMPLE,
        state: Optional[Mapping[str, Any]] = None,
        profile: "gr.OAuthProfile | None" = None,
        request: "gr.Request | None" = None,
    ) -> tuple:
        # Declared, because it is used two lines down. It was not, and the
        # handler raised NameError on every change of the model dropdown, the
        # library checkboxes or the mode: a live crash on three of the most
        # common interactions in the interface, in a handler no test called.
        #
        # `profile` was annotated `Optional[Mapping[str, Any]]`, and Gradio
        # injects a profile only into a parameter annotated with its own
        # `OAuthProfile`. On a Space that is the *only* way anybody is
        # identified, so this handler saw a signed-out visitor no matter who
        # was using it, and the worker table emptied itself every time
        # somebody picked a model. The annotation is now the one Gradio
        # matches on, and the session state is wired in beside it, so this
        # authorises against the same session as every other handler instead
        # of against `{}`.
        #
        # Re-checked here rather than trusted from the last render. A hidden
        # component is still callable through the API, so every handler that
        # can list workers has to ask who is asking.
        try:
            viewer_id = self._authorize(state or {}, profile, request)
        except LoginRequired:
            viewer_id = None
        rows, choices, value, narrowed = self._agent_components(
            model_id, tool_values, selected_agent, mode, viewer_id=viewer_id
        )
        # The selector itself is deliberately not rewritten here. This handler
        # runs on the user's own interaction with the model dropdown or the
        # library checkboxes; writing the checkbox group back would race their
        # next click (two quick ticks and the second is undone by the first
        # tick's stale response). Choice-list changes arrive via the timer,
        # which skips unless the advertised set actually changed.
        return (
            render.worker_cards(rows, narrowed_by=narrowed),
            gr.Dropdown(choices=choices, value=value),
            gr.skip(),
            gr.skip(),
        )

    # Conversations/runs ----------------------------------------------

    def new_conversation(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        try:
            self._authorize(state, profile, request)
            updated, conversation_id = add_conversation(state)
            self.control.create_conversation(updated["session_id"], conversation_id)
            states = self._conversation_states(updated)
            return (
                updated,
                updated,
                gr.Radio(
                    choices=conversation_choices(updated, states),
                    value=conversation_id,
                ),
                render.conversation_tones(states),
                [],
                _status_html("New conversation created."),
                gr.HTML(value="", visible=False),
            )
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    def delete_conversation(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        """Erase the conversation on screen, here and on the server.

        There was no way to do this at all: a conversation could be started and
        never removed, so the rail only ever grew, and "Clear my data" — the one
        control that erased anything — took the whole session with it. Wanting
        one exchange gone is not the same as wanting everything gone.

        No confirmation step. What it erases is already erasable by design and
        gone from the server within a run's lifetime anyway, and a dialog in
        front of every deletion is how people learn to dismiss dialogs. The row
        disappearing is the feedback.
        """

        try:
            self._authorize(state, profile, request)
            removed = str(state.get("active_conversation_id") or "")
            updated = remove_conversation(state, removed)
            # The browser's copy is gone; tell the server so anything it still
            # holds for that conversation goes with it rather than ageing out.
            try:
                self.control.delete_conversation(updated["session_id"], removed)
            except (ControlPlaneError, ValueError):
                # It may already have gone: a conversation whose runs all
                # finished is cleaned up on its own. Nothing here depends on
                # it still being there.
                pass
            if removed in updated["conversations"]:
                # The last one was emptied rather than removed, so the server
                # needs it to exist again for the next run in it.
                self.control.create_conversation(updated["session_id"], removed)
            updated, messages, status, energy, activity = self._render(updated)
            states = self._conversation_states(updated)
            return (
                updated,
                updated,
                gr.Radio(
                    choices=conversation_choices(updated, states),
                    value=updated["active_conversation_id"],
                ),
                render.conversation_tones(states),
                messages,
                _status_html("Conversation deleted."),
                energy,
                activity,
                self._outputs_component(updated),
            )
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    def change_conversation(
        self,
        state: Mapping[str, Any],
        conversation_id: str,
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        try:
            self._authorize(state, profile, request)
            updated = select_conversation(state, conversation_id)
            updated, messages, status, energy, activity = self._render(updated)
            return updated, updated, messages, status, energy, activity
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    def submit_run(
        self,
        prompt: str,
        model_id: str,
        tool_values: Sequence[str],
        target_agent: str,
        consent: bool,
        uploads: Sequence[Any],
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        try:
            user_id = self._authorize(state, profile, request)
            if consent is not True:
                raise ValueError("Acknowledge the community-worker warning before submitting")
            if not isinstance(prompt, str) or not prompt.strip():
                raise ValueError("Enter a request")
            if model_id not in MODEL_CATALOG:
                raise ValueError("Select a supported model")
            tools = self._tool_selections(tool_values, user_id)
            conversation = active_conversation(state)
            parent_job_id = conversation["job_ids"][-1] if conversation["job_ids"] else None
            stored = self._store_attachments(state["session_id"], uploads or ())
            context = self._context_messages(state) + self._attachment_messages(stored)
            view = self.control.submit_job(
                state["session_id"],
                state["active_conversation_id"],
                prompt.strip(),
                model_id,
                tools,
                parent_job_id=parent_job_id,
                target_agent_id=None if target_agent == "auto" else target_agent,
                # The worker appends ``spec.prompt`` exactly once.  This field
                # contains only the preceding conversation snapshot.
                messages=context,
                # Enough calls for a request that needs several tools in
                # sequence. The worker clamps this against its own ceiling, so
                # this is a request rather than a grant.
                inference_limits={"max_output_tokens": 2048, "max_tool_calls": 10},
                viewer_id=user_id,
            )
            note = self._attachment_note(stored)
            recorded = prompt.strip() + (f"\n\n[{note}]" if note else "")
            updated = append_job(state, job_id=view.spec.id, prompt=recorded)
            updated, messages, status, energy, activity = self._render(updated)
            submit_states = self._conversation_states(updated)
            return (
                updated,
                updated,
                messages,
                "",
                [],  # Library selection deliberately resets for the next run.
                status,
                energy,
                gr.Radio(
                    choices=conversation_choices(updated, submit_states),
                    value=updated["active_conversation_id"],
                ),
                render.conversation_tones(submit_states),
                activity,
                # The attachment control clears too: an upload belongs to the
                # request that carried it, and leaving it loaded would attach
                # the same file to the next one without being asked.
                None,
                self._outputs_component(updated),
            )
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    def cancel_latest(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        try:
            self._authorize(state, profile, request)
            jobs = self.control.list_jobs(state["session_id"], state["active_conversation_id"])
            target = next((job for job in reversed(jobs) if job.status not in _TERMINAL), None)
            if target is None:
                raise ValueError("There is no active or queued run to cancel")
            self.control.cancel_job(state["session_id"], target.spec.id)
            updated, messages, status, energy, activity = self._render(state)
            return updated, updated, messages, status, energy, activity
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    #: One `gr.skip()` per output the timer is wired to, for a tick that could
    #: not run at all. Asserted against the wiring in :meth:`build`, because a
    #: length that drifts from the outputs list would make the safety net
    #: itself the error it exists to prevent.
    TIMER_OUTPUTS = 13

    def refresh_session(
        self,
        state: Mapping[str, Any],
        model_id: str,
        tool_values: Sequence[str],
        selected_agent: str,
        mode: str,
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        """One tick of the two-second refresh, which can never fail loudly.

        Everything below degrades in place: a failure inside the render is
        caught where it happens and the rest of the tick still runs. This
        outermost guard is for the failure nobody predicted, and it is a bare
        `Exception` on purpose. A user-initiated handler should be loud,
        because somebody is standing there having just pressed something and a
        toast tells them what happened. Nobody presses this. It runs unwatched
        thirty times a minute in every open tab, so a raise here is not an
        error message, it is an error message repeating forever, and the
        person reading it did nothing to cause it and can do nothing about it.
        """

        try:
            return self._refresh_session(
                state, model_id, tool_values, selected_agent, mode, profile, request
            )
        except Exception:  # noqa: BLE001 - see the docstring; the log keeps it visible
            _LOG.exception("refresh_session skipped a tick")
            return tuple(gr.skip() for _ in range(self.TIMER_OUTPUTS))

    def _refresh_session(
        self,
        state: Mapping[str, Any],
        model_id: str,
        tool_values: Sequence[str],
        selected_agent: str,
        mode: str,
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        updated = dict(state)
        try:
            viewer_id = self._authorize(state, profile, request)
            self.control.cleanup_expired()
            updated, messages, status, energy, activity = self._render(state)
        except LoginRequired:
            viewer_id = None
            messages, status = [], _status_html("Sign in to run a model.")
            energy = _session_energy_html(())
            activity = gr.HTML(value="", visible=False)
        except (ControlPlaneError, ValueError):
            # THE TRANSCRIPT IS LEFT ALONE, AND THE REST OF THE TICK RUNS.
            #
            # `LoginRequired` was caught here and `ControlPlaneError` was not,
            # and the one that actually arrived was `CapacityError: session
            # registry is full` from inside `_authorize` — a fact about the
            # server, raised for every user of it at the same instant, on a
            # handler that fires every two seconds. That is the shape of the
            # "random errors" this was hunted for.
            #
            # Skipping these four outputs leaves what is on screen exactly as
            # it is, which is right: the transcript is the browser's own copy
            # and the server has nothing to add this tick. The worker table
            # below still refreshes, so a server that recovers is visible
            # without a reload.
            _LOG.exception("refresh_session could not render this tick")
            viewer_id = None
            messages, status = gr.skip(), gr.skip()
            energy = gr.skip()
            activity = gr.skip()
        rows, choices, value, narrowed = self._agent_components(
            model_id, tool_values, selected_agent, mode, viewer_id=viewer_id
        )
        # Only push a new library selector when the advertised set actually
        # changed. The timer fires every five seconds, and re-rendering the
        # CheckboxGroup on every tick races the user's own clicks: a box
        # ticked while a tick was in flight would be silently unticked by the
        # stale value coming back.
        current_refs = self._advertised_library_refs(viewer_id)
        cache = getattr(self, "_last_library_refs", None)
        if not isinstance(cache, dict):
            cache = {}
            self._last_library_refs = cache
        if current_refs != cache.get(viewer_id):
            cache[viewer_id] = current_refs
            library, library_note = self._library_components(
                tool_values or None,
                defaults=_library_defaults(updated),
                viewer_id=viewer_id,
            )
        else:
            library, library_note = gr.skip(), gr.skip()
        # THE RAIL HAS TO FOLLOW THE RUNS, AND ONLY WHEN THEY MOVE.
        #
        # Without this the states were written once, when the run was
        # submitted, and never again: a conversation queued at the moment of
        # submission still read "waiting" long after its answer had arrived,
        # which is worse than showing no state at all. A state that is only
        # ever right at the instant it is written is a decoration.
        #
        # Pushed only when the map changes, for the same reason the library
        # selector is: this fires every two seconds, and re-rendering the list
        # on every tick would race a click on it and move the user to another
        # conversation mid-sentence. The comparison is held in the browser's
        # own state rather than on this object, which is shared by every
        # session connected to the server.
        states = self._conversation_states(updated)
        if states != updated.get("rail_states"):
            updated = {**updated, "rail_states": states}
            rail = gr.Radio(
                choices=conversation_choices(updated, states),
                value=updated["active_conversation_id"],
            )
            tones = render.conversation_tones(states)
        else:
            rail, tones = gr.skip(), gr.skip()
        return (
            updated,
            updated,
            messages,
            status,
            energy,
            render.worker_cards(rows, narrowed_by=narrowed),
            gr.Dropdown(choices=choices, value=value),
            library,
            library_note,
            self._outputs_component(updated),
            activity,
            rail,
            tones,
        )

    def reassign_latest(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
    ) -> tuple:
        """Move the newest stranded run in this conversation to another agent."""

        try:
            user_id = self._authorize(state, profile, request)
            self.control.cleanup_expired()
            jobs = self.control.list_jobs(state["session_id"], state["active_conversation_id"])
            target = next(
                (job for job in reversed(jobs) if job.status is JobStatus.STRANDED), None
            )
            if target is None:
                raise ValueError("There is no stranded run to reassign")
            self.control.reassign_job(
                state["session_id"], target.spec.id, viewer_id=user_id
            )
            updated, messages, status, energy, activity = self._render(state)
            return updated, updated, messages, status, energy, activity
        except (ControlPlaneError, LoginRequired, ValueError) as exc:
            raise gr.Error(str(exc)) from exc

    def clear_my_data(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None = None,
        request: gr.Request | None = None,
    ) -> tuple:
        """The working control the interface promises: erase everything now.

        Server session, its jobs, its outputs directory, the browser copy of
        the transcript, and the on-screen state, in one click. What returns is
        indistinguishable from a first visit.

        **This authorises, and it did not.** It was the one wired handler with
        no `profile` and no `request`, on the reasoning that erasing your own
        data needs no permission. The reasoning was wrong twice. The session id
        it erases comes from the caller, and `delete_session` checks no owner,
        so anybody who learned or guessed another person's session id could
        delete their runs and their outputs. And every call mints a fresh
        server-side session, so an unauthenticated caller could fill the
        server's session table on their own. `api_visibility="private"` looks
        like it prevents that and does not: Gradio resolves any handler by
        name.

        A signed-out caller now gets a fresh blank state and touches nothing,
        which is the honest answer to "erase my data" from somebody who has no
        data here.

        **"Touches nothing" was still a sentence rather than the code.** The
        authorisation was added and the `create_session` under it was left
        unconditional, so the second half of the hole described above was
        never actually closed: a signed-out caller looping this handler filled
        the session registry exactly as before, and a full registry makes
        every other handler on the server raise. The fresh session is now
        materialised only for somebody the server knows, and a signed-out
        caller gets the blank state and no server-side trace at all.
        """

        try:
            self._authorize(state, profile, request)
            session_id = (
                str(state.get("session_id", "")) if isinstance(state, Mapping) else ""
            )
        except LoginRequired:
            session_id = ""
        fresh = new_session()
        if session_id:
            self.sessions.release(session_id)
            self.control.delete_session(session_id)
            self._delete_outputs(session_id)
            # One out, one in. The replacement is materialised only when
            # there was something to replace, so this handler cannot add a
            # session to the registry however many times it is called — the
            # hole the docstring above describes and the code left open.
            # A caller with no session gets the blank state and no
            # server-side trace, and the next timer tick creates the session
            # for them if they ever need one.
            self.control.create_session(fresh["session_id"])
            for conversation_id in fresh["conversations"]:
                self.control.create_conversation(fresh["session_id"], conversation_id)
        library, library_note = self._library_components(())
        fresh_states = self._conversation_states(fresh)
        return (
            fresh,
            fresh,
            gr.Radio(
                choices=conversation_choices(fresh, fresh_states),
                value=fresh["active_conversation_id"],
            ),
            render.conversation_tones(fresh_states),
            [],
            _status_html("All session data has been erased, here and in this browser."),
            _session_energy_html(()),
            library,
            library_note,
            self._outputs_component(fresh),
            gr.HTML(value="", visible=False),
        )

    def _context_messages(self, state: Mapping[str, Any]) -> tuple[ChatMessage, ...]:
        """Build follow-up context from the browser's transcript.

        The server no longer holds finished prompts, so this reads the copy the
        browser owns.  For a turn queued behind an unfinished parent the server
        rebuilds the context itself at release time; see
        ``ControlPlane._context_from_parent_locked``.
        """

        messages: list[ChatMessage] = []
        for turn in conversation_turns(state)[-20:]:
            if turn["prompt"]:
                messages.append(ChatMessage("user", turn["prompt"]))
            answer = turn["answer"] or {}
            output = answer.get("output") if isinstance(answer, Mapping) else None
            if isinstance(output, str) and output:
                messages.append(ChatMessage("assistant", output[:65_536]))
        return tuple(messages)

    def _agent_name(self, agent_id: str) -> str:
        try:
            return self.control.get_agent(agent_id).capabilities.name
        except ControlPlaneError:
            return agent_id

    def _answer_payload(self, view: JobView, claimed: Any) -> dict[str, Any]:
        payload = _as_mapping(claimed)
        events = payload.get("tool_events")
        # The per-turn trace, kept rather than reduced to a count (defect A7).
        # Each entry is bounded and worker-supplied, so only known keys are
        # copied and every one is re-truncated here before it can be rendered.
        trace: list[dict[str, Any]] = []
        if isinstance(events, (list, tuple)):
            for event in events[:64]:
                if not isinstance(event, Mapping):
                    continue
                entry: dict[str, Any] = {
                    "tool": str(event.get("tool_id", ""))[:128],
                    "version": str(event.get("version", ""))[:32],
                    "ok": event.get("ok") is True,
                }
                if isinstance(event.get("elapsed_ms"), int):
                    entry["elapsed_ms"] = event["elapsed_ms"]
                if isinstance(event.get("error_code"), str):
                    entry["error_code"] = event["error_code"][:80]
                if isinstance(event.get("artifact"), str):
                    entry["artifact"] = event["artifact"][:120]
                trace.append(entry)
        usage = _as_mapping(payload.get("usage"))
        harness = usage.get("harness")
        return {
            "output": _result_output(claimed) or "The worker returned an empty response.",
            "energy": dict(_as_mapping(payload.get("energy"))),
            "tool_calls": len(trace),
            "trace": trace,
            # What the agent actually did, turn by turn, kept with the answer.
            # The live stream showed this while the run was in flight and then
            # dropped it, so a finished turn could only say what came out and
            # never how. Worker-supplied and re-bounded here, like everything
            # else that crosses the wire.
            "steps": _run_steps(payload.get("steps")),
            "agent": self._agent_name(view.spec.target_agent_id),
            "tools": [
                {"id": tool.id, "version": tool.version} for tool in view.spec.allowed_tools
            ],
            "guard": dict(_as_mapping(_as_mapping(payload.get("usage")).get("guard"))),
            # Kept so the next run of the same model can be given an estimate.
            # Wall-clock from submission to delivery, which is the number the
            # user actually waited, not the worker's self-reported inference
            # time. It stays in the browser with the rest of the transcript.
            "model": view.spec.model_id,
            "duration_seconds": max(0.0, float(view.updated_at) - float(view.spec.created_at)),
            "mode": str(harness)[:64] if isinstance(harness, str) else "",
        }

    # Estimates ---------------------------------------------------------

    def _estimate_seconds(
        self, state: Mapping[str, Any], model_id: str
    ) -> tuple[Optional[float], str]:
        """Expected duration for the next run of ``model_id``, and its basis.

        Built only from runs this browser has already seen finish, because
        that is the only history that exists: the server erases a run when it
        completes. One prior run is enough to say something useful, and the
        note always states how many it is drawn from, so a single sample
        cannot read as a measurement.
        """

        durations: list[float] = []
        conversations = state.get("conversations")
        if not isinstance(conversations, Mapping):
            return None, ""
        for conversation in conversations.values():
            if not isinstance(conversation, Mapping):
                continue
            for answer in (conversation.get("answers") or {}).values():
                if not isinstance(answer, Mapping):
                    continue
                if answer.get("model") != model_id:
                    continue
                value = answer.get("duration_seconds")
                if isinstance(value, (int, float)) and not isinstance(value, bool):
                    if math.isfinite(value) and value > 0:
                        durations.append(float(value))
        if not durations:
            return None, ""
        # The most recent handful, not all of history: a worker that has since
        # been replaced by a faster one should stop dragging the estimate.
        recent = durations[-5:]
        mean = sum(recent) / len(recent)
        label = model_id
        manifest = MODEL_CATALOG.get(model_id)
        if manifest is not None:
            label = manifest.label
        plural = "run" if len(recent) == 1 else "runs"
        return mean, (
            f"Estimate only, from the last {len(recent)} completed {plural} of "
            f"{label} in this browser. A different worker will take a different time."
        )

    # Live activity -----------------------------------------------------

    def _activity(
        self, state: Mapping[str, Any], views: Mapping[str, JobView]
    ) -> str:
        """The in-flight panel, or an empty string when nothing is running.

        Only the newest non-terminal run in the active conversation is shown.
        Earlier queued runs are already visible as positions in the transcript,
        and stacking spinners would say less, not more.
        """

        turns = conversation_turns(state)
        target: Optional[JobView] = None
        for turn in reversed(turns):
            view = views.get(turn["job_id"])
            if view is not None and view.status not in _TERMINAL:
                target = view
                break
        if target is None:
            return ""

        snapshot = None
        if target.spec.target_agent_id and target.spec.target_agent_id != "unassigned":
            try:
                snapshot = self.control.get_agent(target.spec.target_agent_id).snapshot
            except ControlPlaneError:
                snapshot = None

        job_id = target.spec.id
        steps: Sequence[Mapping[str, Any]] = ()
        note = ""
        phase = PHASE_WORKING
        tone = "work"
        if snapshot is not None:
            raw_steps = snapshot.live_steps.get(job_id)
            if isinstance(raw_steps, (list, tuple)):
                steps = [step for step in raw_steps if isinstance(step, Mapping)]
            raw_note = snapshot.progress_notes.get(job_id)
            note = raw_note if isinstance(raw_note, str) else ""
            phase = normalise_phase(snapshot.progress_phase.get(job_id))

        if target.status is JobStatus.STRANDED:
            tone = "wait"
            if target.spec.target_agent_id == "unassigned":
                headline, detail = _phase_copy(PHASE_WAITING_FOR_WORKER)
            else:
                headline = "The worker stopped responding"
                detail = (
                    f"{self._agent_name(target.spec.target_agent_id)} was holding this "
                    "run and has stopped polling. Nothing is lost: wait for it to come "
                    "back, press Reassign to move the run to another compatible worker, "
                    "or cancel it."
                )
            steps = ()
        elif target.status is JobStatus.BLOCKED:
            tone = "wait"
            headline = "Waiting for the previous turn"
            detail = (
                "Turns run in order, so this starts when the one before it finishes."
            )
            steps = ()
        elif target.status is JobStatus.OFFERED:
            headline, detail = _phase_copy(PHASE_OFFERED)
            detail = f"{detail} Worker: {self._agent_name(target.spec.target_agent_id)}."
            steps = ()
        elif target.status in {JobStatus.ACCEPTED, JobStatus.QUEUED}:
            headline, detail = _phase_copy(PHASE_QUEUED_ON_WORKER)
            position = target.queue_position
            if isinstance(position, int) and position > 0:
                detail = f"{detail} Position {position} in its queue."
            steps = ()
        else:
            headline, detail = _phase_copy(phase)
            if note:
                # Worker prose, escaped by render.activity. Kept beside the
                # server's own sentence rather than replacing it, so the
                # explanation never depends on a remote machine's wording.
                detail = f"{detail} Worker reports: {note}."

        started = float(target.spec.created_at)
        if snapshot is not None:
            worker_start = snapshot.started_at.get(job_id)
            if isinstance(worker_start, (int, float)) and worker_start > 0:
                started = min(started, float(worker_start)) or started
        elapsed = max(0.0, time.time() - started)
        estimate, estimate_note = self._estimate_seconds(state, target.spec.model_id)
        # An estimate is only meaningful once the run is actually being worked
        # on. Showing a countdown against a queue of unknown length would be a
        # guess dressed as a measurement.
        if target.status is not JobStatus.RUNNING:
            estimate, estimate_note = None, ""

        return render.activity(
            headline=headline,
            detail=detail,
            steps=steps,
            elapsed_seconds=elapsed,
            estimate_seconds=estimate,
            estimate_note=estimate_note,
            working=True,
            tone=tone,
        )

    def _render(self, state: Mapping[str, Any]) -> tuple[dict, list[dict], str, str, dict]:
        """Render the browser-owned transcript, claiming answers exactly once.

        The server erases a prompt as soon as its run finishes, so the text
        shown here comes from ``gr.State``.  A completed run's answer is
        fetched once via ``claim_result`` and then stored client-side; the
        server drops it on its next cleanup pass.
        """

        updated: dict[str, Any] = dict(state)
        turns = conversation_turns(state)
        views: dict[str, JobView] = {}
        try:
            for view in self.control.list_jobs(
                state["session_id"], state["active_conversation_id"]
            ):
                views[view.spec.id] = view
        except ControlPlaneError:
            pass

        messages: list[dict[str, str]] = []
        status_lines: list[str] = []
        readings: list[EnergyReading] = []
        receipts: list[Mapping[str, Any]] = []
        #: Runs that ended without an answer. Reported beside the transcript,
        #: because a failure is a fact about a run and not something a speaker
        #: said, but never dropped: a question whose run failed silently would
        #: look like the page had lost it.
        outcomes: list[Mapping[str, Any]] = []

        for turn_number, turn in enumerate(turns, start=1):
            job_id = turn["job_id"]
            view = views.get(job_id)
            answer = turn["answer"]
            messages.append({"role": "user", "content": turn["prompt"]})

            if view is not None and view.status is JobStatus.COMPLETED and answer is None:
                try:
                    claimed = self.control.claim_result(state["session_id"], job_id)
                except ControlPlaneError:
                    claimed = None
                if claimed is not None:
                    answer = self._answer_payload(view, claimed)
                    # Artifacts are decoded and stored in the session's
                    # Outputs the moment the answer is claimed; the browser
                    # state keeps only the file names.
                    raw_artifacts = _as_mapping(claimed).get("artifacts")
                    if isinstance(raw_artifacts, (list, tuple)) and raw_artifacts:
                        stored = self._store_artifacts(
                            state["session_id"], job_id, raw_artifacts
                        )
                        answer["artifacts"] = [Path(item).name for item in stored]
                        # WHAT THE RUN MADE, READ ONCE.
                        #
                        # A file the run produced is the answer to "make me a
                        # PDF", and until now the transcript said "1 file" and
                        # left the thing itself in a panel. The preview is
                        # derived here, at claim time, and travels in browser
                        # state with the answer: the transcript re-renders
                        # every two seconds and must never re-open a zip to
                        # do it.
                        answer["made"] = render.artifact_preview_data(stored)
                    updated = record_answer(updated, job_id=job_id, answer=answer)

            if answer is not None:
                reading = _energy_reading({"energy": answer.get("energy") or {}})
                readings.append(reading)
                tools = tuple(
                    ToolSelection(item["id"], item.get("version", "1"))
                    for item in answer.get("tools") or ()
                )
                # THE BUBBLE HOLDS THE ANSWER, AND ONLY THE ANSWER.
                #
                # It used to carry a horizontal rule and five lines of
                # telemetry: the energy, the tools allowed, a numbered trace,
                # the files produced and the worker's name. Every one of them
                # is worth showing and none of them is the reply. Appending
                # them to the model's text put the product's bookkeeping inside
                # the user's own conversation, so a two-sentence answer
                # arrived as a paragraph of receipts and the reply had to be
                # read around them.
                #
                # They now live in the run detail beside the transcript, keyed
                # to the same turn. Nothing is hidden; it is next to the
                # conversation rather than inside it.
                receipts.append(_run_receipt(turn_number, answer, reading, tools))
                # THE TRANSCRIPT IS FOR MESSAGES.
                #
                # This branch is the only one that appends an assistant turn,
                # and it appends the model's own words and nothing else. Every
                # other state a run can be in used to arrive here as a bubble:
                # "Offered to Studio workstation; waiting for its next poll",
                # "Queued #1", "Completed; retrieving the answer". Each was
                # true, and none of them was a message. A conversation of six
                # exchanges was fourteen bubbles, most of them written by the
                # server about itself, and the two that were actually said had
                # to be picked out of the rest.
                #
                # Progress belongs to the run, not to the conversation, so it
                # is reported beside the transcript where it can update in
                # place, and it leaves nothing behind when the run ends.
                # The work first, then the answer, in the order they happened.
                # These render as collapsible panels rather than bubbles, so
                # the transcript still reads as a conversation and the run is
                # still inspectable inside it.
                messages.extend(
                    _step_messages(answer.get("steps") or (), running=False)
                )
                messages.append({"role": "assistant", "content": str(answer.get("output", ""))})
                # The artifact, immediately under the words that announced it.
                # A run whose whole purpose was to produce a file ends on the
                # file, not on a sentence about a file.
                made = render.made_things(answer.get("made") or ())
                if made:
                    messages.append({"role": "assistant", "content": made})
                status_label = "completed"
                outcome = ""
            elif view is None:
                status_label = "erased"
                outcome = (
                    "No longer held by the server. Prompts are erased when a run "
                    "finishes and answers once delivered."
                )
            elif view.status is JobStatus.FAILED:
                status_label = view.status.value
                outcome = f"Failed: {view.error or 'the worker did not give a reason'}"
            elif view.status is JobStatus.CANCELLED:
                status_label = view.status.value
                outcome = "Cancelled."
            elif view.status is JobStatus.EXPIRED:
                status_label = view.status.value
                outcome = "Expired before it completed."
            elif view.status is JobStatus.STRANDED:
                status_label = view.status.value
                if view.spec.target_agent_id == "unassigned":
                    outcome = "Waiting for a compatible worker to come online."
                else:
                    outcome = (
                        f"The worker {self._agent_name(view.spec.target_agent_id)} "
                        "holding this run has stopped polling. Nothing is lost: wait "
                        "for it, reassign it, or cancel it."
                    )
            else:
                # In flight. Nothing to say about the run's outcome yet, but
                # the steps it has taken so far go in now and are replaced by
                # the kept copy when it finishes, so the transcript fills in
                # as the work happens instead of arriving all at once at the
                # end.
                status_label = view.status.value
                outcome = ""
                messages.extend(_step_messages(self._live_steps(view), running=True))

            if outcome:
                # A run that ended without an answer still has to be visible, or
                # the user asks a question and the page appears to forget it.
                # It is a line about the run, in the place runs are reported,
                # rather than a bubble pretending somebody said it.
                outcomes.append({"turn": turn_number, "text": outcome})
            agent_label = (
                answer.get("agent")
                if answer is not None
                else (self._agent_name(view.spec.target_agent_id) if view else "\u2013")
            )
            # Numbered by turn, not identified by job id. `job_7ac89110` is
            # the server's handle for a run and means nothing to the person
            # reading it: three runs produced three lines that differed only
            # in eight random characters. The turn number is the one thing
            # that tells them apart, and it matches the numbering on the
            # receipts below the transcript.
            status_lines.append(f"Run {turn_number} · {status_label} · {agent_label}")

        status = (
            "\n\n".join(status_lines[-5:])
            if status_lines
            else "No runs in this conversation yet."
        )
        # One panel under the transcript: what is happening now, anything that
        # ended without an answer, then what each finished answer cost and did.
        # All three are commentary on the conversation and none belongs inside
        # it.
        activity_html = (
            self._activity(updated, views)
            + render.run_outcomes(outcomes)
            + render.run_receipts(receipts)
        )
        return (
            updated,
            messages,
            _status_html(status),
            _session_energy_html(
                readings,
                overall=_all_readings(updated),
                lifetimes=self._worker_lifetimes(),
            ),
            gr.HTML(value=activity_html, visible=bool(activity_html)),
        )

    def _worker_lifetimes(self) -> list[dict[str, Any]]:
        """Every online worker's self-reported total since it started.

        Every value here originates on a machine this server does not control,
        so nothing is trusted: names are escaped at render time, numbers are
        coerced, and the whole block is labelled unverifiable. It is included
        because "how much has this machine spent since you turned it on" is a
        question the product exists to answer, and refusing to answer it would
        be a different kind of dishonesty.
        """

        entries: list[dict[str, Any]] = []
        try:
            views = self.control.list_agents()
        except (AttributeError, ControlPlaneError):
            return entries
        for view in views:
            snapshot = getattr(view, "snapshot", None)
            if snapshot is None or not getattr(view, "online", False):
                continue
            if not snapshot.lifetime_runs:
                continue
            entries.append(
                {
                    "name": getattr(view.capabilities, "name", "") or view.capabilities.agent_id,
                    "uptime_seconds": snapshot.uptime_seconds,
                    "joules": snapshot.lifetime_joules,
                    "runs": snapshot.lifetime_runs,
                    "measured_runs": snapshot.lifetime_measured_runs,
                    "scope": snapshot.energy_scope,
                }
            )
        return entries

    def _job_phase(self, job: JobView) -> str:
        try:
            snapshot = self.control.get_agent(job.spec.target_agent_id).snapshot
        except ControlPlaneError:
            return PHASE_WORKING
        if snapshot is None:
            return PHASE_WORKING
        return normalise_phase(snapshot.progress_phase.get(job.spec.id))

    def _job_progress(self, job: JobView) -> str:
        try:
            snapshot = self.control.get_agent(job.spec.target_agent_id).snapshot
            value = snapshot.progress.get(job.spec.id) if snapshot else None
            note = snapshot.progress_notes.get(job.spec.id) if snapshot else None
            parts = ""
            if value is not None:
                parts += f" · {value * 100:.0f}%"
            if isinstance(note, str) and note:
                # Worker-supplied display text; backticks and markdown control
                # characters are stripped by _status_html-style escaping at
                # render time because the chatbot sanitises HTML, but the
                # length is still bounded here.
                parts += f" · {note[:160]}"
            return parts
        except ControlPlaneError:
            pass
        return ""

    def _join_panel(self, code: str = "", state: Mapping[str, Any] | None = None) -> str:
        """The address an agent should be pointed at, and the code if minted.

        The port comes from the same policy the launch uses, so the panel
        cannot drift from what the server actually bound.
        """

        from distinct_protocol.netpolicy import (  # noqa: PLC0415
            local_addresses,
            resolve_bind_host,
            resolve_bind_port,
        )

        space_host = os.environ.get("SPACE_HOST", "").strip()
        if os.environ.get("SYSTEM") == "spaces" and space_host:
            # On HuggingFace Spaces the bind address is container-internal
            # (127.0.0.1/0.0.0.0 inside the pod). Agents must dial the public
            # Space URL, which the platform provides in SPACE_HOST.
            addresses = [f"https://{space_host}"]
        else:
            host = resolve_bind_host()
            port = resolve_bind_port()
            addresses = local_addresses(port, host)
        qr = ""
        if addresses:
            try:
                from distinct_agent.direct import qr_svg  # noqa: PLC0415

                qr = qr_svg(addresses[0])
            except Exception:  # noqa: BLE001 - the address alone is enough
                qr = ""
        # The command mirrors the defaults this browser chose on the library
        # page, so what the person copies is what their runs will expect.
        chosen = _library_defaults(state) if state is not None else ()
        tools = ",".join(chosen) if chosen else "local"
        # Where the code lives, taken from the platform rather than written
        # down here. A hard-coded URL is a URL that goes stale the first time
        # this is deployed anywhere else, and the panel would then send
        # volunteers to somebody else's repository.
        space_id = os.environ.get("SPACE_ID", "").strip()
        source = f"https://huggingface.co/spaces/{space_id}" if space_id else ""
        return render.join_this_server(
            addresses=addresses, code=code, qr=qr, tools=tools, source=source
        )

    # Pairing/downloads ------------------------------------------------

    def pairing_code(
        self,
        state: Mapping[str, Any],
        profile: gr.OAuthProfile | None,
        request: gr.Request | None = None,
        *,
        force: bool = False,
    ) -> tuple:
        """Render the setup panel. Nothing is minted, and nothing can fail.

        THIS HANDLER USED TO ISSUE A CREDENTIAL, AND THAT WAS THE PROBLEM.

        Opening a section is navigation. This one minted a server-side pairing
        code every time it ran, which produced three separate faults:

        * The code changed under people. Open the panel, copy the code, open it
          again to re-read the command, and the clipboard no longer matched the
          page.
        * It was an outage waiting to happen. Opening a section is cheap and
          scriptable, live codes came from one pool shared by everybody on this
          server, and they lasted ten minutes -- so a loop here filled the pool
          and stopped anyone else pairing a worker.
        * It could refuse. Signed out it raised, so clicking "Run a community
          agent" to find out what running one involves produced a red toast and
          a 500 in the log for the crime of reading a page.

        Each was patched in turn -- reuse the browser's live code, cap codes per
        owner, catch the refusals -- and the patches were all correct and all
        beside the point. The panel needed a code only because the flow made a
        volunteer carry one from this page to their terminal, and that flow is
        gone: the worker prints its own code and the person types it into the
        box on this page instead.

        So there is nothing to mint, nothing to run out of, nothing to expire
        under anybody, and nothing that needs an account to read. The signature
        keeps ``state`` and ``force`` because the accordion and the tests are
        wired to them, and both are now ignored.
        """

        updated = dict(state) if isinstance(state, Mapping) else {}
        # Kept only so a browser that saved one before this change does not
        # carry it around for ever.
        updated.pop("pairing_code", None)
        updated.pop("pairing_expires", None)
        return updated, updated, self._join_panel("", updated)

    # `new_pairing_code` and the "New code" button are both gone, and so is the
    # minting they existed to trigger. `ControlPlane.create_pairing_code` stays
    # for `--pair`, which scripted setups still use: there a code is minted and
    # consumed by the same program and nobody has to read anything.

    def build(self) -> gr.Blocks:
        # ``head`` belongs to launch() in Gradio 6, not to the constructor; the
        # entrypoints pass HEAD there so Inter is fetched before first paint.
        # Serve the artwork from the repository rather than hotlinking a
        # non-profit's CDN. See scripts/fetch_artwork.py for why. Absolute:
        # a relative path resolves against the working directory, which is the
        # Space root under app.py but not under pytest or scripts/demo_server.py.
        gr.set_static_paths(paths=[Path(__file__).resolve().parent.parent / "assets"])

        # analytics_enabled=False, and separately the environment defaults set
        # in distinct_server/__init__.py. Both are needed: this keyword stops
        # gradio's own version check and start-up ping, and the environment is
        # what stops the huggingface_hub registry fetch underneath it, which
        # answers to no keyword at all.
        with gr.Blocks(title="distinct", fill_width=True, analytics_enabled=False) as demo:
            session_state = gr.State(
                value=new_session,
                time_to_live=8 * 60 * 60,
                delete_callback=self.on_session_delete,
            )
            # The browser's own durable copy of the transcript (defect A13):
            # localStorage via BrowserState, written on every state change and
            # read once at load. The server still holds nothing after
            # delivery; the user holds their own history, and Clear my data
            # erases this copy too.
            browser_state = gr.BrowserState(
                default_value=None,
                storage_key="distinct-transcript",
                secret=BROWSER_STATE_SECRET,
            )
            # One artwork per page load. Chosen here, once, and handed to both
            # the splash and the in-app credit, so the two can never show
            # different images or, worse, an image beside another's credit.
            artwork = render.choose_artwork()

            with gr.Column(elem_classes="c-splash-view") as splash_view:
                with gr.Row(elem_classes="c-splash"):
                    with gr.Column(elem_classes="c-splash__copy"):
                        gr.HTML(render.splash_copy())
                        enter = gr.Button(
                            "Enter the network", variant="primary", scale=0
                        )
                    with gr.Column(elem_classes="c-splash__plate"):
                        gr.HTML(render.splash_figure(artwork))

            with gr.Column(elem_classes="c-shell", visible=False) as app_view:
                gr.HTML('<a class="c-skip" href="#distinct-composer">Skip to the request box</a>')
                # Header, notice, sidebar chrome, worker cards, panels and the
                # footer are all server-rendered markup: see render.py. Gradio
                # components are used only where Python event wiring is needed.
                # Signed out until a handler says otherwise. While nobody is
                # signed in the header stays empty: the Hugging Face login
                # button below is the one sign-in control, so a header link
                # would only duplicate it.
                identity = gr.HTML(_identity_html(""))
                # The quick nav sits directly under the identity box. The
                # library is a destination rather than a section, so its
                # control is a button that swaps the view; "Run an agent" is
                # still a section of this page, so it stays an anchor.
                #
                # THE SIGN-IN CONTROL SITS IN THIS ROW, AND WHY IT IS
                # CONDITIONAL.
                #
                # On a Space, Gradio only wires real OAuth when the interface
                # contains a `LoginButton`: `gradio.routes` gates the whole
                # OAuth setup on `blocks.expects_oauth`, and that attribute is
                # set by the button and by nothing else. This interface had no
                # button, so a Space deployment got no OAuth routes, every
                # `gr.OAuthProfile` arrived as None, and nobody could sign in
                # at all. The access model was not bypassed; it was sealed
                # shut, which looks identical from the outside until somebody
                # tries to use it.
                #
                # Off a Space the same button is actively harmful: Gradio mocks
                # OAuth there and signs every visitor in as a fake profile,
                # which is the exact trap `refuse_mocked_oauth` exists to
                # refuse. So the button appears only where it is real, and the
                # self-hosted deployment uses the `/auth/login` route in
                # `auth_routes.py` instead, which is a link rather than a
                # component because it is an ordinary HTTP redirect.
                #
                # It is inside the nav row rather than under it because it is
                # navigation: left on its own line it read as a stray control
                # sitting between the page's chrome and its content.
                with gr.Row(elem_classes="c-quicknav"):
                    open_library = gr.Button(
                        "Library", size="sm", variant="secondary",
                        elem_classes="c-quicknav__btn", scale=0,
                    )
                    gr.HTML(
                        '<a class="c-quicknav__link" href="#join-agent">Run an agent</a>'
                    )
                    if gradio_oauth_is_real():
                        gr.LoginButton(
                            "Sign in with Hugging Face", size="sm",
                            elem_classes="c-quicknav__auth", scale=0,
                        )

                gr.HTML(render.notice())

                # min_width=0 on both columns is load-bearing, not tidying.
                # Gradio writes its column min-width as an inline style, which
                # no stylesheet rule can lower. Two columns that refuse to
                # shrink inside a row that is not allowed to wrap produce a row
                # wider than its container, and a centred container that
                # overflows spills to the left as well as the right. Left
                # overflow cannot be scrolled to, so it is not merely ugly:
                # it deletes text. Width is decided by the grid in
                # presentation.py and nowhere else.
                with gr.Row(elem_classes="c-main", equal_height=False):
                    # THE LEFT RAIL IS CONVERSATIONS AND NOTHING ELSE.
                    #
                    # It previously shared a column with the worker picker, the
                    # model picker, the worker card, the energy panel and the
                    # file list, which made it a settings panel that happened
                    # to contain a history. Everything that describes the
                    # session rather than the conversation moved to the right.
                    with gr.Column(elem_classes="c-rail", min_width=0):
                        # The other two columns each open with a heading, and
                        # without one here the three columns started on three
                        # different baselines and the rail read as a stray
                        # button rather than as the third part of the layout.
                        gr.HTML(render.section_heading("Conversations"))
                        new_conversation_button = gr.Button(
                            "New conversation", variant="secondary", size="sm"
                        )
                        # A list, not a dropdown. A Radio group is already a
                        # single-selection list with the right semantics and
                        # keyboard behaviour; only its appearance changes.
                        # The per-state washes for the list below. It renders
                        # nothing visible; it carries the rules, because the
                        # radio group has no per-option hook to hang a class
                        # on. Re-rendered wherever the list itself is.
                        rail_tones = gr.HTML(
                            render.conversation_tones({}), elem_classes="c-railtones"
                        )
                        with gr.Column(elem_classes="c-rail__list"):
                            conversation = gr.Radio(
                                label="Conversation",
                                show_label=False,
                                choices=[],
                                container=False,
                                elem_classes="c-convlist",
                            )
                            # DELETING THE ONE YOU ARE LOOKING AT.
                            #
                            # A per-row control would be the obvious design and
                            # is not available: the list is a Radio, and a radio
                            # group gives no per-option hook to hang a button
                            # on. One control acting on the selected
                            # conversation is the same gesture in two steps, and
                            # it is the step order people already use — you open
                            # a conversation, then decide it can go.
                            delete_conversation_button = gr.Button(
                                "Delete this conversation",
                                size="sm",
                                variant="secondary",
                                elem_classes="c-rail__delete",
                            )

                    with gr.Column(elem_classes="c-side", min_width=0):
                        gr.HTML(render.section_heading("Run setup"))
                        with gr.Column(elem_classes="c-runsetup"):
                            # Model first, worker second, because that is the
                            # order the dependency runs in: a worker only
                            # appears here once its operator has approved the
                            # chosen model and downloaded its weights, so the
                            # list below is always the answer to "who can run
                            # this", never a free choice made in ignorance of
                            # the model.
                            model = gr.Dropdown(
                                label="Model",
                                choices=model_choices_with_evidence(),
                                value=DEFAULT_MODEL_ID,
                                container=True,
                            )
                            # HOW THE REQUEST IS ANSWERED.
                            #
                            # This was decided entirely on the worker, by a
                            # flag its operator passed at start-up, and never
                            # shown to the person whose request it was. Two
                            # runs of the same model on two workers could
                            # therefore differ in the thing that matters most
                            # for a small model, with nothing on the page
                            # saying so. It sits between the model and the
                            # worker because that is the order the dependency
                            # runs in: model, then how, then who can do both.
                            mode = gr.Dropdown(
                                label="Agent mode",
                                choices=AGENT_MODE_CHOICES,
                                value=MODE_SIMPLE,
                                container=True,
                            )
                            agent = gr.Dropdown(
                                label="Worker · only those running this model and mode",
                                choices=[("Automatic · least loaded", "auto")],
                                value="auto",
                                container=True,
                            )
                        agent_table = gr.HTML(
                            value=render.worker_cards(()),
                            label="Compatible community agents",
                        )
                        # How a worker somebody else runs becomes one you may
                        # use. The server never decides this: the operator of
                        # the machine hands out the code their worker printed,
                        # and entering it here is that consent arriving.
                        with gr.Row():
                            access_code = gr.Textbox(
                                label="Add a worker · enter the code your worker printed, or one somebody shared with you",
                                placeholder="XXXXX-XXXXX-XXXXX-XXXXX-XXXXX-XXXXX-XXXXX-XXXX",
                                lines=1,
                                container=True,
                            )
                            access_add = gr.Button("Add worker", variant="secondary")
                        access_status = gr.HTML(
                            _status_html(
                                "A worker you started prints a code and waits for you "
                                "to enter it here. That claims it, once. A worker "
                                "somebody else runs prints a code they can share with "
                                "anyone, and entering that one does not use it up."
                            )
                        )
                        # Boundary 1. Boundary 2 lives in its own panel far
                        # below, so the layout itself never invites summing.
                        #
                        # Directly under the worker that produced the figures,
                        # and above the run list rather than at the bottom of
                        # the column: the cost of what you just did is not
                        # small print, and burying it under everything else
                        # was the layout disagreeing with the product.
                        energy_total = gr.HTML(_session_energy_html(()))
                        gr.HTML(render.section_heading("This session"))
                        run_status = gr.HTML(
                            '<p class="c-note"><span class="c-note__text">'
                            "No runs in this conversation yet.</span></p>"
                        )
                        outputs_files = gr.File(
                            value=None,
                            visible=False,
                            label="Files your runs created",
                            file_count="multiple",
                            interactive=False,
                        )
                        clear_data = gr.Button(
                            "Clear my data", variant="secondary", size="sm"
                        )
                        # Pinned to the bottom of the column by margin-top:auto.
                        # It closes the gap that used to sit under the energy
                        # panel and aligns the two columns' bottom edges, and it
                        # is where the licence credit belongs anyway: with the
                        # image, on every page that shows it.
                        gr.HTML(
                            render.credit(artwork, compact=True),
                            elem_classes="c-side__credit",
                        )

                    with gr.Column(elem_classes="c-conversation", min_width=0):
                        # The no-agents veil. Static markup; presentation.py
                        # shows it with :has() only while somebody is signed in
                        # and the worker panel is in its empty state, so there
                        # is no event plumbing to fall out of date. It softly
                        # fades the chat rather than hiding it: the interface
                        # is not broken, it is waiting for a worker.
                        gr.HTML(
                            '<div class="c-veil" role="status">'
                            '<div class="c-veil__panel">'
                            '<p class="c-veil__title">No agents available</p>'
                            '<p class="c-veil__body">Nothing can run until a '
                            "community agent pairs with this server. Set one up "
                            "in a minute, or come back when one is online.</p>"
                            '<a class="c-veil__action" href="#join-agent">'
                            "Set up an agent</a>"
                            "</div></div>"
                        )
                        chat = gr.Chatbot(
                            label="Conversation",
                            show_label=False,
                            # A range, given to the component rather than
                            # imposed on it from the stylesheet. The sheet used
                            # to cap the outer block and make that the scroll
                            # container, which left the component scrolling an
                            # inner element that had nothing to scroll: the
                            # transcript stayed on the first question for ever
                            # while the newest answer sat below the fold.
                            min_height=260,
                            max_height=560,
                            autoscroll=True,
                            # No share, no copy-all, no delete floating over the
                            # first turn. Conversations are managed in the rail
                            # that lists them.
                            buttons=[],
                            layout="bubble",
                            sanitize_html=True,
                            allow_tags=False,
                            elem_classes="c-chat",
                            placeholder=render.empty_state(
                                glyph="◇",
                                title="No runs yet",
                                body=EMPTY_TRANSCRIPT_BODY,
                            ),
                        )
                        # What the run is doing right now: spinner, the phase
                        # named in words, and the model's own tool trace as it
                        # happens. Hidden when nothing is in flight, so an idle
                        # conversation is not decorated with an empty box.
                        activity_panel = gr.HTML(value="", visible=False)
                        # The container IS the field: the Gradio textbox sits
                        # inside it borderless and transparent.
                        with gr.Column(elem_classes="c-composer"):
                            prompt = gr.Textbox(
                                label="Your request",
                                show_label=False,
                                elem_id="distinct-composer",
                                placeholder="Ask something…",
                                lines=2,
                                max_lines=10,
                                container=False,
                            )
                            # Attachments and library selection sit on the
                            # message, not in the sidebar. Both are decisions
                            # about *this* request rather than about the
                            # session, and a control that changes per message
                            # belongs beside the message.
                            with gr.Row(elem_classes="c-composer-tools"):
                                # A drop zone the size of a panel, captioned
                                # "Drop File Here - or - Click to Upload", is
                                # what this component renders by default. Next
                                # to a message box it reads as the main event
                                # rather than as one affordance among several,
                                # and its floating label overlapped its own
                                # body. Presentation collapses it to a pill.
                                attachments = gr.File(
                                    label="Attach files",
                                    show_label=False,
                                    file_count="multiple",
                                    elem_classes="c-attach",
                                    scale=0,
                                )
                                # The label is kept for assistive technology
                                # and hidden from the page: "create pdf · skill"
                                # beside a checkbox already says what it is, so
                                # a heading above the row was a line of type
                                # that told a sighted reader nothing and made
                                # the composer taller.
                                tools = gr.CheckboxGroup(
                                    label="Tools and skills for this request",
                                    show_label=False,
                                    choices=[],
                                    value=[],
                                    visible=False,
                                    interactive=False,
                                    container=True,
                                    scale=3,
                                )
                            library_note = gr.HTML(
                                render.no_tools_note(NO_TOOLS_NOTICE),
                                elem_classes="c-undernote",
                            )
                            with gr.Row(elem_classes="c-composer-foot"):
                                consent = gr.Checkbox(
                                    label=(
                                        "I understand this run sends plaintext to a "
                                        "community-operated worker"
                                    ),
                                    value=False,
                                    container=False,
                                    scale=5,
                                )
                                reassign = gr.Button(
                                    "Reassign", variant="secondary", scale=0, min_width=96
                                )
                                cancel = gr.Button(
                                    "Cancel latest", variant="stop", scale=0, min_width=96
                                )
                                # Disabled until the acknowledgement is ticked.
                                # It used to be always enabled and to raise a
                                # full-width red error on submit, which is the
                                # worst of both: the user does the work of
                                # composing and typing, and only then finds out
                                # the form was never going to accept it. A
                                # control that cannot succeed should say so
                                # before it is pressed.
                                submit = gr.Button(
                                    "Queue run",
                                    variant="primary",
                                    scale=0,
                                    min_width=110,
                                    interactive=False,
                                )

                # The accordions used to sit outside every column, on a fourth
                # left edge of their own. Inside the tail they share the shell's
                # padding box like everything else.
                with gr.Column(elem_classes="c-tail"):
                    with gr.Accordion(
                        "Run a community agent", open=False, elem_id="join-agent"
                    ) as join_section:
                        gr.Markdown(
                            # "The agent owns its bounded FIFO and reports it
                            # on every poll" is true and is a sentence for
                            # somebody reading the queue implementation. The
                            # person here is about to run a worker on their own
                            # machine and wants to know what it will do.
                            "Run the worker on your own machine and it joins this "
                            "server. It takes a fixed number of requests at a time, "
                            "runs them one after another, and tells this server how "
                            "many are waiting."
                        )
                        # The address and the code, filled in. This used to be
                        # a command with the word SERVER in the middle of it,
                        # which asked the reader to find something this
                        # process already knew.
                        join_panel = gr.HTML(self._join_panel())
                        # ONE CONTROL, BECAUSE THERE IS ONE THING TO DO HERE.
                        #
                        # This row used to offer "Agent build: Windows x86-64"
                        # and a Download button beside it. No such build is
                        # produced for this Space, so the button's only
                        # possible outcome was an error telling the volunteer
                        # that an artifact "has not been built for this Space
                        # revision" — a sentence about our release process,
                        # delivered to somebody who wanted to help. The worker
                        # is Python and always was; the panel above now says
                        # so, and the only control left is the one that mints
                        # the code those instructions need.
                        # NO BUTTON IN HERE AT ALL.
                        #
                        # There were two, and both were ceremony. "Generate
                        # pairing code" asked a person to request the one thing
                        # the panel exists to give, and nobody opens this panel
                        # for another reason. "New code" survived it for a
                        # while on the theory that somebody would want to
                        # replace a code that had expired or been used — but
                        # opening the panel already does exactly that: the code
                        # is reused while it is alive and reminted when it is
                        # not, so the button could only ever mint one that was
                        # about to be minted anyway.
                        #
                        # The code stays SERVER-issued, which is not ceremony:
                        # it is what proves an arriving worker was invited by
                        # whoever owns this server. The code the *worker*
                        # makes is the other one — the access code it prints
                        # once paired, which is what you give to the people you
                        # want to let use your machine.

                    with gr.Accordion("Model assessments and their sources", open=False):
                        gr.HTML(
                            '<p class="c-panel__head">'
                            '<span class="c-panel__title">Release assessment</span>'
                            '<span class="c-panel__boundary">boundary 2 of 2</span></p>'
                        )
                        gr.HTML(
                            render.assessment_section(
                                _assessment_cards(),
                                warning=_rubric_warning(),
                                further_reading=_further_reading(),
                            )
                        )

            # THE LIBRARY IS A PAGE, NOT A SECTION.
            #
            # It used to be an accordion at the bottom of the chat page, which
            # is where a person found it after scrolling past the conversation,
            # the composer and the agent panel — that is a footnote, not a
            # destination. It is now a full view of its own, reached from the
            # control under the sign-in box and leaving by the same route.
            #
            # A view rather than a Gradio route, deliberately. A second route
            # is a second Blocks with its own state, and the defaults live in
            # this browser's own storage beside the transcript; two pages
            # writing that one key is how a person loses their history to a
            # visit to the library. One Blocks, two views, one copy of the
            # state.
            with gr.Column(elem_classes="c-libpage", visible=False) as library_view:
                close_library = gr.Button(
                    "\u2190 Back to the conversation", size="sm",
                    variant="secondary", elem_classes="c-libpage__backbtn", scale=0,
                )
                gr.HTML(
                    '<h1 class="c-libpage__title">Library</h1>'
                    '<p class="c-libpage__lede">Every tool and skill this server can '
                    "offer, who wrote it, and what it does. Tap a card to save it as "
                    "a default: it selects itself whenever a worker offers it. Dropping "
                    "it from a single run does not change what is saved.</p>"
                )
                library_search = gr.Textbox(
                    label="Search the library",
                    show_label=False,
                    placeholder="Search by name, reference or what it does…",
                    container=False,
                    elem_classes="c-librarysearch",
                )
                library_summary = gr.HTML(
                    render.library_summary(shown=0, matched=0, total=0, chosen=0)
                )
                library_cards_html = gr.HTML(elem_classes="c-librarycards")
                # THE CARDS ARE THE CONTROL; THIS IS HOW THEY REACH PYTHON.
                #
                # The page used to carry a grid of cards *and* a visible list
                # of tick boxes naming the same members again: two controls for
                # one decision, and the reason the page was unreadable on a
                # phone. The grid is the only control a person sees now. This
                # group is its transport, off the screen, and the click handler
                # forwards a card's click to the box that belongs to it.
                #
                # Forwarding a real click rather than writing a value is the
                # whole trick, and it was arrived at the hard way: setting a
                # hidden field's ``value`` and dispatching ``input`` looks
                # right, changes the DOM, and never reaches the server, because
                # the framework's own binding is what it listens to. Clicking
                # the framework's own control cannot drift from it.
                #
                # Each box is labelled with its bare reference so the mapping
                # from card to box is an exact string match rather than a
                # position, which would silently mis-tick the moment a filter
                # reordered one list and not the other. Hidden in CSS, never
                # with ``visible=False``: gradio 6 leaves an invisible block
                # out of the document altogether, and a transport the page
                # cannot find is a page that silently never saves.
                library_defaults = gr.CheckboxGroup(
                    label="Saved defaults",
                    show_label=False,
                    choices=[],
                    value=[],
                    container=False,
                    elem_id="library-choice",
                    elem_classes="c-hiddenfield",
                )

            # Hidden component bindings form the machine API consumed by gradio_client.
            with gr.Column(visible=False):
                api_pair_code = gr.Textbox()
                api_pair_capabilities = gr.Textbox()
                api_pair_pubkey = gr.Textbox()
                api_pair_access = gr.Textbox()
                api_pair_output = gr.Textbox()
                api_pair_button = gr.Button()
                api_agent_id = gr.Textbox()
                api_timestamp = gr.Number()
                api_nonce = gr.Textbox()
                api_payload = gr.Textbox()
                api_signature = gr.Textbox()
                api_output = gr.Textbox()
                api_sync_button = gr.Button()
                api_accept_button = gr.Button()
                api_complete_button = gr.Button()
                api_catalogue_button = gr.Button()

            demo.load(
                self.load_session,
                inputs=[session_state, browser_state],
                outputs=[
                    session_state,
                    browser_state,
                    identity,
                    conversation,
                    rail_tones,
                    chat,
                    agent_table,
                    agent,
                    run_status,
                    energy_total,
                    tools,
                    library_note,
                    outputs_files,
                    activity_panel,
                    library_defaults,
                    library_summary,
                    library_cards_html,
                ],
                queue=False,
                api_visibility="private",
            )
            # The splash and the app are two views of one Blocks rather than two
            # routes, so the named agent API, the session state and its
            # delete_callback are all untouched by the landing page existing.
            enter.click(
                lambda: (gr.Column(visible=False), gr.Column(visible=True)),
                outputs=[splash_view, app_view],
                queue=False,
                api_visibility="private",
            )
            # THE LIBRARY VIEW SWAP.
            #
            # Two columns, one visible at a time, and the scroll position reset
            # on arrival: a page that opens halfway down is the single clearest
            # sign that nothing really navigated. The transition is instant
            # because both views are already in the document; nothing is
            # fetched to change page.
            _scroll_top = "() => { window.scrollTo({top: 0, behavior: 'instant'}); }"
            open_library.click(
                lambda: (gr.Column(visible=False), gr.Column(visible=True)),
                outputs=[app_view, library_view],
                queue=False,
                api_visibility="private",
                js=_scroll_top,
            )
            close_library.click(
                lambda: (gr.Column(visible=True), gr.Column(visible=False)),
                outputs=[app_view, library_view],
                queue=False,
                api_visibility="private",
                js=_scroll_top,
            )
            delete_conversation_button.click(
                self.delete_conversation,
                inputs=[session_state],
                outputs=[
                    session_state,
                    browser_state,
                    conversation,
                    rail_tones,
                    chat,
                    run_status,
                    energy_total,
                    activity_panel,
                    outputs_files,
                ],
                queue=False,
                api_visibility="private",
            )
            new_conversation_button.click(
                self.new_conversation,
                inputs=[session_state],
                outputs=[
                    session_state,
                    browser_state,
                    conversation,
                    rail_tones,
                    chat,
                    run_status,
                    activity_panel,
                ],
                queue=False,
                api_visibility="private",
            )
            conversation.input(
                self.change_conversation,
                inputs=[session_state, conversation],
                outputs=[
                    session_state,
                    browser_state,
                    chat,
                    run_status,
                    energy_total,
                    activity_panel,
                ],
                queue=False,
                api_visibility="private",
            )
            # The acknowledgement gates the button rather than the submission.
            # .change, not .input: a programmatic reset must move the button
            # too, or the two can disagree.
            consent.change(
                lambda given: gr.Button(interactive=given is True),
                inputs=[consent],
                outputs=[submit],
                queue=False,
                api_visibility="private",
            )
            # .input rather than .change: these callbacks write back into
            # ``tools``, and .change fires on programmatic updates too, which
            # would loop.
            for component in (model, tools, mode):
                component.input(
                    self.refresh_agents,
                    # The session travels with it. Without it the handler
                    # authorised against an empty state, and an empty session
                    # id is the one `create_session` reads as "mint a new
                    # one" — one leaked session per model change, until the
                    # registry filled and the whole server started erroring.
                    inputs=[model, tools, agent, mode, session_state],
                    outputs=[agent_table, agent, tools, library_note],
                    queue=False,
                    api_visibility="private",
                )
            submit.click(
                self.submit_run,
                inputs=[prompt, model, tools, agent, consent, attachments, session_state],
                outputs=[
                    session_state,
                    browser_state,
                    chat,
                    prompt,
                    tools,
                    run_status,
                    energy_total,
                    conversation,
                    rail_tones,
                    activity_panel,
                    attachments,
                    outputs_files,
                ],
                api_name="submit_job",
                api_visibility="private",
            )
            prompt.submit(
                self.submit_run,
                inputs=[prompt, model, tools, agent, consent, attachments, session_state],
                outputs=[
                    session_state,
                    browser_state,
                    chat,
                    prompt,
                    tools,
                    run_status,
                    energy_total,
                    conversation,
                    rail_tones,
                    activity_panel,
                    attachments,
                    outputs_files,
                ],
                api_visibility="private",
            )
            cancel.click(
                self.cancel_latest,
                inputs=[session_state],
                outputs=[
                    session_state,
                    browser_state,
                    chat,
                    run_status,
                    energy_total,
                    activity_panel,
                ],
                queue=False,
                api_visibility="private",
            )
            reassign.click(
                self.reassign_latest,
                inputs=[session_state],
                outputs=[
                    session_state,
                    browser_state,
                    chat,
                    run_status,
                    energy_total,
                    activity_panel,
                ],
                queue=False,
                api_visibility="private",
            )
            clear_data.click(
                self.clear_my_data,
                inputs=[session_state],
                outputs=[
                    session_state,
                    browser_state,
                    conversation,
                    rail_tones,
                    chat,
                    run_status,
                    energy_total,
                    tools,
                    library_note,
                    outputs_files,
                    activity_panel,
                ],
                queue=False,
                api_visibility="private",
            )
            # Opening the section is the request; there is nothing else to ask
            # for in here. The small "New code" control stays for the one case
            # somebody genuinely wants a second one: the first has expired or
            # has already been used.
            join_section.expand(
                self.pairing_code,
                inputs=[session_state],
                # Into the panel, so the address and the code appear in one
                # command rather than in two places the reader has to join up.
                # The state travels with it because the code itself is kept
                # in this browser: the server has only a digest, so re-opening
                # can show the same code rather than mint another.
                outputs=[session_state, browser_state, join_panel],
                queue=False,
                api_visibility="private",
            )
            # Two seconds, not five. The activity panel counts elapsed time
            # and streams the worker's steps, and a spinner that updates
            # every five seconds reads as a stall. The refresh is in-memory
            # and unqueued; the worker's own poll cadence is unchanged.
            timer = gr.Timer(2, active=True)
            timer_outputs = [
                session_state,
                browser_state,
                chat,
                run_status,
                energy_total,
                agent_table,
                agent,
                tools,
                library_note,
                outputs_files,
                activity_panel,
                conversation,
                rail_tones,
            ]
            # A skipped tick returns one `gr.skip()` per output, and a count
            # that disagreed with this list would raise inside the guard whose
            # whole job is to stop the timer raising. Checked here, where both
            # halves are visible, rather than trusted to stay in step.
            assert len(timer_outputs) == self.TIMER_OUTPUTS
            timer.tick(
                self.refresh_session,
                inputs=[session_state, model, tools, agent, mode],
                outputs=timer_outputs,
                queue=False,
                api_visibility="private",
            )

            # The page filters as you type and saves as you tick. Two
            # handlers rather than one, because filtering must never change
            # what is saved: a search box that unticked what it hid would
            # destroy the thing this page exists to keep.
            library_search.input(
                self.update_library_page,
                inputs=[library_search, session_state],
                outputs=[library_defaults, library_summary, library_cards_html],
                queue=False,
                api_visibility="private",
            )
            library_defaults.input(
                self.set_library_defaults,
                inputs=[library_defaults, library_search, session_state],
                outputs=[
                    session_state,
                    browser_state,
                    library_summary,
                    library_cards_html,
                    tools,
                    library_note,
                ],
                queue=False,
                api_visibility="private",
            )

            access_add.click(
                self.redeem_agent_code,
                inputs=[access_code, model, tools, agent, mode, session_state],
                outputs=[access_status, agent, agent_table],
                queue=False,
                api_visibility="private",
            )

            api_pair_button.click(
                self.agent_api.pair,
                inputs=[
                    api_pair_code,
                    api_pair_capabilities,
                    api_pair_pubkey,
                    api_pair_access,
                ],
                outputs=api_pair_output,
                api_name="agent_pair",
                queue=False,
            )
            # THE OTHER DIRECTION. `agent_pair` takes a code the server minted;
            # `agent_register` takes one the worker minted, and creates a record
            # that is inert until somebody signed in claims it. Same inputs, so
            # the same hidden controls carry both.
            api_pair_button.click(
                self.agent_api.register,
                inputs=[
                    api_pair_code,
                    api_pair_capabilities,
                    api_pair_pubkey,
                    api_pair_access,
                ],
                outputs=api_pair_output,
                api_name="agent_register",
                queue=False,
            )
            api_pair_button.click(
                self.agent_api.claim_state,
                inputs=[api_pair_code],
                outputs=api_pair_output,
                api_name="agent_claim_state",
                queue=False,
            )
            signed_inputs = [api_agent_id, api_timestamp, api_nonce, api_payload, api_signature]
            api_catalogue_button.click(
                self.agent_api.catalogue,
                inputs=signed_inputs,
                outputs=api_output,
                api_name="agent_catalogue",
                queue=False,
            )
            api_sync_button.click(
                self.agent_api.sync,
                inputs=signed_inputs,
                outputs=api_output,
                api_name="agent_sync",
                queue=False,
            )
            api_accept_button.click(
                self.agent_api.accept,
                inputs=signed_inputs,
                outputs=api_output,
                api_name="agent_accept",
                queue=False,
            )
            api_complete_button.click(
                self.agent_api.complete,
                inputs=signed_inputs,
                outputs=api_output,
                api_name="agent_complete",
                queue=False,
            )

        # A phone client used to be mounted here: one HTML page at /m and six
        # JSON routes beside it. It has been removed, and the reason is worth
        # keeping so it is not rebuilt the same way.
        #
        # None of those routes consulted an identity. `catalogue` listed every
        # worker to any caller, and `submit` took a session id the client had
        # minted for itself and passed it as the viewer id, so an anonymous
        # visitor could name a worker and spend a volunteer's electricity. The
        # mount was unconditional and failed silently, which meant nobody would
        # have noticed either way. On the Space deployment, which is the one
        # exposed to the internet, the routes were reachable: a red-team run
        # fetched the full worker inventory with no cookie and no header.
        #
        # Sign-in is the server's whole access story. A second front door that
        # did not use it was not a convenience, it was the way in.

        return demo


def build_app(control_plane: Optional[ControlPlane] = None) -> gr.Blocks:
    return DistinctUI(control_plane).build()


def distinct_theme() -> gr.Theme:
    """Build the theme at launch time, as required by Gradio 6.

    ``Base`` rather than ``Soft``: Soft ships its own borders, panel fills and
    shadows, and the redesign then spends its whole budget overriding them.
    Base is close to unstyled, so the tokens below and the CSS in
    :mod:`distinct_server.presentation` are the only sources of appearance.

    Colours here are the accessible shades documented in that module. The
    spec's sage (#6E9B76) is deliberately absent from every text role: it is
    3.0:1 on the canvas and fails AA.
    """

    return gr.themes.Base(
        primary_hue=gr.themes.colors.green,
        secondary_hue=gr.themes.colors.emerald,
        neutral_hue=gr.themes.colors.stone,
        radius_size=gr.themes.sizes.radius_lg,
        spacing_size=gr.themes.sizes.spacing_lg,
        text_size=gr.themes.sizes.text_md,
        font=(
            gr.themes.GoogleFont("Inter"),
            "-apple-system",
            "BlinkMacSystemFont",
            "Segoe UI",
            "sans-serif",
        ),
        font_mono=(gr.themes.GoogleFont("IBM Plex Mono"), "ui-monospace", "monospace"),
    ).set(
        body_background_fill=PALETTE["canvas"],
        body_background_fill_dark=PALETTE["canvas"],
        body_text_color=PALETTE["text"],
        body_text_color_subdued=PALETTE["text_secondary"],
        background_fill_primary=PALETTE["surface"],
        background_fill_secondary=PALETTE["canvas"],
        border_color_primary=PALETTE["hairline"],
        border_color_accent=PALETTE["border"],
        block_background_fill="transparent",
        block_border_width="0px",
        block_shadow="none",
        block_label_background_fill="transparent",
        block_label_text_color=PALETTE["text_secondary"],
        block_title_text_color=PALETTE["text_secondary"],
        panel_background_fill="transparent",
        panel_border_width="0px",
        input_background_fill=PALETTE["surface"],
        input_border_color=PALETTE["border"],
        input_radius="8px",
        button_large_radius="999px",
        button_small_radius="999px",
        button_primary_background_fill=PALETTE["sage_button"],
        button_primary_background_fill_hover="#33523B",
        button_primary_text_color="#FFFFFF",
        button_primary_border_color=PALETTE["sage_button"],
        button_secondary_background_fill=PALETTE["surface"],
        button_secondary_text_color=PALETTE["sage_deep"],
        button_secondary_border_color=PALETTE["border"],
        # The focus ring is the one place the lighter sage earns its keep: it
        # is a 3px non-text indicator, where 3:1 is the applicable threshold.
        input_shadow_focus=f"0 0 0 3px {PALETTE['sage']}",
        checkbox_background_color_selected=PALETTE["sage_button"],
        checkbox_border_color_selected=PALETTE["sage_button"],
        checkbox_border_color_focus=PALETTE["sage"],
    )