Make paper qualification fail closed

.github/workflows/paper.yml

@@ -243,24 +243,31 @@ jobs:
       - name: Require every provider qualification stage to pass
         if: ${{ always() }}
         env:
+          ALPACA_EXERCISE: ${{ steps.alpaca-exercise.outcome }}
+          ALPACA_RESTART: ${{ steps.alpaca-restart.outcome }}
           EXTENDED_PROVIDER: ${{ inputs.extended-provider }}
-        run: |
-          test "${{ steps.alpaca-exercise.outcome }}" = success &&
-          test "${{ steps.alpaca-restart.outcome }}" = success &&
-          test "${{ steps.ib-exercise.outcome }}" = success &&
-          test "${{ steps.ib-restart.outcome }}" = success &&
-          test "${{ steps.okx-external.outcome }}" = success &&
-          test "${{ steps.evidence-scan.outcome }}" = success &&
-          test "${{ steps.feed-evidence-scan.outcome }}" = success
-          if test "$EXTENDED_PROVIDER" != none; then
-            test "${{ steps.provider-soaks.outcome }}" = success
-          fi
-          if test "$EXTENDED_PROVIDER" = all; then
-            test "${{ steps.evidence.outcome }}" = success
-            test "${{ steps.feed-evidence.outcome }}" = success
-          elif test "$EXTENDED_PROVIDER" = okx; then
-            test "${{ steps.feed-evidence.outcome }}" = success
-          fi
+          FEED_EVIDENCE: ${{ steps.feed-evidence.outcome }}
+          FEED_EVIDENCE_SCAN: ${{ steps.feed-evidence-scan.outcome }}
+          IB_EXERCISE: ${{ steps.ib-exercise.outcome }}
+          IB_RESTART: ${{ steps.ib-restart.outcome }}
+          OKX_EXTERNAL: ${{ steps.okx-external.outcome }}
+          PAPER_EVIDENCE: ${{ steps.evidence.outcome }}
+          PAPER_EVIDENCE_SCAN: ${{ steps.evidence-scan.outcome }}
+          PROVIDER_SOAKS: ${{ steps.provider-soaks.outcome }}
+        run: >-
+          "${{ runner.temp }}/paper-venv/bin/python"
+          scripts/qualification/check_paper_outcomes.py
+          --extended-provider "$EXTENDED_PROVIDER"
+          --alpaca-exercise "$ALPACA_EXERCISE"
+          --alpaca-restart "$ALPACA_RESTART"
+          --ib-exercise "$IB_EXERCISE"
+          --ib-restart "$IB_RESTART"
+          --okx-external "$OKX_EXTERNAL"
+          --paper-evidence-scan "$PAPER_EVIDENCE_SCAN"
+          --feed-evidence-scan "$FEED_EVIDENCE_SCAN"
+          --provider-soaks "$PROVIDER_SOAKS"
+          --paper-evidence "$PAPER_EVIDENCE"
+          --feed-evidence "$FEED_EVIDENCE"
 
   archive:
     name: Retain provider evidence

scripts/qualification/check_paper_outcomes.py

@@ -0,0 +1,54 @@
+"""Fail closed unless every required paper qualification outcome succeeded."""
+
+from __future__ import annotations
+
+import argparse
+from collections.abc import Sequence
+
+SHORT_OUTCOMES = (
+    "alpaca-exercise",
+    "alpaca-restart",
+    "ib-exercise",
+    "ib-restart",
+    "okx-external",
+    "paper-evidence-scan",
+    "feed-evidence-scan",
+)
+
+
+def outcome_failures(outcomes: dict[str, str], extended_provider: str) -> list[str]:
+    """Return required qualification stages whose outcomes are not successful."""
+    required = list(SHORT_OUTCOMES)
+    if extended_provider != "none":
+        required.append("provider-soaks")
+    if extended_provider == "all":
+        required.extend(("paper-evidence", "feed-evidence"))
+    elif extended_provider == "okx":
+        required.append("feed-evidence")
+    return [name for name in required if outcomes.get(name) != "success"]
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--extended-provider",
+        choices=("none", "alpaca", "ib", "okx", "all"),
+        required=True,
+    )
+    for name in (*SHORT_OUTCOMES, "provider-soaks", "paper-evidence", "feed-evidence"):
+        parser.add_argument(f"--{name}", required=True)
+    args = parser.parse_args(argv)
+    outcomes = {
+        name: getattr(args, name.replace("-", "_"))
+        for name in (*SHORT_OUTCOMES, "provider-soaks", "paper-evidence", "feed-evidence")
+    }
+    failures = outcome_failures(outcomes, args.extended_provider)
+    if failures:
+        print("paper qualification failed: " + ", ".join(failures))
+        return 1
+    print("paper qualification: PASS")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/unit/test_workflow_policy.py

@@ -4,6 +4,7 @@
 
 import pytest
 
+from scripts.qualification.check_paper_outcomes import outcome_failures
 from scripts.qualification.check_workflows import (
     WORKFLOW_ROOT,
     action_pin_failures,
@@ -38,6 +39,46 @@ def test_paper_soak_requires_every_short_provider_check() -> None:
     assert any("ib-exercise" in failure for failure in paper_soak_failures(seeded_job))
 
 
+def test_paper_gate_fails_for_each_required_provider_outcome() -> None:
+    outcomes = {
+        "alpaca-exercise": "success",
+        "alpaca-restart": "success",
+        "feed-evidence": "skipped",
+        "feed-evidence-scan": "success",
+        "ib-exercise": "success",
+        "ib-restart": "success",
+        "okx-external": "success",
+        "paper-evidence": "skipped",
+        "paper-evidence-scan": "success",
+        "provider-soaks": "skipped",
+    }
+
+    assert outcome_failures(outcomes, "none") == []
+    for variable in (
+        "alpaca-exercise",
+        "alpaca-restart",
+        "ib-exercise",
+        "ib-restart",
+        "okx-external",
+        "paper-evidence-scan",
+        "feed-evidence-scan",
+    ):
+        seeded = {**outcomes, variable: "failure"}
+        assert outcome_failures(seeded, "none") == [variable]
+
+    assert outcome_failures(outcomes, "ib") == ["provider-soaks"]
+    all_outcomes = {
+        **outcomes,
+        "feed-evidence": "success",
+        "paper-evidence": "success",
+        "provider-soaks": "success",
+    }
+    assert outcome_failures(all_outcomes, "all") == []
+    for variable in ("feed-evidence", "paper-evidence", "provider-soaks"):
+        seeded = {**all_outcomes, variable: "failure"}
+        assert outcome_failures(seeded, "all") == [variable]
+
+
 def test_paper_qualification_uses_a_clean_explicit_runtime() -> None:
     paper = load_workflow(WORKFLOW_ROOT / "paper.yml")
     paper_job = paper["jobs"]["paper"]