import os
# Triton saves its kernel tuning to disk, so a new ZeroGPU worker reuses what the start-up warm-up measured
# instead of re-tuning (about 15 s on its first step). Must be set before torch/triton are imported.
os.environ.setdefault("TRITON_CACHE_AUTOTUNING", "1")
from deskforge import (DEFAULT_FAMILY, FAMILIES, annotate, parse_action, predict, # imports spaces first
to_pixels)
import web_models # the web agent's planner (Qwen3.5-9B), resident next to the duel models
import base64
import html
import io
import random
import tempfile
import threading
import time
from pathlib import Path
import gradio as gr
from PIL import Image
import setup_bar
import gpu_bridge
import shot_tool
import web_tab
from desktop_local import THEME_PRESETS, LocalDesktop, describe_theme, random_theme
from duel import CHALLENGES, TOTALS, play
from setup_bar import LOOK_SHUFFLE, RANDOM_TASK
from web_agent import WebAgent
MAX_STEPS = 8
HERE = Path(__file__).parent
desktop = LocalDesktop().start()
desktop.reset("Calculator (Galculator)")
web_agent = WebAgent(web_models.decide) # a headless Chromium, separate from the Xfce desktop
_shot_lock = threading.Lock()
_last_shot = (0.0, None)
def live_view():
"""Current desktop frame, shared by every viewer (at most ~3 grabs/second)."""
global _last_shot
with _shot_lock:
t, img = _last_shot
if img is None or time.time() - t > 0.3:
img = desktop.screenshot()
_last_shot = (time.time(), img)
return img
# ---------------------------------------------------------------------------
# Static pieces: top bar (with the view switch) and the welcome message
# ---------------------------------------------------------------------------
LOGO = """{body} ')
VIEW_ICON = {
"web": _icon('
{VIEW_ICON['web']}Web agent| {VIEW_HINT['web']}
{VIEW_ICON['live']}Live duel| {VIEW_HINT['live']}
{VIEW_ICON['shot']}Try a screenshot| {VIEW_HINT['shot']}
Paper
Dataset
Project
"""
TOPBAR_JS = """
element.addEventListener('click', e => {
const b = e.target.closest('[data-view]');
if (!b) return;
element.querySelectorAll('[data-view]').forEach(x => x.classList.toggle('on', x === b));
const views = { web: ['df-view-web'], live: ['df-view-live', 'df-side'], shot: ['df-view-shot'] };
for (const [view, ids] of Object.entries(views)) for (const id of ids) {
const el = document.getElementById(id);
if (el) el.style.display = view === b.dataset.view ? (id === 'df-view-shot' ? 'block' : 'flex') : 'none';
}
});
"""
INTRO = """
What is this? DeskForge is a small AI model trained to find and click the right
button on a computer screen. Here it races the model it was trained from, on a real Linux desktop running in
this page.
Pick a model, a look and a task at the top, then press Start . Each step appears here:
the instruction, where both models clicked, and whether they hit the right button (the green
box , read from the app itself). DeskForge's click is the one that runs.
One desktop is shared by everyone here, so runs take turns. It has no browser, terminal or internet.
"""
# ---------------------------------------------------------------------------
# Rendering: score and the step-by-step chat
# ---------------------------------------------------------------------------
def _pct(hits, n):
return f"{100 * hits / n:.0f}%" if n else "–"
def render_score(family=DEFAULT_FAMILY, score=None):
score = score or {"deskforge": [0, 0], "base": [0, 0]}
def side(which, name, cls):
hits, n = score[which]
all_hits, all_n = TOTALS[family][which]
return (f'
'
f'
{hits} / {n}
'
f'
{_pct(all_hits, all_n)} for all visitors
{side("deskforge", f"DeskForge {family}", "df-df")}'
f'{side("base", FAMILIES[family].base_label, "df-base")}
')
def _crop(image, step):
"""A zoomed crop centred on the true target (or DeskForge's point when unscored), as a data URI."""
if step.box:
x0, y0, x1, y1 = step.box
cx, cy, span = (x0 + x1) / 2, (y0 + y1) / 2, max(x1 - x0, (y1 - y0) * 2.2)
else:
p = to_pixels(parse_action(step.codes["deskforge"]), image.width, image.height)
if not p:
return None
cx, cy, span = p[0], p[1], 0
w = min(image.width, max(380, span * 3))
h = min(image.height, w / 2.2)
left = int(min(max(cx - w / 2, 0), image.width - w))
top = int(min(max(cy - h / 2, 0), image.height - h))
crop = image.crop((left, top, left + int(w), top + int(h)))
crop.thumbnail((720, 330))
buf = io.BytesIO()
crop.convert("RGB").save(buf, format="JPEG", quality=85)
return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
def _result(which, step, family):
name = "DeskForge" if which == "deskforge" else FAMILIES[family].base_label
hit = step.hits.get(which)
verdict = ('✓ hit ' if hit
else '✗ miss ' if hit is False else "")
code = html.escape(step.codes[which].replace("pyautogui.", "").replace("computer.", ""))
secs = step.times.get(which)
return f"""
{html.escape(name)} {verdict}{f'{secs:.1f}s' if secs else ''}
{code}
"""
def render_feed(steps=(), crops=None, status=None, running=False, error=None, family=DEFAULT_FAMILY):
crops = crops or {}
parts = [] if steps or status else [INTRO]
if status:
parts.append(f'{status}
')
for s in steps:
chip = f'{html.escape(s.note.replace("Theme → ", "New look: "))}
' if s.note.startswith("Theme") else ""
parts.append(f'{chip}
{html.escape(s.instruction)}
' + _result("deskforge", s, family) + _result("base", s, family) + "
")
if "referee" in s.note:
parts.append('DeskForge missed, so the referee clicked the right button to keep going.
')
elif "not scored" in s.note:
parts.append('That button was not on screen, so this step is not scored.
')
if error:
parts.append(f'The run stopped: {html.escape(error)}
')
if running:
parts.append('')
return '' + "".join(parts) + "
"
# ---------------------------------------------------------------------------
# Running duels
# ---------------------------------------------------------------------------
def _apply_look(look):
return desktop.apply_theme(random_theme() if look == LOOK_SHUFFLE else THEME_PRESETS[look])
def run_challenge(family, name, look, roulette=False, tokens=128):
if name == RANDOM_TASK:
name = random.choice(list(CHALLENGES))
challenge = CHALLENGES[name]
status = f"Task: {html.escape(name)} · {html.escape(family)}"
yield render_feed(status=status, running=True, family=family), render_score(family), gr.skip()
desktop.reset(challenge.scene)
_apply_look(look)
yield from _stream(family, status, play(desktop, challenge.steps, family, roulette, tokens))
def run_free(family, scene, steps_text, look, roulette=False, tokens=128):
lines = [s.strip() for s in (steps_text or "").splitlines() if s.strip()]
if not lines:
raise gr.Error("Write at least one instruction, one per line.")
if len(lines) > MAX_STEPS:
gr.Warning(f"Only the first {MAX_STEPS} instructions will run.")
lines = lines[:MAX_STEPS]
status = f"Your instructions on {html.escape(scene)} · not scored"
yield render_feed(status=status, running=True, family=family), render_score(family), gr.skip()
desktop.reset(scene)
_apply_look(look)
yield from _stream(family, status, play(desktop, [(l, None) for l in lines], family, roulette, tokens))
def _stream(family, status, duel):
"""Relay duel steps to the chat; on failure (e.g. GPU quota) keep what ran and say why."""
steps, score, crops = [], None, {}
try:
for image, steps, score in duel:
if steps and steps[-1].n not in crops and image is not None:
crops[steps[-1].n] = _crop(image, steps[-1])
yield (render_feed(steps, crops, status, running=True, family=family), render_score(family, score),
describe_theme(desktop.theme))
yield render_feed(steps, crops, status, family=family), render_score(family, score), describe_theme(desktop.theme)
except Exception as e:
message = gpu_bridge.friendly_error(str(e) or type(e).__name__)
gr.Warning(f"The run stopped: {message}")
yield (render_feed(steps, crops, status, error=message, family=family), render_score(family, score),
describe_theme(desktop.theme))
# ---------------------------------------------------------------------------
# Single screenshot grounding (the original demo)
# ---------------------------------------------------------------------------
EXAMPLES = [
("homebank_file_menu.png", "Select the Export as QIF... option from the File menu."),
("qalculate_factorial.png", "Add the factorial function to the current expression."),
("calculator_financial_mode.png", "Select Financial Mode in the calculator's mode dropdown."),
("mousepad_encoding_dropdown.png", "Open the file encoding dropdown to view available encoding options."),
("image_viewer_menu.png", "Open the Image Viewer menu."),
("applications_menu.png", "Open the Applications menu to view application categories."),
("nautilus_show_sidebar.png", "Enable the sidebar in the file manager."),
("transmission_sort_by_size.png", "Select the Sort by Size option in the Transmission View menu."),
]
gr.set_static_paths(paths=[HERE / "examples"])
def ground(instruction: str, screenshot, max_new_tokens: int = 128, family: str = DEFAULT_FAMILY):
"""Predict the next computer-use action for a desktop screenshot.
Args:
instruction: the task to perform on the screen, e.g. "Open the File menu".
screenshot: a PIL image of the desktop to act on.
max_new_tokens: generation length cap for the emitted action.
family: which DeskForge model to use, "Gemma4-E4B" or "Qwen3.5-4B".
Returns:
Tuple of (annotated screenshot with the predicted click point,
predicted action code, fractional coordinates, inference seconds).
"""
if screenshot is None:
raise gr.Error("Please provide a desktop screenshot first.")
if not instruction or not instruction.strip():
raise gr.Error("Please enter an instruction, e.g. 'Open the File menu'.")
if family not in FAMILIES:
raise gr.Error(f"Unknown model family {family!r}; pick one of {', '.join(FAMILIES)}.")
screenshot = screenshot.convert("RGB")
code, elapsed = predict(instruction, screenshot, int(max_new_tokens), family)
action = parse_action(code)
coords = f"x={action.x:.4f}, y={action.y:.4f}" if action.has_point else "— (no point action)"
return annotate(screenshot, action), code, coords, f"{elapsed:.1f}s"
def ground_screenshot(args: list) -> dict:
"""Server function behind the "Try a screenshot" view; the script's arguments arrive as one list."""
path, instruction, family = (list(args) + ["", "", DEFAULT_FAMILY])[:3]
if path.startswith("example:"):
name = path.split(":", 1)[1]
if name not in dict(EXAMPLES):
raise gr.Error("Unknown example.")
file = HERE / "examples" / name
else:
file = Path(path).resolve()
if not file.is_relative_to(Path(tempfile.gettempdir()).resolve()): # only files uploaded to this app
raise gr.Error("Please upload the screenshot again.")
image, code, coords, secs = ground(instruction, Image.open(file), 128, family)
image.thumbnail((1400, 1400))
buf = io.BytesIO()
image.convert("RGB").save(buf, format="JPEG", quality=88)
return {"image": "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode(),
"code": code, "coords": coords, "time": secs}
# ---------------------------------------------------------------------------
# Theme and styles (palette, type and components from the Docling Agent design)
# ---------------------------------------------------------------------------
HEAD = """
Desktop look: {html.escape(text)}
'
with gr.Blocks(title="DeskForge · Live Desktop Duel", fill_width=True) as demo:
gr.HTML(TOPBAR, js_on_load=TOPBAR_JS, apply_default_css=False)
with gr.Row(elem_id="df-shell"):
with gr.Column(elem_id="df-main", scale=1):
with gr.Column(elem_id="df-view-web"):
gr.HTML(None, html_template=web_tab.TEMPLATE, js_on_load=gpu_bridge.CLIENT + web_tab.SCRIPT,
server_functions=web_agent.server_functions(), suggestions=web_tab.SUGGESTIONS,
logo=web_tab.LOGO, apply_default_css=False)
with gr.Column(elem_id="df-view-live"):
setup = gr.HTML(SETUP_DEFAULT, html_template=setup_bar.TEMPLATE, js_on_load=setup_bar.SCRIPT,
elem_id="df-setup", apply_default_css=False, **SETUP_PROPS)
with gr.Column(elem_classes="df-canvas"):
live = gr.Image(show_label=False, format="jpeg", interactive=False, container=False,
elem_classes="df-sheet", buttons=[])
look_now = gr.HTML(_caption(describe_theme(desktop.theme)), apply_default_css=False)
with gr.Column(elem_id="df-view-shot"):
gr.HTML(None, html_template=shot_tool.TEMPLATE, js_on_load=gpu_bridge.CLIENT + shot_tool.SCRIPT,
server_functions=[ground_screenshot], models=MODELS, apply_default_css=False,
examples=[{"name": n, "text": t, "url": f"gradio_api/file={HERE / 'examples' / n}"} for n, t in EXAMPLES])
with gr.Column(elem_id="df-side"):
gr.HTML('Duel, step by step
', apply_default_css=False)
score = gr.HTML(render_score(), elem_id="df-score-wrap", apply_default_css=False)
feed = gr.HTML(render_feed(), elem_id="df-feed-wrap", apply_default_css=False)
# The original grounding endpoint (API and MCP), kept without any visible UI.
with gr.Column(visible=False):
api_image, api_text = gr.Image(type="pil"), gr.Textbox()
api_tokens, api_family = gr.Number(128), gr.Textbox(DEFAULT_FAMILY)
api_out = [gr.Image(type="pil"), gr.Textbox(), gr.Textbox(), gr.Textbox()]
api_btn = gr.Button()
api_btn.click(ground, [api_text, api_image, api_tokens, api_family], api_out, api_name="predict")
settings = gr.State(SETUP_DEFAULT)
def _relay(gen):
for feed_html, score_html, look_text in gen:
yield feed_html, score_html, _caption(look_text) if isinstance(look_text, str) else look_text
def on_change(evt: gr.EventData):
st = _settings(evt)
return st, render_score(st["model"])
def on_look(evt: gr.EventData):
return _caption(describe_theme(_apply_look(_settings(evt)["look"])))
def on_start(evt: gr.EventData):
st = _settings(evt)
if (st.get("own") or "").strip():
task = random.choice(list(CHALLENGES)) if st["random"] else st["task"]
gen = run_free(st["model"], CHALLENGES[task].scene, st["own"], st["look"], st["roulette"], int(st["tokens"]))
else:
task = RANDOM_TASK if st["random"] else st["task"]
gen = run_challenge(st["model"], task, st["look"], st["roulette"], int(st["tokens"]))
yield from _relay(gen)
def duel_api(family: str, task: str, look: str = LOOK_SHUFFLE) -> tuple[str, str]:
"""Run a scored challenge on the shared desktop; returns the final chat HTML and score HTML."""
feed_html = score_html = ""
for feed_html, score_html, _ in run_challenge(family, task, look):
pass
return feed_html, score_html
gr.Timer(1.0).tick(live_view, outputs=live, show_progress="hidden", concurrency_limit=None,
api_visibility="private")
demo.load(live_view, outputs=live, api_visibility="private")
demo.load(lambda: render_score(DEFAULT_FAMILY), outputs=score, api_visibility="private")
# GPU work requested by the custom views runs here, as a real Gradio event (see gpu_bridge.py).
web_fns = {fn.__name__: fn for fn in web_agent.server_functions()}
gpu_ops = {
"web_plan": lambda d: web_fns["web_plan"](d.get("run", "")),
"ground_screenshot": lambda d: ground_screenshot([d.get("src", ""), d.get("text", ""), d.get("model", DEFAULT_FAMILY)]),
}
bridge = gr.HTML(None, html_template=gpu_bridge.TEMPLATE, js_on_load=gpu_bridge.SCRIPT, elem_id="df-gpu-bridge",
apply_default_css=False)
bridge.submit(gpu_bridge.handler(gpu_ops), None, bridge, concurrency_limit=6, concurrency_id="gpu-bridge",
api_visibility="private", show_progress="hidden", queue=True)
setup.change(on_change, None, [settings, score], queue=False, api_visibility="private")
setup.input(on_look, None, look_now, concurrency_id="desktop", api_visibility="private")
start_event = setup.submit(on_start, None, [feed, score, look_now], concurrency_limit=1,
concurrency_id="desktop", api_visibility="private", show_progress="hidden")
setup.stop(None, cancels=[start_event])
gr.api(duel_api, api_name="duel", concurrency_limit=1, concurrency_id="desktop")
if __name__ == "__main__":
demo.launch(mcp_server=True, theme=THEME, css=CSS, head=HEAD)