Add runner parameter to all agent patterns for pluggable execution

Signed-off-by: Cong Wang <cwang@multikernel.io>

Author: congwang-mk

src/branching/__init__.py

@@ -44,6 +44,7 @@
     "Branch",
     # Process
     "BranchContext",
+    "Runner",
     # Agent patterns
     "Speculate",
     "BestOfN",
@@ -78,6 +79,7 @@
     "Branch": ".core.branch",
     # Process layer
     "BranchContext": ".process.context",
+    "Runner": ".process.runner",
     # Agent layer
     "Speculate": ".agent.speculate",
     "BestOfN": ".agent.patterns",

src/branching/agent/patterns.py

@@ -10,7 +10,7 @@
 from typing import Callable, Optional, Sequence
 
 from ..core.workspace import Workspace
-from ..process.runner import run_in_process
+from ..process.runner import _default_runner, Runner
 from ..exceptions import ConflictError
 from .result import SpeculationResult, SpeculationOutcome
 
@@ -51,12 +51,14 @@ def __init__(
         evaluate: Callable[[Path], float] | None = None,
         timeout: float | None = None,
         commit: bool = True,
+        runner: Runner | None = None,
     ):
         self._candidates = list(candidates)
         self._scores = list(scores) if scores is not None else None
         self._evaluate = evaluate
         self._timeout = timeout
         self._commit = commit
+        self._runner = runner or _default_runner
 
     def _score(self, ret, path, index):
         """Parse candidate return and apply optional evaluator."""
@@ -87,9 +89,8 @@ def _run_candidate(index: int) -> None:
                 ) as b:
                     result.branch_path = b.path
                     try:
-                        ret = run_in_process(
+                        ret = self._runner(
                             self._candidates[index], (b.path,),
-                            workspace=b.path,
                         )
                         success, score = self._score(ret, b.path, index)
                         result.success = success
@@ -182,17 +183,21 @@ def __init__(
         max_retries: int = 3,
         *,
         critique: Optional[Callable[[Path], str]] = None,
+        runner: Runner | None = None,
     ):
         """
         Args:
             task: Callable(path, attempt, feedback) -> success.
                 feedback is None on first attempt, critique output thereafter.
             max_retries: Maximum number of attempts.
             critique: Optional callable(path) -> feedback_string.
+            runner: Execution strategy for candidates. Default forks via
+                BranchContext. Pass a sandlock runner for confinement.
         """
         self._task = task
         self._max_retries = max_retries
         self._critique = critique
+        self._runner = runner or _default_runner
 
     def __call__(self, workspace: Workspace) -> SpeculationOutcome:
         return self._run(workspace)
@@ -211,9 +216,8 @@ def _run(self, workspace: Workspace) -> SpeculationOutcome:
                     branch_name, on_success=None, on_error="abort"
                 ) as b:
                     result.branch_path = b.path
-                    success = run_in_process(
+                    success = self._runner(
                         self._task, (b.path, attempt, feedback),
-                        workspace=b.path,
                     )
                     result.success = bool(success)
                     result.return_value = success
@@ -287,6 +291,7 @@ def __init__(
         ] | None = None,
         max_depth: int = 1,
         timeout: float | None = None,
+        runner: Runner | None = None,
     ):
         """
         Args:
@@ -299,12 +304,14 @@ def __init__(
                 Only used when max_depth > 1.
             max_depth: Maximum exploration depth (1 = single level).
             timeout: Per-level timeout in seconds.
+            runner: Execution strategy for candidates.
         """
         self._strategies = list(strategies)
         self._evaluate = evaluate
         self._expand = expand
         self._max_depth = max_depth
         self._timeout = timeout
+        self._runner = runner or _default_runner
 
     def __call__(self, workspace: Workspace) -> SpeculationOutcome:
         if self._expand is None or self._max_depth <= 1:
@@ -338,9 +345,8 @@ def _run(index: int) -> None:
                 ) as b:
                     result.branch_path = b.path
                     try:
-                        ret = run_in_process(
+                        ret = self._runner(
                             strategies[index], (b.path,),
-                            workspace=b.path,
                         )
                         if isinstance(ret, (tuple, list)):
                             success, score = ret
@@ -506,13 +512,15 @@ def __init__(
         beam_width: int = 3,
         max_depth: int = 2,
         timeout: float | None = None,
+        runner: Runner | None = None,
     ):
         self._strategies = list(strategies)
         self._expand = expand
         self._evaluate = evaluate
         self._beam_width = beam_width
         self._max_depth = max_depth
         self._timeout = timeout
+        self._runner = runner or _default_runner
 
     def _score(self, ret, path):
         """Parse strategy return and apply optional evaluator."""
@@ -563,9 +571,8 @@ def _beam_worker(index: int) -> None:
                     result.branch_path = b.path
                     beam_branches[index] = b
                     try:
-                        ret = run_in_process(
+                        ret = self._runner(
                             self._strategies[index], (b.path,),
-                            workspace=b.path,
                         )
                         result.success, result.score = self._score(
                             ret, b.path
@@ -659,9 +666,8 @@ def _sub_worker(idx: int, _d: int = _depth) -> None:
                         ) as sb:
                             result.branch_path = sb.path
                             try:
-                                ret = run_in_process(
+                                ret = self._runner(
                                     strategy, (sb.path,),
-                                    workspace=sb.path,
                                 )
                                 result.success, result.score = self._score(
                                     ret, sb.path
@@ -775,6 +781,7 @@ def __init__(
         *,
         judge: Callable[[Path, Path], int],
         timeout: float | None = None,
+        runner: Runner | None = None,
     ):
         """
         Args:
@@ -784,10 +791,12 @@ def __init__(
             judge: Callable(path_a, path_b) -> 0 (a wins) or 1 (b wins).
                    Compares two candidates' branches during elimination.
             timeout: Overall timeout in seconds.
+            runner: Execution strategy for candidates.
         """
         self._candidates = list(candidates)
         self._judge = judge
         self._timeout = timeout
+        self._runner = runner or _default_runner
 
     @staticmethod
     def _run_bracket(
@@ -830,9 +839,8 @@ def _run_candidate(index: int) -> None:
                     result.branch_path = b.path
                     branch_paths[index] = b.path
                     try:
-                        success = run_in_process(
+                        success = self._runner(
                             self._candidates[index], (b.path,),
-                            workspace=b.path,
                         )
                         result.success = bool(success)
                         result.return_value = success
@@ -940,6 +948,7 @@ def __init__(
         fan_out: Sequence[int] = (1, 2, 4),
         timeout: float | None = None,
         wave_timeout: float | None = None,
+        runner: Runner | None = None,
     ):
         """
         Args:
@@ -955,11 +964,13 @@ def __init__(
                 ``len(fan_out)``.
             timeout: Overall timeout in seconds across all waves.
             wave_timeout: Per-wave timeout in seconds.
+            runner: Execution strategy for candidates.
         """
         self._task = task
         self._fan_out = list(fan_out)
         self._timeout = timeout
         self._wave_timeout = wave_timeout
+        self._runner = runner or _default_runner
 
     def __call__(self, workspace: Workspace) -> SpeculationOutcome:
         return self._run(workspace)
@@ -1123,14 +1134,13 @@ def _run_task(
         feedback: list[str],
         timeout: Optional[float] = None,
     ) -> tuple[bool, str]:
-        """Run the task in a forked child."""
+        """Run the task via the configured runner."""
         from ..exceptions import ProcessBranchError
 
         try:
-            ret = run_in_process(
+            ret = self._runner(
                 self._task,
                 (path, feedback),
-                workspace=path,
                 timeout=timeout,
             )
             if isinstance(ret, (tuple, list)):

src/branching/agent/speculate.py

@@ -9,6 +9,7 @@
 import threading
 
 from ..core.workspace import Workspace
+from ..process.runner import _default_runner, Runner
 from ..exceptions import ConflictError
 from .result import SpeculationResult, SpeculationOutcome
 
@@ -32,6 +33,7 @@ def __init__(
         first_wins: bool = True,
         max_parallel: int | None = None,
         timeout: float | None = None,
+        runner: Runner | None = None,
     ):
         """
         Args:
@@ -41,11 +43,14 @@ def __init__(
                 abort siblings. If False, run all and commit the first success.
             max_parallel: Maximum parallel workers (default: len(candidates)).
             timeout: Overall timeout in seconds for all candidates.
+            runner: Execution strategy for candidates. Default forks via
+                BranchContext. Pass a sandlock runner for confinement.
         """
         self._candidates = list(candidates)
         self._first_wins = first_wins
         self._max_parallel = max_parallel or len(self._candidates)
         self._timeout = timeout
+        self._runner = runner or _default_runner
 
     def __call__(self, workspace: Workspace) -> SpeculationOutcome:
         return self._run(workspace)
@@ -140,8 +145,7 @@ def _run_candidate(index: int) -> SpeculationResult:
         )
 
     def _run_in_branch(self, path: Path, index: int) -> bool:
-        """Run a candidate in a forked child."""
-        from ..process.runner import run_in_process
+        """Run a candidate via the configured runner."""
         from ..exceptions import ProcessBranchError
 
         per_candidate = (
@@ -151,10 +155,9 @@ def _run_in_branch(self, path: Path, index: int) -> bool:
         )
 
         try:
-            result = run_in_process(
+            result = self._runner(
                 self._candidates[index],
                 (path,),
-                workspace=path,
                 timeout=per_candidate,
             )
             return bool(result)

src/branching/process/__init__.py

@@ -1,3 +1,4 @@
 from .context import BranchContext
+from .runner import Runner, _default_runner
 
-__all__ = ["BranchContext"]
+__all__ = ["BranchContext", "Runner", "_default_runner"]

src/branching/process/runner.py

@@ -11,6 +11,16 @@
 from ..exceptions import ProcessBranchError
 from .context import BranchContext
 
+# Runner protocol: (fn, args, *, timeout) -> Any
+# Used by agent patterns to execute candidates. Swappable to use
+# sandlock or run candidates directly without forking.
+Runner = Callable[..., Any]
+
+
+def _default_runner(fn: Callable, args: tuple, *, timeout: float | None = None) -> Any:
+    """Default runner: fork via BranchContext, pass result via pipe."""
+    return run_in_process(fn, args, workspace=args[0], timeout=timeout)
+
 
 def run_in_process(
     fn: Callable,

tests/test_speculate.py

@@ -13,12 +13,12 @@
 
 
 @pytest.fixture(autouse=True)
-def mock_run_in_process():
+def mock_runner():
     """Run tasks in-process for unit tests (skip fork)."""
-    def _run_inline(fn, args, workspace, **kwargs):
+    def _run_inline(fn, args, **kwargs):
         return fn(*args)
-    with patch("branching.agent.patterns.run_in_process", _run_inline), \
-         patch("branching.process.runner.run_in_process", _run_inline):
+    with patch("branching.agent.patterns._default_runner", _run_inline), \
+         patch("branching.agent.speculate._default_runner", _run_inline):
         yield