from __future__ import annotations import difflib import hashlib import html import inspect import json import os import random import time import uuid from typing import Any, Callable import gradio as gr from demo_controller import DemoRequest, execute_request, load_asset, quick_start_payload from engines.presets import ( BIT_WIDTHS, MODE_HELP, MODE_LABELS, PAGE_SIZES, SHORTLIST_POLICIES, context_choices, model_choices, mode_choices, preset_choices, ) PRESET_BUTTON_LABELS = { "fastest_safe": "โšก Fastest (Safe)", "balanced": "๐ŸŽฏ Balanced (Recommended)", "aggressive": "๐ŸŒถ๏ธ Aggressive", } ARENA_PRESET_BUTTON_LABELS = { "safe_match": "๐Ÿ›ก Safe Match", "balanced_match": "๐ŸŽฏ Systems Match", "fragile_match": "๐Ÿงช Probe Match", } ARENA_PRESET_STATUS_HTML = { "safe_match": "
๐Ÿ›ก Safe MatchStart from the paper's compact-task matrix, where task success stays unchanged while decode gets faster.
", "balanced_match": "", "fragile_match": "
๐Ÿงช Probe MatchAn exploratory 32K/49K probe that shows the longer-context speedups from the paper.
", } LIVE_SAFE_CONTEXT_CHOICES = [4096] BENCHMARK_LIVE_CONTEXT_CHOICES = [1024, 2048] FULL_CONTEXT_CHOICES = context_choices() MAX_UI_LOG_LINES = 80 MAX_UI_LOG_CHARS = 12000 MAX_UI_TRACE_ITEMS = 8 MAX_ARENA_MATCHES = 64 CUSTOM_LIVE_PROMPTS_ENABLED = True LIVE_EXAMPLE_FILLERS = { "retrieval": "Background memo about permit backlogs, bridge closures, zoning appeals, and archive indexing.", "reasoning": "Archived finance notes mention approvals, invoices, compliance dates, and transport budgets across several quarters.", "fragile": "Migration draft notes discuss release sequencing, audit annotations, appendices, and section renumbering in detail.", } FIRST_TIME_HINT_HTML = ( "
๐Ÿ’ก New here?" "Choose a paper section from the preset dropdown, then try ๐ŸŽฏ Balanced (Recommended) for the default comparison profile.
" ) PRESET_STATUS_HTML = { "fastest_safe": ( "
โšก Fastest (Safe)" "Conservative live/demo profile. Keeps the safer selector path while still showing the runtime win.
" + FIRST_TIME_HINT_HTML ), "balanced": ( "" + FIRST_TIME_HINT_HTML ), "aggressive": ( "
๐ŸŒถ๏ธ Aggressive" "Pushes compression/shortlist behavior harder. Faster in some cases, but more likely to drift.
" + FIRST_TIME_HINT_HTML ), } COMPARE_PRESET_QUERY_ALIASES = { "fastest": "fastest_safe", "fastest_safe": "fastest_safe", "safe": "fastest_safe", "compact": "fastest_safe", "balanced": "balanced", "backend": "balanced", "aggressive": "aggressive", "probe": "aggressive", } ARENA_PRESET_QUERY_ALIASES = { "safe": "safe_match", "safe_match": "safe_match", "compact": "safe_match", "balanced": "balanced_match", "balanced_match": "balanced_match", "backend": "balanced_match", "fragile": "fragile_match", "fragile_match": "fragile_match", "probe": "fragile_match", } ARENA_MATCH_STORE: dict[str, dict[str, Any]] = {} def _format_float(value: float, digits: int = 2) -> str: return f"{value:.{digits}f}" def _format_ratio(value: float) -> str: return f"{value:.2f}ร—" def _format_bytes(value: int) -> str: amount = float(value) units = ["B", "KB", "MB", "GB", "TB"] for unit in units: if amount < 1024.0 or unit == units[-1]: if unit == "B": return f"{int(amount)} {unit}" return f"{amount:.1f} {unit}" amount /= 1024.0 return f"{value} B" def _compact_trace(trace: list[dict[str, Any]]) -> list[dict[str, Any]]: if len(trace) <= MAX_UI_TRACE_ITEMS: return trace remaining = len(trace) - MAX_UI_TRACE_ITEMS return [ *trace[:MAX_UI_TRACE_ITEMS], {"name": "trace_truncated", "value": remaining, "unit": "items"}, ] def _compact_result_for_ui(result: dict[str, Any]) -> dict[str, Any]: compact = json.loads(json.dumps(result)) for section in ("baseline", "candidate"): payload = compact.get(section) if isinstance(payload, dict): payload["trace"] = _compact_trace(list(payload.get("trace") or [])) return compact def _ui_logs_text(logs: list[str] | tuple[str, ...] | None) -> str: entries = [str(line) for line in (logs or []) if str(line).strip()] if not entries: return "No logs recorded." clipped_entries = entries[-MAX_UI_LOG_LINES:] text = "\n".join(clipped_entries) truncated = len(clipped_entries) < len(entries) if len(text) > MAX_UI_LOG_CHARS: text = text[-MAX_UI_LOG_CHARS:] truncated = True if truncated: prefix = "Logs trimmed for UI. Showing the most recent output.\n\n" text = prefix + text return text def _render_badge_html(response: Any) -> str: badge = response.run_badge or "setup needed" return _render_run_badge_with_detail(badge, response.request_key, response.source_path) def _render_run_badge_with_detail(badge: str, request_key: str | None = None, detail: str | None = None) -> str: tone_class = ( "badge-precomputed" if badge == "precomputed" else "badge-live" if badge == "live" else "badge-warn" if badge == "live blocked" else "badge-muted" ) detail = detail or "No precomputed artifact path recorded." if request_key: detail = f"{request_key}
{html.escape(detail)}" return ( f"
{html.escape(badge)}" f"{detail}
" ) def _kpi_card(title: str, primary: str, secondary: str, *, tone: str = "neutral") -> str: return ( f"
" f"
{html.escape(title)}
" f"
{html.escape(primary)}
" f"
{html.escape(secondary)}
" "
" ) def _hero_stat(icon: str, value: str, label: str) -> str: return ( "
" f"{html.escape(icon)}" "
" f"
{html.escape(value)}
" f"
{html.escape(label)}
" "
" "
" ) def _is_blocked_result(result: dict[str, Any]) -> bool: return result.get("comparison", {}).get("status") == "blocked" def _speed_display(value: float) -> tuple[str, str]: if abs(value - 1.0) < 0.01: return "at parity", "neutral" if value >= 1.0: return f"{_format_ratio(value)} faster", "good" return f"{_format_ratio(1.0 / max(value, 1e-8))} slower", "bad" def _memory_display(value: float) -> tuple[str, str]: if abs(value - 1.0) < 0.01: return "same KV footprint", "neutral" if value >= 1.0: return f"{_format_ratio(value)} smaller", "good" return f"{_format_ratio(1.0 / max(value, 1e-8))} larger", "bad" def _relative_change_text(value: float, *, good_when_higher: bool = True) -> tuple[str, str]: if abs(value - 1.0) < 0.01: return "about the same", "neutral" if good_when_higher: if value >= 1.0: return f"{_format_ratio(value)} better", "good" return f"{_format_ratio(1 / max(value, 1e-8))} worse", "bad" if value <= 1.0: return f"{_format_ratio(1 / max(value, 1e-8))} lower", "good" return f"{_format_ratio(value)} higher", "bad" def _agreement_state(agreement: float) -> tuple[str, str, str]: if agreement >= 0.95: return "โœ…", "Matches closely", "good" if agreement >= 0.85: return "๐Ÿ‘", "Minor differences", "warn" if agreement >= 0.70: return "โš ๏ธ", "Moderate differences", "warn" return "โŒ", "Significant drift", "bad" def _agreement_badge_text(agreement: float) -> str: icon, label, _ = _agreement_state(agreement) return f"{icon} {label}" def _agreement_metric_text(agreement: float) -> tuple[str, str, str]: icon, label, tone = _agreement_state(agreement) return f"{icon} {label}", f"{100.0 * agreement:.1f}% agreement vs reference", tone def _comparison_label(result: dict[str, Any], key: str, fallback: str) -> str: return str(result.get("comparison", {}).get(key) or fallback) def _baseline_label(result: dict[str, Any]) -> str: return _comparison_label(result, "baseline_label", "Dense (baseline)") def _candidate_label(result: dict[str, Any]) -> str: return _comparison_label(result, "candidate_label", "DotCache (current mode)") def _is_live_result(result: dict[str, Any]) -> bool: return bool(str(result.get("request", {}).get("custom_prompt") or "").strip()) or "live" in { str(note) for note in (result.get("comparison", {}).get("notes") or []) } def _is_paper_fixture(result: dict[str, Any]) -> bool: return bool(result.get("comparison", {}).get("paper_fixture")) def _is_backend_truth_fixture(result: dict[str, Any]) -> bool: return str(result.get("comparison", {}).get("paper_section") or "").strip() == "Backend-truth decode" def _trace_value(payload: dict[str, Any], name: str) -> Any | None: for item in payload.get("trace") or []: if str(item.get("name") or "") == name: return item.get("value") return None def _hero_headline(result: dict[str, Any]) -> str: comparison = result["comparison"] if _is_live_result(result): speed_value, _ = _speed_display(comparison["speedup"]) _, agreement_label, _ = _agreement_state(comparison["agreement"]) candidate_kv = _format_bytes(result["candidate"]["kv_bytes"]) baseline_kv = _format_bytes(result["baseline"]["kv_bytes"]) return ( f"Live Space run: DotCache is {speed_value}, with {candidate_kv} resident KV vs {baseline_kv} dense, " f"and shows {agreement_label.lower()} vs the reference run." ) paper_headline = str(comparison.get("paper_headline") or "").strip() if paper_headline: return paper_headline speed_value, _ = _speed_display(comparison["speedup"]) memory_value, _ = _memory_display(comparison["memory_reduction"]) _, agreement_label, _ = _agreement_state(comparison["agreement"]) if comparison["agreement"] >= 0.95: return f"DotCache preserves output quality while running {speed_value} and using {memory_value} KV memory." return f"DotCache is {speed_value} and uses {memory_value} KV memory, with {agreement_label.lower()} vs the reference row." def _hero_about_link() -> str: return "โ“˜ About" def _render_about_sheet(content_html: str) -> str: return ( "
" "" "" "
" ) def _trend_delta_text(*, dense_value: float, candidate_value: float, better_when_higher: bool, noun: str) -> tuple[str, str]: if dense_value <= 0: return "No dense reference", "neutral" if better_when_higher: ratio = candidate_value / dense_value if abs(ratio - 1.0) < 0.01: return f"same {noun}", "neutral" delta = abs(ratio - 1.0) * 100.0 if ratio >= 1.0: return f"+{delta:.0f}% {noun}", "good" return f"-{delta:.0f}% {noun}", "bad" ratio = candidate_value / dense_value if abs(ratio - 1.0) < 0.01: return f"same {noun}", "neutral" delta = abs(1.0 - ratio) * 100.0 if ratio <= 1.0: return f"-{delta:.0f}% {noun}", "good" return f"+{delta:.0f}% {noun}", "bad" def _render_loading_state(label: str) -> str: return ( "
" "" f"{html.escape(label)}" "
" ) def _render_hero(result: dict[str, Any], response: Any) -> str: baseline = result["baseline"] candidate = result["candidate"] comparison = result["comparison"] is_live = _is_live_result(result) is_paper = _is_paper_fixture(result) if _is_blocked_result(result): reason = comparison.get("blocked_reason") if reason == "hw_limit": subtitle = "This request hit the current hardware/runtime limit before the reference or systems row could execute." status_value = "Hardware limit hit" elif reason == "benchmark_parity": subtitle = "Live compare is disabled until this Space is wired to the same selector artifacts and benchmark profile settings used for the paper runs." status_value = "Benchmark parity required" elif reason == "model_family_not_wired": subtitle = "This model is available in the demo list, but its live execution lane has not been wired yet." status_value = "Live lane not wired" elif reason == "config_limit": subtitle = "This request hit a runtime-safe configuration limit before execution began." status_value = "Config limit hit" else: subtitle = "This request did not execute because the current runtime blocked it before inference started." status_value = "Run blocked" limit_value = ( f"{comparison.get('max_live_context', 0)} token live cap" if comparison.get("max_live_context") else "Current runtime cap" ) hero_stats = "".join( [ _hero_stat("โš ", status_value, "Execution status"), _hero_stat("๐Ÿงฑ", limit_value, "Current live limit"), _hero_stat("๐ŸŽฏ", "Not run", "Comparison metrics"), ] ) return ( "
" "
" "
DotCache Paper Demo
" "
" "

Same output. Less memory. Faster decode.

" f"{_hero_about_link()}" "
" f"

{html.escape(subtitle)}

" "
" f"
{hero_stats}
" f"{_render_badge_html(response)}" "
" ) speed_value, _ = _speed_display(comparison["speedup"]) agreement_badge = _agreement_badge_text(comparison["agreement"]) if is_live: memory_value = _format_bytes(candidate["kv_bytes"]) memory_label = "Runtime resident KV" elif is_paper: memory_value = str(comparison.get("paper_metric_badge") or "Not reported") memory_label = "Paper metric" else: memory_value, _ = _memory_display(comparison["memory_reduction"]) memory_label = "KV footprint" subtitle = str(comparison.get("paper_subtitle") or "").strip() if is_live: subtitle = ( "Live prompt comparison using the current Space runtime. Resident KV bytes here are runtime telemetry " "and are not directly comparable to the paper preset tables." ) elif not subtitle: subtitle = "Reference vs systems long-context inference, summarized from the current run." hero_stats = "".join( [ _hero_stat("โšก", speed_value, "Speed vs reference"), _hero_stat("๐Ÿ’พ", memory_value, memory_label), _hero_stat("๐ŸŽฏ", agreement_badge, f"{100.0 * comparison['agreement']:.0f}% agreement vs reference"), ] ) return ( "
" "
" "
DotCache Paper Demo
" "
" f"

{html.escape(_hero_headline(result))}

" f"{_hero_about_link()}" "
" f"

{html.escape(subtitle)}

" "
" f"
{hero_stats}
" f"{_render_badge_html(response)}" "
" ) def _render_kpis(result: dict[str, Any]) -> str: baseline = result["baseline"] candidate = result["candidate"] comparison = result["comparison"] is_live = _is_live_result(result) is_paper = _is_paper_fixture(result) if _is_blocked_result(result): cards = [ _kpi_card("โšก SPEEDUP", "Not run", "No comparison was executed", tone="neutral"), _kpi_card("๐Ÿ’พ MEMORY REDUCTION", "Not run", "No KV comparison available", tone="neutral"), _kpi_card("๐ŸŽฏ AGREEMENT", "Not run", "Reference vs systems did not execute", tone="neutral"), _kpi_card("โฑ RAW LATENCY", "Not run", "No decode timing recorded", tone="neutral"), _kpi_card("๐Ÿ“ฆ RAW KV SIZE", "Not run", "No cache footprint recorded", tone="neutral"), _kpi_card("๐Ÿงช TOKENS / SEC", "Not run", "Throughput unavailable until execution starts", tone="neutral"), ] return "
" + "".join(cards) + "
" speed_text, speed_tone = _relative_change_text(comparison["speedup"], good_when_higher=True) memory_text, memory_tone = _relative_change_text(comparison["memory_reduction"], good_when_higher=True) tok_ratio = candidate["tok_per_sec"] / baseline["tok_per_sec"] if baseline["tok_per_sec"] > 0 else 1.0 _, tok_tone = _relative_change_text(tok_ratio, good_when_higher=True) agreement_primary, agreement_secondary, agreement_tone = _agreement_metric_text(comparison["agreement"]) latency_tone = "good" if candidate["latency_ms_per_token"] <= baseline["latency_ms_per_token"] else "bad" memory_size_tone = "good" if candidate["kv_bytes"] <= baseline["kv_bytes"] else "bad" if is_live: memory_title = "๐Ÿ’พ EST. KV RESIDENCY" memory_primary = _format_bytes(candidate["kv_bytes"]) memory_secondary = f"runtime-reported vs reference {_format_bytes(baseline['kv_bytes'])}" memory_tone_for_card = "neutral" raw_kv_title = "๐Ÿ“ฆ RAW KV SIZE" raw_kv_primary = _format_bytes(candidate["kv_bytes"]) raw_kv_secondary = f"dense {_format_bytes(baseline['kv_bytes'])}" raw_kv_tone = memory_size_tone elif is_paper: memory_title = "๐Ÿ’พ KV MEMORY" memory_primary = "Not reported" memory_secondary = "This paper row does not publish a KV-memory reduction number." memory_tone_for_card = "neutral" raw_kv_title = "๐Ÿ“„ PAPER METRIC" raw_kv_primary = str(comparison.get("paper_metric_badge") or "Paper-backed row") raw_kv_secondary = "Preset speed and quality come from the paper; KV bytes are not reported there." raw_kv_tone = "neutral" else: memory_title = "๐Ÿ’พ MEMORY REDUCTION" memory_primary = _format_ratio(comparison["memory_reduction"]) memory_secondary = memory_text.replace("better", "smaller than dense").replace("worse", "larger than dense") memory_tone_for_card = memory_tone raw_kv_title = "๐Ÿ“ฆ RAW KV SIZE" raw_kv_primary = _format_bytes(candidate["kv_bytes"]) raw_kv_secondary = f"dense {_format_bytes(baseline['kv_bytes'])}" raw_kv_tone = memory_size_tone cards = [ _kpi_card( "โšก SPEEDUP", _format_ratio(comparison["speedup"]), speed_text.replace("better", "faster than dense").replace("worse", "slower than dense"), tone=speed_tone, ), _kpi_card( memory_title, memory_primary, memory_secondary, tone=memory_tone_for_card, ), _kpi_card( "๐ŸŽฏ AGREEMENT", agreement_primary, agreement_secondary, tone=agreement_tone, ), _kpi_card( "โฑ RAW LATENCY", f"{_format_float(candidate['latency_ms_per_token'], 1)} ms/token", f"dense {_format_float(baseline['latency_ms_per_token'], 1)} ms/token", tone=latency_tone, ), _kpi_card( raw_kv_title, raw_kv_primary, raw_kv_secondary, tone=raw_kv_tone, ), _kpi_card( "๐Ÿงช TOKENS / SEC", _format_ratio(tok_ratio), f"{_format_float(candidate['tok_per_sec'], 1)} vs reference {_format_float(baseline['tok_per_sec'], 1)}", tone=tok_tone, ), ] return "
" + "".join(cards) + "
" def _mini_trend_chart( title: str, dense_value: float, candidate_value: float, *, formatter: Callable[[float], str], better_when_higher: bool, ) -> str: peak = max(dense_value, candidate_value, 1e-6) dense_width = 100.0 * dense_value / peak candidate_width = 100.0 * candidate_value / peak improved = candidate_value >= dense_value if better_when_higher else candidate_value <= dense_value tone = "good" if improved else "bad" delta_text, delta_tone = _trend_delta_text( dense_value=dense_value, candidate_value=candidate_value, better_when_higher=better_when_higher, noun="faster" if better_when_higher else "KV memory", ) return ( "
" f"
{html.escape(title)}
" "
" "
" "
Reference" f"{html.escape(formatter(dense_value))}
" "
" f"
" "
" "
" "
" "
Systems" f"{html.escape(formatter(candidate_value))}
" "
" f"
" "
" "
" "
" f"
{html.escape(delta_text)}
" "
" ) def _render_inline_trends(result: dict[str, Any]) -> str: if _is_blocked_result(result): return ( "
" "
" "
Inline compare unavailable
" "

This request was blocked before the reference or systems row executed, so there is no speed or memory trace to plot yet.

" "
" "
" ) baseline = result["baseline"] candidate = result["candidate"] if _is_paper_fixture(result): return ( "
" + _mini_trend_chart( "Speed", baseline["tok_per_sec"], candidate["tok_per_sec"], formatter=lambda value: f"{_format_float(value, 1)} tok/s", better_when_higher=True, ) + "
" "
KV Memory
" "
Not reported in this paper row
" "

Preset mode uses the paper's speed and quality row only, so placeholder KV bytes are hidden here.

" "
" "
" ) charts = [ _mini_trend_chart( "Speed", baseline["tok_per_sec"], candidate["tok_per_sec"], formatter=lambda value: f"{_format_float(value, 1)} tok/s", better_when_higher=True, ), _mini_trend_chart( "KV Memory", baseline["kv_bytes"] / (1024.0 * 1024.0), candidate["kv_bytes"] / (1024.0 * 1024.0), formatter=lambda value: f"{_format_float(value, 1)} MiB", better_when_higher=False, ), ] return "
" + "".join(charts) + "
" def _render_metrics_band(result: dict[str, Any]) -> str: return ( "
" f"
{_render_kpis(result)}
" f"" "
" ) def _render_story_sentence(result: dict[str, Any]) -> str: comparison = result["comparison"] paper_summary = str(comparison.get("paper_summary") or "").strip() error_summary = str(comparison.get("error_summary") or "").strip() if paper_summary and not _is_blocked_result(result): return paper_summary if _is_blocked_result(result): if comparison.get("blocked_reason") == "hw_limit": return ( f"This request did not run because the selected context ({comparison.get('blocked_context_length', 0)}) " f"exceeds the current live hardware limit of {comparison.get('max_live_context', 0)} tokens." ) if comparison.get("blocked_reason") == "benchmark_parity": return ( "This request did not run because the current Space live lane is not yet using the same selector " "artifacts and benchmark profile settings as the paper-backed presets." ) if comparison.get("blocked_reason") == "model_family_not_wired": return "This model can appear in the demo list, but its custom live runner has not been wired into the v1 Space yet." if comparison.get("blocked_reason") == "auth_required": return "This request did not run because the selected gated model needs HF_TOKEN in the Space settings." if comparison.get("blocked_reason") == "config_limit": return "This request did not run because the current live runner rejects that compression configuration." if comparison.get("blocked_reason") == "cache_download_limit": return ( error_summary or "This request did not run because the Space could not fetch the selected model into its local runtime cache." ) if error_summary: return error_summary return "This request did not run because the current runtime blocked execution before inference started." if _is_live_result(result): speed_value, _ = _speed_display(comparison["speedup"]) _, agreement_label, _ = _agreement_state(comparison["agreement"]) candidate_kv = _format_bytes(result["candidate"]["kv_bytes"]) baseline_kv = _format_bytes(result["baseline"]["kv_bytes"]) return ( f"Live Space run: DotCache is {speed_value}, with {candidate_kv} resident KV vs {baseline_kv} dense, " f"and shows {agreement_label.lower()} compared to the reference run." ) if _is_paper_fixture(result): speed_value, _ = _speed_display(comparison["speedup"]) _, agreement_label, _ = _agreement_state(comparison["agreement"]) return f"This paper row reports {speed_value} with {agreement_label.lower()}; KV-memory reduction is not reported for this preset." speed_value, _ = _speed_display(comparison["speedup"]) memory_value, _ = _memory_display(comparison["memory_reduction"]) _, agreement_label, _ = _agreement_state(comparison["agreement"]) if comparison["agreement"] >= 0.95: return ( f"DotCache preserves output quality while running {speed_value} " f"and using {memory_value} KV memory." ) if comparison["agreement"] >= 0.85: return ( f"DotCache is {speed_value} and uses " f"{memory_value} KV memory, with {agreement_label.lower()} compared to dense." ) return ( f"DotCache is {speed_value} and uses " f"{memory_value} KV memory, with {agreement_label.lower()} compared to dense." ) def _render_delta_strip(result: dict[str, Any]) -> str: comparison = result["comparison"] is_live_prompt = bool(str(result.get("request", {}).get("custom_prompt") or "").strip()) delta_label = "This run" if is_live_prompt else "ฮ” Result" error_summary = str(comparison.get("error_summary") or "").strip() if _is_blocked_result(result): if comparison.get("blocked_reason") == "hw_limit": blocked_text = ( f"Live run blocked at {comparison.get('blocked_context_length', 0)} tokens " f"(current cap {comparison.get('max_live_context', 0)})" ) elif comparison.get("blocked_reason") == "benchmark_parity": blocked_text = "Live run disabled until benchmark-parity selector artifacts and profiles are wired" elif comparison.get("blocked_reason") == "model_family_not_wired": blocked_text = "Live run blocked because this model family is not wired into the v1 runner yet" elif comparison.get("blocked_reason") == "auth_required": blocked_text = "Live run blocked because this gated model needs HF_TOKEN in the Space settings" elif comparison.get("blocked_reason") == "config_limit": blocked_text = "Live run blocked by the current runtime-safe compression limits" elif comparison.get("blocked_reason") == "cache_download_limit": blocked_text = error_summary or "Live run blocked while fetching model weights into the Space cache" else: blocked_text = error_summary or "Live run blocked before the reference or systems row started" if len(blocked_text) > 160: blocked_text = blocked_text[:157].rstrip() + "..." return ( "
" f"
{html.escape(delta_label)}
" f"
{html.escape(blocked_text)}
" "
" ) speed_delta = abs(comparison["speedup"] - 1.0) * 100.0 speed_class = "good" if comparison["speedup"] >= 1.0 else "bad" speed_text = f"+{speed_delta:.0f}% speed" if comparison["speedup"] >= 1.0 else f"-{speed_delta:.0f}% speed" memory_saved = abs(1.0 - (1.0 / max(comparison["memory_reduction"], 1e-8))) * 100.0 memory_class = "good" if comparison["memory_reduction"] >= 1.0 else "bad" memory_text = ( f"-{memory_saved:.0f}% KV memory" if comparison["memory_reduction"] >= 1.0 else f"+{memory_saved:.0f}% KV memory" ) drift = max(0.0, 100.0 * (1.0 - comparison["agreement"])) paper_metric = str(comparison.get("paper_metric_badge") or "").strip() if _is_live_result(result): return ( "
" f"
{html.escape(delta_label)}
" f"
{html.escape(speed_text)}
" f"
resident KV {_format_bytes(result['candidate']['kv_bytes'])}
" f"
~{drift:.0f}% output drift
" "
" ) if paper_metric: return ( "
" f"
{html.escape(delta_label)}
" f"
{speed_text}
" f"
{html.escape(paper_metric)}
" f"
~{drift:.0f}% output drift
" "
" ) return ( "
" f"
{html.escape(delta_label)}
" f"
{speed_text}
" f"
{memory_text}
" f"
~{drift:.0f}% output drift
" "
" ) def _render_response_tab(result: dict[str, Any]) -> str: baseline = html.escape(result["baseline"]["text"]) candidate = html.escape(result["candidate"]["text"]) agreement = result["comparison"]["agreement"] baseline_label = html.escape(_baseline_label(result)) candidate_label = html.escape(_candidate_label(result)) if _is_blocked_result(result): return ( "
" "
" "
" f"
{baseline_label}
" "
Not run
" "
" f"

{baseline}

" "
" "
" "
" f"
{candidate_label}
" "
Blocked by runtime
" "
" f"

{candidate}

" "
" "
" ) if _is_backend_truth_fixture(result): match_badge = _agreement_badge_text(agreement) return ( "
" "
" "
" f"
{baseline_label}
" "
Decode sample
" "
" f"

{baseline}

" "
" "
" "
" f"
{candidate_label}
" f"
= 0.95 else 'warn' if agreement >= 0.70 else 'bad'}'>{html.escape(match_badge)}
" "
" f"

{candidate}

" "
" "
" ) match_badge = _agreement_badge_text(agreement) return ( "
" "
" "
" f"
{baseline_label}
" "
Reference
" "
" f"

{baseline}

" "
" "
" "
" f"
{candidate_label}
" f"
= 0.95 else 'warn' if agreement >= 0.70 else 'bad'}'>{html.escape(match_badge)}
" "
" f"

{candidate}

" "
" "
" ) def _highlight_diff(text: str, other_text: str) -> str: matcher = difflib.SequenceMatcher(a=text.split(), b=other_text.split()) pieces: list[str] = [] words = text.split() for tag, i1, i2, _, _ in matcher.get_opcodes(): chunk = " ".join(words[i1:i2]) if not chunk: continue escaped = html.escape(chunk) if tag == "equal": pieces.append(escaped) else: pieces.append(f"{escaped}") return " ".join(pieces) def _render_diff_tab(result: dict[str, Any]) -> str: baseline_text = result["baseline"]["text"] candidate_text = result["candidate"]["text"] agreement = result["comparison"]["agreement"] baseline_label = html.escape(_baseline_label(result)) candidate_label = html.escape(_candidate_label(result)) if _is_blocked_result(result): return ( "
" "
" "
" f"
{baseline_label}
" "
No diff
" "
" "

The reference row was not executed, so there is no diff to compute for this request.

" "
" "
" "
" f"
{candidate_label} (diff)
" "
Blocked by runtime
" "
" f"

{html.escape(candidate_text)}

" "
" "
" ) return ( "
" "
" "
" f"
{baseline_label}
" "
Reference
" "
" f"

{_highlight_diff(baseline_text, candidate_text)}

" "
" "
" "
" f"
{candidate_label} (diff)
" f"
= 0.95 else 'warn' if agreement >= 0.70 else 'bad'}'>{html.escape(_agreement_badge_text(agreement))}
" "
" f"

{_highlight_diff(candidate_text, baseline_text)}

" "
" "
" ) def _chart_bar(label: str, dense_value: float, candidate_value: float, suffix: str, dense_label: str, candidate_label: str) -> str: peak = max(dense_value, candidate_value, 1e-6) dense_width = 100.0 * dense_value / peak candidate_width = 100.0 * candidate_value / peak return ( "
" f"
{html.escape(label)}
" f"
{html.escape(dense_label)}" f"
" f"{dense_value:.1f}{suffix}
" f"
{html.escape(candidate_label)}" f"
" f"{candidate_value:.1f}{suffix}
" "
" ) def _chart_note_block(label: str, body: str) -> str: return ( "
" f"
{html.escape(label)}
" f"
{html.escape(body)}
" "
" ) def _render_charts_tab(result: dict[str, Any]) -> str: baseline = result["baseline"] candidate = result["candidate"] comparison = result["comparison"] baseline_label = _baseline_label(result) candidate_label = _candidate_label(result) if _is_blocked_result(result): return ( "
" "
" "
Charts unavailable
" "
This request was blocked before execution, so there are no reference or systems traces to chart.
" "
" "
" ) if _is_paper_fixture(result): return ( "
" + _chart_bar("Tokens / sec", baseline["tok_per_sec"], candidate["tok_per_sec"], "", baseline_label, candidate_label) + _chart_bar( "Latency / token", baseline["latency_ms_per_token"], candidate["latency_ms_per_token"], " ms", baseline_label, candidate_label, ) + _chart_note_block( "KV memory", "Not reported in this paper row. The preset uses paper-backed speed and quality only.", ) + _chart_bar( "Agreement", 1.0, comparison["agreement"], "", baseline_label, candidate_label, ) + "
" ) memory_chart_label = "Resident KV" if _is_live_result(result) else "KV memory" return ( "
" + _chart_bar("Tokens / sec", baseline["tok_per_sec"], candidate["tok_per_sec"], "", baseline_label, candidate_label) + _chart_bar( "Latency / token", baseline["latency_ms_per_token"], candidate["latency_ms_per_token"], " ms", baseline_label, candidate_label, ) + _chart_bar( memory_chart_label, baseline["kv_bytes"] / (1024.0 * 1024.0), candidate["kv_bytes"] / (1024.0 * 1024.0), " MiB", baseline_label, candidate_label, ) + _chart_bar( "Agreement", 1.0, comparison["agreement"], "", baseline_label, candidate_label, ) + "
" ) def _render_logs(response: Any) -> str: return _ui_logs_text(getattr(response, "logs", None)) def _build_request( model: str, preset: str, custom_prompt: str, live_mode: bool, context_length: int, mode: str, page_size: int, bits_k: int, bits_v: int, recent_window: int, sink_window: int, shortlist_policy: str, compare_against_dense: bool, ) -> DemoRequest: return DemoRequest( model=model, preset=preset or None, custom_prompt=custom_prompt, live_mode=bool(live_mode), context_length=int(context_length), mode=mode, page_size=int(page_size), bits_k=int(bits_k), bits_v=int(bits_v), recent_window=int(recent_window), sink_window=int(sink_window), shortlist_policy=shortlist_policy, compare_against_dense=compare_against_dense, ) def _run_request( model: str, preset: str, custom_prompt: str, live_mode: bool, context_length: int, mode: str, page_size: int, bits_k: int, bits_v: int, recent_window: int, sink_window: int, shortlist_policy: str, compare_against_dense: bool, ) -> tuple[Any, ...]: request = _build_request( model, preset, custom_prompt, live_mode, context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, compare_against_dense, ) response = execute_request(request) result = response.result return ( _render_hero(result, response), _render_metrics_band(result), "", _render_delta_strip(result), _render_story_sentence(result), _render_response_tab(result), _render_diff_tab(result), _render_charts_tab(result), {"result": result, "mode_help": MODE_HELP[mode], "mode_label": MODE_LABELS[mode]}, _render_logs(response), ) def _apply_quick_start( current_model: str, current_preset: str, current_custom_prompt: str, current_context_length: int, key: str, ) -> tuple[Any, ...]: payload = quick_start_payload(key) return ( current_model, current_preset, current_custom_prompt, current_context_length, payload["mode"], payload["page_size"], payload["bits_k"], payload["bits_v"], payload["recent_window"], payload["sink_window"], payload["shortlist_policy"], ) def _set_compare_loading_state() -> tuple[Any, str]: return gr.update(value="Running comparison...", interactive=False), _render_loading_state("Running comparison...") def _clear_compare_loading_state() -> tuple[Any, str]: return gr.update(value="Compare", interactive=True), "" def _live_context_guard_copy(mode: str) -> str: if mode == "custom": return ( "
" "โšก Live mode (ZeroGPU)Your prompt runs on the real Space system using the selected preset's benchmark configuration. When the runtime allows it, custom live prompts and example buttons can use the 4K context lane to sanity-check that behavior stays in the same ballpark as the cached paper row." "
" ) if mode == "live": return ( "
" "โšก Live mode (ZeroGPU)Live mode exposes the full runtime context ladder up to the current Space cap. Empty-prompt replay is still only bundled for the benchmark rows, so use a custom prompt or example button when you want to run the 4K live lane." "
" ) if mode == "benchmark": return ( "
" "โšก Live benchmark replayThis build replays only the bundled paper/benchmark presets live. Custom prompts and example challenge prompts are disabled here, and context stays on the benchmark-safe 1024 or 2048 rows." "
" ) return ( "
" "Preset modePreset-backed rounds use the bundled benchmark rows directly. Switch to live to replay that benchmark row on the current runtime when the required selector artifact is available." "
" ) def _available_live_context_choices() -> list[int]: max_live_context = int(os.getenv("DOTCACHE_SPACE_MAX_LIVE_CONTEXT", "4096")) choices = [value for value in FULL_CONTEXT_CHOICES if int(value) <= max_live_context] return choices or [min(max_live_context, 1024)] def _available_live_benchmark_context_choices() -> list[int]: benchmark_choices = [value for value in BENCHMARK_LIVE_CONTEXT_CHOICES if int(value) <= int(os.getenv("DOTCACHE_SPACE_MAX_LIVE_CONTEXT", "4096"))] return benchmark_choices or _available_live_context_choices() def _available_live_example_context_choices() -> list[int]: safe_choices = [value for value in _available_live_context_choices() if int(value) in LIVE_SAFE_CONTEXT_CHOICES] return safe_choices or _available_live_context_choices() def _preferred_live_example_context(current_context_length: int) -> int: safe_choices = _available_live_example_context_choices() current_value = int(current_context_length) if current_value in safe_choices: return current_value return int(safe_choices[-1]) def _guard_live_context(prompt_mode: str, custom_prompt: str, current_context_length: int) -> tuple[Any, str]: if _is_live_prompt_mode(prompt_mode): available_choices = _available_live_context_choices() guarded_value = int(current_context_length) if guarded_value not in available_choices: guarded_value = available_choices[min(len(available_choices) - 1, 1)] has_custom_prompt = bool(custom_prompt.strip()) if has_custom_prompt: return ( gr.update(choices=available_choices, value=guarded_value), _live_context_guard_copy("custom"), ) return ( gr.update(choices=available_choices, value=guarded_value), _live_context_guard_copy("live"), ) has_custom_prompt = bool(custom_prompt.strip()) restored_value = int(current_context_length) if restored_value not in FULL_CONTEXT_CHOICES: restored_value = 2048 return ( gr.update(choices=FULL_CONTEXT_CHOICES, value=restored_value), _live_context_guard_copy("preset"), ) def _is_live_prompt_mode(prompt_mode: str) -> bool: return str(prompt_mode or "").strip().lower() == "live" def _effective_custom_prompt(prompt_mode: str, custom_prompt: str) -> str: if _is_live_prompt_mode(prompt_mode): return custom_prompt return "" def _set_compare_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[Any, Any, Any, str]: is_live = _is_live_prompt_mode(prompt_mode) context_update, note = _guard_live_context(prompt_mode, current_custom_prompt if is_live else "", current_context_length) return ( gr.update(visible=not is_live), gr.update(visible=is_live), context_update, note, ) def _set_arena_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[Any, Any, Any, str, Any]: is_live = _is_live_prompt_mode(prompt_mode) context_update, note = _guard_live_context(prompt_mode, current_custom_prompt if is_live else "", current_context_length) button_update = ( _arena_start_button_state(current_custom_prompt) if is_live else gr.update(value="Start Arena Match", interactive=True) ) return ( gr.update(visible=not is_live), gr.update(visible=is_live), context_update, note, button_update, ) def _refresh_compare_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[Any, str]: _, _, context_update, note = _set_compare_prompt_mode(prompt_mode, current_context_length, current_custom_prompt) return context_update, note def _refresh_arena_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[Any, str, Any]: _, _, context_update, note, button_update = _set_arena_prompt_mode(prompt_mode, current_context_length, current_custom_prompt) return context_update, note, button_update def _select_compare_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[str, Any, str]: context_update, note = _refresh_compare_prompt_mode(prompt_mode, current_context_length, current_custom_prompt) return prompt_mode, context_update, note def _select_arena_prompt_mode(prompt_mode: str, current_context_length: int, current_custom_prompt: str) -> tuple[str, Any, str, Any]: context_update, note, button_update = _refresh_arena_prompt_mode(prompt_mode, current_context_length, current_custom_prompt) return prompt_mode, context_update, note, button_update def _run_request_with_prompt_mode( model: str, preset: str, prompt_mode: str, custom_prompt: str, context_length: int, mode: str, page_size: int, bits_k: int, bits_v: int, recent_window: int, sink_window: int, shortlist_policy: str, compare_against_dense: bool, ) -> tuple[Any, ...]: return _run_request( model, preset, _effective_custom_prompt(prompt_mode, custom_prompt), _is_live_prompt_mode(prompt_mode), context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, compare_against_dense, ) def _arena_start_match_with_prompt_mode( model: str, preset: str, prompt_mode: str, custom_prompt: str, context_length: int, mode: str, page_size: int, bits_k: int, bits_v: int, recent_window: int, sink_window: int, shortlist_policy: str, history: list[dict[str, Any]] | None, ) -> tuple[dict[str, Any], list[dict[str, Any]], str, str, str, str, str, str, str, str, str, dict[str, Any], str, Any]: return _arena_start_match( model, preset, _effective_custom_prompt(prompt_mode, custom_prompt), _is_live_prompt_mode(prompt_mode), context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, history, ) def _live_prompt_intro_html() -> str: if not CUSTOM_LIVE_PROMPTS_ENABLED: return ( "
" "
โšก Live benchmark replay
" "

This Qwen demo now replays only bundled paper-backed benchmark rows live.

" "
" "๐Ÿ’ก Choose a preset in the Preset tab" "๐Ÿ’ก Switch to Live to replay that same row on the current runtime" "๐Ÿ’ก Custom prompts are disabled in this benchmark-faithful build" "
" "
" ) return ( "
" "
๐Ÿงช Try your own prompt (live)
" "

Bring your own test and see if DotCache still holds up.

" "
" "๐Ÿ’ก Works best with factual retrieval" "๐Ÿ’ก Multi-step reasoning" "๐Ÿ’ก Questions that depend on earlier context" "
" "
" ) def _live_prompt_challenge_html() -> str: if not CUSTOM_LIVE_PROMPTS_ENABLED: return ( "
" "Benchmark-faithful mode" "The live runner is constrained to the same preset rows used by the paper, so example challenge prompts are hidden here." "
" ) return ( "
" "๐Ÿงจ Try to break it" "Can you find a prompt where DotCache gives a worse answer? The example buttons will use the 4K live lane when this Space runtime allows it." "
" ) def _repeat_live_example_filler(example_key: str, *, target_chars: int) -> str: unit = LIVE_EXAMPLE_FILLERS[example_key] if target_chars <= len(unit): return unit repeats = max(1, (target_chars // max(len(unit), 1)) + 2) return " ".join(unit for _ in range(repeats)) def _build_live_example_prompt(example_key: str, context_length: int) -> str: target_chars = max(1200, int(context_length) * 4) if example_key == "retrieval": before = _repeat_live_example_filler("retrieval", target_chars=int(target_chars * 0.55)) after = _repeat_live_example_filler("retrieval", target_chars=int(target_chars * 0.35)) return ( "Context:\n" f"{before}\n" "Important detail: the archive code is RIVER-58142. Remember it exactly.\n" f"{after}\n\n" "Question: What is the archive code? Return only the exact value on one line." ) if example_key == "reasoning": filler = _repeat_live_example_filler("reasoning", target_chars=int(target_chars * 0.8)) return ( "Planning notes:\n" f"{filler}\n" "Decision memo: Start with 17. Add 26. Subtract 9. Add 14. " "Think silently if needed, but the visible output must be exactly one line in the form FINAL: .\n\n" "Question: What is the final total?" ) filler = _repeat_live_example_filler("fragile", target_chars=int(target_chars * 0.75)) return ( "Document excerpt:\n" f"{filler}\n" "Section 4 - Migration Plan: The API transition is staged over three weeks.\n" "Section 7 - Audit Notes: In the original draft, this section was numbered Section 5.\n" "Citation: [See Section 7 for the renumbering note.]\n\n" "Question: Which section is referenced in the citation, and what was the original section number?" ) def _apply_live_example_prompt(example_key: str, context_length: int) -> str: return _build_live_example_prompt(example_key, context_length) def _apply_compare_live_example(example_key: str, current_context_length: int) -> tuple[Any, str, str, Any, str]: example_context = _preferred_live_example_context(current_context_length) prompt = _apply_live_example_prompt(example_key, example_context) context_update, note = _guard_live_context("live", prompt, example_context) return gr.update(selected="live"), "live", prompt, context_update, note def _apply_arena_live_example(example_key: str, current_context_length: int) -> tuple[Any, str, str, Any, str, Any]: example_context = _preferred_live_example_context(current_context_length) prompt = _apply_live_example_prompt(example_key, example_context) context_update, note = _guard_live_context("live", prompt, example_context) return ( gr.update(selected="live"), "live", prompt, context_update, note, _arena_start_button_state(prompt), ) def _arena_start_button_state(custom_prompt: str) -> Any: if not CUSTOM_LIVE_PROMPTS_ENABLED: return gr.update(value="Start Arena Match", interactive=True) if custom_prompt.strip(): return gr.update(value="Start Arena Match", interactive=True) return gr.update(value="Enter a prompt or choose an example", interactive=False) def _workflow_nav_html(selected_mode: str) -> str: compare_class = "workflow-link active" if selected_mode == "Compare" else "workflow-link" arena_class = "workflow-link active" if selected_mode == "Arena" else "workflow-link" return ( "
" f"Compare" f"Arena" "
" ) def _query_params(request: gr.Request | None) -> dict[str, Any]: if request is None: return {} return dict(getattr(request, "query_params", {}) or {}) def _deep_link_mode(query_params: dict[str, Any]) -> str: requested_mode = str(query_params.get("mode") or "").strip().lower() if requested_mode == "compare": return "Compare" return "Arena" def _deep_link_compare_preset(query_params: dict[str, Any]) -> str: requested_preset = str(query_params.get("preset") or "").strip().lower() return COMPARE_PRESET_QUERY_ALIASES.get(requested_preset, "balanced") def _deep_link_arena_preset(query_params: dict[str, Any]) -> str: requested_preset = str(query_params.get("preset") or "").strip().lower() return ARENA_PRESET_QUERY_ALIASES.get(requested_preset, "balanced_match") def _compare_preset_ui(selected_key: str) -> tuple[Any, Any, Any, str]: return ( gr.update(value=PRESET_BUTTON_LABELS["fastest_safe"], variant="primary" if selected_key == "fastest_safe" else "secondary"), gr.update(value=PRESET_BUTTON_LABELS["balanced"], variant="primary" if selected_key == "balanced" else "secondary"), gr.update(value=PRESET_BUTTON_LABELS["aggressive"], variant="primary" if selected_key == "aggressive" else "secondary"), PRESET_STATUS_HTML[selected_key], ) def _arena_preset_ui(selected_key: str) -> tuple[Any, Any, Any, str]: return ( gr.update(value=ARENA_PRESET_BUTTON_LABELS["safe_match"], variant="primary" if selected_key == "safe_match" else "secondary"), gr.update(value=ARENA_PRESET_BUTTON_LABELS["balanced_match"], variant="primary" if selected_key == "balanced_match" else "secondary"), gr.update(value=ARENA_PRESET_BUTTON_LABELS["fragile_match"], variant="primary" if selected_key == "fragile_match" else "secondary"), ARENA_PRESET_STATUS_HTML[selected_key], ) def _profile_values_from_quick_start(key: str) -> tuple[Any, Any, Any, Any, Any, Any, Any]: payload = quick_start_payload(key) return ( payload["mode"], payload["page_size"], payload["bits_k"], payload["bits_v"], payload["recent_window"], payload["sink_window"], payload["shortlist_policy"], ) def _arena_profile_values_from_preset(key: str) -> tuple[Any, Any, Any, Any, Any, Any, Any]: mapping = { "safe_match": "fastest_safe", "balanced_match": "balanced", "fragile_match": "aggressive", } return _profile_values_from_quick_start(mapping[key]) def _initial_page_state(request: gr.Request | None = None) -> tuple[Any, ...]: query_params = _query_params(request) selected_mode = _deep_link_mode(query_params) compare_preset_key = _deep_link_compare_preset(query_params) arena_preset_key = _deep_link_arena_preset(query_params) compare_visible = selected_mode == "Compare" arena_visible = selected_mode == "Arena" return ( _workflow_nav_html(selected_mode), gr.update(visible=compare_visible), gr.update(visible=arena_visible), *_compare_preset_ui(compare_preset_key), *_profile_values_from_quick_start(compare_preset_key), *_arena_preset_ui(arena_preset_key), *_arena_profile_values_from_preset(arena_preset_key), ) def _initial_compare_results(request: gr.Request | None = None) -> tuple[Any, ...]: query_params = _query_params(request) if _deep_link_mode(query_params) != "Compare": return ("", "", "", "", "", "", "", "", {}, "") compare_preset_key = _deep_link_compare_preset(query_params) payload = quick_start_payload(compare_preset_key) return _run_request( payload["model"], payload["preset"], "", False, payload["context_length"], payload["mode"], payload["page_size"], payload["bits_k"], payload["bits_v"], payload["recent_window"], payload["sink_window"], payload["shortlist_policy"], True, ) def _arena_base_hero() -> str: return ( "
" "
" "
Arena Mode
" "
" "

DotCache Arena

" f"{_hero_about_link()}" "
" "

Blind compare dense vs compressed long-context inference, then reveal speed, memory, and agreement.

" "
Read first, measure second.
" "
" "
" ) def _arena_empty_match() -> str: return ( "
" "Start Arena Match" "Generate a blind A/B round to compare two anonymous outputs before the metrics appear." "
" ) def _arena_pre_reveal_helper() -> str: return ( "
" "Read both outputs before revealing which system produced them." "Can you tell which one is compressed?" "
" ) def _arena_post_reveal_helper() -> str: return ( "
" "Now compare your judgment against the measured agreement and efficiency gains." "Arena Mode is designed to make quality tradeoffs visible, not hide them." "
" ) def _arena_vote_label(vote: str | None) -> str: labels = { "a_better": "A is better", "b_better": "B is better", "tie": "Tie / basically the same", "not_sure": "Not sure", } return labels.get(str(vote or ""), "No vote yet") def _arena_config_hash(request: DemoRequest) -> str: payload = request.to_dict() canonical = json.dumps(payload, sort_keys=True) return hashlib.sha1(canonical.encode("utf-8")).hexdigest()[:12] def _arena_assignment(seed: int) -> tuple[str, str]: roles = ["baseline", "candidate"] random.Random(seed).shuffle(roles) return roles[0], roles[1] def _store_arena_match(full_state: dict[str, Any]) -> str: match_id = uuid.uuid4().hex ARENA_MATCH_STORE[match_id] = full_state while len(ARENA_MATCH_STORE) > MAX_ARENA_MATCHES: oldest = next(iter(ARENA_MATCH_STORE)) if oldest == match_id: break ARENA_MATCH_STORE.pop(oldest, None) return match_id def _resolve_arena_state(state: dict[str, Any] | None) -> dict[str, Any] | None: if not state: return None match_id = str(state.get("arena_match_id") or "").strip() if not match_id: return state stored = ARENA_MATCH_STORE.get(match_id) if stored is None: return state merged = dict(stored) for key in ( "arena_match_id", "config_hash", "blind_shuffle_seed", "output_a_role", "output_b_role", "revealed", "user_vote", "compressed_guess", "vote_timestamp", "voted_before_reveal", "started_at", ): if key in state: merged[key] = state[key] return merged def _build_arena_state(request: DemoRequest, response: Any) -> dict[str, Any]: seed = random.randint(0, 2**31 - 1) output_a_role, output_b_role = _arena_assignment(seed) compact_result = _compact_result_for_ui(response.result) full_state = { "result": compact_result, "response_meta": { "run_badge": response.run_badge, "logs_text": _ui_logs_text(list(response.logs or [])), }, "request": request.to_dict(), "config_hash": _arena_config_hash(request), "blind_shuffle_seed": seed, "output_a_role": output_a_role, "output_b_role": output_b_role, "revealed": False, "user_vote": None, "compressed_guess": "not_sure", "vote_timestamp": None, "voted_before_reveal": False, "started_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), } match_id = _store_arena_match(full_state) return { "arena_match_id": match_id, "config_hash": full_state["config_hash"], "blind_shuffle_seed": seed, "output_a_role": output_a_role, "output_b_role": output_b_role, "revealed": False, "user_vote": None, "compressed_guess": "not_sure", "vote_timestamp": None, "voted_before_reveal": False, "started_at": full_state["started_at"], } def _arena_output_payload(state: dict[str, Any], slot: str) -> dict[str, Any]: result = state["result"] role = state[f"output_{slot.lower()}_role"] return result["baseline"] if role == "baseline" else result["candidate"] def _arena_output_identity(state: dict[str, Any], slot: str) -> str: role = state[f"output_{slot.lower()}_role"] comparison = state["result"].get("comparison", {}) if role == "baseline": return str(comparison.get("baseline_label") or "Reference row") return str(comparison.get("candidate_label") or "Systems row") def _arena_output_badge(state: dict[str, Any], slot: str) -> tuple[str, str]: role = state[f"output_{slot.lower()}_role"] if role == "baseline": return f"Was Output {slot}", "neutral" agreement = float(state["result"]["comparison"]["agreement"]) _, label, tone = _agreement_state(agreement) return f"Was Output {slot}", tone def _arena_stats_label(text: str) -> str: word_count = len(text.split()) return f"{word_count} words" def _render_arena_match_html(state: dict[str, Any] | None) -> str: if not state: return _arena_empty_match() result = state["result"] blocked = _is_blocked_result(result) cards: list[str] = [] for slot in ("A", "B"): payload = _arena_output_payload(state, slot) text = html.escape(str(payload.get("text") or "")) if state.get("revealed"): header = _arena_output_identity(state, slot) badge_text, badge_tone = _arena_output_badge(state, slot) meta = f"
{html.escape(_arena_stats_label(str(payload.get('text') or '')))}
" badge = f"
{html.escape(badge_text)}
" role = state[f"output_{slot.lower()}_role"] card_class = f"arena-output-card revealed role-{role}" else: header = f"Output {slot}" meta = f"
{html.escape(_arena_stats_label(str(payload.get('text') or '')))}
" badge = "
Blind
" card_class = "arena-output-card blind" cards.append( f"
" "
" f"
{html.escape(header)}
{meta}
" f"{badge}" "
" f"

{text}

" "
" ) wrapper_class = "arena-match-grid blocked" if blocked else "arena-match-grid" header = ( "
" "๐Ÿ” Blind comparison" "Read both outputs before revealing." "
" if not state.get("revealed") else "" ) return header + f"
" + "".join(cards) + "
" def _arena_session_summary(history: list[dict[str, Any]]) -> str: if not history: return "" revealed_votes = [entry for entry in history if entry.get("revealed")] if not revealed_votes: return "" dotcache_or_tie = 0 for entry in revealed_votes: vote = entry.get("user_vote") chosen_dotcache = entry.get("chosen_role") == "candidate" if vote == "tie": dotcache_or_tie += 1 elif vote in {"a_better", "b_better"} and chosen_dotcache: dotcache_or_tie += 1 percent = round((100.0 * dotcache_or_tie) / max(len(revealed_votes), 1)) return f"In this session, users picked DotCache or Tie {percent}% of the time." def _arena_candidate_slot(state: dict[str, Any]) -> str: return "A" if state.get("output_a_role") == "candidate" else "B" def _arena_guess_label(value: str | None) -> str: labels = { "a": "Compressed is Output A", "b": "Compressed is Output B", "not_sure": "I can't tell", None: "I can't tell", } return labels.get(value, "I can't tell") def _render_arena_vote_status(state: dict[str, Any] | None, history: list[dict[str, Any]] | None = None) -> str: history = history or [] if not state: return "
Start a match, then vote before reveal.
" vote_text = _arena_vote_label(state.get("user_vote")) guess_text = _arena_guess_label(state.get("compressed_guess")) if not state.get("revealed"): return ( "
" f"Your read: {html.escape(vote_text)}" f"Compressed guess: {html.escape(guess_text)}" "You can change your vote before reveal." "
" ) voted_before = "Yes" if state.get("voted_before_reveal") else "No" return ( "
" f"Your read: {html.escape(vote_text)}" f"Your compressed guess: {html.escape(guess_text)}" f"Voted before reveal: {voted_before}" "
" ) def _arena_reveal_sentence(result: dict[str, Any]) -> str: comparison = result["comparison"] speed_value, _ = _speed_display(comparison["speedup"]) agreement = float(comparison["agreement"]) if _is_live_result(result): candidate_kv = _format_bytes(result["candidate"]["kv_bytes"]) baseline_kv = _format_bytes(result["baseline"]["kv_bytes"]) if agreement >= 0.95: return f"DotCache matched the dense response closely while running {speed_value}, with {candidate_kv} resident KV vs {baseline_kv} dense." if agreement >= 0.85: return f"This live round traded some speed for lower resident KV, with only minor wording differences after reveal." if agreement >= 0.70: return f"This live round cut resident KV, but moderate output drift appeared in the revealed answer." return f"This live round shows a tradeoff: lower resident KV on this Space runtime, but answer quality diverged noticeably." memory_value, _ = _memory_display(comparison["memory_reduction"]) if agreement >= 0.95: return f"DotCache matched the dense response closely while running {speed_value} and using {memory_value} KV memory." if agreement >= 0.85: return f"DotCache improved efficiency substantially, with minor wording differences visible after reveal." if agreement >= 0.70: return f"DotCache still improved efficiency here, but moderate output drift appeared in this matchup." return f"This round shows a genuine tradeoff: DotCache used {memory_value} KV memory, but answer quality diverged noticeably." def _arena_compact_summary(result: dict[str, Any]) -> str: comparison = result["comparison"] _, agreement_label, _ = _agreement_state(float(comparison["agreement"])) speed_value, _ = _speed_display(comparison["speedup"]) if _is_live_result(result): return f"{agreement_label} โ€ข {speed_value} โ€ข {_format_bytes(result['candidate']['kv_bytes'])} resident KV" memory_value, _ = _memory_display(comparison["memory_reduction"]) return f"{agreement_label} โ€ข {speed_value} โ€ข {memory_value} KV" def _arena_reveal_reaction(state: dict[str, Any]) -> tuple[str, str]: vote = str(state.get("user_vote") or "") guess = str(state.get("compressed_guess") or "not_sure") actual_slot = _arena_candidate_slot(state).lower() if vote == "tie": return "๐Ÿ‘ Fair call โ€” very close outputs", "warm" if guess == actual_slot: return "โœ… Nice โ€” you spotted it", "good" if guess == "not_sure": return "๐Ÿค” Tricky โ€” harder than it looks", "warn" return "๐Ÿค” Tricky โ€” harder than it looks", "warn" def _render_arena_reveal_banner(state: dict[str, Any] | None, history: list[dict[str, Any]] | None = None) -> str: if not state or not state.get("revealed"): return "" history = history or [] result = state["result"] comparison = result["comparison"] icon, label, tone = _agreement_state(float(comparison["agreement"])) actual_slot = _arena_candidate_slot(state) guess_text = _arena_guess_label(state.get("compressed_guess")) correct_guess = str(state.get("compressed_guess") or "") == actual_slot.lower() guess_mark = "โœ…" if correct_guess else "โŒ" if str(state.get("compressed_guess") or "") != "not_sure" else "โ€ข" reaction_text, reaction_tone = _arena_reveal_reaction(state) session_summary = _arena_session_summary(history) session_pill = "" if session_summary: session_text = session_summary.replace("In this session, users picked DotCache or Tie ", "").replace( "% of the time.", "% chose DotCache or Tie", ) session_pill = f"
๐Ÿ”ฅ In this session: {html.escape(session_text)}
" return ( f"
" f"
{html.escape(f'๐ŸŽฏ DotCache was Output {actual_slot}')}
" f"
{html.escape(f'{icon} {label}')}
" f"
{html.escape(f'You guessed: {guess_text} {guess_mark}')}
" f"
Actual: Output {html.escape(actual_slot)} was DotCache
" f"
{html.escape(reaction_text)}
" f"
{html.escape(_arena_compact_summary(result))}
" f"
{html.escape(_arena_reveal_sentence(result))}
" f"{session_pill}" "
" ) def _arena_logs_text(state: dict[str, Any]) -> str: logs_text = str(state.get("response_meta", {}).get("logs_text") or "").strip() return logs_text or "No logs recorded." def _render_arena_inline_reveal(state: dict[str, Any], history: list[dict[str, Any]] | None = None) -> str: result = state["result"] run_badge = str(state.get("response_meta", {}).get("run_badge") or "").strip() badge_html = _render_run_badge_with_detail(run_badge, None, "Live Arena result") intro = ( "
" f"{badge_html}" "Live reveal uses the compact summary view so the reveal stays instant." "
" if run_badge == "live" else "" ) return ( "
" f"{intro}" f"{_render_arena_reveal_banner(state, history)}" f"{_render_metrics_band(result)}" f"{_render_delta_strip(result)}" f"
{html.escape(_render_story_sentence(result))}
" "
" ) def _render_arena_state( state: dict[str, Any] | None, history: list[dict[str, Any]] | None = None, ) -> tuple[str, str, str, str, str, str, str, str, str, dict[str, Any], str, Any]: history = history or [] if not state: return ( _arena_empty_match(), _render_arena_vote_status(None, history), _arena_pre_reveal_helper(), "", "", "", "", "", "", {}, "", gr.update(visible=False), ) result = state["result"] if not state.get("revealed"): return ( _render_arena_match_html(state), _render_arena_vote_status(state, history), _arena_pre_reveal_helper(), "", "", "", "", "", "", {}, "", gr.update(visible=False), ) return ( _render_arena_match_html(state), _render_arena_vote_status(state, history), _arena_post_reveal_helper(), _render_arena_reveal_banner(state, history), _render_metrics_band(result), _render_delta_strip(result), _render_story_sentence(result), _render_diff_tab(result), _render_charts_tab(result), { "result": _compact_result_for_ui(result), "arena": { k: state.get(k) for k in ("output_a_role", "output_b_role", "blind_shuffle_seed", "revealed", "user_vote", "config_hash") }, }, _arena_logs_text(state), gr.update(visible=True), ) def _arena_history_entry(state: dict[str, Any]) -> dict[str, Any]: chosen_role = None vote = state.get("user_vote") if vote == "a_better": chosen_role = state.get("output_a_role") elif vote == "b_better": chosen_role = state.get("output_b_role") return { "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "config_hash": state.get("config_hash"), "user_vote": vote, "compressed_guess": state.get("compressed_guess"), "chosen_role": chosen_role, "voted_before_reveal": bool(state.get("voted_before_reveal")), "revealed": bool(state.get("revealed")), } def _arena_start_match( model: str, preset: str, custom_prompt: str, live_mode: bool, context_length: int, mode: str, page_size: int, bits_k: int, bits_v: int, recent_window: int, sink_window: int, shortlist_policy: str, history: list[dict[str, Any]] | None, ) -> tuple[dict[str, Any], list[dict[str, Any]], str, str, str, str, str, str, str, str, str, dict[str, Any], str, Any]: request = _build_request( model, preset, custom_prompt, live_mode, context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, True, ) response = execute_request(request) client_state = _build_arena_state(request, response) rendered = _render_arena_state(_resolve_arena_state(client_state), history) return (client_state, history or [], *rendered) def _arena_vote( state: dict[str, Any] | None, history: list[dict[str, Any]] | None, vote: str, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], str]: history = history or [] if not state: return state, history, _render_arena_vote_status(None, history) state = dict(state) state["user_vote"] = vote state["vote_timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%S%z") if not state.get("revealed"): state["voted_before_reveal"] = True return state, history, _render_arena_vote_status(state, history) def _arena_set_guess( state: dict[str, Any] | None, history: list[dict[str, Any]] | None, guess: str, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], str]: history = history or [] if not state: return state, history, _render_arena_vote_status(None, history) state = dict(state) state["compressed_guess"] = guess return state, history, _render_arena_vote_status(state, history) def _arena_reveal( state: dict[str, Any] | None, history: list[dict[str, Any]] | None, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], str, str, str, str, str, str, str, str, str, dict[str, Any], str, Any]: history = history or [] if not state: rendered = _render_arena_state(None, history) return (state, history, *rendered) client_state = dict(state) first_reveal = not client_state.get("revealed") client_state["revealed"] = True if first_reveal: updated_entry = _arena_history_entry(client_state) updated_entry["revealed"] = True history = [*history, updated_entry] full_state = _resolve_arena_state(client_state) if full_state is None: rendered = _render_arena_state(None, history) return (client_state, history, *rendered) match_id = str(client_state.get("arena_match_id") or "").strip() if match_id: ARENA_MATCH_STORE[match_id] = full_state result = full_state["result"] return ( client_state, history, _render_arena_match_html(full_state), _render_arena_vote_status(client_state, history), _arena_post_reveal_helper(), _render_arena_reveal_banner(full_state, history), _render_metrics_band(result), _render_delta_strip(result), _render_story_sentence(result), "
Reveal complete. Diff is temporarily disabled for live Arena matches.
", "
Reveal complete. Charts are temporarily disabled for live Arena matches.
", { "result": {"status": "reveal_complete"}, "arena": {"revealed": True, "details": "temporarily_disabled"}, }, "Detailed reveal logs are temporarily disabled for live Arena matches.", gr.update(visible=True), ) def _arena_reveal_visible( state: dict[str, Any] | None, history: list[dict[str, Any]] | None, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], str, str, str, Any, Any, Any, Any, str, Any]: full = _arena_reveal(state, history) full_state = _resolve_arena_state(full[0]) run_badge = str((full_state or {}).get("response_meta", {}).get("run_badge") or "") match_html_output: Any = full[2] reveal_banner_output: Any = full[5] metrics_output: Any = full[6] delta_output: Any = full[7] summary_output: Any = full[8] inline_reveal_output = "" post_panel_output: Any = full[13] if run_badge == "live": match_html_output = gr.skip() reveal_banner_output = gr.skip() metrics_output = gr.skip() delta_output = gr.skip() summary_output = gr.skip() inline_reveal_output = _render_arena_inline_reveal(full_state, full[1]) if full_state else "" post_panel_output = gr.update(visible=False) else: reveal_banner_output = gr.skip() metrics_output = gr.skip() delta_output = gr.skip() summary_output = gr.skip() inline_reveal_output = _render_arena_inline_reveal(full_state, full[1]) if full_state else "" post_panel_output = gr.update(visible=True) return ( full[0], full[1], match_html_output, full[3], full[4], reveal_banner_output, metrics_output, delta_output, summary_output, inline_reveal_output, post_panel_output, ) def _apply_arena_preset(current_model: str, key: str) -> tuple[Any, ...]: mapping = { "safe_match": "fastest_safe", "balanced_match": "balanced", "fragile_match": "aggressive", } payload = quick_start_payload(mapping[key]) return ( current_model, payload["preset"], payload["custom_prompt"], payload["context_length"], payload["mode"], payload["page_size"], payload["bits_k"], payload["bits_v"], payload["recent_window"], payload["sink_window"], payload["shortlist_policy"], ) def _set_arena_loading_state() -> tuple[Any, str]: return gr.update(value="Starting Arena Match...", interactive=False), _render_loading_state("Building a blind A/B arena match...") def _clear_arena_loading_state() -> tuple[Any, str]: return gr.update(value="Start Arena Match", interactive=True), "" CSS = """ @import url('https://fonts.googleapis.com/css2?family=IBM+Plex+Sans:wght@400;500;600;700&family=IBM+Plex+Mono:wght@400;500&display=swap'); :root { --ink: #0b1f2a; --muted: #52636d; --surface: rgba(255, 255, 255, 0.88); --surface-strong: rgba(255, 255, 255, 0.97); --line: rgba(14, 84, 101, 0.14); --accent: #0f8b8d; --accent-deep: #115f6b; --accent-soft: #d9f3f0; --warm: #f2a65a; --bg-top: #f7fbfd; --bg-bottom: #ecf3e8; --good: #197c52; --good-soft: #e4f5ec; --warn: #b26a00; --warn-soft: #fff1da; --bad: #a52d38; --bad-soft: #fde8ea; --panel-controls: rgba(248, 251, 252, 0.72); --panel-results: rgba(255, 255, 255, 0.84); } .gradio-container { font-family: "IBM Plex Sans", "Avenir Next", sans-serif; color: var(--ink); background: radial-gradient(circle at top left, rgba(15, 139, 141, 0.14), transparent 30%), radial-gradient(circle at top right, rgba(242, 166, 90, 0.12), transparent 28%), linear-gradient(180deg, var(--bg-top), var(--bg-bottom)); } .mono, .code, code, pre { font-family: "IBM Plex Mono", "SFMono-Regular", monospace; } .workflow-nav { display: inline-flex; gap: 0.65rem; margin: 0 0 1rem 0; padding: 0.25rem; border-radius: 999px; background: rgba(255, 255, 255, 0.72); border: 1px solid rgba(17, 95, 107, 0.12); box-shadow: 0 12px 28px rgba(9, 45, 54, 0.06); } .workflow-link { min-width: 140px; padding: 0.72rem 1.1rem; border-radius: 999px; text-align: center; text-decoration: none; color: var(--ink); font-weight: 700; transition: background 0.18s ease, color 0.18s ease, transform 0.18s ease; } .workflow-link:hover { background: rgba(17, 95, 107, 0.08); transform: translateY(-1px); } .workflow-link.active { background: linear-gradient(135deg, #ff8c2a, #ff6f0f); color: white; box-shadow: 0 10px 24px rgba(255, 111, 15, 0.22); } .hero-shell { border: 1px solid rgba(17, 95, 107, 0.14); border-radius: 28px; padding: 1.4rem 1.5rem; background: radial-gradient(circle at top right, rgba(15, 139, 141, 0.12), transparent 30%), linear-gradient(135deg, rgba(255,255,255,0.98), rgba(239,248,247,0.92)); box-shadow: 0 22px 56px rgba(9, 45, 54, 0.1); } .hero-shell .eyebrow { font-size: 0.82rem; text-transform: uppercase; letter-spacing: 0.09em; color: var(--accent-deep); font-weight: 700; } .hero-title-row { display: flex; align-items: flex-start; justify-content: space-between; gap: 1rem; } .hero-shell h1 { margin: 0.3rem 0 0.35rem 0; font-size: clamp(2rem, 3vw, 3rem); line-height: 1.02; } .hero-shell .hero-subtitle { margin: 0; color: var(--muted); } .hero-about-link { display: inline-flex; align-items: center; justify-content: center; padding: 0.55rem 0.9rem; border-radius: 999px; border: 1px solid rgba(17, 95, 107, 0.14); background: rgba(255, 255, 255, 0.9); color: var(--accent-deep); text-decoration: none; font-weight: 700; white-space: nowrap; transition: transform 0.18s ease, box-shadow 0.18s ease, background 0.18s ease; } .hero-about-link:hover { background: rgba(240, 248, 247, 0.98); box-shadow: 0 12px 28px rgba(9, 45, 54, 0.08); transform: translateY(-1px); } .about-panel-overlay { position: fixed; inset: 0; display: none; z-index: 2000; } .about-panel-overlay:target { display: block; } .about-panel-backdrop { position: absolute; inset: 0; background: rgba(7, 20, 30, 0.42); backdrop-filter: blur(4px); } .about-panel-sheet { position: absolute; top: 0; right: 0; width: min(520px, 92vw); height: 100%; overflow-y: auto; padding: 1.2rem 1.2rem 1.5rem 1.2rem; background: linear-gradient(180deg, rgba(255,255,255,0.99), rgba(241,248,247,0.98)); border-left: 1px solid rgba(17, 95, 107, 0.12); box-shadow: -22px 0 48px rgba(9, 45, 54, 0.16); } .about-panel-head { display: flex; align-items: center; justify-content: space-between; gap: 1rem; margin-bottom: 1rem; } .about-panel-title { font-size: 1rem; font-weight: 700; color: var(--ink); } .about-panel-close { display: inline-flex; align-items: center; justify-content: center; padding: 0.5rem 0.8rem; border-radius: 999px; background: rgba(17, 95, 107, 0.08); color: var(--accent-deep); text-decoration: none; font-weight: 700; } .about-sheet-copy { color: var(--ink); } .about-sheet-eyebrow { font-size: 0.8rem; text-transform: uppercase; letter-spacing: 0.08em; color: var(--accent-deep); font-weight: 700; } .about-sheet-copy h2 { margin: 0.4rem 0 0.7rem 0; font-size: 1.7rem; line-height: 1.1; } .about-sheet-copy h3 { margin: 1.2rem 0 0.5rem 0; font-size: 1rem; } .about-sheet-copy p, .about-sheet-copy li { color: var(--muted); line-height: 1.65; } .about-sheet-copy ul { margin: 0; padding-left: 1.15rem; } .about-sheet-links { display: flex; flex-wrap: wrap; gap: 0.65rem; margin-top: 1.15rem; } .about-divider { height: 1px; margin: 1rem 0 0.25rem 0; background: linear-gradient(90deg, rgba(17, 95, 107, 0.16), rgba(17, 95, 107, 0)); } .about-sheet-links a { display: inline-flex; align-items: center; padding: 0.55rem 0.8rem; border-radius: 999px; background: rgba(17, 95, 107, 0.08); color: var(--accent-deep); text-decoration: none; font-weight: 700; } .arena-shell { border-color: rgba(178, 106, 0, 0.18); background: radial-gradient(circle at top left, rgba(242, 166, 90, 0.14), transparent 28%), linear-gradient(135deg, rgba(255,255,255,0.98), rgba(255,247,235,0.92)); } .arena-hero-pill { display: inline-flex; align-items: center; margin-top: 0.9rem; padding: 0.4rem 0.8rem; border-radius: 999px; background: rgba(255, 245, 226, 0.95); color: #8a5408; font-weight: 700; font-size: 0.88rem; } .hero-stats { display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 0.8rem; margin: 1rem 0 1.1rem 0; } .hero-stat { display: flex; gap: 0.8rem; align-items: center; border-radius: 18px; padding: 0.85rem 0.95rem; background: rgba(255,255,255,0.82); border: 1px solid rgba(17, 95, 107, 0.1); } .hero-stat-icon { font-size: 1.25rem; } .hero-stat-value { font-weight: 700; font-size: 1.1rem; } .hero-stat-label { color: var(--muted); font-size: 0.88rem; } .control-panel, .results-panel { border: 1px solid var(--line); border-radius: 24px; padding: 1rem; box-shadow: 0 16px 40px rgba(9, 45, 54, 0.06); } .control-panel { background: var(--panel-controls); } .arena-panel .control-panel { background: rgba(248, 251, 252, 0.62); border-color: rgba(17, 95, 107, 0.1); } .results-panel { background: var(--panel-results); } .arena-placeholder { display: flex; flex-direction: column; gap: 0.3rem; padding: 1.1rem 1.15rem; border-radius: 20px; border: 1px dashed rgba(17, 95, 107, 0.18); background: rgba(255, 255, 255, 0.72); color: var(--muted); } .preset-row button { min-height: 64px; border-radius: 18px !important; transition: transform 0.16s ease, box-shadow 0.16s ease, background 0.16s ease; box-shadow: 0 10px 20px rgba(9, 45, 54, 0.06); } .preset-row button:hover { transform: translateY(-1px); box-shadow: 0 14px 26px rgba(9, 45, 54, 0.1); } .preset-status-card { display: flex; flex-direction: column; gap: 0.18rem; padding: 0.75rem 0.9rem; border: 1px solid rgba(17, 95, 107, 0.12); border-radius: 16px; background: rgba(255, 255, 255, 0.72); margin-bottom: 1rem; } .preset-status-card.recommended { background: linear-gradient(135deg, rgba(255, 245, 226, 0.95), rgba(255,255,255,0.78)); } .preset-status-card span { color: var(--muted); font-size: 0.92rem; } .preset-guide { display: flex; flex-direction: column; gap: 0.2rem; padding: 0.7rem 0.9rem; border-radius: 14px; background: rgba(255, 255, 255, 0.76); border: 1px dashed rgba(17, 95, 107, 0.14); margin-top: 0.7rem; } .preset-guide span { color: var(--muted); font-size: 0.9rem; } .arena-helper { display: flex; flex-direction: column; gap: 0.25rem; margin: 0.35rem 0 0.85rem 0; padding: 0.8rem 0.95rem; border-radius: 16px; background: rgba(255, 255, 255, 0.78); border: 1px solid rgba(17, 95, 107, 0.1); } .arena-helper span { color: var(--muted); } .arena-helper.revealed { background: rgba(255, 250, 241, 0.85); border-color: rgba(178, 106, 0, 0.16); } .arena-match-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 1.2rem; margin-top: 0.35rem; } .arena-blind-header { display: flex; flex-direction: column; gap: 0.2rem; margin: 0.1rem 0 0.7rem 0; } .arena-blind-header span { color: var(--muted); } .arena-output-card { border-radius: 22px; padding: 1.15rem 1.15rem 1.2rem 1.15rem; border: 1px solid rgba(17, 95, 107, 0.14); background: rgba(255, 255, 255, 0.88); box-shadow: 0 18px 38px rgba(9, 45, 54, 0.08); transition: transform 0.2s ease, box-shadow 0.2s ease; } .arena-output-card:hover { transform: translateY(-1px); box-shadow: 0 22px 42px rgba(9, 45, 54, 0.11); } .arena-output-card.blind { border-color: rgba(17, 95, 107, 0.2); background: radial-gradient(circle at top right, rgba(15, 139, 141, 0.08), transparent 26%), rgba(255, 255, 255, 0.92); } .arena-output-card.revealed { animation: arenaRevealCard 0.42s ease; } .arena-output-card.role-candidate { border-color: rgba(25, 124, 82, 0.18); } .arena-output-card.role-baseline { border-color: rgba(84, 102, 110, 0.18); } .arena-card-head { display: flex; justify-content: space-between; gap: 0.8rem; align-items: flex-start; } .arena-card-label { font-size: 1.08rem; font-weight: 800; letter-spacing: 0.01em; } .arena-card-meta { margin-top: 0.18rem; color: var(--muted); font-size: 0.82rem; } .arena-blind-badge { border-radius: 999px; padding: 0.32rem 0.62rem; background: rgba(84, 102, 110, 0.12); color: #51616d; font-size: 0.8rem; font-weight: 700; } .arena-output-card p { margin: 0.8rem 0 0 0; line-height: 1.68; white-space: pre-wrap; } .arena-decision-row { margin-top: 0.9rem; align-items: stretch; } .arena-preference, .arena-guess { padding: 0.85rem 0.9rem; border-radius: 18px; background: rgba(255, 255, 255, 0.82); border: 1px solid rgba(17, 95, 107, 0.1); } .arena-preference .wrap, .arena-guess .wrap { gap: 0.55rem; } .arena-preference label, .arena-guess label { border-radius: 999px !important; } .arena-reveal-button button { min-height: 100%; height: 100%; min-width: 100%; border-radius: 18px !important; font-size: 1rem !important; font-weight: 800 !important; } .arena-vote-status { display: flex; flex-direction: column; gap: 0.18rem; margin-top: 0.7rem; padding: 0.75rem 0.95rem; border-radius: 16px; background: rgba(255, 255, 255, 0.78); border: 1px solid rgba(17, 95, 107, 0.1); } .arena-vote-status span { color: var(--muted); font-size: 0.88rem; } .arena-vote-status.revealed { border-color: rgba(178, 106, 0, 0.16); } .arena-reveal-banner { border-radius: 18px; padding: 1rem 1.05rem; margin-bottom: 0.9rem; border: 1px solid rgba(17, 95, 107, 0.16); background: rgba(255,255,255,0.86); animation: arenaRevealBanner 0.45s ease; box-shadow: 0 20px 38px rgba(9, 45, 54, 0.08); } .arena-reveal-banner .arena-reveal-kicker { font-size: 0.82rem; text-transform: uppercase; letter-spacing: 0.07em; color: var(--accent-deep); font-weight: 800; } .arena-reveal-banner .arena-reveal-label { font-weight: 800; font-size: 1.14rem; margin-top: 0.14rem; } .arena-reveal-banner .arena-reveal-guess { margin-top: 0.18rem; color: var(--muted); font-weight: 600; } .arena-reveal-banner .arena-reveal-copy { margin-top: 0.35rem; line-height: 1.58; } .arena-reveal-banner .arena-reveal-copy.compact { margin-top: 0.45rem; display: inline-flex; padding: 0.42rem 0.78rem; border-radius: 999px; background: rgba(15, 139, 141, 0.08); color: var(--accent-deep); font-weight: 800; } .arena-reveal-reaction { display: inline-flex; margin-top: 0.45rem; padding: 0.38rem 0.72rem; border-radius: 999px; font-weight: 800; } .arena-reveal-reaction.good { background: var(--good-soft); color: var(--good); } .arena-reveal-reaction.warn, .arena-reveal-reaction.warm { background: var(--warn-soft); color: var(--warn); } .arena-session-pill { display: inline-flex; margin-top: 0.55rem; padding: 0.38rem 0.72rem; border-radius: 999px; background: rgba(255, 245, 226, 0.95); color: #8a5408; font-weight: 800; } .meta-strip { display: flex; gap: 0.8rem; align-items: center; flex-wrap: wrap; margin-bottom: 0.9rem; } .live-context-note { margin: 0.45rem 0 0.2rem 0; padding: 0.7rem 0.85rem; border-radius: 14px; border: 1px dashed rgba(17, 95, 107, 0.16); background: rgba(255, 255, 255, 0.62); color: var(--muted); font-size: 0.9rem; } .live-context-note.active { border-color: rgba(242, 166, 90, 0.4); background: rgba(255, 244, 232, 0.92); color: #8a5408; } .live-context-note strong { display: block; margin-bottom: 0.18rem; color: var(--ink); } .live-context-note.active strong { color: #8a5408; } .live-context-note code { background: rgba(255, 255, 255, 0.82); border-radius: 999px; padding: 0.1rem 0.35rem; } .live-prompt-shell, .live-challenge-note { margin: 0.25rem 0 0.7rem 0; padding: 0.9rem 1rem; border-radius: 18px; border: 1px solid rgba(17, 95, 107, 0.12); background: rgba(255, 255, 255, 0.82); } .live-prompt-title { font-size: 1rem; font-weight: 700; color: var(--ink); margin-bottom: 0.2rem; } .live-prompt-shell p, .live-challenge-note span { margin: 0; color: var(--muted); } .live-prompt-bullets { display: flex; flex-wrap: wrap; gap: 0.45rem; margin-top: 0.65rem; } .live-prompt-bullets span { padding: 0.3rem 0.6rem; border-radius: 999px; background: rgba(15, 139, 141, 0.08); color: var(--accent-deep); font-size: 0.86rem; font-weight: 600; } .live-challenge-note { background: rgba(255, 245, 236, 0.92); border-color: rgba(242, 166, 90, 0.25); } .live-challenge-note strong { display: block; margin-bottom: 0.18rem; color: #8a5408; } .example-row { gap: 0.5rem; margin: 0.2rem 0 0.55rem 0; } .example-row button { min-height: 42px; } .run-badge { display: inline-flex; align-items: center; padding: 0.3rem 0.7rem; border-radius: 999px; font-size: 0.8rem; text-transform: uppercase; letter-spacing: 0.06em; font-weight: 700; } .badge-precomputed { background: #e0f5f2; color: #0d6768; } .badge-live { background: #fff0db; color: #9a5b06; } .badge-muted { background: #edf2f4; color: #51616d; } .badge-warn { background: #fff1da; color: #9a5b06; } .meta-detail { color: var(--muted); font-size: 0.9rem; line-height: 1.4; } .kpi-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(170px, 1fr)); gap: 1rem; } .metrics-band { display: grid; grid-template-columns: minmax(0, 2.2fr) minmax(280px, 1fr); gap: 0.9rem; align-items: start; } .metrics-kpis, .metrics-trends { min-width: 0; } .trend-grid { display: grid; grid-template-columns: 1fr; gap: 0.8rem; } .trend-card { border: 1px solid var(--line); border-radius: 20px; background: rgba(255, 255, 255, 0.82); box-shadow: 0 16px 34px rgba(9, 45, 54, 0.07); padding: 0.9rem 1rem 0.85rem 1rem; } .trend-card.blocked p { margin: 0.45rem 0 0 0; color: var(--muted); line-height: 1.55; } .trend-title { color: var(--muted); font-size: 0.82rem; text-transform: uppercase; letter-spacing: 0.06em; font-weight: 700; } .bar-compare { display: grid; gap: 0.75rem; margin-top: 0.6rem; } .bar-row { display: flex; flex-direction: column; gap: 0.35rem; } .bar-meta { display: flex; justify-content: space-between; align-items: baseline; gap: 0.75rem; } .bar-track { width: 100%; height: 14px; border-radius: 999px; background: #e5efee; overflow: hidden; } .bar-fill { height: 100%; border-radius: 999px; } .bar-fill.dense { background: linear-gradient(90deg, #7f9aa8, #55707d); } .bar-fill.candidate.good { background: linear-gradient(90deg, #1fa86d, #197c52); } .bar-fill.candidate.bad { background: linear-gradient(90deg, #d45862, #a52d38); } .trend-name { color: var(--muted); font-size: 0.8rem; } .trend-delta { display: inline-flex; margin-top: 0.7rem; padding: 0.35rem 0.65rem; border-radius: 999px; font-size: 0.82rem; font-weight: 700; } .trend-delta.good { background: var(--good-soft); color: var(--good); } .trend-delta.bad { background: var(--bad-soft); color: var(--bad); } .trend-delta.neutral { background: rgba(84, 102, 110, 0.12); color: #51616d; } .kpi-card, .response-card, .chart-block { border: 1px solid var(--line); border-radius: 20px; background: var(--surface); box-shadow: 0 16px 34px rgba(9, 45, 54, 0.07); } .kpi-card { padding: 0.95rem 1rem; } .kpi-label, .response-label, .chart-label { color: var(--muted); font-size: 0.85rem; text-transform: uppercase; letter-spacing: 0.05em; font-weight: 700; } .kpi-primary { font-size: 1.55rem; font-weight: 700; margin-top: 0.2rem; } .kpi-secondary { color: var(--muted); font-size: 0.9rem; margin-top: 0.3rem; } .tone-good { border-color: rgba(25, 124, 82, 0.16); background: linear-gradient(180deg, rgba(255,255,255,0.96), var(--good-soft)); } .tone-warn { border-color: rgba(178, 106, 0, 0.16); background: linear-gradient(180deg, rgba(255,255,255,0.96), var(--warn-soft)); } .tone-bad { border-color: rgba(165, 45, 56, 0.16); background: linear-gradient(180deg, rgba(255,255,255,0.96), var(--bad-soft)); } .delta-strip { display: flex; gap: 0.65rem; align-items: center; flex-wrap: wrap; padding: 0.85rem 1rem; border: 1px solid rgba(17, 95, 107, 0.12); border-radius: 18px; background: rgba(255,255,255,0.8); animation: resultPulse 0.48s ease; } .delta-label { font-size: 0.85rem; text-transform: uppercase; letter-spacing: 0.07em; color: var(--muted); font-weight: 700; } .delta-item { padding: 0.44rem 0.82rem; border-radius: 999px; font-weight: 700; font-size: 0.9rem; } .delta-item.good { background: #d7f3e4; color: var(--good); } .delta-item.warn { background: var(--warn-soft); color: var(--warn); } .delta-item.bad { background: #f8dde1; color: var(--bad); } .story-sentence { margin: 0.3rem 0 1rem 0; font-size: 1.03rem; line-height: 1.55; animation: resultFade 0.35s ease; } .arena-inline-reveal { margin: 0.9rem 0 0.4rem 0; display: grid; gap: 0.75rem; padding: 0.9rem 1rem 1rem 1rem; border: 1px solid rgba(17, 95, 107, 0.12); border-radius: 22px; background: linear-gradient(180deg, rgba(255,255,255,0.96), rgba(240,248,247,0.92)); box-shadow: 0 16px 34px rgba(9, 45, 54, 0.08); } .arena-inline-intro { display: flex; align-items: center; gap: 0.65rem; flex-wrap: wrap; color: var(--muted); font-size: 0.92rem; } .arena-inline-intro .run-badge { margin-right: 0.15rem; } .response-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); gap: 0.9rem; } .response-card { padding: 1rem; } .response-head { display: flex; justify-content: space-between; gap: 0.8rem; align-items: flex-start; } .response-rule { height: 2px; width: 72px; border-radius: 999px; background: linear-gradient(90deg, var(--accent), transparent); margin-top: 0.35rem; } .response-badge { border-radius: 999px; padding: 0.35rem 0.65rem; font-size: 0.82rem; font-weight: 700; white-space: nowrap; } .response-badge.good { background: var(--good-soft); color: var(--good); } .response-badge.warn { background: var(--warn-soft); color: var(--warn); } .response-badge.bad { background: var(--bad-soft); color: var(--bad); } .response-badge.neutral { background: rgba(84, 102, 110, 0.12); color: #51616d; } .results-state { display: inline-flex; align-items: center; gap: 0.6rem; margin: 0.15rem 0 0.55rem 0; padding: 0.55rem 0.85rem; border-radius: 999px; background: rgba(255, 245, 226, 0.85); color: #8a5408; border: 1px solid rgba(178, 106, 0, 0.18); font-weight: 600; } .results-spinner { width: 14px; height: 14px; border-radius: 999px; border: 2px solid rgba(178, 106, 0, 0.22); border-top-color: #c7770b; animation: spinnerSpin 0.8s linear infinite; } .response-card p { margin: 0.7rem 0 0 0; line-height: 1.65; white-space: pre-wrap; } .response-note { margin-top: 0.72rem; color: var(--muted); font-size: 0.84rem; } .response-card mark { background: rgba(242, 166, 90, 0.26); color: inherit; padding: 0.08rem 0.16rem; border-radius: 0.2rem; } .charts-grid { display: grid; gap: 0.9rem; } .chart-block { padding: 0.95rem 1rem; } .chart-row { display: grid; grid-template-columns: 70px 1fr 80px; gap: 0.7rem; align-items: center; margin-top: 0.7rem; font-size: 0.92rem; } .chart-track { width: 100%; height: 12px; border-radius: 999px; background: #e5efee; overflow: hidden; } .chart-fill { height: 100%; border-radius: 999px; } .chart-fill.dense { background: linear-gradient(90deg, #7f9aa8, #55707d); } .chart-fill.candidate { background: linear-gradient(90deg, #0f8b8d, #115f6b); } .helper-box { margin-top: 1rem; padding: 0.85rem 1rem; border-radius: 16px; background: rgba(255,255,255,0.68); border: 1px dashed rgba(17, 95, 107, 0.16); } .arena-panel .helper-box { opacity: 0.78; background: rgba(255,255,255,0.56); } .helper-box p, .helper-box ul, .helper-box li, .helper-box h3 { margin: 0; } .helper-box h3 { font-size: 0.95rem; margin-bottom: 0.35rem; } .gradio-container label { font-weight: 500 !important; } .control-panel .gr-accordion { opacity: 0.68; filter: saturate(0.7); font-size: 0.95rem; } .prompt-mode-tabs { margin: 0.15rem 0 0.55rem 0; } .prompt-mode-tabs button { border-radius: 999px !important; font-weight: 700 !important; min-width: 112px; } .prompt-mode-tabs button[aria-selected="true"] { background: linear-gradient(135deg, #ff8c2a, #ff6f0f) !important; color: white !important; } .hero-shell, .metrics-band, .response-grid, .charts-grid { animation: resultFade 0.35s ease; } @keyframes spinnerSpin { from { transform: rotate(0deg); } to { transform: rotate(360deg); } } @keyframes resultFade { from { opacity: 0; transform: translateY(4px); } to { opacity: 1; transform: translateY(0); } } @keyframes resultPulse { 0% { transform: scale(0.992); box-shadow: 0 0 0 rgba(15, 139, 141, 0.0); } 55% { transform: scale(1.004); box-shadow: 0 14px 28px rgba(15, 139, 141, 0.1); } 100% { transform: scale(1); box-shadow: none; } } @keyframes arenaRevealCard { 0% { opacity: 0.2; transform: rotateY(-8deg) translateY(6px); } 100% { opacity: 1; transform: rotateY(0deg) translateY(0); } } @keyframes arenaRevealBanner { 0% { opacity: 0; transform: translateY(8px); } 65% { opacity: 1; transform: translateY(-1px); } 100% { transform: translateY(0); } } @media (max-width: 980px) { .hero-title-row { flex-direction: column; align-items: flex-start; } .metrics-band { grid-template-columns: 1fr; } } """ def _env_flag(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} def _env_int(name: str) -> int | None: value = os.getenv(name) if value is None or not value.strip(): return None return int(value) def _textbox_copy_kwargs() -> dict[str, Any]: signature = inspect.signature(gr.Textbox) if "buttons" in signature.parameters: return {"buttons": ["copy"]} if "show_copy_button" in signature.parameters: return {"show_copy_button": True} return {} with gr.Blocks(title="DotCache Paper Demo") as demo: about_panel_html = load_asset("about_panel.html") workflow_nav = gr.HTML(_workflow_nav_html("Arena")) about_panel = gr.HTML(_render_about_sheet(about_panel_html)) with gr.Column(visible=False) as compare_panel: hero_panel = gr.HTML() with gr.Row(equal_height=True, elem_classes="preset-row"): fastest_safe = gr.Button(PRESET_BUTTON_LABELS["fastest_safe"], variant="secondary") balanced = gr.Button(PRESET_BUTTON_LABELS["balanced"], variant="primary") aggressive = gr.Button(PRESET_BUTTON_LABELS["aggressive"], variant="secondary") preset_status = gr.HTML(PRESET_STATUS_HTML["balanced"]) with gr.Row(): with gr.Column(scale=4, elem_classes="control-panel"): model = gr.Dropdown( choices=model_choices(), value="qwen35_9b_hf", label="Model", ) compare_prompt_mode = gr.State(value="preset") with gr.Tabs(selected="preset", elem_classes="prompt-mode-tabs") as compare_prompt_tabs: with gr.Tab("Preset", id="preset") as compare_preset_tab: preset = gr.Dropdown( choices=preset_choices(), value="backend_truth", label="Prompt preset", ) with gr.Tab("Live prompt", id="live") as compare_live_tab: live_prompt_intro = gr.HTML(_live_prompt_intro_html()) with gr.Row(elem_classes="example-row", visible=CUSTOM_LIVE_PROMPTS_ENABLED): compare_example_retrieval = gr.Button("Long context retrieval", variant="secondary") compare_example_reasoning = gr.Button("Multi-step reasoning", variant="secondary") compare_example_fragile = gr.Button("Fragile case", variant="secondary") custom_prompt = gr.Textbox( label="Challenge DotCache With Your Own Test" if CUSTOM_LIVE_PROMPTS_ENABLED else "Custom live prompt", lines=5, placeholder=( "Bring your own test and see if DotCache still holds up." if CUSTOM_LIVE_PROMPTS_ENABLED else "Custom live prompts are disabled in this benchmark-faithful build." ), interactive=CUSTOM_LIVE_PROMPTS_ENABLED, ) live_prompt_challenge = gr.HTML(_live_prompt_challenge_html()) context_length = gr.Dropdown( choices=context_choices(), value=2048, label="Context length", ) live_context_note = gr.HTML(_live_context_guard_copy("preset")) mode = gr.Dropdown( choices=mode_choices(), value="m0_selective_k", label="Compression mode", info="Fastest Safe, Balanced, and Aggressive change this profile. In preset mode it controls the displayed reference/systems lane; in live mode it controls the actual runtime path.", ) with gr.Row(): run_button = gr.Button("Run", variant="secondary") compare_button = gr.Button("Compare", variant="primary") with gr.Accordion("Advanced", open=False): page_size = gr.Dropdown(choices=list(PAGE_SIZES), value=16, label="Page size") bits_k = gr.Dropdown(choices=list(BIT_WIDTHS), value=4, label="bits_k") bits_v = gr.Dropdown(choices=list(BIT_WIDTHS), value=4, label="bits_v") recent_window = gr.Slider(minimum=0, maximum=32, step=1, value=8, label="recent_window") sink_window = gr.Slider(minimum=0, maximum=8, step=1, value=0, label="sink_window") shortlist_policy = gr.Dropdown( choices=list(SHORTLIST_POLICIES), value="benchmark", label="shortlist policy", ) with gr.Column(scale=6, elem_classes="results-panel"): metrics_band = gr.HTML() results_state = gr.HTML() delta_strip = gr.HTML() summary = gr.Markdown(elem_classes="story-sentence") with gr.Tabs(): with gr.TabItem("Response"): response_html = gr.HTML() with gr.TabItem("Side-by-side diff"): diff_html = gr.HTML() with gr.TabItem("Charts"): charts_html = gr.HTML() with gr.TabItem("Metrics JSON"): metrics_json = gr.JSON() with gr.TabItem("Logs"): logs_output = gr.Textbox(lines=14, max_lines=24, label="Logs", **_textbox_copy_kwargs()) request_inputs = [ model, preset, compare_prompt_mode, custom_prompt, context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, ] request_outputs = [hero_panel, metrics_band, results_state, delta_strip, summary, response_html, diff_html, charts_html, metrics_json, logs_output] run_button.click( fn=lambda *args: _run_request_with_prompt_mode(*args, compare_against_dense=False), inputs=request_inputs, outputs=request_outputs, show_progress="minimal", ) compare_button.click( fn=_set_compare_loading_state, outputs=[compare_button, results_state], show_progress="hidden", queue=False, ).then( fn=lambda *args: _run_request_with_prompt_mode(*args, compare_against_dense=True), inputs=request_inputs, outputs=request_outputs, show_progress="minimal", ).then( fn=_clear_compare_loading_state, outputs=[compare_button, results_state], show_progress="hidden", queue=False, ) compare_preset_tab.select( fn=lambda current_context_length, current_custom_prompt: _select_compare_prompt_mode("preset", current_context_length, current_custom_prompt), inputs=[context_length, custom_prompt], outputs=[compare_prompt_mode, context_length, live_context_note], show_progress="hidden", queue=False, ) compare_live_tab.select( fn=lambda current_context_length, current_custom_prompt: _select_compare_prompt_mode("live", current_context_length, current_custom_prompt), inputs=[context_length, custom_prompt], outputs=[compare_prompt_mode, context_length, live_context_note], show_progress="hidden", queue=False, ) custom_prompt.change( fn=_refresh_compare_prompt_mode, inputs=[compare_prompt_mode, context_length, custom_prompt], outputs=[context_length, live_context_note], show_progress="hidden", queue=False, ) if CUSTOM_LIVE_PROMPTS_ENABLED: for button, key in ( (compare_example_retrieval, "retrieval"), (compare_example_reasoning, "reasoning"), (compare_example_fragile, "fragile"), ): button.click( fn=lambda current_context_length, example_key=key: _apply_compare_live_example( example_key, int(current_context_length), ), inputs=[context_length], outputs=[compare_prompt_tabs, compare_prompt_mode, custom_prompt, context_length, live_context_note], show_progress="hidden", queue=False, ).then( fn=lambda *args: _run_request_with_prompt_mode(*args, compare_against_dense=True), inputs=request_inputs, outputs=request_outputs, show_progress="minimal", ) quick_start_outputs = [ model, preset, custom_prompt, context_length, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, ] preset_button_outputs = [fastest_safe, balanced, aggressive, preset_status] for button, key in ( (fastest_safe, "fastest_safe"), (balanced, "balanced"), (aggressive, "aggressive"), ): button.click( fn=lambda current_model, current_preset, current_custom_prompt, current_context_length, preset_key=key: _apply_quick_start( current_model, current_preset, current_custom_prompt, current_context_length, preset_key, ), inputs=[model, preset, custom_prompt, context_length], outputs=quick_start_outputs, ).then( fn=lambda: (gr.update(selected="preset"), "preset"), outputs=[compare_prompt_tabs, compare_prompt_mode], show_progress="hidden", queue=False, ).then( fn=_refresh_compare_prompt_mode, inputs=[compare_prompt_mode, context_length, custom_prompt], outputs=[context_length, live_context_note], show_progress="hidden", queue=False, ).then( fn=lambda preset_key=key: _compare_preset_ui(preset_key), outputs=preset_button_outputs, ).then( fn=_set_compare_loading_state, outputs=[compare_button, results_state], show_progress="hidden", queue=False, ).then( fn=lambda *args: _run_request_with_prompt_mode(*args, compare_against_dense=True), inputs=request_inputs, outputs=request_outputs, show_progress="minimal", ).then( fn=_clear_compare_loading_state, outputs=[compare_button, results_state], show_progress="hidden", queue=False, ) demo.load( fn=_initial_compare_results, outputs=request_outputs, ) with gr.Column(visible=False, elem_classes="arena-panel") as arena_panel: arena_state = gr.State(value=None) arena_history = gr.State(value=[]) arena_hero = gr.HTML(_arena_base_hero()) with gr.Row(equal_height=True, elem_classes="preset-row"): arena_safe = gr.Button(ARENA_PRESET_BUTTON_LABELS["safe_match"], variant="secondary") arena_balanced = gr.Button(ARENA_PRESET_BUTTON_LABELS["balanced_match"], variant="primary") arena_fragile = gr.Button(ARENA_PRESET_BUTTON_LABELS["fragile_match"], variant="secondary") arena_preset_status = gr.HTML(ARENA_PRESET_STATUS_HTML["balanced_match"]) with gr.Row(): with gr.Column(scale=4, elem_classes="control-panel"): arena_model = gr.Dropdown( choices=model_choices(), value="qwen35_9b_hf", label="Model", ) arena_prompt_mode = gr.State(value="preset") with gr.Tabs(selected="preset", elem_classes="prompt-mode-tabs") as arena_prompt_tabs: with gr.Tab("Preset", id="preset") as arena_preset_tab: arena_preset = gr.Dropdown( choices=preset_choices(), value="backend_truth", label="Prompt preset", ) with gr.Tab("Live prompt", id="live") as arena_live_tab: arena_live_prompt_intro = gr.HTML(_live_prompt_intro_html()) with gr.Row(elem_classes="example-row", visible=CUSTOM_LIVE_PROMPTS_ENABLED): arena_example_retrieval = gr.Button("Long context retrieval", variant="secondary") arena_example_reasoning = gr.Button("Multi-step reasoning", variant="secondary") arena_example_fragile = gr.Button("Fragile case", variant="secondary") arena_custom_prompt = gr.Textbox( label="Challenge DotCache With Your Own Test" if CUSTOM_LIVE_PROMPTS_ENABLED else "Custom live prompt", lines=5, placeholder=( "Bring your own test and see if DotCache still holds up." if CUSTOM_LIVE_PROMPTS_ENABLED else "Custom live prompts are disabled in this benchmark-faithful build." ), interactive=CUSTOM_LIVE_PROMPTS_ENABLED, ) arena_live_prompt_challenge = gr.HTML(_live_prompt_challenge_html()) arena_context_length = gr.Dropdown( choices=context_choices(), value=2048, label="Context length", ) arena_live_context_note = gr.HTML(_live_context_guard_copy("preset")) arena_mode = gr.Dropdown( choices=mode_choices(), value="m0_selective_k", label="Compression mode", info="Arena still compares the reference lane against the systems lane; this controls which runtime profile the blind candidate uses.", ) arena_start_button = gr.Button("Start Arena Match", variant="primary", interactive=True) with gr.Accordion("Advanced", open=False): arena_page_size = gr.Dropdown(choices=list(PAGE_SIZES), value=16, label="Page size") arena_bits_k = gr.Dropdown(choices=list(BIT_WIDTHS), value=4, label="bits_k") arena_bits_v = gr.Dropdown(choices=list(BIT_WIDTHS), value=4, label="bits_v") arena_recent_window = gr.Slider(minimum=0, maximum=32, step=1, value=8, label="recent_window") arena_sink_window = gr.Slider(minimum=0, maximum=8, step=1, value=0, label="sink_window") arena_shortlist_policy = gr.Dropdown( choices=list(SHORTLIST_POLICIES), value="benchmark", label="shortlist policy", ) with gr.Column(scale=6, elem_classes="results-panel"): arena_results_state = gr.HTML() arena_match_html = gr.HTML(_arena_empty_match()) with gr.Row(elem_classes="arena-decision-row"): arena_preference = gr.Radio( choices=[ ("Output A is better", "a_better"), ("Output B is better", "b_better"), ("Tie (hard to tell)", "tie"), ("Not sure", "not_sure"), ], value=None, label="Which do you prefer?", elem_classes="arena-preference", ) arena_guess = gr.Radio( choices=[ ("Compressed is Output A", "a"), ("Compressed is Output B", "b"), ("I can't tell", "not_sure"), ], value="not_sure", label="Which output do you think is compressed?", elem_classes="arena-guess", ) with gr.Column(scale=1, min_width=140): arena_reveal = gr.Button("Reveal Results", variant="primary", elem_classes="arena-reveal-button") arena_vote_status = gr.HTML(_render_arena_vote_status(None, [])) arena_helper_html = gr.HTML(_arena_pre_reveal_helper()) arena_inline_reveal = gr.HTML() with gr.Column(visible=False) as arena_post_panel: arena_reveal_banner = gr.HTML() arena_post_metrics = gr.HTML() arena_post_delta = gr.HTML() arena_post_summary = gr.Markdown(elem_classes="story-sentence") with gr.Tabs(): with gr.TabItem("Side-by-side diff"): arena_post_diff = gr.HTML() with gr.TabItem("Metrics"): arena_post_metrics_json = gr.JSON() with gr.TabItem("Charts"): arena_post_charts = gr.HTML() with gr.TabItem("Logs"): arena_post_logs = gr.Textbox(lines=14, max_lines=24, label="Logs", **_textbox_copy_kwargs()) arena_inputs = [ arena_model, arena_preset, arena_prompt_mode, arena_custom_prompt, arena_context_length, arena_mode, arena_page_size, arena_bits_k, arena_bits_v, arena_recent_window, arena_sink_window, arena_shortlist_policy, ] arena_render_outputs = [ arena_match_html, arena_vote_status, arena_helper_html, arena_reveal_banner, arena_post_metrics, arena_post_delta, arena_post_summary, arena_post_diff, arena_post_charts, arena_post_metrics_json, arena_post_logs, arena_post_panel, ] arena_reveal_outputs = [ arena_state, arena_history, arena_match_html, arena_vote_status, arena_helper_html, arena_reveal_banner, arena_post_metrics, arena_post_delta, arena_post_summary, arena_inline_reveal, arena_post_panel, ] arena_start_outputs = [arena_state, arena_history, *arena_render_outputs] arena_start_button.click( fn=_set_arena_loading_state, outputs=[arena_start_button, arena_results_state], show_progress="hidden", queue=False, ).then( fn=lambda: (gr.update(value=None), gr.update(value="not_sure"), "", gr.update(visible=False)), outputs=[arena_preference, arena_guess, arena_inline_reveal, arena_post_panel], show_progress="hidden", queue=False, ).then( fn=_arena_start_match_with_prompt_mode, inputs=[*arena_inputs, arena_history], outputs=arena_start_outputs, show_progress="minimal", ).then( fn=_clear_arena_loading_state, outputs=[arena_start_button, arena_results_state], show_progress="hidden", queue=False, ) arena_preset_tab.select( fn=lambda current_context_length, current_custom_prompt: _select_arena_prompt_mode("preset", current_context_length, current_custom_prompt), inputs=[arena_context_length, arena_custom_prompt], outputs=[arena_prompt_mode, arena_context_length, arena_live_context_note, arena_start_button], show_progress="hidden", queue=False, ) arena_live_tab.select( fn=lambda current_context_length, current_custom_prompt: _select_arena_prompt_mode("live", current_context_length, current_custom_prompt), inputs=[arena_context_length, arena_custom_prompt], outputs=[arena_prompt_mode, arena_context_length, arena_live_context_note, arena_start_button], show_progress="hidden", queue=False, ) arena_custom_prompt.change( fn=_refresh_arena_prompt_mode, inputs=[arena_prompt_mode, arena_context_length, arena_custom_prompt], outputs=[arena_context_length, arena_live_context_note, arena_start_button], show_progress="hidden", queue=False, ) if CUSTOM_LIVE_PROMPTS_ENABLED: for button, key in ( (arena_example_retrieval, "retrieval"), (arena_example_reasoning, "reasoning"), (arena_example_fragile, "fragile"), ): button.click( fn=lambda current_context_length, example_key=key: _apply_arena_live_example( example_key, int(current_context_length), ), inputs=[arena_context_length], outputs=[ arena_prompt_tabs, arena_prompt_mode, arena_custom_prompt, arena_context_length, arena_live_context_note, arena_start_button, ], show_progress="hidden", queue=False, ) arena_preference.change( fn=_arena_vote, inputs=[arena_state, arena_history, arena_preference], outputs=[arena_state, arena_history, arena_vote_status], show_progress="hidden", queue=False, ) arena_guess.change( fn=_arena_set_guess, inputs=[arena_state, arena_history, arena_guess], outputs=[arena_state, arena_history, arena_vote_status], show_progress="hidden", queue=False, ) arena_reveal.click( fn=_arena_reveal_visible, inputs=[arena_state, arena_history], outputs=arena_reveal_outputs, show_progress="hidden", queue=False, ) arena_preset_outputs = [ arena_model, arena_preset, arena_custom_prompt, arena_context_length, arena_mode, arena_page_size, arena_bits_k, arena_bits_v, arena_recent_window, arena_sink_window, arena_shortlist_policy, ] arena_preset_button_outputs = [arena_safe, arena_balanced, arena_fragile, arena_preset_status] for button, key in ( (arena_safe, "safe_match"), (arena_balanced, "balanced_match"), (arena_fragile, "fragile_match"), ): button.click( fn=lambda current_model, preset_key=key: _apply_arena_preset(current_model, preset_key), inputs=[arena_model], outputs=arena_preset_outputs, ).then( fn=lambda: (gr.update(selected="preset"), "preset"), outputs=[arena_prompt_tabs, arena_prompt_mode], show_progress="hidden", queue=False, ).then( fn=_refresh_arena_prompt_mode, inputs=[arena_prompt_mode, arena_context_length, arena_custom_prompt], outputs=[arena_context_length, arena_live_context_note, arena_start_button], show_progress="hidden", queue=False, ).then( fn=lambda preset_key=key: _arena_preset_ui(preset_key), outputs=arena_preset_button_outputs, ) demo.load( fn=_initial_page_state, outputs=[ workflow_nav, compare_panel, arena_panel, fastest_safe, balanced, aggressive, preset_status, mode, page_size, bits_k, bits_v, recent_window, sink_window, shortlist_policy, arena_safe, arena_balanced, arena_fragile, arena_preset_status, arena_mode, arena_page_size, arena_bits_k, arena_bits_v, arena_recent_window, arena_sink_window, arena_shortlist_policy, ], show_progress="hidden", queue=False, ) if __name__ == "__main__": launch_kwargs = { "share": _env_flag("DOTCACHE_SPACE_SHARE", default=False), "server_name": os.getenv("DOTCACHE_SPACE_HOST"), "server_port": _env_int("DOTCACHE_SPACE_PORT"), "show_error": True, } if "css" in inspect.signature(demo.launch).parameters: launch_kwargs["css"] = CSS if "ssr_mode" in inspect.signature(demo.launch).parameters: launch_kwargs["ssr_mode"] = False demo.launch(**launch_kwargs)