Add Stage/Pipeline for piping between sandboxes

Sandbox.cmd() returns a lazy Stage.  Stages chain with | into a
Pipeline.  Each stage runs in its own Landlock+seccomp sandbox;
inter-stage data flows through kernel pipe buffers the parent
never holds.

This enables XOA (Execute-Only Agents): a planner with LLM access
pipes a script to an executor with data access.  Disjoint policies
ensure untrusted data never reaches the LLM.

    result = (
        Sandbox(planner_policy).cmd(["python3", "plan.py"])
        | Sandbox(executor_policy).cmd(["python3", "-"])
    ).run()

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

Author: congwang-mk

src/sandlock/__init__.py

@@ -8,7 +8,7 @@
 from ._version import __version__
 from .policy import Policy, FsIsolation, BranchAction, parse_ports
 from .sandbox import Sandbox
-from ._runner import Result
+from ._runner import Result, Stage, Pipeline
 from ._checkpoint import Checkpoint
 from ._notif_policy import NotifPolicy, NotifAction, PathRule
 from ._profile import load_profile, list_profiles
@@ -32,6 +32,8 @@
     "__version__",
     # Core API
     "Sandbox",
+    "Stage",
+    "Pipeline",
     "Policy",
     "FsIsolation",
     "BranchAction",

src/sandlock/_runner.py

@@ -7,12 +7,15 @@
 import os
 import signal
 from dataclasses import dataclass, field
-from typing import Any, Callable
+from typing import Any, Callable, TYPE_CHECKING
 
 from .exceptions import SandboxError
 from ._context import SandboxContext
 from .policy import Policy
 
+if TYPE_CHECKING:
+    from .sandbox import Sandbox
+
 
 @dataclass
 class Result:
@@ -217,3 +220,331 @@ def _drain_and_close(fd: int) -> None:
         os.close(fd)
     except OSError:
         pass
+
+
+# --- Stage / Pipeline ---
+
+
+class Stage:
+    """A sandbox bound to a command, not yet running.
+
+    Created by :meth:`Sandbox.cmd`.  Stages can be chained into a
+    pipeline with the ``|`` operator::
+
+        result = (
+            Sandbox(policy_a).cmd(["producer"])
+            | Sandbox(policy_b).cmd(["consumer"])
+        ).run()
+    """
+
+    __slots__ = ("sandbox", "args")
+
+    def __init__(self, sandbox: Sandbox, args: list[str]):
+        self.sandbox = sandbox
+        self.args = args
+
+    def __or__(self, other: Stage | Pipeline) -> Pipeline:
+        if isinstance(other, Pipeline):
+            return Pipeline([self] + other.stages)
+        if isinstance(other, Stage):
+            return Pipeline([self, other])
+        return NotImplemented
+
+    def run(self, *, timeout: float | None = None) -> Result:
+        """Run this single stage.  Equivalent to ``sandbox.run(args)``."""
+        return self.sandbox.run(self.args, timeout=timeout)
+
+
+class Pipeline:
+    """A chain of stages connected by pipes.
+
+    Each stage's stdout is wired to the next stage's stdin.  The
+    parent process never holds the inter-stage pipe data — it flows
+    through kernel buffers only.
+
+    Created by ``stage_a | stage_b``.
+    """
+
+    __slots__ = ("stages",)
+
+    def __init__(self, stages: list[Stage]):
+        if len(stages) < 2:
+            raise ValueError("Pipeline requires at least 2 stages")
+        self.stages = stages
+
+    def __or__(self, other: Stage | Pipeline) -> Pipeline:
+        if isinstance(other, Stage):
+            return Pipeline(self.stages + [other])
+        if isinstance(other, Pipeline):
+            return Pipeline(self.stages + other.stages)
+        return NotImplemented
+
+    def run(
+        self,
+        *,
+        stdout: int | None = None,
+        timeout: float | None = None,
+    ) -> Result:
+        """Execute the pipeline.
+
+        Each stage runs in its own sandbox.  Stdout of stage N is piped
+        to stdin of stage N+1.  The parent never reads inter-stage data.
+
+        Args:
+            stdout: File descriptor for the final stage's stdout.
+                When set, the last stage writes directly to this fd
+                and ``result.stdout`` is empty.  Use ``sys.stdout.fileno()``
+                to send output to the terminal.
+            timeout: Maximum seconds to wait for the entire pipeline.
+
+        Returns:
+            Result from the last stage.  ``stdout`` contains the last
+            stage's output only when the *stdout* parameter is not set.
+        """
+        return run_pipeline(self.stages, stdout=stdout, timeout=timeout)
+
+
+def run_pipeline(
+    stages: list[Stage],
+    *,
+    stdout: int | None = None,
+    timeout: float | None = None,
+) -> Result:
+    """Execute a pipeline of sandboxed stages connected by pipes.
+
+    The parent creates pipes between adjacent stages and closes its
+    copies immediately after fork.  Data flows through kernel buffers;
+    the parent never reads inter-stage data.
+
+    Args:
+        stages: Ordered list of stages to execute.
+        stdout: File descriptor for the final stage's stdout.
+        timeout: Maximum seconds to wait for the entire pipeline.
+
+    Returns:
+        Result from the last stage.
+    """
+    n = len(stages)
+    if n == 0:
+        raise ValueError("Pipeline requires at least 1 stage")
+    if n == 1:
+        return stages[0].run(timeout=timeout)
+
+    # Create inter-stage pipes: pipe[i] connects stage i → stage i+1
+    # Each pipe is (read_fd, write_fd)
+    pipes = [os.pipe() for _ in range(n - 1)]
+
+    # Create stderr pipe for last stage
+    last_stderr_r, last_stderr_w = os.pipe()
+
+    # Capture stdout of last stage (unless caller provided an fd)
+    capture_stdout = stdout is None
+    if capture_stdout:
+        last_stdout_r, last_stdout_w = os.pipe()
+    else:
+        last_stdout_r, last_stdout_w = -1, -1
+
+    contexts = [None] * n    # type: list
+    opened_fds = []           # type: list  # track fds we need to clean up on error
+
+    try:
+        for i, stage in enumerate(stages):
+            # Determine this stage's stdin/stdout fds
+            stdin_fd = pipes[i - 1][0] if i > 0 else -1
+            if i < n - 1:
+                stdout_fd = pipes[i][1]
+            elif capture_stdout:
+                stdout_fd = last_stdout_w
+            else:
+                stdout_fd = os.dup(stdout)
+                opened_fds.append(stdout_fd)
+            stderr_fd = last_stderr_w if i == n - 1 else -1
+
+            # Build the target function for this stage.  Capture fds
+            # and pipe list by value to avoid closure issues.
+            cmd = stage.args
+            _stdin = stdin_fd
+            _stdout = stdout_fd
+            _stderr = stderr_fd
+            _pipes = pipes
+            _last_stderr_w = last_stderr_w
+            _last_stdout_r = last_stdout_r
+            _last_stdout_w = last_stdout_w
+
+            def _make_target(cmd, _stdin, _stdout, _stderr,
+                             _pipes, _last_stderr_w,
+                             _last_stdout_r, _last_stdout_w):
+                def _target():
+                    # Close all pipe fds the child doesn't need
+                    for r, w in _pipes:
+                        if r != _stdin:
+                            os.close(r)
+                        if w != _stdout:
+                            os.close(w)
+                    if _last_stderr_w >= 0 and _last_stderr_w != _stderr:
+                        os.close(_last_stderr_w)
+                    if _last_stdout_r >= 0:
+                        os.close(_last_stdout_r)
+                    if _last_stdout_w >= 0 and _last_stdout_w != _stdout:
+                        os.close(_last_stdout_w)
+
+                    # Wire stdin
+                    if _stdin >= 0:
+                        os.dup2(_stdin, 0)
+                        os.close(_stdin)
+
+                    # Wire stdout
+                    os.dup2(_stdout, 1)
+                    if _stdout > 2:
+                        os.close(_stdout)
+
+                    # Wire stderr (last stage only)
+                    if _stderr >= 0:
+                        os.dup2(_stderr, 2)
+                        if _stderr > 2:
+                            os.close(_stderr)
+
+                    try:
+                        os.execvp(cmd[0], cmd)
+                    except OSError as e:
+                        os.write(2, f"exec failed: {e}\n".encode())
+                        os._exit(127)
+
+                return _target
+
+            target = _make_target(cmd, _stdin, _stdout, _stderr,
+                                  _pipes, _last_stderr_w,
+                                  _last_stdout_r, _last_stdout_w)
+
+            sb = stage.sandbox
+            branch = sb._setup_branch()
+            policy = sb._effective_policy()
+            inner_policy = dataclasses.replace(policy, close_fds=False)
+            for attr in ('_overlay_branch', '_cow_branch'):
+                val = getattr(policy, attr, None)
+                if val is not None:
+                    object.__setattr__(inner_policy, attr, val)
+
+            ctx = SandboxContext(target, inner_policy, sb._id)
+            ctx.__enter__()
+            contexts[i] = ctx
+
+        # Parent: close ALL pipe fds.  Data flows kernel-only.
+        for r, w in pipes:
+            os.close(r)
+            os.close(w)
+        pipes = []  # prevent double-close in error path
+
+        os.close(last_stderr_w)
+        last_stderr_w = -1
+
+        if last_stdout_w >= 0:
+            os.close(last_stdout_w)
+            last_stdout_w = -1
+
+        for fd in opened_fds:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        opened_fds = []
+
+        # Wait for all stages.  Timeout applies to the whole pipeline.
+        exit_codes = [0] * n
+        timed_out = False
+        for i in range(n):
+            ctx = contexts[i]
+            if ctx is None:
+                exit_codes[i] = -1
+                continue
+            try:
+                exit_codes[i] = ctx.wait(timeout=timeout)
+            except TimeoutError:
+                timed_out = True
+                ctx.abort()
+                exit_codes[i] = -1
+                # Abort remaining stages
+                for j in range(i + 1, n):
+                    if contexts[j] is not None:
+                        contexts[j].abort()
+                        exit_codes[j] = -1
+                break
+
+        # Clean up contexts
+        for i in range(n):
+            if contexts[i] is not None:
+                try:
+                    contexts[i].__exit__(None, None, None)
+                except Exception:
+                    pass
+                contexts[i] = None
+
+        # Finish branches
+        for i, stage in enumerate(stages):
+            stage.sandbox._finish_branch(error=exit_codes[i] != 0)
+
+        # Read last stage's captured output
+        if capture_stdout and last_stdout_r >= 0:
+            stdout_data = _read_all_fd(last_stdout_r)
+            os.close(last_stdout_r)
+            last_stdout_r = -1
+        else:
+            stdout_data = b""
+            if last_stdout_r >= 0:
+                _drain_and_close(last_stdout_r)
+                last_stdout_r = -1
+
+        stderr_data = _read_all_fd(last_stderr_r)
+        os.close(last_stderr_r)
+        last_stderr_r = -1
+
+        last_exit = exit_codes[-1]
+        if timed_out:
+            return Result(
+                success=False, exit_code=-1,
+                error="Pipeline timed out",
+                stderr=stderr_data,
+            )
+
+        return Result(
+            success=(last_exit == 0),
+            exit_code=last_exit,
+            stdout=stdout_data,
+            stderr=stderr_data,
+        )
+
+    except BaseException:
+        # Cleanup on error: close all fds, abort all contexts
+        for r, w in pipes:
+            try:
+                os.close(r)
+            except OSError:
+                pass
+            try:
+                os.close(w)
+            except OSError:
+                pass
+        for fd in (last_stderr_r, last_stderr_w, last_stdout_r, last_stdout_w):
+            if fd >= 0:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+        for fd in opened_fds:
+            try:
+                os.close(fd)
+            except OSError:
+                pass
+        for ctx in contexts:
+            if ctx is not None:
+                try:
+                    ctx.abort()
+                    ctx.__exit__(None, None, None)
+                except Exception:
+                    pass
+        for stage in stages:
+            try:
+                stage.sandbox._finish_branch(error=True)
+            except Exception:
+                pass
+        raise