"""Hugging Face Space: watch an agent play Prime Rush through MCP tools.
Honest about what it is. The six tools, their schemas, the scoring and the
outcome file are the real ones - this Space imports the same ``indus_mcp``
package a lab would run. What it is NOT is the real game: a Space has no
Unity server, so the tools here drive ``tests/fake_game.py``, the stand-in
that speaks the same bridge protocol and reproduces the real game's response
characteristics, so behaviour here matches behaviour there.
The map is drawn from the tool responses alone - position, circle centre,
circle radius - which is the point: everything the viewer sees is everything
the agent was told. Nothing is rendered that the agent did not have.
"""
from __future__ import annotations
import html
import json
import math
import os
import sys
import tempfile
from pathlib import Path
import gradio as gr
from findings import FINDINGS, FINDINGS_CSS
sys.path.insert(0, str(Path(__file__).resolve().parent))
BUDGET = 100.0
INK, INK3 = "#15161a", "#8a8d94"
BLUE, ORANGE = "#2a78d6", "#eb6834"
GRID, SURF = "#e8e6e1", "#ffffff"
# ---------------------------------------------------------------- the map
def render_map(path: list[tuple[float, float]], sit: dict | None) -> str:
"""Top-down view, drawn only from what the tools reported."""
W = H = 420
if not sit:
return (f'')
cx, cz = sit["zone_center_x"], sit["zone_center_z"]
r = max(sit["zone_radius"], 1.0)
nr = max(sit.get("next_zone_radius") or r, 1.0)
px, pz = sit["position_x"], sit["position_z"]
# Frame the circle, the next circle and the whole path walked so far.
xs = [cx - r, cx + r, px] + [p[0] for p in path]
zs = [cz - r, cz + r, pz] + [p[1] for p in path]
pad = max(40.0, 0.08 * max(max(xs) - min(xs), max(zs) - min(zs)))
x0, x1 = min(xs) - pad, max(xs) + pad
z0, z1 = min(zs) - pad, max(zs) + pad
span = max(x1 - x0, z1 - z0, 1.0)
x0 = (x0 + x1) / 2 - span / 2
z0 = (z0 + z1) / 2 - span / 2
k = W / span
def sx(x: float) -> float: return (x - x0) * k
def sy(z: float) -> float: return H - (z - z0) * k # +z is up
inside = bool(sit.get("inside_safe_zone"))
trail = " ".join(f"{sx(a):.1f},{sy(b):.1f}" for a, b in path)
ncx = sit.get("next_zone_center_x", cx)
ncz = sit.get("next_zone_center_z", cz)
bar = 100.0
while bar * k < 48:
bar *= 2
grid = "".join(
f''
f''
for i in range(1, 6))
return f""""""
def render_calls(items: list[dict]) -> str:
if not items:
return '
tool calls appear here
'
rows = []
for it in items[-16:]:
args = ", ".join(f"{k}={v}" for k, v in (it["args"] or {}).items())
cls = "pr-call pr-free" if it["free"] else "pr-call"
rows.append(
f'
")
# ---------------------------------------------------------------- the agent
def _fresh_session(port: int, seed: int):
"""Point the MCP session at a new match. The session is a singleton by
design - one server, one match - so a Space that plays repeatedly has to
retire it between runs."""
import indus_mcp.session as S
outcome = Path(tempfile.gettempdir()) / f"space-outcome-{seed}.json"
os.environ.update(
INDUS_BRIDGE_HOST="127.0.0.1", INDUS_BRIDGE_PORT=str(port),
INDUS_TASK="zone_survival", INDUS_BUDGET_SECONDS=str(BUDGET),
INDUS_FF_RADIUS="350", INDUS_OUTCOME_PATH=str(outcome),
)
S._SESSION = None
return S.session()
def _start_fake(seed: int, port: int):
import random as _r
from tests.fake_game import FakeGame
rng = _r.Random(seed)
return FakeGame(port=port,
zone_center=(rng.uniform(-400, 400), rng.uniform(-400, 400)),
radius0=2000.0, shrink_per_tick=0.3).start()
def scripted_agent(sit: dict) -> list[tuple[str, dict]]:
"""Work out the bearing, face it, run. The whole policy, and the control
every trained policy here is measured against."""
dx = sit["zone_center_x"] - sit["position_x"]
dz = sit["zone_center_z"] - sit["position_z"]
bearing = math.degrees(math.atan2(dx, dz)) % 360
if sit["inside_safe_zone"]:
return [("hold_position", {"seconds": 10})]
return [("turn_to_bearing", {"bearing": round(bearing, 1)}),
("move", {"seconds": 12, "forward": 1.0, "run": True})]
async def play(seed: int):
from fastmcp import Client
from indus_mcp.main import mcp
seed = int(seed)
port = 9100 + (seed % 200)
game = _start_fake(seed, port)
session = _fresh_session(port, seed)
feed: list[dict] = []
path: list[tuple[float, float]] = []
sit: dict | None = None
def frame():
return render_map(path, sit), render_calls(feed), render_stats(session.outcome())
try:
async with Client(mcp) as c:
for _ in range(40):
r = await c.call_tool("get_situation", {})
sit = r.structured_content
path.append((sit["position_x"], sit["position_z"]))
feed.append({"name": "get_situation", "args": {}, "free": True,
"text": (r.content[0].text if r.content else "").strip()})
yield frame()
if session.finished:
break
for name, args in scripted_agent(sit):
r = await c.call_tool(name, args)
d = r.structured_content or {}
if d.get("position_x") is not None:
path.append((d["position_x"], d["position_z"]))
sit = _merge(sit, d, False)
feed.append({"name": name, "args": args, "free": False,
"text": (r.content[0].text if r.content else "").strip()})
yield frame()
if session.finished:
break
if session.finished:
break
if not session.finished:
session.finish("out of turns")
yield frame()
finally:
game.stop()
# ---------------------------------------------------------------- leaderboard
RESULTS = Path(__file__).resolve().parent / "results" / "summary.json"
def _board_table(d: dict) -> str:
"""One task's table: agents across the top, measures down the side.
Reading a column tells you about one agent, reading a row compares them
on one thing, and the best in each row is bold. The footnotes say what
was held constant, because a number without them is not a result.
"""
rows = sorted(d.get("rows", []), key=lambda r: -r.get("score", 0))
if not rows:
return ""
def label(r):
n = r["model"].split(":", 1)[-1].split("/")[-1]
return f"{n}1" if r.get("reference") else n
measures = [
("Score", d.get("measure", "Task score"), lambda r: f'{r["score"]*100:.1f}', True),
("95% interval", "Bootstrap over the scored zones",
lambda r: (f'{r["ci95"][0]*100:.1f} – {r["ci95"][1]*100:.1f}'
if r.get("ci95") else "—"), False),
("Median", "Per-zone median", lambda r: f'{r.get("median", 0)*100:.1f}', True),
("Scored above 50%", "Zones", lambda r: str(r.get("reached", "—")), True),
("Decisions", "Tool calls per episode",
lambda r: (f'{r["tool_calls_mean"]:.1f}' if r.get("tool_calls_mean")
else "—"), False),
]
head = "
" + "".join(f"
{label(r)}
" for r in rows) + "
"
body = []
for name, gloss, fmt, mark in measures:
vals = [fmt(r) for r in rows]
best = -1
if mark:
try:
best = [float(v.split("/")[0]) for v in vals].index(
max(float(v.split("/")[0]) for v in vals))
except ValueError:
best = -1
cells = "".join(f'
{"" + v + "" if i == best else v}
'
for i, v in enumerate(vals))
body.append(f'
{name}{gloss}
{cells}
')
title = d.get("task", "task").replace("_", " ")
meta = html.escape(d.get("note", ""))
gloss = d.get("describe", "")
return (f'
{title}
'
+ (f'
{gloss}
' if gloss else "")
+ f'
{head}{"".join(body)}
'
+ (f'
{meta}
' if meta else ""))
def render_board() -> str:
"""Read whatever has been published into results/summary.json.
The file is absent until there are numbers worth standing behind, and the
page says so rather than showing an empty table. Publishing is a commit
of one file, not a change to this app.
"""
if not RESULTS.exists():
return ('
'
'
No scores published yet.
'
'
A ranking is only worth reading once every zone is '
'winnable. The numbers go here when they are.
')
d = json.loads(RESULTS.read_text(encoding="utf-8"))
boards = d if isinstance(d, list) else d.get("tasks") or [d]
notes = [
"The best result in each row is shown in bold, where a higher number is "
"better. — indicates not applicable: the reference policies do not "
"call tools.",
"Every agent is driven by one tool-calling loop; the only per-provider "
"difference is the endpoint.",
"Episodes are truncated at the action budget rather than at match end, so a "
"score is not a win rate.",
"Intervals are wide at these sample sizes. Rows whose intervals overlap are "
"not separated by this data, whatever their order in the table.",
]
return ("".join(_board_table(b) for b in boards)
+ '' + "".join(f"
{n}
" for n in notes) + ""
+ '
1 Reference policies, not models: a '
"twelve-line hand-written policy, and uniform random actions as the "
"floor. A benchmark whose floor is high is not measuring anything.
")
BOARD_INTRO = """
Leaderboard
Every row plays the same zones through the same six tools and is scored by the same loop.
Rows marked as references are not models: a random policy, a short hand-written one, and a
policy trained in the environment, included so the model scores have a floor and a
comparison.
"""
# ---------------------------------------------------------------- replays
REPLAY_DIR = Path(__file__).resolve().parent / "replays"
# Only get_situation reports the circle. Movement tools report the agent and a
# couple of derived fields, so merging everything from every call leaves a
# stale next-circle sitting beside a fresh current one - which draws a "next"
# ring larger than the one it is shrinking towards.
ZONE_FIELDS = ("zone_center_x", "zone_center_z", "zone_radius", "zone_round",
"next_zone_center_x", "next_zone_center_z", "next_zone_radius")
def _merge(sit: dict | None, result: dict, is_situation: bool) -> dict:
keep = {k: v for k, v in result.items()
if v is not None and (is_situation or k not in ZONE_FIELDS)}
return {**(sit or {}), **keep}
def _replays() -> dict[str, Path]:
"""Recorded episodes, newest naming first. Each file is the trajectory a
real run wrote: the model, the zone, every tool call and its response."""
out = {}
for f in sorted(REPLAY_DIR.glob("*.json")) if REPLAY_DIR.exists() else []:
try:
d = json.loads(f.read_text(encoding="utf-8"))
except ValueError:
continue
o = d.get("outcome") or {}
out[f"{d.get('model', '?')} โ zone {d.get('seed', '?')} โ "
f"{o.get('frac_inside_zone', 0)*100:.0f}% inside"] = f
return out
async def play_replay(label: str):
"""Step through a recorded episode at reading speed.
This is a real run against a real game server, not the stand-in: the
positions and the tool responses are what the model actually got.
"""
import asyncio
files = _replays()
f = files.get(label)
if f is None:
yield render_map([], None), render_calls([]), render_stats(None)
return
d = json.loads(f.read_text(encoding="utf-8"))
calls, path, sit, feed = d.get("calls", []), [], None, []
o = d.get("outcome") or {}
for c in calls:
r = c.get("result") or {}
if r.get("position_x") is not None:
path.append((r["position_x"], r["position_z"]))
sit = _merge(sit, r, c.get("tool") == "get_situation")
feed.append({"name": c.get("tool", "?"), "args": c.get("input") or {},
"free": c.get("tool") == "get_situation",
"text": _summarise(c)})
if sit and sit.get("zone_center_x") is not None:
yield render_map(path, sit), render_calls(feed), render_stats({})
await asyncio.sleep(0.45)
yield render_map(path, sit), render_calls(feed), render_stats(o)
def _summarise(call: dict) -> str:
"""The transcripts store structured results, not the text block, so
rebuild a readable line rather than dumping json at the viewer."""
r = call.get("result") or {}
bits = []
if r.get("position_x") is not None:
bits.append(f"at x={r['position_x']}, z={r['position_z']}")
if r.get("facing_yaw") is not None:
bits.append(f"facing {r['facing_yaw']}")
if r.get("distance_travelled") is not None:
bits.append(f"moved {r['distance_travelled']} units")
if r.get("degrees_off") is not None:
bits.append(f"{r['degrees_off']} off the bearing asked for")
if r.get("inside_safe_zone") is not None:
bits.append("inside the circle" if r["inside_safe_zone"] else "outside the circle")
if r.get("zone_radius") is not None:
bits.append(f"radius {r['zone_radius']}")
if r.get("seconds_remaining") is not None:
bits.append(f"{r['seconds_remaining']}s left")
return " ยท ".join(bits) or "(no fields)"
WATCH_INTRO = """
Recorded runs
Episodes from real matches on a real game server. The positions and the tool responses are
what the model actually received. Pick one and it plays back a step at a time.
"""
NO_REPLAYS = """
No recordings published yet.
Runs against the real game go here. The tab beside this one drives the
same tools against a stand-in, which needs no game server.
An agent plays a battle-royale match through six tools. A safe circle
shrinks and the agent has to be inside it. It is given its own coordinates and the
circle’s, and has to work out the direction itself.
six toolsruns under ArchipelagoApache-2.0
"""
WHY = """
Timing
The match only advances inside a tool call, by the amount that call asks for. A model can
take as long as it likes to decide and the game waits. The same seed and the same calls
give the same match.
What the agent gets
`get_situation` returns coordinates: where it is, which way it is facing, where the circle
is and how big. There is no `move_to` and no bearing. Movement is relative to the direction
it is facing, so it has to turn first.
"""
NOTE = """
Scoring
Each call that advances the match writes `outcome.json` from the game state. The score is
read from that file, not from the transcript, and a run that stops early is still scored on
what it did. The task pack and a reference grader are in the Files tab.
What this page runs
The tools, their schemas and the scoring are the ones a lab would run. The game is not: a
Space has no Unity server, so the tools drive a stand-in that speaks the same protocol. The
map shows only what the tools returned.
"""
FOOT = """
SuperGaming · code, task pack and tests in the Files tab
· Apache-2.0
"""
with gr.Blocks(title="Prime Rush โ a battle royale as an MCP environment") as demo:
gr.HTML(HERO)
_reps = _replays()
with gr.Tab("Watch a model play"):
gr.Markdown(WATCH_INTRO)
if _reps:
with gr.Row():
with gr.Column(scale=3):
rmap = gr.HTML(render_map([], None))
rstats = gr.HTML(render_stats(None))
with gr.Row():
pick = gr.Dropdown(list(_reps), value=list(_reps)[0],
label="Episode", scale=3)
rgo = gr.Button("Play", variant="primary", scale=0)
with gr.Column(scale=2):
gr.HTML('
What the model called
')
rfeed = gr.HTML(render_calls([]))
rgo.click(play_replay, inputs=[pick], outputs=[rmap, rfeed, rstats])
else:
gr.HTML(NO_REPLAYS)
with gr.Tab("Try the tools"):
with gr.Row():
with gr.Column(scale=3):
gr.HTML('
The match
')
map_out = gr.HTML(render_map([], None))
stats_out = gr.HTML(render_stats(None))
with gr.Row():
seed = gr.Slider(2000, 2029, value=2004, step=1, label="Zone layout (seed)")
go = gr.Button("Play an episode", variant="primary", scale=0)
with gr.Column(scale=2):
gr.HTML('
What the agent called
')
feed_out = gr.HTML(render_calls([]))
with gr.Row():
with gr.Column():
gr.Markdown(WHY)
with gr.Column():
gr.Markdown(NOTE)
with gr.Tab("Findings"):
gr.HTML(FINDINGS)
with gr.Tab("Leaderboard"):
gr.Markdown(BOARD_INTRO)
gr.HTML(render_board())
gr.HTML(FOOT)
go.click(play, inputs=[seed], outputs=[map_out, feed_out, stats_out])
if __name__ == "__main__":
demo.launch(css=CSS, theme=gr.themes.Soft(
primary_hue="blue", neutral_hue="slate",
font=["ui-sans-serif", "system-ui", "sans-serif"]))