feat: complete the eval engine — real worker + OEB-aware queue reader

The arena's submission loop was open: nothing produced leaderboard results.

- run_eval_worker.py is now a complete one-shot worker (run on external
GPU/SLURM infra, not the Space): PENDING -> mark RUNNING -> oeb.benchmark()
-> map_oeb_results_to_contents -> append to public braindecode/contents
(single train.parquet, deduped by model+adapter) -> mark FINISHED/FAILED.
Supports --device, --dry-run, --slurm, --infra-folder, --limit. Writes
parquet via pandas + upload_file to stay robust across datasets/pyarrow.
- _refresh_models_cache now parses the OpenEEGBench request schema
(model_name/submitter/benchmark_kwargs) instead of the legacy LLM fields
(model/revision/sender), which previously KeyError'd on every entry so
submissions never appeared in the queue UI. Legacy fields kept as fallback.
- worker-requirements.txt pins open-eeg-bench for the worker env.

Verified end-to-end against throwaway HF repos: publish() round-trips through
load_dataset(...)['train'] (what the leaderboard reads), dedup works, and the
reader surfaces OEB submissions as pending.

backend/app/services/models.py

@@ -214,25 +214,42 @@ class ModelService(HuggingFaceService):
 
                         # Calculate wait time
                         try:
-                            submit_time = datetime.fromisoformat(content["submitted_time"].replace("Z", "+00:00"))
+                            submitted_time = content.get("submitted_time")
+                            if not submitted_time:
+                                progress.update()
+                                continue
+                            submit_time = datetime.fromisoformat(submitted_time.replace("Z", "+00:00"))
                             if submit_time.tzinfo is None:
                                 submit_time = submit_time.replace(tzinfo=timezone.utc)
                             current_time = datetime.now(timezone.utc)
                             wait_time = current_time - submit_time
 
+                            # OpenEEGBench request schema, with a fallback to the
+                            # legacy LLM-leaderboard fields for any old entries.
+                            bk = content.get("benchmark_kwargs", {}) or {}
+                            name = content.get("model_name") or content.get("model") or "Unknown"
+                            submitter = content.get("submitter") or content.get("sender") or "Unknown"
+                            weights = (
+                                bk.get("hub_repo") or bk.get("checkpoint_url")
+                                or content.get("revision") or "—"
+                            )
+                            strategies = bk.get("finetuning_strategies") or []
+                            adapter = strategies[0] if strategies else content.get("precision", "—")
+
                             model_info = {
-                                "name": content["model"],
-                                "submitter": content.get("sender", "Unknown"),
-                                "revision": content["revision"],
+                                "name": name,
+                                "submitter": submitter,
+                                "revision": weights,
                                 "wait_time": f"{wait_time.total_seconds():.1f}s",
-                                "submission_time": content["submitted_time"],
+                                "submission_time": submitted_time,
                                 "status": target_status,
-                                "precision": content.get("precision", "Unknown")
+                                "precision": adapter,
                             }
 
-                            # Use (model_id, revision, precision) as key to track latest submission
-                            key = (content["model"], content["revision"], content.get("precision", "Unknown"))
-                            if key not in model_submissions or submit_time > datetime.fromisoformat(model_submissions[key]["submission_time"].replace("Z", "+00:00")):
+                            # One entry per (model, submitter, weights) — keep the latest.
+                            key = (name, submitter, weights)
+                            prev = model_submissions.get(key)
+                            if prev is None or submit_time > datetime.fromisoformat(prev["submission_time"].replace("Z", "+00:00")):
                                 model_submissions[key] = model_info
 
                         except (ValueError, TypeError) as e:

backend/scripts/run_eval_worker.py

@@ -1,68 +1,178 @@
-"""OpenEEGBench eval worker (stub).
+"""OpenEEGBench eval worker — the engine behind the arena.
 
-Polls the requests queue for PENDING submissions, runs oeb.benchmark(), maps
-results to the contents schema, and writes them back. The GPU run is guarded:
-without --run it only previews. open_eeg_bench is imported lazily so this file
-imports without the heavy dependency installed.
+This is NOT run by the Space (a CPU web server). Run it on a machine with a GPU
+(or SLURM access) that has ``open-eeg-bench`` installed and an ``HF_TOKEN`` with
+write access to the ``braindecode`` org:
 
-Usage:
-    python -m scripts.run_eval_worker --dry-run
-    python -m scripts.run_eval_worker --run --device cuda
+    pip install -r scripts/worker-requirements.txt   # open-eeg-bench (+ torch, braindecode)
+    export HF_TOKEN=hf_xxx
+    python -m scripts.run_eval_worker --dry-run                  # list pending, do nothing
+    python -m scripts.run_eval_worker --device cuda              # process the queue once
+    python -m scripts.run_eval_worker --slurm --infra-folder ./oeb-cache --device cuda
+
+For each PENDING request it:
+  1. marks the request RUNNING (in braindecode/requests),
+  2. runs ``oeb.benchmark(**benchmark_kwargs)``,
+  3. maps the result DataFrame to the leaderboard schema (results_mapping),
+  4. appends the row(s) to the public braindecode/contents dataset,
+  5. marks the request FINISHED (or FAILED with the error).
+
+It is one-shot by design (process the current queue, then exit) so it can be run
+from cron / a systemd timer / a SLURM job. ``open_eeg_bench`` is imported lazily,
+so this module imports fine without it (e.g. for --dry-run or tests).
 """
 import argparse
 import json
 import logging
-import tempfile
+from datetime import datetime, timezone
 from pathlib import Path
 
 from app.config.base import HF_TOKEN
 from app.config.hf_config import API as hf_api, QUEUE_REPO, AGGREGATED_REPO
 from app.services.results_mapping import map_oeb_results_to_contents
 
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+logger = logging.getLogger("eval_worker")
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def load_pending(limit=None):
+    """Return ``[(path_in_repo, entry), ...]`` for PENDING open-eeg-bench requests.
 
+    A missing queue dataset (no submissions yet) is treated as empty.
+    """
+    try:
+        local_dir = hf_api.snapshot_download(
+            repo_id=QUEUE_REPO, repo_type="dataset", token=HF_TOKEN
+        )
+    except Exception as e:
+        logger.info("Queue %s not available (%s); nothing to do.", QUEUE_REPO, e)
+        return []
 
-def load_pending():
-    local_dir = hf_api.snapshot_download(repo_id=QUEUE_REPO, repo_type="dataset", token=HF_TOKEN)
     pending = []
-    for p in Path(local_dir).glob("**/*.json"):
+    for p in sorted(Path(local_dir).glob("**/*.json")):
         try:
             entry = json.loads(p.read_text())
         except Exception:
             continue
-        if entry.get("status") == "PENDING" and entry.get("framework") == "open-eeg-bench":
-            pending.append((p, entry))
-    return pending
+        if entry.get("framework") == "open-eeg-bench" and entry.get("status") == "PENDING":
+            pending.append((str(p.relative_to(local_dir)), entry))
+    return pending[:limit] if limit else pending
 
 
-def run_one(entry, device):
-    import open_eeg_bench as oeb  # lazy heavy import
-    df = oeb.benchmark(device=device, **entry["benchmark_kwargs"])
+def set_status(path_in_repo, entry, status, **extra):
+    """Re-upload the request JSON with an updated status (RUNNING/FINISHED/FAILED)."""
+    updated = {**entry, "status": status, **extra}
+    hf_api.upload_file(
+        path_or_fileobj=json.dumps(updated, indent=2).encode("utf-8"),
+        path_in_repo=path_in_repo,
+        repo_id=QUEUE_REPO,
+        repo_type="dataset",
+        token=HF_TOKEN,
+        commit_message=f"{updated.get('model_name', '?')}: {status}",
+    )
+    return updated
+
+
+def run_benchmark(entry, device, infra):
+    """Run oeb.benchmark() for a request and map the result to contents rows."""
+    import open_eeg_bench as oeb  # heavy, lazy import (torch/braindecode)
+
+    df = oeb.benchmark(device=device, infra=infra, **entry["benchmark_kwargs"])
     return map_oeb_results_to_contents(df, entry)
 
 
+def _load_contents_df():
+    """Read the current contents (the single ``train.parquet``); empty if absent."""
+    import pandas as pd
+    from huggingface_hub import hf_hub_download
+
+    try:
+        path = hf_hub_download(
+            repo_id=AGGREGATED_REPO, filename="train.parquet",
+            repo_type="dataset", token=HF_TOKEN,
+        )
+        return pd.read_parquet(path)
+    except Exception:
+        return pd.DataFrame()
+
+
+def publish(rows):
+    """Append rows to the public contents dataset (create it if needed).
+
+    Dedupes by (model, adapter) keeping the latest, so re-submissions update in
+    place. Stored as a single ``train.parquet`` that ``load_dataset(repo)["train"]``
+    reads — written with pandas to stay robust across datasets/pyarrow versions.
+    """
+    import os
+    import tempfile
+    import pandas as pd
+
+    combined = pd.concat([_load_contents_df(), pd.DataFrame(rows)], ignore_index=True)
+    if {"fullname", "adapter"}.issubset(combined.columns):
+        combined = combined.drop_duplicates(subset=["fullname", "adapter"], keep="last")
+
+    hf_api.create_repo(
+        repo_id=AGGREGATED_REPO, repo_type="dataset",
+        private=False, exist_ok=True, token=HF_TOKEN,
+    )
+    with tempfile.NamedTemporaryFile(suffix=".parquet", delete=False) as f:
+        tmp = f.name
+    try:
+        combined.to_parquet(tmp, index=False)
+        hf_api.upload_file(
+            path_or_fileobj=tmp, path_in_repo="train.parquet",
+            repo_id=AGGREGATED_REPO, repo_type="dataset", token=HF_TOKEN,
+            commit_message="Update leaderboard contents",
+        )
+    finally:
+        os.unlink(tmp)
+    return len(combined)
+
+
 def main():
-    ap = argparse.ArgumentParser()
-    ap.add_argument("--dry-run", action="store_true", help="List pending submissions only.")
-    ap.add_argument("--run", action="store_true", help="Actually run oeb.benchmark (needs GPU + open-eeg-bench).")
-    ap.add_argument("--device", default="cpu")
+    ap = argparse.ArgumentParser(
+        description="Run queued OpenEEGBench submissions and publish results."
+    )
+    ap.add_argument("--dry-run", action="store_true", help="List pending submissions and exit.")
+    ap.add_argument("--device", default="cuda", help="Torch device for benchmark() (default: cuda).")
+    ap.add_argument("--limit", type=int, default=None, help="Max submissions to process this run.")
+    ap.add_argument("--infra-folder", default=None, help="oeb cache/results folder (enables caching + SLURM).")
+    ap.add_argument("--slurm", action="store_true", help="Submit experiments via SLURM (oeb infra cluster=slurm).")
     args = ap.parse_args()
 
-    pending = load_pending()
-    logger.info("Found %d pending OpenEEGBench submissions.", len(pending))
-    for path, entry in pending:
-        logger.info("PENDING: %s (%s)", entry.get("model_name"), entry["benchmark_kwargs"].get("model_cls"))
-        if args.dry_run or not args.run:
-            continue
+    pending = load_pending(args.limit)
+    logger.info("Found %d pending submission(s).", len(pending))
+
+    if args.dry_run:
+        for _, e in pending:
+            logger.info("PENDING %s — %s", e.get("model_name"), (e.get("benchmark_kwargs") or {}).get("model_cls"))
+        return
+
+    infra = {}
+    if args.infra_folder:
+        infra["folder"] = args.infra_folder
+    if args.slurm:
+        infra["cluster"] = "slurm"
+    infra = infra or None
+
+    for path_in_repo, entry in pending:
+        name = entry.get("model_name", "?")
+        logger.info("RUNNING %s", name)
+        set_status(path_in_repo, entry, "RUNNING", started_time=_now())
         try:
-            rows = run_one(entry, args.device)
-            logger.info("Mapped %d leaderboard row(s) for %s.", len(rows), entry.get("model_name"))
-            # NOTE: appending to AGGREGATED_REPO (contents) is intentionally left for the
-            # production worker; print the rows here so the contract is observable.
-            print(json.dumps(rows, indent=2, default=str))
-        except Exception as e:
-            logger.error("Run failed for %s: %s", entry.get("model_name"), e)
+            rows = run_benchmark(entry, args.device, infra)
+            if not rows:
+                raise RuntimeError("benchmark produced no completed results")
+            total = publish(rows)
+            set_status(path_in_repo, entry, "FINISHED", finished_time=_now(), n_rows=len(rows))
+            logger.info("FINISHED %s — published %d row(s); contents now has %d.", name, len(rows), total)
+        except Exception as e:  # noqa: BLE001 — record the failure on the request and move on
+            logger.exception("FAILED %s", name)
+            set_status(path_in_repo, entry, "FAILED", finished_time=_now(), error=str(e)[:500])
 
 
 if __name__ == "__main__":

backend/scripts/worker-requirements.txt

@@ -0,0 +1,9 @@
+# Extra dependency for the eval worker (scripts/run_eval_worker.py).
+# The worker runs OUTSIDE the Space, on a GPU/SLURM machine. Install it on top
+# of the backend deps (the worker imports app.config.* / app.services.*):
+#
+#   cd backend && pip install -e . -r scripts/worker-requirements.txt
+#   # or: poetry install && pip install -r scripts/worker-requirements.txt
+#
+# open-eeg-bench pulls torch + braindecode transitively.
+open-eeg-bench>=0.6.0