glas-harness/src/glas_harness/transport.py

"""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._-]*@)?[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 or not _SSH_TARGET.fullmatch(self.ssh_target):
                raise TransportError(
                    "SSH target must be a single non-option host or user@host 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 non-option 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")