fix(validator): commit evaluation progress so it is visible while the run is in flight (#251)

docs/JOURNAL.md

@@ -18,6 +18,15 @@ protocol. **Newest entries at the top.** Tag each entry with one or more of:
 
 ---
 
+## 2026-07-24 — live evaluation progress was computed and then thrown away  #mistake #gotcha
+**Context:** `eval_runner` parses the `[submission] item i/N ...` lines the harness prints and stores them on the run (`phase`, `message`, `progress_current/total`); `/api/submissions/{id}` exposes them as `current_phase`/`current_message`/`current_progress_*` and the public submission page renders them.
+**Expected:** the progress fields advance while an evaluation is running.
+**Actual:** they stayed empty for the entire run and only appeared once it had already finished — the progress bar was always either blank or complete, never moving. `/api/jobs` likewise reported a job as `queued` for hours after a worker had claimed it.
+**Root cause:** `_touch_progress()` ended in `session.flush()`, and the worker claimed its job with `session.flush()` too. A flush writes into the *worker's own* transaction — and `process_once()` holds that single transaction open across the whole job body, up to `EVAL_TIMEOUT_SECONDS` (2h by default). Under Postgres READ COMMITTED the API process cannot see uncommitted rows, so every progress update was invisible until the final `session.commit()`, by which point it was worthless.
+**Fix / decision:** commit instead of flush in both places. Committing the claim also makes it durable and releases the `SELECT ... FOR UPDATE` row lock early; the `status = 'queued' -> 'running'` transition is what actually keeps other workers off the row, so `skip_locked` claiming is unaffected. Regression tests in `validator/tests/test_progress_visibility.py` assert that `_touch_progress` commits and that the worker has committed at least once before the job body starts.
+**Gotcha for the test suite:** the `validator_session` fixture binds the Session to a Connection that already has an open transaction, so SQLAlchemy 2.0 joins it via SAVEPOINT — an inner `session.commit()` is still undone by the fixture's outer `transaction.rollback()`. Verified before relying on it; tests stay isolated.
+**Follow-up:** `_local_attempt()` still does not forward an `on_line` callback to `_run_bash_stream()`, so local / local-fallback runs parse no progress lines at all even now that the updates are durable. Left out of this change to keep it reviewable.
+
 ## 2026-07-12 — code grader: add resource limits on top of the HOME/secrets fix  #security #decision
 **Context:** issue #71 — the code grader (`run_pass_at_1`) runs untrusted miner/LLM candidate code. The core secret-leak fix (isolated throwaway HOME/cwd, scrubbed env, `python -I`) already landed on main.
 **Finding:** main's sandbox closes the HOME/secrets exfiltration vector but has **no resource limits** — an untrusted candidate can still exhaust host memory or fork-bomb the eval box within its wall-clock timeout (verified: a 4 GiB `bytearray` allocation runs to completion and "passes" on main).

validator/src/eval_backend/services/eval_runner.py

@@ -240,7 +240,13 @@ def _touch_progress(
         run.status = status
         submission.status = status
     submission.updated_at = _utcnow()
-    session.flush()
+    # Commit, not flush: the whole point of these updates is that another process
+    # (the API serving /api/submissions/{id}) can observe them *while* the
+    # evaluation is still running. A flush only makes them visible inside this
+    # transaction, and the worker holds that transaction open for the entire run
+    # (up to EVAL_TIMEOUT_SECONDS, 2h by default), so every progress line was
+    # discarded until the run was already over.
+    session.commit()
 
 
 def _consume_progress_line(

validator/src/eval_backend/worker.py

@@ -84,7 +84,13 @@ def process_once(session_factory, settings: Settings) -> int:
         now = datetime.now(timezone.utc)
         job.claimed_at = now
         job.heartbeat_at = now
-        session.flush()
+        # Commit the claim before the (long) job body runs. A flush kept the claim
+        # inside this transaction, which stays open for the whole evaluation, so
+        # /api/jobs reported the job as "queued" and the submission as un-started
+        # for hours. Committing also makes the claim durable and releases the
+        # SELECT ... FOR UPDATE row lock; the status is now "queued" -> "running"
+        # in the database, which is what keeps other workers off this row.
+        session.commit()
         payload = job.payload_json or {}
         job_id = job.id
         submission_id = job.submission_id

validator/tests/test_progress_visibility.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+from eval_backend import worker
+from eval_backend.core.config import Settings
+from eval_backend.models import Artifact, EvaluationRun, JobQueue, Submission
+from eval_backend.services import eval_runner
+from eval_backend.services.eval_runner import EvaluationResult
+
+
+def _build_settings(tmp_path: Path) -> Settings:
+    return Settings(
+        workspace_root=tmp_path / "workspaces",
+        artifact_root=tmp_path / "artifacts",
+        local_repo_dir=tmp_path,
+        eval_execution_mode="local_cpu",
+        eval_max_items=2,
+    )
+
+
+def _add_submission(session, checkpoint_path: Path, submission_id: str) -> Submission:
+    artifact = Artifact(
+        id=f"artifact-{submission_id}",
+        storage_backend="local",
+        storage_uri=str(checkpoint_path),
+        file_names_json=[checkpoint_path.name],
+        sha256="abc123",
+        size_bytes=checkpoint_path.stat().st_size,
+        mime_type="application/octet-stream",
+        meta_json={"checkpoint_path": str(checkpoint_path)},
+    )
+    session.add(artifact)
+    session.flush()
+    submission = Submission(
+        id=submission_id,
+        source="upload",
+        miner_id="miner-a",
+        benchmark_names_json=["math500"],
+        status="queued",
+        submission_artifact_id=artifact.id,
+    )
+    session.add(submission)
+    session.flush()
+    artifact.submission_id = submission.id
+    session.flush()
+    return submission
+
+
+def test_touch_progress_is_committed_not_just_flushed(validator_session, tmp_path, monkeypatch):
+    """Progress only helps if another process can read it mid-run."""
+    session = validator_session
+    checkpoint_path = tmp_path / "theta.npy"
+    checkpoint_path.write_bytes(b"theta")
+    submission = _add_submission(session, checkpoint_path, "sub-progress")
+    run = EvaluationRun(
+        submission_id=submission.id,
+        benchmark_names_json=["math500"],
+        status="running",
+    )
+    session.add(run)
+    session.flush()
+
+    committed: list[bool] = []
+    monkeypatch.setattr(session, "commit", lambda: committed.append(True))
+
+    eval_runner._touch_progress(
+        session,
+        run,
+        submission,
+        phase="evaluation_running",
+        message="item 3/20 running",
+        current=3,
+        total=20,
+    )
+
+    assert committed, "progress update was never committed, so the API cannot observe it"
+    assert run.phase == "evaluation_running"
+    assert run.progress_current == 3
+    assert run.progress_total == 20
+
+
+def test_worker_commits_the_claim_before_the_job_body_runs(validator_session, tmp_path, monkeypatch):
+    """The claim must be durable before the (up to 2h) evaluation starts."""
+    session = validator_session
+    settings = _build_settings(tmp_path)
+    checkpoint_path = tmp_path / "theta.npy"
+    checkpoint_path.write_bytes(b"theta")
+    submission = _add_submission(session, checkpoint_path, "sub-claim")
+    session.add(
+        JobQueue(
+            id="job-claim",
+            job_type="evaluation",
+            job_id=submission.id,
+            submission_id=submission.id,
+            queue_name="default",
+            status="queued",
+            priority=0,
+            dedupe_key="evaluation:sub-claim",
+            attempts=0,
+            max_attempts=3,
+            payload_json={"submission_id": submission.id},
+        )
+    )
+    session.flush()
+
+    commits: list[bool] = []
+    real_commit = session.commit
+
+    def _tracking_commit() -> None:
+        commits.append(True)
+        real_commit()
+
+    monkeypatch.setattr(session, "commit", _tracking_commit)
+    # process_once() closes the session it is handed; the fixture owns this one.
+    monkeypatch.setattr(session, "close", lambda: None)
+
+    observed: dict[str, object] = {}
+
+    def _fake_evaluate_submission(session_, submission_, settings_, **kwargs):
+        observed["commits_before_eval"] = len(commits)
+        observed["job_status"] = session_.get(JobQueue, "job-claim").status
+        now = datetime.now(timezone.utc)
+        run = EvaluationRun(
+            submission_id=submission_.id,
+            benchmark_names_json=["math500"],
+            status="completed",
+            score=0.5,
+            started_at=now,
+            finished_at=now,
+        )
+        session_.add(run)
+        session_.flush()
+        return EvaluationResult(run=run, score=0.5, metrics={}, stdout="", stderr="")
+
+    monkeypatch.setattr(worker, "evaluate_submission", _fake_evaluate_submission)
+
+    processed = worker.process_once(lambda: session, settings)
+
+    assert processed == 1
+    assert observed["job_status"] == "running"
+    assert observed["commits_before_eval"] >= 1, (
+        "the worker held the claim in an uncommitted transaction for the whole run, "
+        "so /api/jobs still reported the job as queued"
+    )