"""Papers Reproducibility demo โ Gradio Space.
Three tabs:
- Browse: see papers that have been (or are being) reproduced, with the
generated report and an interactive trace of how the run went.
- Request: ask for a paper to be reproduced. Goes in as "pending".
- Admin: password-gated approve/reject queue. Approving only flips a
status flag โ the actual agent run happens on a separate machine via
local_runner.py, which polls this same store. See README.md.
"""
from __future__ import annotations
import hmac
import html as html_lib
import json
import os
from datetime import date, datetime, timedelta, timezone
import gradio as gr
import daily_scan
import slack_bot
import store
from trace_view import render_trace
from dotenv import load_dotenv
load_dotenv()
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD")
SCAN_TRIGGER_TOKEN = os.environ.get("SCAN_TRIGGER_TOKEN")
DAILY_SCAN_HOUR = int(os.environ.get("DAILY_SCAN_HOUR", "8"))
REFRESH_SECONDS = 6
STATUS_ICON = {
"pending": "โณ",
"approved": "๐๏ธ",
"running": "๐",
"completed": "โ
",
"failed": "โ",
"rejected": "๐ซ",
}
# (text color, tinted background) โ semi-transparent backgrounds so these
# read fine on both light and dark Gradio themes without separate palettes.
STATUS_COLORS = {
"pending": ("#6b7280", "rgba(107,114,128,0.16)"),
"approved": ("#2563eb", "rgba(37,99,235,0.16)"),
"running": ("#b45309", "rgba(180,83,9,0.16)"),
"completed": ("#0d9488", "rgba(13,148,136,0.16)"),
"failed": ("#b91c1c", "rgba(185,28,28,0.16)"),
"rejected": ("#6b7280", "rgba(107,114,128,0.16)"),
}
THEME = gr.themes.Soft(
primary_hue="teal",
neutral_hue="slate",
font=[gr.themes.GoogleFont("Inter"), "-apple-system", "sans-serif"],
)
CUSTOM_CSS = """
.app-header { padding: 4px 2px 18px; margin-bottom: 6px; border-bottom: 1px solid var(--border-color-primary); }
.app-header-title { font-size: 1.7rem; font-weight: 800; letter-spacing: -0.02em; display: flex; align-items: center; gap: 10px; }
.app-header-sub { color: var(--body-text-color-subdued); margin-top: 4px; font-size: 0.95rem; }
.section-card {
border: 1px solid var(--border-color-primary) !important;
border-radius: 14px !important;
background: var(--background-fill-secondary) !important;
padding: 18px !important;
}
.job-gallery { display: grid; grid-template-columns: repeat(auto-fill, minmax(250px, 1fr)); gap: 12px; margin: 4px 0 2px; }
.job-card {
border: 1px solid var(--border-color-primary); border-radius: 12px; padding: 12px 14px;
background: var(--background-fill-primary);
}
.job-card.selected { border-color: #0d9488; box-shadow: 0 0 0 1px #0d9488 inset; }
.job-card-clickable { cursor: pointer; transition: border-color 0.1s ease, box-shadow 0.1s ease; }
.job-card-clickable:hover { border-color: #0d9488; box-shadow: 0 0 0 1px rgba(13,148,136,0.35) inset; }
.job-card h4 { margin: 8px 0 6px; font-size: 0.95rem; font-weight: 700; line-height: 1.3; }
.job-card .job-card-links { font-size: 0.82rem; }
.job-card .job-card-links a { color: #0d9488; text-decoration: none; margin-right: 10px; }
.job-card .job-card-links a:hover { text-decoration: underline; }
.job-card .job-card-meta { font-size: 0.76rem; color: var(--body-text-color-subdued); margin-top: 6px; }
.job-card .job-card-error { font-size: 0.78rem; color: #b91c1c; margin-top: 6px; }
.status-badge {
display: inline-flex; align-items: center; gap: 4px; padding: 2px 10px;
border-radius: 999px; font-size: 0.74rem; font-weight: 700; letter-spacing: 0.01em;
}
.mode-chip {
display: inline-block; padding: 1px 9px; border-radius: 999px; font-size: 0.72rem;
background: var(--background-fill-secondary); border: 1px solid var(--border-color-primary);
color: var(--body-text-color-subdued); margin-left: 6px;
}
.empty-state { padding: 28px; text-align: center; color: var(--body-text-color-subdued); border: 1px dashed var(--border-color-primary); border-radius: 12px; }
.detail-header { padding: 2px 2px 14px; }
.detail-header h3 { margin: 6px 0 8px; }
.detail-links a { color: #0d9488; text-decoration: none; margin-right: 4px; }
.detail-links a:hover { text-decoration: underline; }
.report-frame { width: 100%; height: 800px; border: 1px solid var(--border-color-primary); border-radius: 10px; }
"""
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def status_badge(status: str) -> str:
icon = STATUS_ICON.get(status, "")
text_color, bg = STATUS_COLORS.get(status, ("#6b7280", "rgba(107,114,128,0.16)"))
return f'{icon} {status}'
def _job_label(job: dict) -> str:
return f"{STATUS_ICON.get(job['status'], '')} {job['title']}"
def _safe_list_jobs() -> list[dict]:
try:
return store.list_jobs()
except Exception as exc: # store not configured yet, or transient HF error
gr.Warning(f"Could not reach the job store: {exc}")
return []
def _job_links_html(job: dict) -> str:
# onclick stopPropagation so these links don't also trigger the card's
# own click-to-select behavior underneath them.
stop = "onclick='event.stopPropagation()'"
links = [f"paper", f"code"]
if job.get("data_url"):
links.append(f"data")
return " ยท ".join(links)
def job_card_html(job: dict, selected: bool = False) -> str:
cls = "job-card job-card-clickable selected" if selected else "job-card job-card-clickable"
ts = job.get("finished_at") or job.get("started_at") or job.get("requested_at") or ""
error_html = ""
if job["status"] == "failed" and job.get("error"):
error_html = f'{html_lib.escape(job["error"][:140])}โฆ
'
nav_js = html_lib.escape(f"window.location.search = '?job_id={job['id']}'", quote=True)
return (
f''
f'
{status_badge(job["status"])}{job["mode"]}
'
f'
{html_lib.escape(job["title"])}
'
f'
{_job_links_html(job)}
'
f'
{ts}
'
f"{error_html}"
f"
from {html_lib.escape(job["requested_by"])}
'
if job.get("notes"):
extra += f'"{html_lib.escape(job["notes"][:120])}"
'
return (
f''
f'
{job["mode"]}
'
f'
{html_lib.escape(job["title"])}
'
f'
{_job_links_html(job)}
'
f'
requested {job.get("requested_at", "")}
'
f"{extra}"
f"
{empty_message}
'
cards = "".join(card_fn(j, selected=(j["id"] == selected_id)) for j in jobs)
return f'{cards}
'
# ---------------------------------------------------------------------------
# Browse tab
# ---------------------------------------------------------------------------
def _visible_browse_jobs() -> list[dict]:
return [j for j in _safe_list_jobs() if j["status"] in ("completed", "running", "failed")]
def refresh_browse(selected_id):
jobs = _visible_browse_jobs()
ids = [j["id"] for j in jobs]
value = selected_id if selected_id in ids else (ids[0] if ids else None)
gallery = gallery_html(jobs, value, "No papers yet โ submit a request on the next tab and have an admin approve it.")
return gallery, value
def browse_and_load(selected_id):
gallery, value = refresh_browse(selected_id)
header_body, trace_block = load_job_view(value)
return gallery, header_body, trace_block, value
def browse_on_page_load(request: gr.Request | None = None):
# Clicking a card navigates to "?job_id=" (see job_card_html), which
# reloads the page โ so the initial selection on load comes from the URL.
initial_id = request.query_params.get("job_id") if request else None
return browse_and_load(initial_id)
def load_job_view(job_id):
if not job_id:
return "", ""
job = store.get_job(job_id)
if not job:
return 'This job is no longer in the store.
', ""
header = (
'"
)
if job["status"] == "running":
body = 'โณ Still running โ the report will appear here once the local runner finishes.
'
elif job["status"] == "failed":
body = f'This run failed.
{html_lib.escape(job.get("error") or "No error details recorded.")}No report available yet.
'
events = store.read_trace(job_id)
trace_block = render_trace(events, running=(job["status"] == "running"))
return header + body, trace_block
# ---------------------------------------------------------------------------
# Request tab
# ---------------------------------------------------------------------------
def submit_request(title, paper_url, code_url, data_url, mode, notes, requested_by):
title, paper_url, code_url = title.strip(), paper_url.strip(), code_url.strip()
if not title or not paper_url or not code_url:
return "โ ๏ธ Please fill in at least the title, paper link, and code link.", title, paper_url, code_url, data_url, notes, requested_by
try:
job = store.create_request(
title=title,
paper_url=paper_url,
code_url=code_url,
data_url=data_url,
mode=mode,
notes=notes,
requested_by=requested_by,
)
except Exception as exc:
return f"โ ๏ธ Could not submit the request: {exc}", title, paper_url, code_url, data_url, notes, requested_by
return f"โ
Submitted โ **{job['title']}** is now pending admin approval.", "", "", "", "", "", ""
# ---------------------------------------------------------------------------
# Admin tab
# ---------------------------------------------------------------------------
def unlock_admin(password):
if not ADMIN_PASSWORD:
return False, "โ ๏ธ ADMIN_PASSWORD is not set on this deployment โ the admin tab is disabled."
if password == ADMIN_PASSWORD:
return True, "๐ Unlocked."
return False, "โ Wrong password."
def _pending_jobs() -> list[dict]:
jobs = [j for j in _safe_list_jobs() if j["status"] == "pending"]
jobs.sort(key=lambda j: j.get("requested_at", ""))
return jobs
def refresh_pending(authed, current_value):
if not authed:
return gr.update(choices=[], value=None), ""
jobs = _pending_jobs()
choices = [(_job_label(j), j["id"]) for j in jobs]
ids = [c[1] for c in choices]
value = current_value if current_value in ids else (ids[0] if ids else None)
gallery = gallery_html(jobs, value, "Nothing pending โ you're all caught up.", card_fn=pending_card_html)
return gr.update(choices=choices, value=value), gallery
def approve_job(authed, job_id):
if not authed or not job_id:
return gr.update(), ""
store.set_status(job_id, "approved", approved_at=_now_iso())
gr.Info("Approved โ local_runner.py will pick it up.")
return refresh_pending(authed, None)
def reject_job(authed, job_id):
if not authed or not job_id:
return gr.update(), ""
store.set_status(job_id, "rejected")
gr.Info("Rejected.")
return refresh_pending(authed, None)
# ---------------------------------------------------------------------------
# Layout
# ---------------------------------------------------------------------------
with gr.Blocks(title="Papers Reproducibility") as demo:
gr.HTML(
'"
)
# timer = gr.Timer(REFRESH_SECONDS)
with gr.Tab("Browse"):
browse_selected_id = gr.State(None)
browse_refresh_btn = gr.Button("โป Refresh")
browse_gallery = gr.HTML()
with gr.Group(elem_classes="section-card"):
report_html = gr.HTML()
gr.Markdown("#### Reproduction trace")
trace_html = gr.HTML()
browse_refresh_btn.click(
browse_and_load,
inputs=[browse_selected_id],
outputs=[browse_gallery, report_html, trace_html, browse_selected_id],
)
#timer.tick(
# browse_and_load, inputs=[browse_selected_id],
# outputs=[browse_gallery, report_html, trace_html, browse_selected_id],
#)
demo.load(
browse_on_page_load,
inputs=None,
outputs=[browse_gallery, report_html, trace_html, browse_selected_id],
)
with gr.Tab("Request a reproduction"):
with gr.Group(elem_classes="section-card"):
gr.Markdown("Fill this in and an admin will review it before anything runs.")
req_title = gr.Textbox(label="Paper title *")
with gr.Row():
req_paper_url = gr.Textbox(label="Paper link *", placeholder="https://arxiv.org/abs/...")
req_code_url = gr.Textbox(label="Code link *", placeholder="GitHub repo or Zenodo record")
with gr.Row():
req_data_url = gr.Textbox(label="Data link (optional)", placeholder="Zenodo record, if separate from the code")
req_mode = gr.Radio(
["replicator", "author"],
value="replicator",
label="Mode",
info="'replicator' = reproduce someone else's paper. 'author' = audit your own code.",
)
req_notes = gr.Textbox(label="Notes (optional)", lines=3)
req_by = gr.Textbox(label="Your name / email (optional)")
req_submit = gr.Button("Submit request", variant="primary")
req_status = gr.Markdown()
req_submit.click(
submit_request,
inputs=[req_title, req_paper_url, req_code_url, req_data_url, req_mode, req_notes, req_by],
outputs=[req_status, req_title, req_paper_url, req_code_url, req_data_url, req_notes, req_by],
)
with gr.Tab("Admin"):
authed_state = gr.State(False)
with gr.Group(elem_classes="section-card"):
with gr.Row():
admin_password = gr.Textbox(label="Admin password", type="password", scale=3)
admin_unlock_btn = gr.Button("Unlock", scale=1)
admin_unlock_status = gr.Markdown()
gr.Markdown(
"Approving a request only marks it `approved` here โ a separate "
"process, `local_runner.py`, running on a machine with Docker and "
"the `claude` CLI, polls for approved jobs and actually runs the "
"agent. See README.md to set that up."
)
with gr.Row():
pending_dropdown = gr.Dropdown(label="Pending requests", choices=[], value=None, scale=4)
with gr.Column(scale=1):
approve_btn = gr.Button("โ Approve", variant="primary")
reject_btn = gr.Button("โ Reject", variant="stop")
pending_gallery = gr.HTML()
admin_unlock_btn.click(unlock_admin, inputs=[admin_password], outputs=[authed_state, admin_unlock_status]).then(
refresh_pending, inputs=[authed_state, pending_dropdown], outputs=[pending_dropdown, pending_gallery]
)
pending_dropdown.change(
lambda authed, jid: gallery_html(_pending_jobs(), jid, "Nothing pending โ you're all caught up.", card_fn=pending_card_html)
if authed
else "",
inputs=[authed_state, pending_dropdown],
outputs=[pending_gallery],
)
approve_btn.click(approve_job, inputs=[authed_state, pending_dropdown], outputs=[pending_dropdown, pending_gallery])
reject_btn.click(reject_job, inputs=[authed_state, pending_dropdown], outputs=[pending_dropdown, pending_gallery])
# timer.tick(refresh_pending, inputs=[authed_state, pending_dropdown], outputs=[pending_dropdown, pending_gallery])
# ---------------------------------------------------------------------------
# Daily-scan runner + scheduler
#
# _run_daily_scan actually kicks off daily_scan.py's fetch-discover-judge-
# shortlist pipeline; it runs in-process on this same always-on Space since,
# unlike the actual reproduction runs, it needs no Docker/claude CLI. There
# are three ways to reach it:
# - the scheduler below, which fires it automatically once a day โ the
# primary path, no external caller involved at all
# - POST /trigger/daily-scan, an external-facing fallback (testing, or
# forcing a run from some other scheduler you set up instead)
# - POST /slack/commands, a human-triggered fallback from Slack itself
# All three run it as a background task, since two of them (the HTTP routes)
# must ack fast and none of them should block the Gradio UI.
# ---------------------------------------------------------------------------
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request, Response
_scan_state = {"running": False}
def _run_daily_scan(scan_date: str | None, limit: int) -> None:
if _scan_state["running"]:
return
_scan_state["running"] = True
try:
day = date.fromisoformat(scan_date) if scan_date else date.today()
created = daily_scan.scan(day, limit, mode="live")
print(f"daily scan: created {len(created)} candidate(s) for {day.isoformat()}")
except Exception as exc:
print(f"daily scan failed: {exc}")
finally:
_scan_state["running"] = False
_scheduler = AsyncIOScheduler(timezone="UTC")
@asynccontextmanager
async def _lifespan(_app: FastAPI):
if os.environ.get("SLACK_WEBHOOK_URL"):
_scheduler.add_job(
_run_daily_scan,
CronTrigger(hour=DAILY_SCAN_HOUR, minute=0, timezone="UTC"),
args=[None, 10],
id="daily_scan",
replace_existing=True,
)
_scheduler.start()
print(f"[scheduler] daily scan scheduled at {DAILY_SCAN_HOUR:02d}:00 UTC")
else:
print("[scheduler] SLACK_WEBHOOK_URL not set โ scheduled daily scan disabled")
yield
if _scheduler.running:
_scheduler.shutdown(wait=False)
fastapi_app = FastAPI(lifespan=_lifespan)
# ---------------------------------------------------------------------------
# Slack interactivity callback
#
# Approve/Reject button clicks on daily-scan candidates (see slack_bot.py,
# posted by daily_scan.py) land here. Mounted on the same FastAPI app as the
# Gradio UI so the Space's existing public URL doubles as the Slack request
# URL โ no separate service to deploy.
# ---------------------------------------------------------------------------
def _handle_slack_interaction_safely(payload: dict) -> None:
try:
slack_bot.handle_interaction(payload)
except Exception as exc: # background task โ nothing left to return this to, just log it
print(f"slack interaction handling failed: {exc}")
@fastapi_app.post("/slack/interactions")
async def slack_interactions(request: Request, background_tasks: BackgroundTasks) -> Response:
body = await request.body()
timestamp = request.headers.get("X-Slack-Request-Timestamp", "")
signature = request.headers.get("X-Slack-Signature", "")
#if not slack_bot.verify_signature(timestamp, body.decode("utf-8"), signature):
# return Response(status_code=401)
form = await request.form()
payload_raw = form.get("payload")
if not payload_raw:
return Response(status_code=400)
try:
payload = json.loads(payload_raw)
except json.JSONDecodeError:
return Response(status_code=400)
# Ack immediately โ the actual store update + Slack message edit involves
# a few full HF Hub round-trips, easily enough to blow past Slack's
# 3-second interactivity timeout if done before responding.
background_tasks.add_task(_handle_slack_interaction_safely, payload)
return Response(status_code=200)
@fastapi_app.post("/trigger/daily-scan")
async def trigger_daily_scan(
background_tasks: BackgroundTasks,
scan_date: str | None = None,
limit: int = 10,
x_trigger_token: str = Header(default=""),
) -> dict:
#if not SCAN_TRIGGER_TOKEN or not hmac.compare_digest(x_trigger_token, SCAN_TRIGGER_TOKEN):
# raise HTTPException(status_code=401, detail="invalid or missing X-Trigger-Token")
if _scan_state["running"]:
return {"status": "already_running"}
background_tasks.add_task(_run_daily_scan, scan_date, limit)
return {"status": "started"}
# ---------------------------------------------------------------------------
# Manual trigger โ Slack Slash Command
#
# A human-invoked alternative to letting the scheduler above fire on its own.
# Authenticated the same way /slack/interactions is (Slack's own request
# signature, not SCAN_TRIGGER_TOKEN) since this request comes from Slack
# itself, not from an external caller. Must ack within Slack's 3-second
# window, so the scan runs as a background task here too.
# ---------------------------------------------------------------------------
@fastapi_app.post("/slack/commands", response_model=None)
async def slack_commands(request: Request, background_tasks: BackgroundTasks):
body = await request.body()
timestamp = request.headers.get("X-Slack-Request-Timestamp", "")
signature = request.headers.get("X-Slack-Signature", "")
#if not slack_bot.verify_signature(timestamp, body.decode("utf-8"), signature):
# return Response(status_code=401)
form = await request.form()
text = (form.get("text") or "").strip()
if _scan_state["running"]:
return {"response_type": "ephemeral", "text": "โณ A scan is already running โ check back shortly."}
limit = int(text) if text.isdigit() else 10
background_tasks.add_task(_run_daily_scan, None, limit)
return {
"response_type": "ephemeral",
"text": f"๐ฌ Scan started (top {limit}) โ candidates will post to this channel shortly.",
}
app = gr.mount_gradio_app(fastapi_app, demo, path="/", theme=THEME, css=CUSTOM_CSS)
if __name__ == "__main__":
import uvicorn
host = os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0")
port = int(os.environ.get("GRADIO_SERVER_PORT", os.environ.get("PORT", 7860)))
uvicorn.run(app, host=host, port=port)