Fix submission intake reliability

.github/workflows/score-from-r2.yml

@@ -31,18 +31,14 @@ permissions:
 env:
   PR_REF: v0.6.0
 
+concurrency:
+  group: score-from-r2
+  cancel-in-progress: false
+
 jobs:
   score:
     runs-on: ubuntu-latest
-    # R2_* are repository secrets (Settings → Secrets → Actions). No environment / pre-run
-    # approval — the single human checkpoint is reviewing + merging the aggregate PR this
-    # job opens. cron (1x/day) + the bearer key on the intake Worker bound abuse.
-    env:
-      R2_ENDPOINT: https://${{ secrets.R2_ACCOUNT_ID }}.r2.cloudflarestorage.com
-      BUCKET: ${{ secrets.R2_BUCKET }}
-      AWS_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
-      AWS_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
-      AWS_DEFAULT_REGION: auto
+    timeout-minutes: 360
     steps:
       - uses: actions/checkout@v7
 
@@ -64,6 +60,13 @@ jobs:
           cache-to: type=gha,mode=max
 
       - name: Pull pending submissions + prior scored detail from R2
+        id: pull
+        env:
+          R2_ENDPOINT: https://${{ secrets.R2_ACCOUNT_ID }}.r2.cloudflarestorage.com
+          BUCKET: ${{ secrets.R2_BUCKET }}
+          AWS_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+          AWS_DEFAULT_REGION: auto
         run: |
           mkdir -p incoming results/scored
           # reset_results (manual): wipe prior scored detail so the board is rebuilt from
@@ -72,30 +75,59 @@ jobs:
             echo "reset_results: wiping s3://$BUCKET/results"
             aws s3 rm "s3://$BUCKET/results" --recursive --endpoint-url "$R2_ENDPOINT" || true
           fi
-          # incoming/ = raw submissions to score; results/ = all prior scored detail, so the
-          # rebuilt board reflects the FULL history, not just this batch.
-          aws s3 sync "s3://$BUCKET/incoming" ./incoming --endpoint-url "$R2_ENDPOINT"
+          # Snapshot the exact queue keys before downloading. Only this immutable manifest
+          # may be archived later; an object arriving while scoring remains in incoming/.
+          aws s3api list-objects-v2 --bucket "$BUCKET" --prefix incoming/ \
+            --endpoint-url "$R2_ENDPOINT" --query 'Contents[].Key' --output json \
+            | jq -r '.[] | select(endswith(".json"))' > queue-manifest.txt
+          while IFS= read -r key; do
+            [ -n "$key" ] || continue
+            rel="${key#incoming/}"
+            [ "$rel" != "$key" ] && [ -n "$rel" ] || continue
+            mkdir -p "incoming/$(dirname "$rel")"
+            aws s3 cp "s3://$BUCKET/$key" "incoming/$rel" --endpoint-url "$R2_ENDPOINT"
+          done < queue-manifest.txt
+
+          # results/ = all prior scored detail, so the rebuilt board reflects full history.
           aws s3 sync "s3://$BUCKET/results"  ./results/scored --endpoint-url "$R2_ENDPOINT"
-          echo "pending: $(ls -1 incoming/*.json 2>/dev/null | wc -l) file(s)"
+          echo "pending snapshot: $(wc -l < queue-manifest.txt) object(s)"
 
       - name: Re-verify with pred (zero-trust + provenance), rebuild the leaderboard
+        id: score
+        continue-on-error: true
         run: |
           docker run --rm --network none \
             -v "$PWD/incoming:/app/submissions" \
             -v "$PWD/results:/app/results" \
             --entrypoint python prb-scoring:latest \
-            -m benchmark.backend_score --local /app/submissions /app/results/scored
+            -m benchmark.backend_score --official \
+            --local /app/submissions /app/results/scored
 
-      - name: Persist scored detail privately (R2), archive processed submissions
+      - name: Persist results and transition only the snapshotted queue objects
+        if: always() && steps.pull.outcome == 'success'
+        env:
+          R2_ENDPOINT: https://${{ secrets.R2_ACCOUNT_ID }}.r2.cloudflarestorage.com
+          BUCKET: ${{ secrets.R2_BUCKET }}
+          AWS_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+          AWS_DEFAULT_REGION: auto
         run: |
           # Scored detail carries certificates — it goes back to PRIVATE R2, never to git.
           aws s3 sync ./results/scored "s3://$BUCKET/results" --endpoint-url "$R2_ENDPOINT"
-          # Move the now-scored raw submissions out of the queue. Runs only when the scoring
-          # step exited 0 (no FAILED), so we never archive an un-scored item. Limitation: the
-          # archive is coarse (whole incoming/ at once), so a permanently-malformed submission
-          # FAILs every run and jams the queue until deleted from R2 incoming/ by hand.
-          aws s3 mv "s3://$BUCKET/incoming" "s3://$BUCKET/processed" \
-            --recursive --endpoint-url "$R2_ENDPOINT" || true
+
+          # FINISHED objects move to processed/. Permanent input failures move to failed/
+          # with their diagnostic status. Retryable or unclassified failures stay queued.
+          python -m benchmark.r2_queue \
+            --manifest queue-manifest.txt --incoming-dir incoming > queue-transitions.tsv
+          while IFS=$'\t' read -r source destination status_path status_destination; do
+            [ -n "$source" ] || continue
+            if [ "$status_destination" != "-" ]; then
+              aws s3 cp "$status_path" "s3://$BUCKET/$status_destination" \
+                --endpoint-url "$R2_ENDPOINT"
+            fi
+            aws s3 mv "s3://$BUCKET/$source" "s3://$BUCKET/$destination" \
+              --endpoint-url "$R2_ENDPOINT"
+          done < queue-transitions.tsv
 
       - name: Open ONE PR per new/changed submission entry (main is protected)
         env:
@@ -141,3 +173,10 @@ jobs:
           done
           echo "opened/updated $opened submission PR(s)"
           git checkout -q "$BASE" 2>/dev/null || true
+
+      - name: Report retryable scoring failure
+        if: always() && steps.score.outcome == 'failure'
+        run: |
+          echo "One or more submissions had a retryable verifier/infrastructure failure."
+          echo "They remain under R2 incoming/ and will be retried on the next run."
+          exit 1

CONTRIBUTING.md

@@ -210,15 +210,17 @@ python -m benchmark.backend_score --local submissions/ results/scored/
 
 For each `PENDING` submission it:
 
-1. flips status `PENDING → RUNNING`,
+1. validates the current schema, pinned library commit, and clean-run status, then flips
+   status `PENDING → RUNNING`,
 2. re-runs `benchmark/verify_submission.py` — which calls `verify()` on **every**
    certificate and re-derives the bundle from `pred`, so a fabricated or tampered
    counterexample is rejected, and checks new runs against the bounded submit ledger
    (legacy submissions retain trajectory provenance),
 3. recomputes `bugs_found` as **distinct rules with a confirmed bug** (many certificates
    for one rule collapse to one — no count padding),
-4. writes the scored result + a ranked `leaderboard.json`, and sets status
-   `FINISHED` (or `FAILED` with a reason).
+4. writes the scored result + a ranked `leaderboard.json`, and sets status `FINISHED`; a
+   permanent input error is isolated under R2 `failed/`, while a retryable verifier or
+   infrastructure failure remains in `incoming/` for the next run.
 
 In production this runs unattended inside GitHub Actions: `score-from-r2.yml` pulls pending
 submissions from private R2, re-verifies them, and opens a PR with the refreshed aggregate;

benchmark/backend_score.py

@@ -23,9 +23,16 @@
 import sys
 from pathlib import Path
 
+from benchmark.env_setup import pinned_commit
+from benchmark.submit import validate_submission
 from benchmark.verify_submission import leaderboard_entry, score_submission
 
 STATUS_SUFFIX = ".status.json"
+OFFICIAL_SCHEMA_VERSION = "2.1"
+
+
+class PermanentSubmissionError(ValueError):
+    """A submission-data error that retrying the same object cannot fix."""
 
 # One PUBLIC file per submission lives at site/results/<slug>.json, where the slug ties the
 # file (and its PR branch) to that specific run: model + submission time + a short id. The
@@ -156,15 +163,53 @@ def _pending_submissions(subs_dir: Path) -> list[Path]:
         if p.name.endswith(STATUS_SUFFIX):
             continue
         st = _read_status(p)
-        if st and st.get("status") == "FINISHED":
-            continue
+        if st:
+            if st.get("status") == "FINISHED":
+                continue
+            if st.get("status") == "FAILED" and st.get("retryable") is False:
+                continue
         out.append(p)
     return out
 
 
 # ── scoring one submission ────────────────────────────────────────────────────
 
-def score_one(sub_path: Path, results_dir: Path, repo_dir: str | None = None) -> dict:
+def _validate_for_scoring(submission: object, *, official: bool,
+                          expected_commit: str | None) -> dict:
+    """Validate the durable queue envelope before invoking ``pred``.
+
+    Basic validation applies to local and production scoring so malformed input is a
+    permanent per-object failure instead of an exception that jams the whole queue.  The
+    official gate additionally fixes the schema and target commit, and rejects partial runs
+    from the public leaderboard. Test submissions may be partial because they never publish.
+    """
+    if not isinstance(submission, dict):
+        raise PermanentSubmissionError("submission is not a JSON object")
+
+    problems = validate_submission(submission)
+
+    if official:
+        if submission.get("schema_version") != OFFICIAL_SCHEMA_VERSION:
+            problems.append(
+                f"official submissions require schema_version {OFFICIAL_SCHEMA_VERSION}")
+        target_commit = expected_commit or pinned_commit()
+        if submission.get("library_commit") != target_commit:
+            problems.append(
+                "library_commit does not match this benchmark round "
+                f"(expected {target_commit})")
+        if submission.get("run_error") and not submission.get("test"):
+            problems.append(
+                "partial run has run_error; resubmit a clean run or upload it with --test")
+
+    if problems:
+        # Preserve order while removing duplicate messages emitted by overlapping checks.
+        unique = list(dict.fromkeys(problems))
+        raise PermanentSubmissionError("; ".join(unique))
+    return submission
+
+
+def score_one(sub_path: Path, results_dir: Path, repo_dir: str | None = None, *,
+              official: bool = False, expected_commit: str | None = None) -> dict:
     """Score a single submission file. Returns its leaderboard entry.
 
     Writes the scored results.json (results.schema-shaped) to results_dir/<stem>.json and
@@ -173,7 +218,12 @@ def score_one(sub_path: Path, results_dir: Path, repo_dir: str | None = None) ->
     results_dir.mkdir(parents=True, exist_ok=True)
     _write_status(sub_path, "RUNNING")
     try:
-        submission = json.loads(sub_path.read_text(encoding="utf-8"))
+        try:
+            raw_submission = json.loads(sub_path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, UnicodeError) as e:
+            raise PermanentSubmissionError(f"invalid submission JSON: {e}") from e
+        submission = _validate_for_scoring(
+            raw_submission, official=official, expected_commit=expected_commit)
         scored, report = score_submission(submission, repo_dir)
         entry = leaderboard_entry(submission, scored)
         (results_dir / f"{sub_path.stem}.json").write_text(
@@ -182,8 +232,9 @@ def score_one(sub_path: Path, results_dir: Path, repo_dir: str | None = None) ->
                       model=scored["model"], bugs_found=scored["bugs_found"],
                       verdicts=report)
         return entry
-    except Exception as e:  # any failure → FAILED status with the reason (user feedback)
-        _write_status(sub_path, "FAILED", error=str(e))
+    except Exception as e:  # isolate permanent input failures from retryable infra failures
+        retryable = not isinstance(e, PermanentSubmissionError)
+        _write_status(sub_path, "FAILED", error=str(e), retryable=retryable)
         raise
 
 
@@ -230,19 +281,28 @@ def aggregate_leaderboard(results_dir: Path) -> list[dict]:
 
 # ── local queue ───────────────────────────────────────────────────────────────
 
-def process_local(subs_dir: str, results_dir: str, repo_dir: str | None = None) -> list[dict]:
+def process_local(subs_dir: str, results_dir: str, repo_dir: str | None = None, *,
+                  official: bool = False,
+                  expected_commit: str | None = None) -> list[dict]:
     """Score all pending submissions in subs_dir; return a per-submission summary."""
     _assert_pred_version()
     subs = Path(subs_dir)
     results = Path(results_dir)
     summary = []
     for sub_path in _pending_submissions(subs):
         try:
-            entry = score_one(sub_path, results, repo_dir)
+            entry = score_one(
+                sub_path, results, repo_dir,
+                official=official, expected_commit=expected_commit)
             summary.append({"submission": sub_path.name, "status": "FINISHED",
                             "model": entry["model"], "bugs_found": entry["bugs_found"]})
         except Exception as e:
-            summary.append({"submission": sub_path.name, "status": "FAILED", "error": str(e)})
+            summary.append({
+                "submission": sub_path.name,
+                "status": "FAILED",
+                "error": str(e),
+                "retryable": not isinstance(e, PermanentSubmissionError),
+            })
     aggregate_leaderboard(results)
     # Public per-submission entry files (one PR each downstream); test entries are excluded.
     write_board_entries(results, results / "board")
@@ -258,6 +318,12 @@ def main() -> None:
                       help="Aggregate per-submission entry files (site/results/*.json) into "
                            "a ranked leaderboard JSON (the deployed site/results.json)")
     parser.add_argument("--repo-dir", default=None, help="problem-reductions repo (default: pred on PATH)")
+    parser.add_argument(
+        "--official", action="store_true",
+        help="Require the current schema, pinned library commit, and a clean non-test run")
+    parser.add_argument(
+        "--expected-commit", default=None,
+        help="Override the official target commit (tests/operations; default: image pin)")
     args = parser.parse_args()
 
     if args.build_board:
@@ -266,7 +332,9 @@ def main() -> None:
         print(f"built {args.build_board[1]}: {len(board)} model(s)")
         return
 
-    summary = process_local(args.local[0], args.local[1], args.repo_dir)
+    summary = process_local(
+        args.local[0], args.local[1], args.repo_dir,
+        official=args.official, expected_commit=args.expected_commit)
 
     for s in summary:
         line = f"{s['status']:8}  {s['submission']}"
@@ -276,13 +344,14 @@ def main() -> None:
             line += f"  ({s.get('error', '')})"
         print(line)
     n_failed = sum(1 for s in summary if s["status"] == "FAILED")
+    n_retryable = sum(
+        1 for s in summary if s["status"] == "FAILED" and s.get("retryable", True))
     print(f"\n{sum(1 for s in summary if s['status'] == 'FINISHED')} scored, "
-          f"{n_failed} failed")
-    # A FAILED status is an infra/verification error (crash, pred error), NOT a legit
-    # "no bug" verdict (that is FINISHED with bugs_found=0). Exit non-zero so the caller
-    # (score-from-r2.yml) stops BEFORE archiving incoming/ → processed/ — an un-scored
-    # submission must stay queued for retry, never be silently lost.
-    if n_failed:
+          f"{n_failed} failed ({n_retryable} retryable)")
+    # Permanent input failures are quarantined object-by-object by the R2 workflow. Only a
+    # retryable failure makes the process non-zero, so transient verifier/infra failures stay
+    # in incoming/ while other submissions can still be persisted and published.
+    if n_retryable:
         sys.exit(1)