submit: stream live eval progress to the submitter's status panel

Adds a small in-process progress registry (progress.py) that the eval
worker publishes stage notes to (queued -> waiting for GPU -> evaluating
-> collecting results -> done/failed, plus an "X of N chunks done" count
for sharded runs). handle_submit keeps its generator alive after queuing
and streams those notes into the personal status panel until a terminal
stage or a backstop deadline. The shared leaderboard table stays coarse;
this is the personal-view feedback only, so it adds no write load to the
results file. Also fixes the _with_hub_retries tests for the newer
huggingface_hub HfHubHTTPError(response=...) signature.

Co-authored-by: Cursor <cursoragent@cursor.com>

progress.py

@@ -0,0 +1,111 @@
+# Copyright 2026 Hugging Face
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""In-process live-progress registry for the submitter's status panel.
+
+This is the *personal-view* half of the progress story (the shared
+leaderboard table stays deliberately coarse, driven only by the
+``status`` field on the results row). The background eval worker
+publishes short, human-readable stage notes here as it advances a
+submission; the Submit-tab generator observes them and streams them
+into that submitter's status panel until the submission reaches a
+terminal stage.
+
+Deliberately **ephemeral + in-memory**: it is *not* the source of
+truth for a submission's outcome (that's the row the worker writes to
+``results.jsonl``, which the leaderboard table reads). A Space restart,
+or a submitter whose request is served by a different process, simply
+loses the fine-grained notes and the personal view falls back to the
+coarse row state. Keeping this layer out of the shared file is exactly
+what lets the progress feedback be granular without adding write load
+to the leaderboard's single source of truth.
+"""
+from __future__ import annotations
+
+import threading
+import time
+from dataclasses import dataclass
+
+# Coarse lifecycle states a submission moves through *after* it has been
+# accepted and queued. ``QUEUED`` / ``RUNNING`` are transient; the
+# ``message`` carried alongside is what actually varies (e.g. "waiting
+# for a GPU" vs "evaluating" vs "3 of 8 chunks done"). ``COMPLETED`` /
+# ``FAILED`` are terminal: once a submission reaches one, the observing
+# generator stops streaming.
+QUEUED = "queued"
+RUNNING = "running"
+COMPLETED = "completed"
+FAILED = "failed"
+
+_TERMINAL = frozenset({COMPLETED, FAILED})
+
+# Entries untouched for this long are pruned on the next publish so the
+# registry can't grow unbounded across a long-lived Space process. Well
+# above any realistic eval wall-time, so a still-streaming submission is
+# never pruned out from under its observer.
+ENTRY_TTL_SECONDS = 60 * 60
+
+
+@dataclass(frozen=True)
+class Snapshot:
+    """An immutable point-in-time view of one submission's progress."""
+
+    state: str
+    message: str
+    updated_at: float
+
+
+_LOCK = threading.Lock()
+_ENTRIES: dict[str, Snapshot] = {}
+
+
+def is_terminal(state: str) -> bool:
+    """True for states the observer should stop streaming on."""
+    return state in _TERMINAL
+
+
+def publish(submission_id: str, state: str, message: str) -> None:
+    """Record the latest progress note for *submission_id*.
+
+    Overwrites any prior note (the registry keeps only the most recent
+    snapshot per submission). Prunes stale entries opportunistically so
+    no separate reaper thread is needed.
+    """
+    now = time.time()
+    with _LOCK:
+        _ENTRIES[submission_id] = Snapshot(state, message, now)
+        _prune_locked(now)
+
+
+def get(submission_id: str) -> Snapshot | None:
+    """Return the latest snapshot for *submission_id*, or ``None``."""
+    with _LOCK:
+        return _ENTRIES.get(submission_id)
+
+
+def clear() -> None:
+    """Drop every entry. Test helper; not used by the app at runtime."""
+    with _LOCK:
+        _ENTRIES.clear()
+
+
+def _prune_locked(now: float) -> None:
+    """Remove entries older than the TTL. Caller must hold ``_LOCK``."""
+    stale = [
+        sid
+        for sid, snap in _ENTRIES.items()
+        if now - snap.updated_at > ENTRY_TTL_SECONDS
+    ]
+    for sid in stale:
+        del _ENTRIES[sid]

submit.py

@@ -115,6 +115,7 @@ from huggingface_hub import (
 )
 from huggingface_hub.errors import EntryNotFoundError, HfHubHTTPError
 
+import progress
 from leaderboard import HF_DATA_REPO, HF_ORG, HF_SUBMISSIONS_REPO
 
 logger = logging.getLogger(__name__)
@@ -172,6 +173,18 @@ EVAL_JOB_NAMESPACE = "michaelr27"
 EVAL_JOB_TIMEOUT = "30m"
 EVAL_JOB_WORKER_COUNT = "8"
 
+# Live personal-view progress streaming. After a submission is queued,
+# handle_submit keeps its generator alive and re-reads the in-process
+# progress registry (which the background worker writes to) every few
+# seconds, pushing each new note into the submitter's status panel. The
+# deadline is a generous backstop: the stream normally ends the moment
+# the worker publishes a terminal note, well before this trips. If it
+# does trip (worker died, very long sharded run), the panel tells the
+# submitter the eval continues in the background and to watch the
+# leaderboard, rather than hanging forever.
+PROGRESS_STREAM_POLL_SECONDS = 3
+PROGRESS_STREAM_DEADLINE_SECONDS = 45 * 60
+
 # Poll cadence + outer deadline guarding inspect_job. 5 s is fast
 # enough that a 60 s eval lands in <10 s of completion, slow enough
 # that we don't hammer the API. Deadline matches the Job's own
@@ -288,13 +301,60 @@ def _submit_status(state: str, message: str) -> str:
     The panel is the durable counterpart to the transient ``gr.Info`` /
     ``gr.Error`` toasts: a submitter always sees the current stage and
     any rejection reason without having to catch an ephemeral toast.
-    *state* picks the leading glyph (``working`` / ``queued`` /
-    ``error``).
+    *state* picks the leading glyph (``working`` / ``queued`` / ``done``
+    / ``error``).
     """
-    glyph = {"working": "⏳", "queued": "✅", "error": "❌"}.get(state, "•")
+    glyph = {"working": "⏳", "queued": "✅", "done": "🎉", "error": "❌"}.get(
+        state, "•"
+    )
     return f"{glyph} {message}"
 
 
+# Maps the progress registry's coarse state to the `_submit_status`
+# glyph state. The registry's transient states (queued waiting for a
+# slot, running on the GPU) both read as "in progress" in the panel;
+# the terminal ones get the celebratory / error glyph.
+_PROGRESS_PANEL_STATE = {
+    progress.QUEUED: "queued",
+    progress.RUNNING: "working",
+    progress.COMPLETED: "done",
+    progress.FAILED: "error",
+}
+
+
+def _running_message_for_stage(stage: str) -> str:
+    """Friendly note for a non-terminal HF Jobs stage.
+
+    The Jobs API exposes a stage string per poll. We only care about
+    one distinction the submitter actually feels: actively *running* vs
+    still *waiting for a machine*. Treating any non-RUNNING, non-terminal
+    stage as "queued on the GPU" keeps the message robust to the exact
+    set of intermediate stage names the API may use.
+    """
+    if stage == "RUNNING":
+        return "Evaluating your submission on a GPU…"
+    return "Evaluation queued on a GPU — waiting for a free machine…"
+
+
+def _completed_progress_message(summary: dict[str, Any]) -> str:
+    """Terminal success note, surfacing the headline score when present."""
+    score = summary.get("aggregate_score")
+    if isinstance(score, (int, float)):
+        return (
+            f"Done — scored {float(score):.4f}. Your row is on the "
+            f"Unvalidated leaderboard."
+        )
+    return "Done — your row is on the Unvalidated leaderboard."
+
+
+def _failed_progress_message(reason: str | None) -> str:
+    """Terminal failure note, appending the short reason when there is one."""
+    reason = (reason or "").strip()
+    if reason:
+        return f"Evaluation failed: {reason}"
+    return "Evaluation failed."
+
+
 def handle_submit(
     zip_file,
     profile: gr.OAuthProfile | None,
@@ -389,18 +449,64 @@ def handle_submit(
             yield _submit_status("error", msg)
             raise gr.Error(msg)
 
+        # Seed the registry so the stream below has something to show
+        # in the gap before the worker publishes its first stage note.
+        progress.publish(
+            submission_id,
+            progress.QUEUED,
+            f"Queued ({len(fixture_names)} fixtures) — waiting for the "
+            f"evaluation to start…",
+        )
         _spawn_worker(submission_id, blob_url, sorted(fixture_names))
         yield _submit_status(
             "queued",
             f"Submission `{submission_id}` queued ({len(fixture_names)} "
             f"fixtures). The eval runs on an HF Jobs GPU; your row appears on "
             f"the **Unvalidated** leaderboard and flips to completed when the "
-            f"job finishes (typically 1–3 minutes).",
+            f"job finishes (typically 1–3 minutes). Live progress below.",
         )
+        # Keep the generator alive, observing the in-process progress
+        # registry the worker writes to, until the submission reaches a
+        # terminal stage (or the backstop deadline). This is the
+        # personal-view live feedback; the shared table stays coarse.
+        yield from _stream_submission_progress(submission_id)
     finally:
         shutil.rmtree(tmp, ignore_errors=True)
 
 
+def _stream_submission_progress(submission_id: str):
+    """Yield panel markdown as the worker advances *submission_id*.
+
+    Polls the in-process :mod:`progress` registry every
+    :data:`PROGRESS_STREAM_POLL_SECONDS` and yields a fresh status panel
+    only when the human-readable note changes (so the panel updates on
+    real transitions, not every tick). Returns when the submission
+    reaches a terminal state, or yields a "still running in the
+    background" note and returns if the backstop deadline trips first
+    (worker death, an unusually long sharded run, etc.).
+    """
+    deadline = time.monotonic() + PROGRESS_STREAM_DEADLINE_SECONDS
+    last_message: str | None = None
+    while True:
+        snap = progress.get(submission_id)
+        if snap is not None and snap.message != last_message:
+            last_message = snap.message
+            yield _submit_status(
+                _PROGRESS_PANEL_STATE.get(snap.state, "working"), snap.message,
+            )
+        if snap is not None and progress.is_terminal(snap.state):
+            return
+        if time.monotonic() >= deadline:
+            yield _submit_status(
+                "queued",
+                "Evaluation is taking longer than expected; it continues in "
+                "the background. Check the **Unvalidated** leaderboard for "
+                "the final result.",
+            )
+            return
+        time.sleep(PROGRESS_STREAM_POLL_SECONDS)
+
+
 def _validate_form(zip_file) -> str | None:
     """Form-level check before any zip parsing.
 
@@ -871,16 +977,34 @@ def _run_worker(
                 submission_id, submission_blob_url, fixture_names,
             )
             return
+        progress.publish(
+            submission_id,
+            progress.RUNNING,
+            "Evaluation dispatched — waiting for a GPU…",
+        )
         job_id = _dispatch_eval_job(submission_id, submission_blob_url)
         logger.info("Dispatched eval job %s for %s", job_id, submission_id)
         stage, status_message = _poll_until_done(job_id, submission_id)
         if stage == "COMPLETED":
+            progress.publish(
+                submission_id,
+                progress.RUNNING,
+                "Evaluation finished — collecting results…",
+            )
             summary = _fetch_run_summary_from_report(submission_id)
             _flip_row_to_completed(submission_id, summary)
+            progress.publish(
+                submission_id,
+                progress.COMPLETED,
+                _completed_progress_message(summary),
+            )
             logger.info("Worker completed for %s", submission_id)
             return
         reason = _job_failure_reason(job_id, stage, status_message)
         _flip_row_to_failed(submission_id, reason)
+        progress.publish(
+            submission_id, progress.FAILED, _failed_progress_message(reason),
+        )
         logger.warning(
             "Eval job %s for %s ended %s: %s",
             job_id, submission_id, stage, reason,
@@ -888,6 +1012,9 @@ def _run_worker(
     except Exception as e:  # noqa: BLE001 - broad on purpose; we map to row state
         logger.exception("Worker failed for %s", submission_id)
         reason = f"{type(e).__name__}: {str(e)}"[:FAILURE_REASON_MAX_CHARS]
+        progress.publish(
+            submission_id, progress.FAILED, _failed_progress_message(reason),
+        )
         try:
             _flip_row_to_failed(submission_id, reason)
         except Exception:
@@ -931,6 +1058,11 @@ def _run_worker_sharded(
         "Sharded eval for %s: %d fixtures -> %d shard(s)",
         submission_id, len(fixture_names), len(shards),
     )
+    progress.publish(
+        submission_id,
+        progress.RUNNING,
+        f"Evaluation split into {len(shards)} chunks — dispatching to GPUs…",
+    )
     for shard_id, st in shards.items():
         _dispatch_shard(submission_id, submission_blob_url, shard_id, st)
 
@@ -942,13 +1074,24 @@ def _run_worker_sharded(
             :FAILURE_REASON_MAX_CHARS
         ]
         _flip_row_to_failed(submission_id, reason)
+        progress.publish(
+            submission_id, progress.FAILED, _failed_progress_message(reason),
+        )
         logger.warning("Sharded eval for %s failed: %s", submission_id, reason)
         return
 
+    progress.publish(
+        submission_id,
+        progress.RUNNING,
+        "All chunks evaluated — merging results…",
+    )
     summary = _merge_shards_and_publish(
         submission_id, list(shards.keys()), fixture_names,
     )
     _flip_row_to_completed(submission_id, summary)
+    progress.publish(
+        submission_id, progress.COMPLETED, _completed_progress_message(summary),
+    )
     logger.info("Sharded worker completed for %s", submission_id)
     _cleanup_shard_artifacts(submission_id)
 
@@ -1058,11 +1201,23 @@ def _poll_shards_until_done(
     """
     deadline = time.monotonic() + SHARD_POLL_DEADLINE_SECONDS
     consecutive_errors = 0
+    last_done = -1
+    total = len(shards)
     while True:
         running = [
             sid for sid, st in shards.items()
             if st["stage"] not in ("COMPLETED", "FAILED")
         ]
+        # Push an "N of M chunks done" note to the submitter's panel
+        # whenever the completed count advances.
+        done = sum(1 for st in shards.values() if st["stage"] == "COMPLETED")
+        if done != last_done:
+            last_done = done
+            progress.publish(
+                submission_id,
+                progress.RUNNING,
+                f"Evaluating… {done} of {total} chunks done.",
+            )
         if not running:
             break
         for shard_id in running:
@@ -1137,6 +1292,7 @@ def _poll_until_done(
     """
     deadline = time.monotonic() + JOB_POLL_DEADLINE_SECONDS
     consecutive_errors = 0
+    last_stage: str | None = None
     while True:
         try:
             info = inspect_job(job_id=job_id)
@@ -1157,6 +1313,16 @@ def _poll_until_done(
         message = info.status.message
         if stage in ("COMPLETED", "ERROR"):
             return stage, message
+        # Surface the running-vs-waiting distinction to the submitter's
+        # panel, but only when the stage actually changes (not every
+        # tick), so the personal view reflects real transitions.
+        if stage != last_stage:
+            last_stage = stage
+            progress.publish(
+                submission_id,
+                progress.RUNNING,
+                _running_message_for_stage(stage),
+            )
         if time.monotonic() >= deadline:
             return "ERROR", (
                 f"Space-side poll deadline exceeded "

tests/test_progress.py

@@ -0,0 +1,133 @@
+"""Unit tests for the live personal-view progress feedback.
+
+Two surfaces:
+
+1. :mod:`progress` - the in-process registry the eval worker writes to
+   (publish / get / terminal classification / TTL pruning).
+2. :mod:`submit`'s observer side - the message helpers and the
+   :func:`submit._stream_submission_progress` generator that streams
+   registry notes into the submitter's status panel until a terminal
+   stage or the backstop deadline.
+
+All time + registry access is monkeypatched, so the suite has zero
+network traffic and never actually sleeps.
+"""
+from __future__ import annotations
+
+import progress
+import submit
+
+
+def test_publish_get_roundtrip():
+    progress.clear()
+    progress.publish("a", progress.RUNNING, "evaluating")
+    snap = progress.get("a")
+    assert snap is not None
+    assert snap.state == progress.RUNNING
+    assert snap.message == "evaluating"
+    assert progress.get("does-not-exist") is None
+
+
+def test_publish_overwrites_prior_note():
+    progress.clear()
+    progress.publish("a", progress.QUEUED, "queued")
+    progress.publish("a", progress.RUNNING, "running")
+    snap = progress.get("a")
+    assert snap is not None
+    assert snap.state == progress.RUNNING
+    assert snap.message == "running"
+
+
+def test_is_terminal_classification():
+    assert progress.is_terminal(progress.COMPLETED)
+    assert progress.is_terminal(progress.FAILED)
+    assert not progress.is_terminal(progress.QUEUED)
+    assert not progress.is_terminal(progress.RUNNING)
+
+
+def test_prune_drops_stale_entries(monkeypatch):
+    progress.clear()
+    base = 1_000.0
+    monkeypatch.setattr(progress.time, "time", lambda: base)
+    progress.publish("old", progress.RUNNING, "x")
+    # A later publish past the TTL window prunes the untouched entry.
+    monkeypatch.setattr(
+        progress.time, "time", lambda: base + progress.ENTRY_TTL_SECONDS + 1
+    )
+    progress.publish("new", progress.RUNNING, "y")
+    assert progress.get("old") is None
+    assert progress.get("new") is not None
+
+
+def test_running_message_distinguishes_running_from_waiting():
+    running = submit._running_message_for_stage("RUNNING")
+    waiting = submit._running_message_for_stage("QUEUED")
+    assert "Evaluating" in running
+    assert "waiting" in waiting.lower()
+    assert running != waiting
+
+
+def test_completed_message_surfaces_score_when_present():
+    msg = submit._completed_progress_message({"aggregate_score": 0.8086})
+    assert "0.8086" in msg
+    # No score -> still a clean terminal note, no crash.
+    assert "Done" in submit._completed_progress_message({})
+
+
+def test_failed_message_appends_reason():
+    assert "boom" in submit._failed_progress_message("boom")
+    assert submit._failed_progress_message(None) == "Evaluation failed."
+    assert submit._failed_progress_message("  ") == "Evaluation failed."
+
+
+def test_stream_yields_on_change_and_stops_on_terminal(monkeypatch):
+    """The panel updates on real transitions and stops at a terminal state."""
+    monkeypatch.setattr(submit.time, "sleep", lambda *_: None)
+    snaps = [
+        progress.Snapshot(progress.QUEUED, "queued msg", 0.0),
+        progress.Snapshot(progress.RUNNING, "running msg", 0.0),
+        # Duplicate message -> no new yield.
+        progress.Snapshot(progress.RUNNING, "running msg", 0.0),
+        progress.Snapshot(progress.COMPLETED, "done msg", 0.0),
+    ]
+    it = iter(snaps)
+    monkeypatch.setattr(submit.progress, "get", lambda _sid: next(it))
+
+    out = list(submit._stream_submission_progress("x"))
+
+    assert len(out) == 3
+    assert "queued msg" in out[0]
+    assert "running msg" in out[1]
+    assert "done msg" in out[2]
+    # Terminal success gets the celebratory glyph.
+    assert out[2].startswith("🎉")
+
+
+def test_stream_emits_failure_glyph_on_failed_terminal(monkeypatch):
+    monkeypatch.setattr(submit.time, "sleep", lambda *_: None)
+    it = iter([progress.Snapshot(progress.FAILED, "it broke", 0.0)])
+    monkeypatch.setattr(submit.progress, "get", lambda _sid: next(it))
+
+    out = list(submit._stream_submission_progress("x"))
+
+    assert len(out) == 1
+    assert out[0].startswith("❌")
+    assert "it broke" in out[0]
+
+
+def test_stream_backstop_deadline_emits_background_note(monkeypatch):
+    """If the registry never goes terminal, the stream ends gracefully."""
+    monkeypatch.setattr(submit.time, "sleep", lambda *_: None)
+    # First monotonic() sets the deadline; the second trips it.
+    ticks = iter([0.0, submit.PROGRESS_STREAM_DEADLINE_SECONDS + 1])
+    monkeypatch.setattr(
+        submit.time,
+        "monotonic",
+        lambda: next(ticks, submit.PROGRESS_STREAM_DEADLINE_SECONDS + 1),
+    )
+    monkeypatch.setattr(submit.progress, "get", lambda _sid: None)
+
+    out = list(submit._stream_submission_progress("x"))
+
+    assert len(out) == 1
+    assert "background" in out[0].lower()

tests/test_submit.py

@@ -22,12 +22,18 @@ import submit
 def _hub_http_error(status: int, headers: dict | None = None) -> submit.HfHubHTTPError:
     """An ``HfHubHTTPError`` with a minimal response carrying *status*.
 
-    Built without going through the real Hub: construct the exception
-    with no response, then attach a stand-in so ``_with_hub_retries``
-    can read ``response.status_code`` / ``response.headers``.
+    Built without going through the real Hub: a ``SimpleNamespace``
+    stands in for the httpx response so ``_with_hub_retries`` can read
+    ``response.status_code`` / ``response.headers``. Newer
+    ``huggingface_hub`` makes ``response`` a required keyword-only
+    constructor argument, so it's passed in directly (and re-assigned
+    afterwards for the older positional-optional signature too).
     """
-    err = submit.HfHubHTTPError(f"HTTP {status}")
-    err.response = SimpleNamespace(status_code=status, headers=headers or {})
+    response = SimpleNamespace(
+        status_code=status, headers=headers or {}, request=None,
+    )
+    err = submit.HfHubHTTPError(f"HTTP {status}", response=response)
+    err.response = response
     return err