app.py

"""Gradio control-plane UI — the demo front end.

This is the browser-facing demo for the whole project. It does **no** governance
of its own: it is a thin presentation layer over the real control plane assembled
in :mod:`control_plane.scenarios`. Every decision shown here is produced by the
Microsoft-AGT policy gate, every signature by real Ed25519 keys, and every audit
row by the tamper-evident hash chain. The UI's only job is to make the *governed
loop* visible.

What the operator can do:

* **Run the seven acceptance scenarios** with one click each.
* **Type a free-form task** and run it against the live model (when an API key is
  configured), choosing the autonomy tier and which agent identity signs.
* **Toggle the global kill switch** and watch otherwise-allowed actions get blocked.
* **Approve or reject** the one action that pauses for human sign-off.

What the operator sees for every run (the "governed loop visible" requirement):

* the **proposed action** (human summary *and* the raw structured form),
* the **backend path** it was routed to,
* the **governance decision**,
* the **verified agent identity** beside the decision, including any
  claimed-vs-verified mismatch (impersonation),
* the **execution result fed back to the agent** — and, when an action is blocked,
  the agent *adapting* on its next turn rather than halting, and
* the **tamper-evident audit log** with its integrity check.

The module is import-safe with no API key (the live model is built lazily, only
when a custom task is actually run), so a UI smoke test can build the whole
interface in CI without secrets or network.
"""

from __future__ import annotations

import gradio as gr

from control_plane.governance import GovernanceDecision
from control_plane.scenarios import (
    SCENARIOS_BY_KEY,
    ScenarioResult,
)
from control_plane.schema import AutonomyTier

# The autonomy tiers, in ascending order, as (label, enum-value) pairs for the
# operator's tier selector. The label spells out what each ceiling permits.
_TIER_CHOICES: list[tuple[str, str]] = [
    ("L0 · Read-only", AutonomyTier.L0_READ_ONLY.value),
    ("L1 · Recommend / draft", AutonomyTier.L1_RECOMMEND_ONLY.value),
    ("L2 · Bounded action", AutonomyTier.L2_BOUNDED_ACTION.value),
    ("L3 · Approval-required action", AutonomyTier.L3_APPROVAL_REQUIRED_ACTION.value),
]

# The two demo identities the operator can have sign a custom task, as
# ``(display, value)`` pairs: the menu shows the role in plain English (so a
# first-time viewer grasps the per-identity least-privilege difference at a
# glance), while the *value* stays the registered agent name the gate verifies
# against (``cp.identities[value]``) — so the label change is purely cosmetic.
# soc-responder may take containment actions; soc-triage is read/draft only
# (see ``build_demo_identities`` in control_plane/identity.py).
_IDENTITY_CHOICES = [
    ("Responder (can act)", "soc-responder"),
    ("Triage (read-only)", "soc-triage"),
]

# An example query per governance test case (keyed by scenario key). Clicking one
# loads its query into the task box AND sets the tier / kill-switch needed to
# reproduce that case with the LIVE agent — so each test case has a runnable
# free-text equivalent, not just its scripted button.
#
# Test case 7 (impersonation) has no live example on purpose: it requires an action
# signed with a *different* key than it claims, which a free-text query cannot
# express. It is demonstrable only via its scripted button.
# ``backend`` is the execution path the query is *designed* to exercise (the live
# model ultimately picks, but the query is written to steer it there).
_EXAMPLE_QUERIES: dict[str, dict] = {
    "l0_read": {
        "query": "Look up the asset record for ASSET-001.",
        "kill_switch": False,
        "backend": "Direct API (read)",
    },
    "l1_draft": {
        "query": "Draft remediation for CVE-2021-44228.",
        "kill_switch": False,
        "backend": "Function Call (advisory draft)",
    },
    "l2_ticket": {
        "query": "Open a low-risk tracking ticket for incident INC-1001.",
        "kill_switch": False,
        "backend": "Direct API (write)",
    },
    "l3_approval": {
        "query": "Stop the suspicious container beef5678cd90 to contain the threat.",
        "kill_switch": False,
        "backend": "Safe CLI (simulated)",
    },
    "denied": {
        "query": "Delete the host logs for incident INC-1001.",
        "kill_switch": False,
        "backend": "Safe CLI (blocked before run)",
    },
    "kill_switch": {
        # Engages the kill switch so the live run actually shows KILL_SWITCH_BLOCKED.
        "query": "Open a follow-up ticket for incident INC-1001.",
        "kill_switch": True,
        "backend": "Direct API (blocked by kill switch)",
    },
}

# One coloured badge per decision, so the verdict is readable at a glance.
_DECISION_BADGE: dict[GovernanceDecision, str] = {
    GovernanceDecision.ALLOW: "🟢 ALLOW",
    GovernanceDecision.DENY: "🔴 DENY",
    GovernanceDecision.REQUIRE_APPROVAL: "🟡 REQUIRE_APPROVAL",
    GovernanceDecision.KILL_SWITCH_BLOCKED: "⛔ KILL_SWITCH_BLOCKED",
}


# --------------------------------------------------------------------------- #
# Rendering helpers — turn a run's result into the panels the UI shows         #
# --------------------------------------------------------------------------- #


def _decision_badge(decision: GovernanceDecision) -> str:
    """The coloured label for a single decision (falls back to the raw name)."""
    return _DECISION_BADGE.get(decision, decision.value)


def _final_decision(result: ScenarioResult) -> GovernanceDecision | None:
    """The decision the *first* governed turn resolved to — the headline verdict.

    For most scenarios there is one turn; for the adapt-after-deny scenario the
    first turn (the blocked one) is the point being demonstrated.
    """
    return result.outcome.turns[0].decision if result.outcome.turns else None


def _identity_line(claimed: str, verified: str | None) -> str:
    """One line of claimed-vs-verified identity, flagging any mismatch.

    This is the identity attribution made visible: a verified DID means the signature
    checked out against the *claimed* agent's registered key; a missing DID on a
    denied action is the signature of impersonation.
    """
    if verified:
        return f"claimed `{claimed}` → ✅ verified `{verified}`"
    return f"claimed `{claimed}` → ❌ **not verified** (impersonation / unsigned)"


def _render_decision_banner(result: ScenarioResult) -> str:
    """The headline panel: what was demonstrated and the resulting verdict."""
    scenario = result.scenario
    decision = _final_decision(result)
    verdict = _decision_badge(decision) if decision else "—"
    # number 0 is the live free-text run; the numbered cases are the governance
    # test cases. Title each accordingly so neither reads as "a canned scenario".
    heading = (
        scenario.title
        if scenario.number == 0
        else f"Governance test case {scenario.number} — {scenario.title}"
    )
    return f"### {verdict}\n\n**{heading}**\n\n{scenario.demonstrates}"


# How each run-ending status reads to a human (used by the scorecard).
_STATUS_LABEL: dict[str, str] = {
    "running": "⏳ running… (live)",
    "final_answer": "✅ agent finished",
    "stopped_repeat": "⚠️ stopped — agent repeated a completed action",
    "max_turns": "⚠️ hit the turn limit",
    "error": "❌ error",
}


def _render_scorecard(result: ScenarioResult) -> str:
    """A one-glance summary of the whole run, shown above the turn-by-turn loop.

    A long run (especially a live one) can be dozens of turns; this line lets the
    operator grasp the shape of it — how many actions, how many were allowed vs
    blocked, how many actually executed, and how the run ended — without scrolling.
    """
    turns = result.outcome.turns
    allow = sum(1 for t in turns if t.decision is GovernanceDecision.ALLOW)
    blocked = len(turns) - allow
    executed = sum(1 for t in turns if t.executed)
    status = _STATUS_LABEL.get(result.outcome.status, result.outcome.status)
    return (
        f"**Run summary —** {len(turns)} turn(s) · 🟢 {allow} allowed · "
        f"🔴 {blocked} blocked · ▶ {executed} executed · status: {status}"
    )


def _render_turn(i: int, turn) -> str:
    """Render ONE turn as a collapsed, click-to-expand block.

    The headline (turn number, decision, action) is always visible; the detail —
    the proposal, the raw structured action, the decision reason, and the result
    fed back to the agent — is tucked inside a ``<details>`` so a long loop stays
    scannable. Expanding a turn reveals the full governed-loop story.
    """
    action = turn.action
    headline = (
        f"Turn {i} · {_decision_badge(turn.decision)} · "
        f"<code>{action.action_name}</code> on <code>{action.backend.value}</code>"
    )

    body: list[str] = []
    if turn.summary:
        body.append(f"**Agent proposed:** {turn.summary}")
    body.append(
        f"**Action:** `{action.action_name}`  ·  "
        f"**Backend:** `{action.backend.value}`  ·  "
        f"**Proposed tier:** `{action.autonomy_tier.value}`"
    )
    # The exact structured action, so the proposal is fully inspectable.
    body.append(f"```json\n{_pretty_json(action.model_dump(mode='json'))}\n```")
    body.append(f"**Decision:** {_decision_badge(turn.decision)} — {turn.reason}")
    if turn.approval_record is not None:
        rec = turn.approval_record
        body.append(
            f"**Human approval:** operator `{rec.approver}` **{rec.decision}** "
            f"this action — {rec.reason}"
        )
    body.append(f"**Fed back to the agent →** {_feedback_text(turn)}")

    inner = "\n\n".join(body)
    return f"<details><summary>{headline}</summary>\n\n{inner}\n\n</details>"


def _render_loop(result: ScenarioResult) -> str:
    """Render the governed loop: a scorecard, each turn collapsed, then the answer.

    Each turn is one click-to-expand block (see :func:`_render_turn`), so the loop
    is readable whether it is 1 turn or 30. When a turn is blocked, the *next* turn
    shows the agent adapting — the closed feedback loop the system requires to be
    visible, not internal-only.
    """
    lines: list[str] = ["## Governed loop", _render_scorecard(result), ""]

    for i, turn in enumerate(result.outcome.turns, start=1):
        lines.append(_render_turn(i, turn))

    # Call out the adapt-after-block beat explicitly (the headline of scenario 5).
    if _agent_adapted(result):
        lines.append(
            "> 🔁 **The loop kept going:** after the block, the agent adapted and "
            "proposed an allowed action instead of halting."
        )

    # The agent's terminal answer — collapsed, because a live model's answer can be
    # long. The operator opens it when they want the prose; the decision and the
    # loop above already tell the governance story.
    if result.outcome.final_answer:
        lines.append(
            "<details><summary>📝 <b>Final answer</b> (click to read)</summary>\n\n"
            f"{result.outcome.final_answer}\n\n</details>"
        )
    elif result.outcome.status == "error":
        lines.append(f"### ❌ Error\n{result.outcome.error}")

    return "\n\n".join(lines)


def _feedback_text(turn) -> str:
    """A short description of what the agent was told after this turn.

    Mirrors the harness's own feedback builder closely enough to make the loop
    legible: the backend result on a successful run, or the block/approval outcome
    otherwise. (The harness builds the real feedback string internally; this is the
    human-readable echo of it.)
    """
    if turn.executed:
        return f"ALLOWED and executed. Backend returned: `{_short(turn.result)}`"
    if turn.execution_error is not None:
        return f"ALLOWED but the backend errored: {turn.execution_error}"
    if turn.approval_record is not None and turn.approval_record.decision == "rejected":
        return "REJECTED by the operator — nothing ran; the agent must adapt."
    return (
        f"NOT executed ({turn.decision.value}). No backend was touched; "
        "the agent must adapt or explain."
    )


def _agent_adapted(result: ScenarioResult) -> bool:
    """True when a blocked turn was followed by a further proposal (visible adapt)."""
    turns = result.outcome.turns
    return (
        len(turns) >= 2
        and turns[0].decision is not GovernanceDecision.ALLOW
        and not turns[0].executed
    )


def _render_identity(result: ScenarioResult) -> str:
    """The identity panel: claimed vs verified, deduplicated.

    Reads straight off the audit records so the panel and the log agree by
    construction. A long run repeats the same identity many times, so rather than
    one line per decision we list each *unique* ``claimed → verified`` pair once,
    with how many actions it covered, and surface any ❌ mismatch (the signature of
    impersonation) first so it can't be missed.
    """
    # Collapse to unique (claimed, verified) pairs, counting and preserving order.
    seen: dict[tuple[str, str | None], int] = {}
    for rec in result.audit_records:
        key = (rec.claimed_agent_id, rec.verified_agent_did)
        seen[key] = seen.get(key, 0) + 1

    if not seen:
        return "## Verified agent identity\n\n_No decisions recorded._"

    # Mismatches (no verified DID) first — that is the thing worth seeing.
    ordered = sorted(seen.items(), key=lambda kv: kv[0][1] is not None)

    lines = ["## Verified agent identity"]
    for (claimed, verified), count in ordered:
        suffix = f"  ·  {count} action(s)" if count > 1 else ""
        lines.append(f"- {_identity_line(claimed, verified)}{suffix}")
    return "\n".join(lines)


def _render_audit_rows(result: ScenarioResult) -> list[list[str]]:
    """The audit log as table rows — every decision, in order."""
    rows: list[list[str]] = []
    for i, rec in enumerate(result.audit_records, start=1):
        rows.append(
            [
                str(i),
                rec.policy_decision,
                rec.engine,
                rec.action_name,
                rec.backend,
                rec.operator_tier,
                rec.claimed_agent_id,
                (rec.verified_agent_did or "—"),
                _short(rec.reason, limit=90),
            ]
        )
    return rows


def _render_integrity(result: ScenarioResult) -> str:
    """The hash-chain integrity line — proof the audit log is tamper-evident."""
    if result.audit_valid:
        return (
            f"🔒 **Audit chain verified** — {len(result.audit_records)} record(s), "
            "hash chain intact (tamper-evident)."
        )
    return f"⚠️ **Audit chain verification FAILED:** {result.audit_error}"


def _render_all(result: ScenarioResult) -> tuple:
    """Render one run into the full tuple of UI outputs (order matches the bindings)."""
    return (
        _render_decision_banner(result),
        _render_loop(result),
        _render_identity(result),
        _render_audit_rows(result),
        _render_integrity(result),
    )


# Small formatting utilities -------------------------------------------------- #


def _pretty_json(value: object) -> str:
    import json

    return json.dumps(value, indent=2, sort_keys=True)


def _short(value: object, limit: int = 240) -> str:
    """A compact, one-line preview of a backend result for the feedback line."""
    text = str(value)
    text = " ".join(text.split())  # collapse whitespace/newlines
    return text if len(text) <= limit else text[: limit - 1] + "…"


# --------------------------------------------------------------------------- #
# Event handlers                                                              #
# --------------------------------------------------------------------------- #


def _on_custom_task(
    task: str,
    tier: str,
    identity_name: str,
    kill_switch: bool,
    approval_choice: str,
):
    """Run a free-form task against the **live** model, **streaming** progress.

    This is a *generator*: Gradio renders the UI on every ``yield``, so the operator
    sees instant feedback the moment they click and then watches the governed loop
    fill in turn by turn — instead of staring at a frozen screen while the model
    thinks (the first model call alone can take a minute or two).

    The real OpenRouter-backed agent decides what to do, one governed step at a
    time, while the same gate / dispatcher / audit log govern it. It needs an
    ``OPENROUTER_API_KEY``; without one we show a friendly message, so the rest of
    the demo still works offline.
    """
    if not task or not task.strip():
        yield _info_panels("Enter a task above, or load one of the example queries.")
        return

    import queue
    import threading
    import time

    from agents.soc_agent import TurnResult
    from control_plane.scenarios import build_control_plane

    # The model run happens on a BACKGROUND thread; this generator stays free to
    # re-render the page ~once a second. That is what makes streaming actually work:
    # the agent does most of its work as advisory tool calls *inside one model turn*,
    # so between-turn yields alone never update mid-turn. Instead we tap the
    # per-action progress hook (govern_action → on_turn), which fires for EVERY step
    # — advisory or structured — and push each onto a queue the UI drains live.
    cp = build_control_plane()
    progress: "queue.Queue[object]" = queue.Queue()
    live_turns: list = []
    state: dict = {}

    def on_turn(turn: "TurnResult") -> None:
        # Runs on the worker thread as each action is governed. Record it and poke
        # the UI thread to re-render. Kept trivial (no rendering here).
        live_turns.append(turn)
        progress.put("turn")

    def worker() -> None:
        try:
            state["outcome"] = _run_live_loop(
                task.strip(),
                tier=tier,
                identity_name=identity_name,
                kill_switch=kill_switch,
                approval_choice=approval_choice,
                cp=cp,
                on_turn=on_turn,
            )
        except Exception as exc:  # surfaced to the UI below, never crashes the app.
            state["error"] = exc
        finally:
            progress.put("done")

    # 1) Instant feedback the moment the button is clicked — the screen changes
    #    immediately, so the model warm-up never looks like a hang.
    yield _live_progress_panels(cp, live_turns, elapsed=0, building=True)

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

    # 2) Drain the progress queue. We block up to 1s for the next step; on timeout
    #    we still re-render so the elapsed-time counter visibly ticks (motion even
    #    during a slow ~10–15s model round-trip). Each governed step appears as it
    #    lands.
    start = time.monotonic()
    while True:
        try:
            signal = progress.get(timeout=1.0)
        except queue.Empty:
            signal = None
        if signal == "done":
            break
        yield _live_progress_panels(
            cp, live_turns, elapsed=int(time.monotonic() - start), building=False
        )

    # 3) Final render — the real terminal outcome (status, final answer), or a
    #    friendly message if the live model couldn't run (usually a missing key).
    if "error" in state:
        exc = state["error"]
        if isinstance(exc, RuntimeError):
            yield _info_panels(
                f"**Live model unavailable:** {exc}\n\n"
                "Set `OPENROUTER_API_KEY` to run free-form tasks, or use the one-click "
                "test cases, which run fully offline."
            )
        else:
            yield _info_panels(f"**The live run failed:** {type(exc).__name__}: {exc}")
        return
    yield _render_all(_live_result(task, tier, identity_name, cp, state["outcome"]))


def _run_live_loop(
    task: str,
    *,
    tier: str,
    identity_name: str,
    kill_switch: bool,
    approval_choice: str,
    cp,
    on_turn,
):
    """Build and run the live governed loop to completion (called on a worker thread).

    Wired with all four backends (including MCP) via the canonical dispatcher
    factory, so a custom task can use any execution path. The live model is built
    lazily here — importing this module never needs a key. Returns the final
    :class:`~agents.soc_agent.LoopOutcome`; progress is reported via ``on_turn``.
    """
    # Imported lazily so app import (and the smoke test) never require a live model.
    from agents.soc_agent import _SYSTEM_PROMPT, GovernedAgentLoop, build_brain
    from control_plane.approval import ApprovalGate, auto_approve, auto_reject
    from control_plane.kill_switch import KillSwitch
    from control_plane.settings import load_settings
    from execution_backends.dispatcher import build_default_dispatcher

    # The live-run turn cap is read from configuration (OPENROUTER_MAX_TURNS in
    # .env) — a single, code-free knob for the demo's worst-case run time.
    settings = load_settings()
    identity = cp.identities.get(identity_name)
    switch = KillSwitch(engaged=bool(kill_switch))

    # The operator's standing approve/reject choice resolves any REQUIRE_APPROVAL
    # the live model triggers (single-operator synchronous gate).
    operator = (
        auto_approve(approver="operator", reason="approved in the demo UI")
        if approval_choice == "approve"
        else auto_reject(approver="operator", reason="rejected in the demo UI")
    )
    approval_gate = ApprovalGate(operator, audit_sink=cp.audit)

    # Demo-tuned prompt: the live model's wall-clock cost is dominated by how many
    # tool calls it chains (each is a separate ~10–15s round-trip). For a live demo
    # we ask it to be decisive — at most one advisory lookup before it proposes an
    # action or concludes. This only steers the free-text path; governance is unchanged.
    demo_prompt = (
        _SYSTEM_PROMPT
        + "\n\nDEMO EFFICIENCY (important): You are running in a live, interactive "
        "demo. Be decisive and fast. Do EXACTLY what the task asks and nothing more. "
        "Most demo tasks are a SINGLE concrete step (e.g. open a ticket, look up a "
        "record, draft remediation): for those, take that one step — propose the one "
        "governed action, or make the one advisory call that IS the task — and the "
        "moment it succeeds, give your FinalAnswer immediately. Do NOT add any extra "
        "investigation, containment, remediation, risk-scoring or follow-up steps the "
        "task did not explicitly ask for. Make AT MOST ONE advisory tool call. Never "
        "repeat a lookup or re-create something you already did. Finish within 2–3 turns."
    )
    brain = build_brain(system_prompt=demo_prompt)  # raises RuntimeError if no API key.

    # All four real backends; the context manager releases the MCP connection after.
    with build_default_dispatcher(cp.gate.gate_verifier) as dispatcher:
        loop = GovernedAgentLoop(
            brain=brain,
            gate=cp.gate,
            dispatcher=dispatcher,
            operator_tier=tier,
            kill_switch=switch,
            identity=identity,
            approval_gate=approval_gate,
            incident_id="INC-1001",
            max_turns=settings.max_turns,
            stop_on_repeat=True,
            on_turn=on_turn,  # report every governed step the instant it happens.
        )
        return loop.run(task)


def _live_result(task: str, tier: str, identity_name: str, cp, outcome):
    """Wrap a live outcome in the same shape a scenario uses, so renderers are shared."""
    from control_plane.scenarios import Scenario, _finish

    pseudo = Scenario(
        key="custom",
        number=0,
        title="Custom task (live model)",
        demonstrates="A free-form task driven by the live agent under full governance.",
        request=task,
        operator_tier=AutonomyTier(tier),
        expected_decision=_final_or_allow(outcome),
        identity_name=identity_name,
        build_turns=lambda _identity: [],
    )
    return _finish(pseudo, outcome, cp.audit)


def _live_progress_panels(cp, live_turns: list, *, elapsed: int, building: bool) -> tuple:
    """Render an in-progress live run: a ticking banner plus the steps so far.

    Reuses the normal loop/identity/audit renderers on a synthetic "running"
    outcome built from the steps governed so far, so mid-run the page looks exactly
    like a finished run — just still filling in. The banner carries an elapsed-time
    counter so there is visible motion even during a slow model round-trip.
    """
    from agents.soc_agent import LoopOutcome
    from control_plane.scenarios import ScenarioResult

    if building and not live_turns:
        banner = (
            "### ⏳ Live run starting…\n\n"
            "Building the model and contacting OpenRouter. The first governed step "
            "usually appears in ~15–45s (each step is a model round-trip). Steps "
            "stream in below as they happen."
        )
        return (banner, "", "", [], "")

    outcome = LoopOutcome(status="running")
    outcome.turns = list(live_turns)
    # The worker thread is concurrently appending audit records; snapshot defensively
    # so a rare mid-write read can never break the progress render.
    try:
        records = cp.audit.records
        valid, error = cp.audit.verify()
    except Exception:
        records, valid, error = (), True, None
    result = ScenarioResult(
        scenario=None,  # banner is overridden below, so no scenario is needed.
        outcome=outcome,
        audit_records=records,
        audit_valid=valid,
        audit_error=error,
    )
    banner = (
        f"### ⏳ Live run in progress — {elapsed}s elapsed · "
        f"{len(live_turns)} step(s) governed so far\n\n"
        "_Each step is a real model round-trip (~10–15s); it appears the moment the "
        "gate decides it. The final answer arrives when the agent is done._"
    )
    # Call the per-panel renderers directly (not _render_all): the decision banner
    # needs a Scenario, which we don't have mid-run — we supply our own banner above.
    return (
        banner,
        _render_loop(result),
        _render_identity(result),
        _render_audit_rows(result),
        _render_integrity(result),
    )


def _final_or_allow(outcome) -> GovernanceDecision:
    """Best-effort headline decision for a custom run (ALLOW if it produced nothing)."""
    return outcome.turns[0].decision if outcome.turns else GovernanceDecision.ALLOW


def _info_panels(message: str) -> tuple:
    """Render an informational message into every output panel (no run happened)."""
    return (message, "", "", [], "")


# Plain-English meaning of each autonomy tier, for the example-query table's
# "Sets" column (so the user understands what the example configures, not just a
# tier code).
_TIER_BLURB: dict[str, str] = {
    "L0_READ_ONLY": "Autonomy **L0** — agent may only read",
    "L1_RECOMMEND_ONLY": "Autonomy **L1** — may also draft / recommend",
    "L2_BOUNDED_ACTION": "Autonomy **L2** — may also take low-risk actions",
    "L3_APPROVAL_REQUIRED_ACTION": "Autonomy **L3** — may also request high-risk actions (need approval)",
}


def _example_query_table_md() -> str:
    """A markdown table documenting what each example-query button loads.

    Built from the same ``_EXAMPLE_QUERIES`` data the buttons use, so the table and
    the buttons can never drift apart. The "Sets" column spells out, in plain
    English, what each example configures and why.
    """
    rows = [
        "| # | Click loads this query | …and sets these controls | How it runs |",
        "|:--:|---|---|---|",
    ]
    for key, cfg in _EXAMPLE_QUERIES.items():
        sc = SCENARIOS_BY_KEY[key]
        sets = _TIER_BLURB[sc.operator_tier.value]
        if cfg["kill_switch"]:
            sets += " · **Kill switch ON** — execution globally paused"
        rows.append(f"| {sc.number} | \"{cfg['query']}\" | {sets} | {cfg['backend']} |")
    rows.append(
        "| 7 | _(impersonation — needs a forged signature, which a typed query "
        "can't express)_ | _use the scripted **7. Identity · Impersonation** button_ "
        "| Direct API (denied at identity check) |"
    )
    rows.append("")
    rows.append(
        "<sub>**How it runs** is the backend / execution path each query is *written* "
        "to take — **Direct API**, **Function Call**, **MCP**, or **Safe CLI**. The "
        "live agent makes the final call, so a run may take a different one.</sub>"
    )
    return "\n".join(rows)


def _waiting_panels(message: str) -> tuple:
    """The instant "working…" state shown the moment a live run is kicked off.

    Goes in the decision-banner slot (top of the page) with the other panels
    cleared, so the screen visibly changes immediately instead of looking frozen
    while the model warms up.
    """
    return (message, "🔄 _Starting the governed loop…_", "", [], "")


# --------------------------------------------------------------------------- #
# The interface                                                               #
# --------------------------------------------------------------------------- #

# The intro is split into three pieces so the dense "kill switch → identity →
# tier → backend → policy" chain becomes a glanceable picture instead of a wall
# of jargon: a one-line value prop (markdown), the gate diagram (inline SVG), and
# a one-line "try it" (markdown). Two short text lines bookend the visual.
_HEADLINE = """
# 🛡️ SOC Agent Control Plane
**Governed security agents that operate strictly within your limits, with full action verification and logging**
"""


# The governance pipeline as an inline SVG (via gr.HTML, so it is self-contained
# and survives deployment to a Space — no asset path / allowed_paths to wire, and
# it stays crisp at any size). Colours are chosen to read on both the light and
# dark themes: solid-fill boxes with white text, light chips with dark text.
_GATE_SVG = """
<svg viewBox="0 66 800 188" width="100%" style="max-width:780px;height:auto;display:block;margin:0 auto 6px"
     xmlns="http://www.w3.org/2000/svg" font-family="ui-sans-serif,system-ui,sans-serif">
  <defs>
    <marker id="arr" markerWidth="8" markerHeight="8" refX="5.5" refY="3" orient="auto">
      <path d="M0,0 L6,3 L0,6 Z" fill="#94a3b8"/>
    </marker>
  </defs>

  <!-- 1. the agent proposes an action -->
  <rect x="6" y="100" width="120" height="52" rx="10" fill="#6366f1"/>
  <text x="66" y="122" text-anchor="middle" fill="#fff" font-size="12.5" font-weight="600">Agent proposes</text>
  <text x="66" y="139" text-anchor="middle" fill="#fff" font-size="12.5" font-weight="600">an action</text>
  <line x1="128" y1="126" x2="158" y2="126" stroke="#94a3b8" stroke-width="2" marker-end="url(#arr)"/>

  <!-- 2. one governance gate, four sequential checks. Each chip carries a tiny
       plain-English caption under its name so a non-expert grasps what the check
       does (the bare term alone — "Tier" — means nothing to them). The Policy
       chip is widened and emphasised: it is the per-action rulebook, and it names
       the actual data file (policies/agt_policy.yaml) so a reader sees the rules
       live in editable config, not in code. -->
  <rect x="160" y="72" width="434" height="108" rx="12" fill="none" stroke="#6366f1" stroke-width="1.6"/>
  <text x="377" y="90" text-anchor="middle" fill="#6366f1" font-size="11" font-weight="700" letter-spacing="0.6">GOVERNANCE GATE</text>
  <g text-anchor="middle">
    <rect x="170" y="102" width="72" height="58" rx="8" fill="#eef2ff" stroke="#c7d2fe"/>
    <text x="206" y="126" fill="#4338ca" font-size="11.5" font-weight="700">Kill switch</text><text x="206" y="143" fill="#6366f1" font-size="8.5">global stop</text>
    <rect x="259" y="102" width="72" height="58" rx="8" fill="#eef2ff" stroke="#c7d2fe"/>
    <text x="295" y="126" fill="#4338ca" font-size="11.5" font-weight="700">Identity</text><text x="295" y="143" fill="#6366f1" font-size="8.5">who's acting</text>
    <rect x="348" y="102" width="72" height="58" rx="8" fill="#eef2ff" stroke="#c7d2fe"/>
    <text x="384" y="126" fill="#4338ca" font-size="11.5" font-weight="700">Tier</text><text x="384" y="143" fill="#6366f1" font-size="8.5">autonomy cap</text>
    <rect x="436" y="100" width="148" height="62" rx="8" fill="#e0e7ff" stroke="#a5b4fc"/>
    <text x="510" y="121" fill="#3730a3" font-size="11.5" font-weight="700">Policy — the rulebook</text>
    <text x="510" y="137" fill="#4338ca" font-size="9">allow · deny · approval</text>
    <text x="510" y="151" fill="#4f46e5" font-size="8.3" font-family="ui-monospace,monospace">policies/agt_policy.yaml</text>
  </g>
  <g fill="#a5b4fc" font-size="15" text-anchor="middle">
    <text x="250" y="135">›</text><text x="339" y="135">›</text><text x="428" y="135">›</text>
  </g>
  <line x1="594" y1="126" x2="622" y2="126" stroke="#94a3b8" stroke-width="2" marker-end="url(#arr)"/>

  <!-- 3. outcome: only ALLOW reaches a backend -->
  <rect x="624" y="98" width="62" height="27" rx="13" fill="#16a34a"/>
  <text x="655" y="116" text-anchor="middle" fill="#fff" font-size="11.5" font-weight="700">ALLOW</text>
  <text x="694" y="116" fill="#16a34a" font-size="11.5" font-weight="600">→ executes</text>
  <rect x="624" y="133" width="62" height="27" rx="13" fill="#dc2626"/>
  <text x="655" y="151" text-anchor="middle" fill="#fff" font-size="11.5" font-weight="700">DENY</text>
  <text x="694" y="151" fill="#dc2626" font-size="11.5" font-weight="600">→ blocked</text>

  <!-- 4. every decision (allow or deny) is recorded -->
  <line x1="377" y1="180" x2="377" y2="206" stroke="#94a3b8" stroke-width="2" marker-end="url(#arr)"/>
  <line x1="690" y1="162" x2="690" y2="206" stroke="#94a3b8" stroke-width="2" marker-end="url(#arr)"/>
  <rect x="160" y="208" width="640" height="38" rx="10" fill="#475569"/>
  <text x="480" y="232" text-anchor="middle" fill="#fff" font-size="12" font-weight="600">Tamper-evident audit log — every decision recorded</text>
</svg>
"""

_AUDIT_HEADERS = [
    "#",
    "Decision",
    "Engine",
    "Action",
    "Backend",
    "Operator tier",
    "Claimed agent",
    "Verified DID",
    "Reason",
]


def build_demo() -> gr.Blocks:
    """Construct the Gradio interface (no server started — the caller launches it).

    Kept as a pure builder so the smoke test can assemble the whole UI in CI
    without calling ``launch`` (no port, no network).
    """
    with gr.Blocks(title="Agent Control Plane", theme=gr.themes.Soft()) as demo:
        # Gradio renders the dataframe's built-in "copy" and "fullscreen" toolbar
        # icons (top-right of the audit table) tiny and transparent, so first-time
        # users miss them. We enlarge them and give them a visible bordered/filled
        # button look so they read as clickable controls. A <style> tag injected via
        # gr.HTML applies wherever it lands in the DOM, so this also works on the
        # deployed Space (Gradio 6 moved the Blocks ``css=`` arg to ``launch()``,
        # which build_demo() does not call). Scoped to ``#audit-log-table`` so no
        # other component's icons are affected. ``--size-*`` etc. are Gradio theme
        # vars, so the styling tracks the active theme.
        gr.HTML(
            """
            <style>
            /* Position/space the toolbar icon cluster a touch off the table edge. */
            #audit-log-table .icon-buttons { top: 6px; right: 6px; gap: 6px; }
            /* Make each icon a clearly-clickable, bordered button (was ~24px,
               transparent). */
            #audit-log-table .icon-button {
                width: var(--size-9); height: var(--size-9);
                background: var(--background-fill-secondary) !important;
                border: 1px solid var(--border-color-accent);
                border-radius: var(--radius-md);
            }
            #audit-log-table .icon-button:hover {
                background: var(--color-accent-soft) !important;
            }
            /* Scale the SVG glyph up to match the larger button. */
            #audit-log-table .icon-button svg {
                width: var(--size-6); height: var(--size-6);
            }
            /* Paint the primary "Run" button green (go), overriding the indigo
               theme accent so the main call-to-action stands out. */
            #run-btn {
                background: #16a34a !important;
                border-color: #15803d !important;
                color: #fff !important;
            }
            #run-btn:hover { background: #15803d !important; }
            </style>
            """
        )
        # Short value prop → visual gate pipeline → short "try it" (see the
        # _HEADLINE / _GATE_SVG / _TRY_IT comments above).
        gr.Markdown(_HEADLINE)
        gr.HTML(_GATE_SVG)

        with gr.Row():
            # -- Left column: the operator's controls ------------------------ #
            with gr.Column(scale=1):
                gr.Markdown("### Operator controls")
                # Default to L2 for the live box: the example tasks involve reading
                # *and* drafting (needs L1) and low-risk actions (needs L2), so an L2
                # ceiling lets a free-form task actually succeed end to end. (At the
                # lower L0 default a "draft …" task is correctly denied, which reads
                # as broken to a first-time user.) The seven test-case buttons set
                # their own tier, so this default only affects the custom task.
                tier = gr.Dropdown(
                    choices=_TIER_CHOICES,
                    value=AutonomyTier.L2_BOUNDED_ACTION.value,
                    label="Autonomy tier for a custom task (the ceiling the gate enforces)",
                    # info supports markdown (Gradio 6), so the tier guide renders
                    # as a real bulleted list instead of one dense run-on line.
                    info=(
                        "**Tier guide** — pick one that allows what your task asks for:\n"
                        "- **L0** — read only\n"
                        "- **L1** — also draft / recommend\n"
                        "- **L2** — also low-risk actions\n"
                        "- **L3** — also high-risk, with human approval"
                    ),
                )
                # Only L3 tasks ever pause for human sign-off, so this control is
                # meaningless for L0–L2 and would just confuse a first-time user. It
                # starts hidden (default tier is L2) and is revealed by the tier
                # dropdown's change handler only when L3 is selected (wired below).
                # It is a *standing pre-authorization*: the verdict applied if/when
                # the run hits REQUIRE_APPROVAL — framed honestly so its purpose is clear.
                approval = gr.Radio(
                    choices=["approve", "reject"],
                    value="approve",
                    label="Pre-authorize high-risk (L3) actions",
                    info=(
                        "An L3 task pauses for operator sign-off. This is the standing "
                        "verdict the gate applies when that happens — approve lets the "
                        "(simulated) action run; reject blocks it and the agent adapts."
                    ),
                    visible=False,
                )
                identity = gr.Dropdown(
                    # (display, value) tuples: the human-readable role shows in the
                    # menu while the underlying value stays the registered agent name
                    # the gate verifies against, so no downstream logic changes.
                    choices=_IDENTITY_CHOICES,
                    value="soc-responder",
                    label="Select agent role",
                    # Kept to one short, scannable sentence (a recruiter skims, not
                    # reads): the demo recipe alone conveys the per-identity
                    # least-privilege point without the cryptography lecture.
                    info=(
                        "Each role has limited powers. Try Triage → ask it to stop a "
                        "container → the gate denies it."
                    ),
                )
                kill_switch = gr.Checkbox(
                    value=False,
                    label="🔴 Kill switch (globally block all non-read execution)",
                )
                task = gr.Textbox(
                    label="Enter Task (type a task or load an example below)",
                    placeholder="e.g. Draft remediation for CVE-2021-44228.",
                    lines=3,
                )
                # Example query per test case: one click loads its query into the box
                # and sets the tier (+ kill switch) needed to reproduce that case with
                # the LIVE agent. The operator still presses Run. Default-arg binding
                # captures each key; the handler returns into [task, tier, kill_switch].
                gr.Markdown(
                    "<sub>Example query — click to load it, then press Run:</sub>"
                )
                with gr.Row():
                    for _key, _cfg in _EXAMPLE_QUERIES.items():
                        _sc = SCENARIOS_BY_KEY[_key]
                        gr.Button(_sc.title, size="sm").click(
                            fn=lambda key=_key: (
                                _EXAMPLE_QUERIES[key]["query"],
                                SCENARIOS_BY_KEY[key].operator_tier.value,
                                _EXAMPLE_QUERIES[key]["kill_switch"],
                            ),
                            outputs=[task, tier, kill_switch],
                        )
                # A reference table so the operator can see what each example loads
                # without clicking. Collapsed by default to keep the panel tidy.
                with gr.Accordion("ℹ️ What each example query loads", open=False):
                    gr.Markdown(_example_query_table_md())

                with gr.Row():
                    # elem_id lets the injected CSS paint this primary button green
                    # (go = run), distinct from the indigo theme accent used elsewhere.
                    run_task = gr.Button("Run", variant="primary", elem_id="run-btn")
                    # Clear the task box so a new task can be typed (empties the field).
                    gr.Button("Clear", variant="secondary").click(
                        fn=lambda: "", outputs=task
                    )

            # -- Right column: the governed-loop views ----------------------- #
            with gr.Column(scale=2):
                decision_banner = gr.Markdown("### Run a task to see a decision.")
                identity_panel = gr.Markdown()
                loop_panel = gr.Markdown()
                gr.Markdown("## Audit log")
                integrity = gr.Markdown()
                audit = gr.Dataframe(
                    headers=_AUDIT_HEADERS,
                    wrap=True,
                    interactive=False,
                    label="Every governance decision, in order",
                    # Tag the component so the CSS below (injected via gr.HTML at the
                    # top of the page) can target *this* table's toolbar icons only.
                    elem_id="audit-log-table",
                )

        # -- Wiring: custom live task -------------------------------------- #
        run_task.click(
            fn=_on_custom_task,
            inputs=[task, tier, identity, kill_switch, approval],
            outputs=[decision_banner, loop_panel, identity_panel, audit, integrity],
        )

        # -- Wiring: reveal the pre-authorization control only for L3 ------- #
        # The approve/reject verdict only takes effect on an L3 task (the one tier
        # that pauses for sign-off), so show it exactly when L3 is selected and hide
        # it otherwise. This fires on programmatic tier changes too, so loading an
        # L3 example query (#4, #5) reveals it automatically.
        tier.change(
            fn=lambda t: gr.update(
                visible=(t == AutonomyTier.L3_APPROVAL_REQUIRED_ACTION.value)
            ),
            inputs=tier,
            outputs=approval,
        )

    return demo


def main() -> None:
    """Launch the demo (used when running ``python app.py`` locally / on a Space)."""
    build_demo().launch()


if __name__ == "__main__":
    main()