fix: enforce sandbox execution boundary

src/glas_harness/contract.py

@@ -205,13 +205,16 @@ class Rein(ABC):
     @abstractmethod
     def start_session(
         self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
-    ) -> dict[str, str]:
+    ) -> dict[str, Any]:
         """Begin an agent session bound to a sandbox."""
 
     @abstractmethod
-    def dispatch_tool(self, session: dict[str, str], tool_call: ToolCall) -> ToolResult:
+    def dispatch_tool(self, session: dict[str, Any], tool_call: ToolCall) -> ToolResult:
         """Run one tool call under the session policy."""
 
     @abstractmethod
-    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
+    def end_session(self, session: dict[str, Any]) -> ExecutionSummary:
         """Close the session and return normalized rein evidence."""
+
+    def cleanup_session(self, session: dict[str, Any]) -> None:
+        """Remove rein-owned ephemeral material after any terminal path."""

src/glas_harness/gateway.py

@@ -95,6 +95,7 @@ def run_execution(
 
     manager = manager or SandboxManager()
     status = None
+    session = None
     try:
         try:
             status = manager.create(
@@ -128,7 +129,6 @@ def run_execution(
                 {
                     "title": request.title,
                     "description": request.description,
-                    "target_repo": request.repo,
                     "request_id": request_id,
                 },
                 sandbox,
@@ -165,6 +165,14 @@ def run_execution(
         # return normalized evidence rather than leaking a provider exception.
         pass
     finally:
+        if session is not None:
+            try:
+                selected_rein.cleanup_session(session)
+            except Exception as exc:
+                if outcome == "succeeded" or not error:
+                    outcome = "failed"
+                    failure_stage = "teardown"
+                    error = str(exc)
         if status is not None:
             try:
                 manager.destroy(status.sandbox_id)

src/glas_harness/reins/_shared.py

@@ -3,31 +3,9 @@
 from __future__ import annotations
 
 import json
-import subprocess
-import tempfile
-from pathlib import Path
 from typing import Any
 
 
-def git_head(repo: str) -> str:
-    proc = subprocess.run(
-        ["git", "-C", str(Path(repo).expanduser()), "rev-parse", "HEAD"],
-        capture_output=True,
-        text=True,
-    )
-    return proc.stdout.strip() if proc.returncode == 0 else ""
-
-
-def write_task_file(title: str, description: str, target_repo: str, **extra: Any) -> str:
-    task_spec = {"title": title, "description": description, "target_repo": target_repo, **extra}
-    fd = tempfile.NamedTemporaryFile(
-        mode="w", suffix=".json", prefix="glas-harness-task-", delete=False
-    )
-    json.dump(task_spec, fd)
-    fd.close()
-    return fd.name
-
-
 def parse_json_object(text: str) -> dict[str, Any]:
     """Parse a rein CLI's final JSON object from otherwise human-readable output."""
 

src/glas_harness/reins/rein_aharness.py

@@ -21,8 +21,6 @@ the contract itself).
 from __future__ import annotations
 
 import json
-import shutil
-import subprocess
 from typing import Any
 
 from glas_harness.contract import (
@@ -33,7 +31,8 @@ from glas_harness.contract import (
     ToolCall,
     ToolResult,
 )
-from glas_harness.reins._shared import git_head, parse_json_object, write_task_file
+from glas_harness.reins._shared import parse_json_object
+from glas_harness.transport import ExecutionTransport, transport_from_sandbox
 
 
 class ReinAharnessNotInstalled(RuntimeError):
@@ -56,43 +55,43 @@ class ReinAharness(Rein):
         self.budget_tokens = budget_tokens
         self._last_result: dict[str, Any] = {}
 
-    def _bin(self) -> str:
-        resolved = shutil.which(self.cli_bin)
-        if not resolved:
-            raise ReinAharnessNotInstalled(
-                f"'{self.cli_bin}' not found on PATH — install rein-aharness first"
-            )
-        return resolved
+    def _bin(self, transport: ExecutionTransport) -> str:
+        try:
+            return transport.resolve_executable(self.cli_bin)
+        except RuntimeError as exc:
+            raise ReinAharnessNotInstalled(str(exc)) from exc
 
     def start_session(
         self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
-    ) -> dict[str, str]:
-        target_repo = (
-            inputs.get("target_repo")
-            or sandbox.reachability.get("workspace_dir")
-            or sandbox.reachability.get("remote_dir")
-        )
-        if not target_repo:
-            raise ValueError("no target_repo resolvable from inputs or sandbox reachability")
-
-        task_file = inputs.get("task_file") or write_task_file(
-            inputs["title"],
-            inputs["description"],
-            target_repo,
-            agent=inputs.get("agent", "coach"),
-            timeout_seconds=profile.limits.timeout_seconds or 600,
+    ) -> dict[str, Any]:
+        transport = transport_from_sandbox(sandbox)
+        target_repo = transport.workspace
+        head_before = transport.git_head()
+        generated_task_file = not bool(inputs.get("task_file"))
+        task_file = inputs.get("task_file") or transport.write_task_file(
+            {
+                "title": inputs["title"],
+                "description": inputs["description"],
+                "target_repo": target_repo,
+                "agent": inputs.get("agent", "coach"),
+                "timeout_seconds": profile.limits.timeout_seconds or 600,
+            }
         )
 
         return {
             "sandbox_id": sandbox.sandbox_id,
+            "transport": transport,
             "task_file": task_file,
+            "generated_task_file": generated_task_file,
             "target_repo": target_repo,
-            "head_before": git_head(target_repo),
+            "head_before": head_before,
+            "timeout_seconds": profile.limits.timeout_seconds or 600,
         }
 
-    def dispatch_tool(self, session: dict[str, str], tool_call: ToolCall) -> ToolResult:
+    def dispatch_tool(self, session: dict[str, Any], tool_call: ToolCall) -> ToolResult:
+        transport: ExecutionTransport = session["transport"]
         argv = [
-            self._bin(),
+            self._bin(transport),
             "run",
             "--task-file",
             session["task_file"],
@@ -107,7 +106,7 @@ class ReinAharness(Rein):
             argv += ["--tool-profile", self.tool_profile]
         if self.budget_tokens:
             argv += ["--budget-tokens", str(self.budget_tokens)]
-        proc = subprocess.run(argv, capture_output=True, text=True)
+        proc = transport.run(argv, timeout=session["timeout_seconds"])
         ok = proc.returncode == 0
         output, events = self._split_stream_events(proc.stdout)
         self._last_result = parse_json_object(output)
@@ -146,8 +145,9 @@ class ReinAharness(Rein):
             remaining_lines.append(line)
         return "\n".join(remaining_lines), events
 
-    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
-        head_after = git_head(session["target_repo"])
+    def end_session(self, session: dict[str, Any]) -> ExecutionSummary:
+        transport: ExecutionTransport = session["transport"]
+        head_after = transport.git_head()
         committed = bool(head_after) and head_after != session.get("head_before")
         reported_ok = bool(self._last_result.get("ok", committed))
         reason = self._last_result.get("reason") or None
@@ -168,3 +168,8 @@ class ReinAharness(Rein):
                 "persona_source": self._last_result.get("persona_source"),
             },
         )
+
+    def cleanup_session(self, session: dict[str, Any]) -> None:
+        if session.get("generated_task_file") and session.get("task_file"):
+            transport: ExecutionTransport = session["transport"]
+            transport.remove_file(session["task_file"])

src/glas_harness/reins/rein_openweights.py

@@ -9,8 +9,6 @@ run into a single call.
 
 from __future__ import annotations
 
-import shutil
-import subprocess
 from typing import Any
 
 from glas_harness.contract import (
@@ -21,7 +19,8 @@ from glas_harness.contract import (
     ToolCall,
     ToolResult,
 )
-from glas_harness.reins._shared import git_head, parse_json_object, write_task_file
+from glas_harness.reins._shared import parse_json_object
+from glas_harness.transport import ExecutionTransport, transport_from_sandbox
 
 
 class ReinOpenWeightsNotInstalled(RuntimeError):
@@ -44,41 +43,47 @@ class ReinOpenWeights(Rein):
         self.tool_profile = tool_profile
         self._last_result: dict[str, Any] = {}
 
-    def _bin(self) -> str:
-        resolved = shutil.which(self.cli_bin)
-        if not resolved:
-            raise ReinOpenWeightsNotInstalled(
-                f"'{self.cli_bin}' not found on PATH — install rein-openweights first"
-            )
-        return resolved
+    def _bin(self, transport: ExecutionTransport) -> str:
+        try:
+            return transport.resolve_executable(self.cli_bin)
+        except RuntimeError as exc:
+            raise ReinOpenWeightsNotInstalled(str(exc)) from exc
 
     def start_session(
         self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
-    ) -> dict[str, str]:
-        target_repo = (
-            inputs.get("target_repo")
-            or sandbox.reachability.get("workspace_dir")
-            or sandbox.reachability.get("remote_dir")
-        )
-        if not target_repo:
-            raise ValueError("no target_repo resolvable from inputs or sandbox reachability")
-
-        task_file = inputs.get("task_file") or write_task_file(
-            inputs["title"],
-            inputs["description"],
-            target_repo,
-            timeout_seconds=profile.limits.timeout_seconds or 600,
+    ) -> dict[str, Any]:
+        transport = transport_from_sandbox(sandbox)
+        target_repo = transport.workspace
+        head_before = transport.git_head()
+        generated_task_file = not bool(inputs.get("task_file"))
+        task_file = inputs.get("task_file") or transport.write_task_file(
+            {
+                "title": inputs["title"],
+                "description": inputs["description"],
+                "target_repo": target_repo,
+                "timeout_seconds": profile.limits.timeout_seconds or 600,
+            }
         )
 
         return {
             "sandbox_id": sandbox.sandbox_id,
+            "transport": transport,
             "task_file": task_file,
+            "generated_task_file": generated_task_file,
             "target_repo": target_repo,
-            "head_before": git_head(target_repo),
+            "head_before": head_before,
+            "timeout_seconds": profile.limits.timeout_seconds or 600,
         }
 
-    def dispatch_tool(self, session: dict[str, str], tool_call: ToolCall) -> ToolResult:
-        argv = [self._bin(), "run", "--task-file", session["task_file"], "--no-hub"]
+    def dispatch_tool(self, session: dict[str, Any], tool_call: ToolCall) -> ToolResult:
+        transport: ExecutionTransport = session["transport"]
+        argv = [
+            self._bin(transport),
+            "run",
+            "--task-file",
+            session["task_file"],
+            "--no-hub",
+        ]
         if self.model:
             argv += ["--model", self.model]
         if self.max_turns:
@@ -87,7 +92,7 @@ class ReinOpenWeights(Rein):
             argv += ["--budget-tokens", str(self.budget_tokens)]
         if self.tool_profile:
             argv += ["--tool-profile", self.tool_profile]
-        proc = subprocess.run(argv, capture_output=True, text=True)
+        proc = transport.run(argv, timeout=session["timeout_seconds"])
         ok = proc.returncode == 0
         self._last_result = parse_json_object(proc.stdout)
         return ToolResult(
@@ -104,8 +109,9 @@ class ReinOpenWeights(Rein):
             },
         )
 
-    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
-        head_after = git_head(session["target_repo"])
+    def end_session(self, session: dict[str, Any]) -> ExecutionSummary:
+        transport: ExecutionTransport = session["transport"]
+        head_after = transport.git_head()
         committed = bool(head_after) and head_after != session.get("head_before")
         reported_ok = bool(self._last_result.get("ok", committed))
         reason = self._last_result.get("reason") or None
@@ -126,3 +132,8 @@ class ReinOpenWeights(Rein):
                 "tool_profile": self._last_result.get("tool_profile") or self.tool_profile,
             },
         )
+
+    def cleanup_session(self, session: dict[str, Any]) -> None:
+        if session.get("generated_task_file") and session.get("task_file"):
+            transport: ExecutionTransport = session["transport"]
+            transport.remove_file(session["task_file"])

src/glas_harness/transport.py

@@ -0,0 +1,184 @@
+"""Execution transports derived from sand-boxer reachability descriptors.
+
+The source checkout is an input to sandbox provisioning, never an execution
+workspace. Once sand-boxer returns READY, every rein subprocess crosses exactly
+one declared boundary: nsenter for a same-host namespace or SSH for a remote
+workspace.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import shlex
+import subprocess
+import uuid
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any, Literal, Sequence
+
+from glas_harness.contract import SandboxHandle
+
+
+class TransportError(RuntimeError):
+    """The sandbox reachability descriptor cannot safely execute commands."""
+
+
+_SSH_TARGET = re.compile(r"^(?:[A-Za-z0-9._-]+@)?[A-Za-z0-9._-]+$")
+
+
+@dataclass(frozen=True)
+class ExecutionTransport:
+    kind: Literal["local_namespace", "ssh"]
+    workspace: str
+    pid: int | None = None
+    ssh_target: str | None = None
+
+    def command(self, argv: Sequence[str]) -> list[str]:
+        if not argv:
+            raise TransportError("cannot execute an empty command")
+        command = [str(item) for item in argv]
+        scoped_command = [
+            "sh",
+            "-c",
+            'cd "$1" && shift && exec "$@"',
+            "sh",
+            self.workspace,
+            *command,
+        ]
+        if self.kind == "local_namespace":
+            if self.pid is None:
+                raise TransportError("local namespace transport is missing pid")
+            return [
+                "nsenter",
+                "--target",
+                str(self.pid),
+                "--mount",
+                "--pid",
+                "--net",
+                "--uts",
+                "--ipc",
+                "--",
+                *scoped_command,
+            ]
+        if self.kind == "ssh":
+            if not self.ssh_target:
+                raise TransportError("SSH transport is missing target")
+            return ["ssh", self.ssh_target, shlex.join(scoped_command)]
+        raise TransportError(f"unsupported execution transport: {self.kind}")
+
+    def run(
+        self,
+        argv: Sequence[str],
+        *,
+        input_text: str | None = None,
+        timeout: float | None = None,
+    ) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            self.command(argv),
+            input=input_text,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+        )
+
+    def resolve_executable(self, name: str, *, timeout: float = 15.0) -> str:
+        if not name or "/" in name:
+            raise TransportError("rein executable must be a command name, not a host path")
+        proc = self.run(
+            ["sh", "-c", 'command -v -- "$1"', "sh", name], timeout=timeout
+        )
+        resolved = proc.stdout.strip()
+        if proc.returncode != 0 or not resolved:
+            raise TransportError(
+                f"'{name}' is not installed inside the selected sandbox transport"
+            )
+        return resolved
+
+    def git_head(self, *, timeout: float = 15.0) -> str:
+        proc = self.run(
+            ["git", "-C", self.workspace, "rev-parse", "HEAD"], timeout=timeout
+        )
+        head = proc.stdout.strip()
+        if proc.returncode != 0 or not head:
+            raise TransportError(
+                "cannot read git HEAD inside the selected sandbox transport"
+            )
+        return head
+
+    def write_task_file(self, payload: dict[str, Any]) -> str:
+        task_name = f".glas-harness-task-{uuid.uuid4().hex}.json"
+        # Keep the task spec under Git's private metadata directory so an
+        # agent's broad `git add -A` cannot accidentally commit its prompt.
+        task_path = str(PurePosixPath(self.workspace) / ".git" / task_name)
+        content = json.dumps(payload)
+        if self.kind == "local_namespace":
+            path = Path(task_path)
+            try:
+                descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+                with os.fdopen(descriptor, "w") as task_file:
+                    task_file.write(content)
+            except OSError as exc:
+                raise TransportError(f"cannot create sandbox task file: {exc}") from exc
+            return task_path
+
+        script = 'umask 077; cat > "$1"'
+        proc = self.run(["sh", "-c", script, "sh", task_path], input_text=content, timeout=15)
+        if proc.returncode != 0:
+            raise TransportError(
+                "cannot create remote sandbox task file: "
+                + (proc.stderr.strip() or f"exit {proc.returncode}")
+            )
+        return task_path
+
+    def remove_file(self, path: str) -> None:
+        if self.kind == "local_namespace":
+            Path(path).unlink(missing_ok=True)
+            return
+        proc = self.run(["rm", "-f", "--", path], timeout=15)
+        if proc.returncode != 0:
+            raise TransportError(
+                "cannot remove remote sandbox task file: "
+                + (proc.stderr.strip() or f"exit {proc.returncode}")
+            )
+
+
+def transport_from_sandbox(sandbox: SandboxHandle) -> ExecutionTransport:
+    reachability = sandbox.reachability
+    pid = reachability.get("pid")
+    workspace = reachability.get("workspace_dir")
+    ssh_target = reachability.get("ssh")
+    remote_dir = reachability.get("remote_dir")
+
+    has_local = pid is not None or workspace is not None
+    has_remote = ssh_target is not None or remote_dir is not None
+    if has_local and has_remote:
+        raise TransportError("ambiguous sandbox reachability: local and SSH fields coexist")
+    if has_local:
+        if not pid or not workspace:
+            raise TransportError(
+                "incomplete local sandbox reachability: pid and workspace_dir are required"
+            )
+        try:
+            parsed_pid = int(str(pid))
+        except ValueError as exc:
+            raise TransportError("local sandbox reachability pid must be an integer") from exc
+        if parsed_pid <= 0 or not Path(str(workspace)).is_absolute():
+            raise TransportError("local sandbox reachability has invalid pid or workspace_dir")
+        return ExecutionTransport(
+            kind="local_namespace", workspace=str(workspace), pid=parsed_pid
+        )
+    if has_remote:
+        if not ssh_target or not remote_dir:
+            raise TransportError(
+                "incomplete remote sandbox reachability: ssh and remote_dir are required"
+            )
+        if not _SSH_TARGET.fullmatch(str(ssh_target)):
+            raise TransportError("SSH reachability must be a single host or user@host target")
+        if not PurePosixPath(str(remote_dir)).is_absolute():
+            raise TransportError("remote sandbox workspace must be an absolute path")
+        return ExecutionTransport(
+            kind="ssh", workspace=str(remote_dir), ssh_target=str(ssh_target)
+        )
+    raise TransportError("sandbox exposes no supported execution reachability")