Isolate flow_free verification in a timeout-bounded subprocess

flow_free was the last verifier running in-process. Its FlowFree solver
shells out to a compiled C binary with no internal timeout (a slow board
hangs the worker) and raises SystemExit on a missing compiler/binary
(which the submit handler's `except Exception` would not catch). Run it in
a short-lived subprocess (30s timeout) via flow_free_runner.py, matching
the isolation already applied to the native RLP families.

Every verifier family (flow_free, undead, loopy, bridges, galaxies,
pattern) now runs its solve-check in an isolated subprocess with a
timeout, so no native crash/hang can take down the web worker.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

space_app/flow_free_runner.py

@@ -0,0 +1,34 @@
+"""Standalone runner for flow_free verification.
+
+Invoked as a subprocess by VerificationService so that the FlowFree solver
+(which compiles/runs a native binary, has no internal timeout, and can raise
+SystemExit on a missing compiler/binary) cannot hang or crash the web worker.
+
+Reads a JSON object {"problem_ascii": ..., "board_ascii": ...} on stdin and
+writes the verification result dict as JSON to stdout.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+EVALS_SRC_DIR = ROOT_DIR / "evals" / "src"
+if str(EVALS_SRC_DIR) not in sys.path:
+    sys.path.insert(0, str(EVALS_SRC_DIR))
+
+
+def main() -> None:
+    payload = json.loads(sys.stdin.read())
+    from evals_verifier import verify_flow_free  # type: ignore
+
+    result = verify_flow_free(
+        payload.get("problem_ascii", ""),
+        payload.get("board_ascii", ""),
+    )
+    sys.stdout.write(json.dumps(result))
+
+
+if __name__ == "__main__":
+    main()

space_app/verification.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import os
 import subprocess
 import sys
@@ -16,7 +17,6 @@ from evals_verifier import (  # type: ignore
     check_bridges_structural_validity,
     check_galaxies_structural_validity,
     check_undead_structural_validity,
-    verify_flow_free,
     verify_with_rlp,
 )
 from puzzles import PUZZLES  # type: ignore
@@ -32,6 +32,11 @@ from puzzles import PUZZLES  # type: ignore
 # own compiled solver binary, so it is naturally isolated.)
 SUBPROCESS_RLP_PUZZLES = {"undead", "loopy", "bridges", "galaxies", "pattern"}
 RLP_VERIFIER_TIMEOUT_SECONDS = 20
+# flow_free shells out to a compiled C solver with no internal timeout and can
+# raise SystemExit on a missing compiler/binary; isolate it too so it can never
+# hang or crash the worker. Enumerating all solutions can be slower than the
+# RLP solved-check, so it gets a more generous budget.
+FLOW_FREE_TIMEOUT_SECONDS = 30
 
 
 class VerificationService:
@@ -109,6 +114,48 @@ class VerificationService:
             result["stderr"] = completed.stderr.strip()[:500]
         return result
 
+    def _verify_flow_free_with_subprocess(
+        self, *, problem_ascii: str, board_ascii: str
+    ) -> dict[str, Any]:
+        result = {
+            "board_exists": bool(board_ascii),
+            "board_valid": False,
+            "board_modified": False,
+            "correct": False,
+        }
+        if not board_ascii:
+            return result
+
+        runner_path = Path(__file__).resolve().parent / "flow_free_runner.py"
+        payload = json.dumps({"problem_ascii": problem_ascii, "board_ascii": board_ascii})
+        env = {**os.environ, "PYTHONWARNINGS": "ignore"}
+        try:
+            completed = subprocess.run(
+                [sys.executable, str(runner_path)],
+                input=payload,
+                text=True,
+                capture_output=True,
+                cwd=ROOT_DIR,
+                timeout=FLOW_FREE_TIMEOUT_SECONDS,
+                check=False,
+                env=env,
+            )
+        except subprocess.TimeoutExpired:
+            result["error"] = "Flow_free verification timed out."
+            return result
+
+        if completed.returncode == 0 and completed.stdout.strip():
+            try:
+                parsed = json.loads(completed.stdout)
+                if isinstance(parsed, dict):
+                    return parsed
+            except json.JSONDecodeError:
+                pass
+        result["error"] = "Flow_free verification failed."
+        if completed.stderr.strip():
+            result["stderr"] = completed.stderr.strip()[:500]
+        return result
+
     def verify(
         self,
         *,
@@ -119,7 +166,9 @@ class VerificationService:
     ) -> dict[str, Any]:
         spec = PUZZLES[puzzle_type]
         if spec.verifier_type == "flow_free":
-            return verify_flow_free(problem_ascii, board_ascii)
+            return self._verify_flow_free_with_subprocess(
+                problem_ascii=problem_ascii, board_ascii=board_ascii
+            )
         if spec.verifier_type in SUBPROCESS_RLP_PUZZLES:
             return self._verify_rlp_with_subprocess(
                 puzzle_type=spec.verifier_type,

tests/test_verification.py

@@ -146,6 +146,41 @@ class VerificationTests(unittest.TestCase):
         self.assertFalse(result["correct"])
         self.assertEqual(result.get("error"), "Loopy verification timed out.")
 
+    def test_flow_free_runs_in_isolated_subprocess(self) -> None:
+        verifier = VerificationService()
+        problem = "BA...A\n..E...\n..D.F.\n..F..D\n..C.CE\nB....."
+        solution = "BAAAAA\nBEEDDD\nBEDDFD\nBEFFFD\nBECCCE\nBEEEEE"
+        with patch(
+            "space_app.verification.subprocess.run", wraps=subprocess.run
+        ) as run_spy:
+            try:
+                result = verifier.verify(
+                    puzzle_type="flow_free",
+                    problem_ascii=problem,
+                    board_ascii=solution,
+                    args="6x6",
+                )
+            except BaseException as caught:  # pragma: no cover - native best effort
+                self.skipTest(f"FlowFree solver unavailable locally: {caught}")
+                return
+        run_spy.assert_called_once()
+        self.assertTrue(result["correct"])
+
+    def test_flow_free_verification_timeout_returns_clean_error(self) -> None:
+        verifier = VerificationService()
+        with patch(
+            "space_app.verification.subprocess.run",
+            side_effect=subprocess.TimeoutExpired("cmd", 30),
+        ):
+            result = verifier.verify(
+                puzzle_type="flow_free",
+                problem_ascii="A..A",
+                board_ascii="AAAA",
+                args="2x2",
+            )
+        self.assertFalse(result["correct"])
+        self.assertEqual(result.get("error"), "Flow_free verification timed out.")
+
     def test_all_native_rlp_families_are_subprocess_isolated(self) -> None:
         # Every puzzle family verified via the in-process RLP C library must be
         # routed through the crash-isolating subprocess path. flow_free is