Fail closed when harness pass@end is zero

docs/evaluation-protocol.md

@@ -129,7 +129,11 @@ omit hostnames and raw completions.
 
 Report `pass@1` separately from `pass@end`. Escalation rate is the
 fraction of cases that called `strong`. Stub `mcp_ms` is not GPU
-latency. Live `mcp_ms` is.
+latency. Live `mcp_ms` is. `scripts/run_harness.py` still prints the
+table, then exits non-zero when `pass@end` is 0 (or below
+`--min-pass-end`). Printed rows are not a pass. GitHub Actions stub
+MCP golden / apply / failover steps invoke that same script, so a
+stub that refuses every call reddens those steps.
 
 That harness is **local failover after a task was already delegated**.
 It does not prove that the premium model chose to delegate, nor that it

scripts/run_harness.py

@@ -15,8 +15,11 @@
     sys.path.insert(0, str(SRC))
 
 from local_coding_slm.eval.harness import (  # noqa: E402
+    campaign_pass_end,
     format_orchestrated,
     format_summary,
+    harness_exit_code,
+    orchestrated_pass_end,
     run_campaign,
     run_orchestrated_campaign,
 )
@@ -37,6 +40,12 @@ def main() -> None:
         action="store_true",
         help="Route + MCP local loop + premium apply gate (scripted reviewer)",
     )
+    parser.add_argument(
+        "--min-pass-end",
+        type=float,
+        default=0.0,
+        help="Fail unless pass@end is at least this. pass@end of 0 always fails.",
+    )
     parser.add_argument(
         "--out",
         default="",
@@ -76,7 +85,8 @@ def main() -> None:
                 encoding="utf-8",
             )
             print(f"wrote {dest / 'orchestrated.json'}")
-        raise SystemExit(0 if results else 1)
+        pass_end, n = orchestrated_pass_end(results)
+        raise SystemExit(harness_exit_code(pass_end, n, args.min_pass_end))
     rows = asyncio.run(
         run_campaign(
             backend=args.backend,
@@ -97,7 +107,8 @@ def main() -> None:
             encoding="utf-8",
         )
         print(f"wrote {dest / 'attempts.jsonl'}")
-    raise SystemExit(0 if rows else 1)
+    pass_end, n = campaign_pass_end(rows)
+    raise SystemExit(harness_exit_code(pass_end, n, args.min_pass_end))
 
 
 if __name__ == "__main__":

src/local_coding_slm/eval/harness.py

@@ -274,6 +274,31 @@ async def _one_attempt(
     return LocalAttempt(record=record, text=text, scored=scored)
 
 
+def campaign_pass_end(rows: list[AttemptRecord]) -> tuple[float, int]:
+    stats = summarize(rows)
+    return float(stats["pass_end"]), int(stats["cases"])
+
+
+def orchestrated_pass_end(results: list[JobResult]) -> tuple[float, int]:
+    delegated = [item for item in results if item.delegated]
+    n = len(delegated)
+    if n == 0:
+        return 0.0, 0
+    passed = sum(1 for item in delegated if item.local_passed)
+    return passed / n, n
+
+
+def harness_exit_code(
+    pass_end: float,
+    n: int,
+    min_pass_end: float = 0.0,
+) -> int:
+    """Fail-closed: no rows or pass@end of 0 is never success."""
+    if n <= 0 or pass_end <= 0.0 or pass_end < min_pass_end:
+        return 1
+    return 0
+
+
 def format_summary(rows: list[AttemptRecord]) -> str:
     stats = summarize(rows)
     lines = [
@@ -296,9 +321,11 @@ def format_summary(rows: list[AttemptRecord]) -> str:
 
 
 def format_orchestrated(results: list[JobResult]) -> str:
+    pass_end, delegated_n = orchestrated_pass_end(results)
     lines = [
         f"jobs={len(results)} applied={sum(1 for item in results if item.applied)} "
-        f"delegated={sum(1 for item in results if item.delegated)}"
+        f"delegated={sum(1 for item in results if item.delegated)}",
+        f"pass@end={pass_end:.2f} delegated_jobs={delegated_n}",
     ]
     for item in results:
         models = list(item.local_models) or "-"

src/local_coding_slm/eval/stub_ollama.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from threading import Thread
@@ -69,6 +70,8 @@ def reply(self, model_tag: str, user: str) -> str:
         delay = self.strong_ms if choice == "strong" else self.fast_ms
         if delay > 0:
             time.sleep(delay / 1000.0)
+        if os.environ.get("LOCAL_CODING_SLM_STUB_REFUSE") == "1":
+            return "REFUSED"
         if case_id == "unknown":
             return "I cannot match that task."
         return scripted_content(case_id, choice, self.visits[key], self.profile)

tests/test_harness_exit.py

@@ -0,0 +1,99 @@
+"""Harness exits on pass@end, not row count. Stub sabotage, no GPU."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+from local_coding_slm.eval.harness import harness_exit_code
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKFLOW = ROOT / ".github" / "workflows" / "tests.yml"
+STUB_MCP_STEPS = (
+    "Stub MCP golden",
+    "Stub MCP + apply gate",
+    "Stub MCP observed failover",
+)
+
+
+def _run_harness(*args: str, refuse: bool = False) -> subprocess.CompletedProcess[str]:
+    env = os.environ.copy()
+    env["PYTHONPATH"] = str(ROOT / "src") + os.pathsep + env.get("PYTHONPATH", "")
+    if refuse:
+        env["LOCAL_CODING_SLM_STUB_REFUSE"] = "1"
+    else:
+        env.pop("LOCAL_CODING_SLM_STUB_REFUSE", None)
+    return subprocess.run(
+        [sys.executable, str(ROOT / "scripts" / "run_harness.py"), *args],
+        cwd=str(ROOT),
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def _workflow_steps(text: str) -> list[dict[str, str]]:
+    steps: list[dict[str, str]] = []
+    current: dict[str, str] | None = None
+    for raw in text.splitlines():
+        stripped = raw.strip()
+        if stripped.startswith("- name:"):
+            if current is not None:
+                steps.append(current)
+            current = {"name": stripped.split(":", 1)[1].strip()}
+        elif current is not None and stripped.startswith("run:"):
+            current["run"] = stripped.split(":", 1)[1].strip()
+        elif current is not None and stripped.startswith("continue-on-error:"):
+            current["continue-on-error"] = stripped.split(":", 1)[1].strip()
+    if current is not None:
+        steps.append(current)
+    return steps
+
+
+class HarnessExitCodeTests(unittest.TestCase):
+    def test_empty_or_zero_pass_end_is_failure(self) -> None:
+        self.assertEqual(harness_exit_code(0.0, 0), 1)
+        self.assertEqual(harness_exit_code(0.0, 4), 1)
+
+    def test_positive_pass_end_respects_minimum(self) -> None:
+        self.assertEqual(harness_exit_code(1.0, 4), 0)
+        self.assertEqual(harness_exit_code(0.5, 4, min_pass_end=1.0), 1)
+        self.assertEqual(harness_exit_code(1.0, 4, min_pass_end=1.0), 0)
+
+
+class HarnessRefuseExitTests(unittest.TestCase):
+    def test_stub_refuse_every_call_exits_nonzero(self) -> None:
+        proc = _run_harness(
+            "--backend",
+            "stub",
+            "--profile",
+            "golden",
+            "--fast-ms",
+            "1",
+            "--strong-ms",
+            "1",
+            "--case",
+            "whitespace_extract",
+            refuse=True,
+        )
+        self.assertNotEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("pass@end=0.00", proc.stdout)
+        self.assertIn("whitespace_extract#1", proc.stdout)
+
+    def test_ci_stub_mcp_steps_invoke_harness_without_continue(self) -> None:
+        text = WORKFLOW.read_text(encoding="utf-8")
+        by_name = {step["name"]: step for step in _workflow_steps(text)}
+        for name in STUB_MCP_STEPS:
+            with self.subTest(step=name):
+                self.assertIn(name, by_name)
+                run = by_name[name].get("run", "")
+                self.assertIn("scripts/run_harness.py", run)
+                self.assertNotIn("continue-on-error", by_name[name])
+
+
+if __name__ == "__main__":
+    unittest.main()