fix(runners): scope schema-validation crashes to the failing case

conformance/scripts/flow_runner.py

@@ -613,6 +613,28 @@ def run_flow_case(
     return result
 
 
+def run_flow_case_safely(
+    client: AdapterClient,
+    base_url: str,
+    flow_case: dict[str, Any],
+    cases_by_path: dict[str, dict[str, Any]],
+    verbose: bool,
+) -> dict[str, Any]:
+    """Contain per-case crashes so one bad response cannot void the whole suite.
+
+    AdapterClient.call raises (rather than returning an error response) when an
+    adapter's output fails schema validation; without this boundary that single
+    case would abort every remaining flow for the adapter.
+    """
+    name = str(flow_case.get("name"))
+    try:
+        return run_flow_case(client, base_url, flow_case, cases_by_path, verbose)
+    except Exception as exc:
+        if verbose:
+            print(f"[{client.adapter.name}] {name}: case error {exc}", file=sys.stderr)
+        return flow_error(name, 0, f"case_error: {exc}")
+
+
 def run_adapter_flows(adapter: AdapterConfig, base_url: str, verbose: bool) -> list[dict[str, Any]]:
     build_error = build_adapter(adapter)
     if build_error:
@@ -621,7 +643,7 @@ def run_adapter_flows(adapter: AdapterConfig, base_url: str, verbose: bool) -> l
     flow_cases = load_flow_cases()
     cases_by_path = {str(flow_case.get("path", "/")): flow_case for flow_case in flow_cases}
     return [
-        run_flow_case(client, base_url, flow_case, cases_by_path, verbose)
+        run_flow_case_safely(client, base_url, flow_case, cases_by_path, verbose)
         for flow_case in flow_cases
         if not flow_case.get("server_only")
     ]
@@ -788,6 +810,7 @@ def main() -> int:
             server.wait(timeout=5)
         except subprocess.TimeoutExpired:
             server.kill()
+            server.wait()
 
 
 if __name__ == "__main__":

conformance/scripts/server_verify_runner.py

@@ -66,11 +66,22 @@ def run_adapter(adapter: AdapterConfig, cases: list[dict[str, Any]]) -> list[Run
     results: list[RunResult] = []
     for case in cases:
         name = str(case.get("name"))
-        response = client.call(
-            "server.verify",
-            case.get("input"),
-            context={"caseName": name},
-        )
+        try:
+            response = client.call(
+                "server.verify",
+                case.get("input"),
+                context={"caseName": name},
+            )
+        except Exception as exc:
+            # Schema-validation failures raise out of call(); keep them scoped
+            # to the case instead of aborting the adapter's remaining cases.
+            results.append(RunResult(
+                adapter=adapter.name,
+                name=name,
+                passed=False,
+                error=str(exc),
+            ))
+            continue
         if not response.get("ok"):
             error = response.get("error") or {}
             results.append(RunResult(

conformance/scripts/test_flow_case_isolation.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Tests for per-case error containment in the flow runner."""
+
+from __future__ import annotations
+
+import unittest
+
+import flow_runner
+from harness import AdapterClient, AdapterConfig
+
+
+class RunFlowCaseSafelyTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.client = AdapterClient(AdapterConfig(name="go", command=["true"], capabilities=[]))
+        self.flow_case = {"name": "charge-success", "path": "/charge/success"}
+
+    def test_exception_becomes_per_case_failure(self) -> None:
+        original = flow_runner.run_flow_case
+
+        def explode(*args: object, **kwargs: object) -> dict[str, object]:
+            raise ValueError("go challenge.parse value failed schema validation")
+
+        flow_runner.run_flow_case = explode
+        try:
+            result = flow_runner.run_flow_case_safely(
+                self.client, "http://127.0.0.1:1", self.flow_case, {}, verbose=False
+            )
+        finally:
+            flow_runner.run_flow_case = original
+
+        self.assertEqual(result["name"], "charge-success")
+        self.assertFalse(result["outcome"]["ok"])
+        self.assertIn("case_error", result["outcome"]["error_type"])
+        self.assertIn("schema validation", result["outcome"]["error_type"])
+
+    def test_successful_case_passes_through_unchanged(self) -> None:
+        original = flow_runner.run_flow_case
+        expected = {"name": "charge-success", "outcome": {"ok": True, "status": 200}}
+        flow_runner.run_flow_case = lambda *args, **kwargs: expected
+        try:
+            result = flow_runner.run_flow_case_safely(
+                self.client, "http://127.0.0.1:1", self.flow_case, {}, verbose=False
+            )
+        finally:
+            flow_runner.run_flow_case = original
+
+        self.assertIs(result, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()