"""Harbor environment backed by the sandbox gateway: `harbor run` here, sandboxes in Kubernetes.

    pip install ./clients/harbor        # or copy this file next to your job
    export SANDBOX_GATEWAY_URL=https://sandboxes.example.com SANDBOX_GATEWAY_TOKEN=...
    harbor run -p path/to/task -a oracle \\
        --env sandbox_gateway_harbor:SandboxGatewayEnvironment

`--ek url=... --ek token=...` override the environment variables. Needs Harbor and httpx only.

One trial = one sandbox (a Pod). The gateway API (gateway/main.go):
  POST   /images                  body: build context (tar.gz); 202 {"id"}  -- Dockerfile tasks
  GET    /images/{id}?wait=30     {"state": building|ready|failed, "image", "log"}
  POST   /sandboxes               {"image", "cpus", "memory_mb", "storage_mb", "env", "compose", "compose_env", "name"}
  GET    /sandboxes/{id}?wait=30  {"state": queued|starting|ready|failed, "reason"}
  POST   /sandboxes/{id}/exec     sandboxd protocol; ?container=<compose service>
  /sandboxes/{id}/sessions        background processes (sandboxd/sessions.go)
  DELETE /sandboxes/{id}

Exec protocol (sandboxd/main.go): body = one JSON line {"argv", "timeout_sec"} then stdin;
reply body = stdout then stderr, headers X-Return-Code, X-Stdout-Length, X-Timed-Out.
Commands longer than DETACH_SEC run as sessions and survive a gateway restart.
Files move as tar through exec, like `kubectl cp`, streamed through temp files. Harbor downloads the
trial's logs, verifier output and artifacts into jobs/<job>/<trial>/ on this machine.
"""

from __future__ import annotations

import asyncio
import gzip
import hashlib
import io
import json
import math
import os
import re
import shlex
import tarfile
import tempfile
import time
import zlib
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
from typing import NamedTuple, override

import httpx
from harbor.environments.base import BaseEnvironment, ExecResult
from harbor.environments.capabilities import (
    EnvironmentCapabilities,
    EnvironmentResourceCapabilities,
)
from harbor.environments.compose_service_ops import (
    ComposeServiceOpsMixin,
    ComposeServiceTransport,
)
from harbor.environments.definition import require_agent_environment_definition
from harbor.environments.tar_transfer import extract_dir_from_file
from harbor.models.trial.config import ResourceMode
from harbor.utils.env import resolve_env_vars

GZIP_MIN = 16 << 10
# Retries of idempotent calls while a gateway replica restarts: 0.5+1+2+4+8+8 s.
RETRIES = 6
RETRY_STATUS = (502, 503, 504)
# How long one GET waits on the gateway before the client asks again.
POLL_SEC = 30
# Commands that may outlive a gateway replica's drain (~10 min) run detached as sandboxd sessions:
# they survive a gateway restart and their output is re-read from sandboxd by offset.
DETACH_SEC = 300
# A just-created sandbox may be unknown to another replica until its informer sees the Pod.
CREATE_GRACE_SEC = 60
# POST /images again after a replica that did not know the build answered.
BUILD_RETRIES = 5
# Archives packed or moved at once per event loop: each one is a temp file and a stream.
TRANSFERS = 4
CHUNK = 1 << 20
DIAGNOSTICS_TIMEOUT_SEC = 5


_ENV_KEY = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")


_version_checked: set[str] = set()

class RunResult(NamedTuple):
    return_code: int
    stdout: bytes
    stderr: bytes
    timed_out: bool


def _body(data: bytes, headers: dict) -> tuple[bytes, dict]:
    """Large exec bodies (file uploads) go gzipped; the gateway unpacks them."""
    if len(data) < GZIP_MIN:
        return data, headers
    return gzip.compress(data, compresslevel=1), {**headers, "Content-Encoding": "gzip"}


class GatewayError(RuntimeError):
    """The gateway refused a call; for exec, the command did not run."""


class GatewayNotFound(GatewayError):
    """404 from the gateway: the object is gone, or the endpoint does not exist (an older gateway)."""
    pass


# One HTTP client per event loop, shared by every trial: with `h2` installed all trials share
# one TLS connection (HTTP/2), otherwise a pool of keep-alive connections.
_clients: dict[asyncio.AbstractEventLoop, httpx.AsyncClient] = {}


_transfers: dict[asyncio.AbstractEventLoop, asyncio.Semaphore] = {}


def _transfer_slot() -> asyncio.Semaphore:
    loop = asyncio.get_running_loop()
    return _transfers.setdefault(loop, asyncio.Semaphore(TRANSFERS))


async def _gzip_stream(head: bytes, f):
    """head, then the file, gzipped chunk by chunk off the event loop."""
    z = zlib.compressobj(1, wbits=31)
    yield z.compress(head)
    while chunk := await asyncio.to_thread(f.read, CHUNK):
        yield await asyncio.to_thread(z.compress, chunk)
    yield z.flush()


def _client() -> httpx.AsyncClient:
    loop = asyncio.get_running_loop()
    if loop not in _clients:
        try:
            import h2  # noqa: F401
            http2 = True
        except ImportError:
            http2 = False
        _clients[loop] = httpx.AsyncClient(
            http2=http2,
            timeout=httpx.Timeout(60, connect=10),
            limits=httpx.Limits(max_connections=None, max_keepalive_connections=256),
        )
    return _clients[loop]


def build_context(environment_dir: Path) -> bytes:
    """The environment directory as a reproducible tar.gz: same files, same bytes, same image."""
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb", mtime=0) as gz, tarfile.open(fileobj=gz, mode="w") as tar:
        for path in sorted(p for p in environment_dir.rglob("*")):
            info = tar.gettarinfo(str(path), arcname=str(path.relative_to(environment_dir)))
            info.mtime, info.uid, info.gid, info.uname, info.gname = 0, 0, 0, "", ""
            if info.isfile():
                with path.open("rb") as f:
                    tar.addfile(info, f)
            else:
                tar.addfile(info)
    return buf.getvalue()


class SandboxGatewayEnvironment(ComposeServiceOpsMixin, BaseEnvironment):
    def __init__(self, *args, url: str | None = None, token: str | None = None,
                 build_secrets: str | None = None, diagnostics: bool = False, **kwargs):
        # Before super().__init__: base init reads `capabilities` and `_validate_definition`.
        environment_dir = Path(kwargs.get("environment_dir") or args[0])
        self._compose_mode = (environment_dir / "docker-compose.yaml").exists()
        super().__init__(*args, **kwargs)
        self._compose_mode = self._compose_mode or bool(self.extra_docker_compose_paths)

        self.url = (url or os.environ.get("SANDBOX_GATEWAY_URL") or "").rstrip("/")
        if not self.url:
            raise ValueError("sandbox gateway URL missing: --ek url=... or SANDBOX_GATEWAY_URL")
        token = token or os.environ.get("SANDBOX_GATEWAY_TOKEN")
        self._headers = {"Authorization": f"Bearer {token}"} if token else {}
        # Names of this machine's environment variables the Dockerfile reads as BuildKit secrets
        # (RUN --mount=type=secret,id=NAME): --ek build_secrets=GIT_TOKEN,PIP_TOKEN.
        names = [n.strip() for n in (build_secrets or os.environ.get("SANDBOX_GATEWAY_BUILD_SECRETS") or "").split(",") if n.strip()]
        unset = [n for n in names if not os.environ.get(n)]
        if unset:
            raise ValueError(f"build_secrets: {', '.join(unset)} not set in this environment")
        self._build_secrets = [("X-Build-Secret", f"{n}={os.environ[n]}") for n in names]

        # The gateway requests (and Kueue admits) by cpus and memory_mb; storage_mb is optional.
        self.cpus = self._resource_request_value("cpu", auto_mode=ResourceMode.REQUEST)
        self.memory_mb = self._resource_request_value("memory", auto_mode=ResourceMode.REQUEST)
        self.storage_mb = self._effective_storage_mb
        missing = [n for n, v in (("cpus", self.cpus), ("memory_mb", self.memory_mb)) if v is None]
        if missing:
            raise ValueError(f"task.toml [environment] must set {', '.join(missing)}: sandbox resources are requested from them")

        self.sandbox_id: str | None = None
        self._stopping = False
        self._diagnostics = diagnostics
        self._diagnostic_spec: dict = {}
        self._sandbox_state: dict = {}

    # ── Harbor plumbing ──────────────────────────────────────────────────

    @staticmethod
    @override
    def type() -> str:
        return "sandbox-gateway"

    @classmethod
    @override
    def resource_capabilities(cls) -> EnvironmentResourceCapabilities:
        return EnvironmentResourceCapabilities(cpu_limit=False, cpu_request=True, memory_limit=True, memory_request=True)

    @property
    @override
    def capabilities(self) -> EnvironmentCapabilities:
        return EnvironmentCapabilities(gpus=False, tpus=False, docker_compose=True)

    @property
    @override
    def _uses_compose(self) -> bool:
        return self._compose_mode

    @override
    def _validate_definition(self):
        require_agent_environment_definition(
            self.environment_dir,
            docker_image=self.task_env_config.docker_image,
            extra_docker_compose_paths=self.extra_docker_compose_paths,
        )

    @override
    def _compose_service_transport(self, service: str | None) -> ComposeServiceTransport:
        if not self._compose_mode:
            raise self._compose_unsupported(service)
        return _Services(self)

    # ── gateway calls ────────────────────────────────────────────────────

    async def _call(self, method: str, path: str, **kwargs) -> dict:
        r = await self._request(method, path, **kwargs)
        return r.json() if r.content else {}

    async def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
        """One gateway call. GET and DELETE are retried while the gateway is briefly unreachable (a
        replica restarting: 502/503/504 or a connection error); POST is not retried."""
        extra = kwargs.pop("headers", {})  # a dict, or (name, value) pairs for a repeated header
        headers = [*self._headers.items(), *(extra.items() if isinstance(extra, dict) else extra)]
        retries = RETRIES if method in ("GET", "DELETE") else 0
        for attempt in range(retries + 1):
            try:
                r = await _client().request(method, f"{self.url}{path}", headers=headers, **kwargs)
            except httpx.TransportError as e:
                if attempt == retries:
                    raise GatewayError(f"sandbox gateway: {method} {path}: {e!r}") from e
            else:
                if r.status_code not in RETRY_STATUS or attempt == retries:
                    break
            self.logger.debug(f"{method} {path}: retrying after a transient failure")
            await asyncio.sleep(min(2**attempt * 0.5, 8))
        if r.status_code >= 400:
            err = GatewayNotFound if r.status_code == 404 else GatewayError
            raise err(f"sandbox gateway: {method} {path}: {r.status_code} {r.text[:1000]}")
        return r

    async def _wait(self, path: str, done: tuple[str, ...], grace_sec: float = 0) -> dict:
        """Long-polls GET path until its state is in `done` or failed. For `grace_sec` a 404 is waited
        out: the id was just created, another replica may not know it yet."""
        last, logged = None, ""
        grace = time.monotonic() + grace_sec
        while True:
            try:
                s = await self._call("GET", path, params=httpx.URL(path).params.merge({"wait": POLL_SEC}), timeout=POLL_SEC + 30)
            except GatewayNotFound:
                if time.monotonic() >= grace:
                    raise
                await asyncio.sleep(1)
                continue
            # Where the sandbox stands (Kueue quota, scheduling, image pull) goes to the trial log.
            now = (s["state"], s.get("reason", "") or s.get("phase", ""))
            if path.startswith("/sandboxes/"):
                self._sandbox_state = s
            if now != last:
                self.logger.info(f"{path}: {now[0]}" + (f": {now[1]}" if now[1] else ""))
                last = now
            # The build's output so far (log=1): new lines go to the trial log.
            log = s.get("log") or ""
            if s["state"] == "building" and log:
                # The gateway sends the tail of the log: new text follows the last lines already shown.
                at = log.rfind(logged) if logged else -1
                for line in (log[at + len(logged):] if at >= 0 else log).splitlines():
                    self.logger.info(f"build: {line}")
                logged = log[-300:]
            if s["state"] in done or s["state"] == "failed":
                return s

    async def run(self, argv: list[str], stdin: bytes = b"", timeout_sec: float | None = None,
                  container: str | None = None, stdin_file=None, stdout_file=None,
                  env: dict[str, str] | None = None) -> RunResult:
        """argv to completion in the sandbox (`container`: a compose service instead of main).
        stdin_file / stdout_file stream from and to files instead of memory (stdout is then b"")."""
        if self.sandbox_id is None:
            raise GatewayError("the sandbox was never created (see the start error above)")
        head = json.dumps({"argv": argv, "timeout_sec": timeout_sec or 0, "env": env or {}}).encode() + b"\n"
        if stdin_file is not None:
            content, headers = _gzip_stream(head, stdin_file), {**self._headers, "Content-Encoding": "gzip"}
        else:
            content, headers = _body(head + stdin, self._headers)
        async with _client().stream(
            "POST", f"{self.url}/sandboxes/{self.sandbox_id}/exec",
            params={"container": container} if container else None,
            headers=headers,
            content=content,
            # The sandbox enforces timeout_sec itself; this only catches a gateway that stopped answering.
            timeout=httpx.Timeout(timeout_sec + 60 if timeout_sec else None, connect=10),
        ) as r:
            if r.status_code != 200:
                await r.aread()
                raise GatewayError(f"exec: {r.status_code} {r.text[:500]}")
            n = int(r.headers["X-Stdout-Length"])
            code, timed_out = int(r.headers["X-Return-Code"]), r.headers.get("X-Timed-Out") == "1"
            if stdout_file is None:
                data = await r.aread()
                return RunResult(code, data[:n], data[n:], timed_out)
            left, err = n, bytearray()
            async for chunk in r.aiter_bytes(CHUNK):
                if left:
                    await asyncio.to_thread(stdout_file.write, chunk[:left])
                    chunk, left = chunk[left:], max(0, left - len(chunk))
                err += chunk
            if left:
                raise GatewayError(f"exec: stdout cut short by {left} bytes")
            return RunResult(code, b"", bytes(err), timed_out)

    async def run_detached(self, argv: list[str], timeout_sec: float | None = None,
                           env: dict[str, str] | None = None) -> RunResult:
        """argv as a sandboxd session: a gateway restart only interrupts the log stream, which resumes
        from its offset on any replica. Never re-run: the session id is the command's identity."""
        base = f"/sandboxes/{self.sandbox_id}/sessions"
        sid = (await self._call("POST", base, json={"argv": argv, "env": env or {}}))["id"]
        out = bytearray()
        try:
            async with asyncio.timeout(timeout_sec):
                while True:
                    try:
                        async with _client().stream(
                                "GET", f"{self.url}{base}/{sid}/logs", headers=self._headers,
                                params={"stream": "stdout", "offset": len(out), "follow": 1},
                                timeout=httpx.Timeout(None, connect=10)) as r:
                            if r.status_code in RETRY_STATUS:
                                raise httpx.TransportError(f"{r.status_code}")
                            if r.status_code >= 400:
                                await r.aread()
                                raise GatewayError(f"session logs: {r.status_code} {r.text[:500]}")
                            async for chunk in r.aiter_bytes():
                                out += chunk
                    except (httpx.TransportError, httpx.TimeoutException):
                        self.logger.debug(f"session {sid}: log stream interrupted, resuming at {len(out)}")
                        await asyncio.sleep(1)
                        continue
                    # follow ends when the process has exited and the log is read, or when a
                    # draining replica closes it early: only the session's state tells.
                    info = await self._call("GET", f"{base}/{sid}")
                    if not info["running"]:
                        if info.get("output_error"):
                            raise GatewayError("session output: " + info["output_error"])
                        break
                err = await self._request("GET", f"{base}/{sid}/logs", params={"stream": "stderr"},
                                          timeout=httpx.Timeout(None, connect=10))
                return RunResult(info["exit_code"], bytes(out), err.content, False)
        except TimeoutError:
            return RunResult(124, bytes(out), b"", True)
        finally:
            try:
                await self._call("DELETE", f"{base}/{sid}")
            except (GatewayError, httpx.HTTPError) as e:
                self.logger.warning(f"could not remove session {sid}: {e}")

    # ── lifecycle ────────────────────────────────────────────────────────

    async def _check_harbor_version(self) -> None:
        if self.url in _version_checked:
            return
        try:
            response = await _client().get(f"{self.url}/version", timeout=5)
            response.raise_for_status()
            data = response.json()
        except (httpx.HTTPError, ValueError) as error:
            # The gateway may be briefly unreachable: do not remember the URL, check again next start.
            self.logger.warning(f"Harbor version check unavailable: {error}; continuing")
            return
        _version_checked.add(self.url)  # the gateway answered: this result is conclusive
        try:
            expected = data.get("harbor_version") if isinstance(data, dict) else None
            if not isinstance(expected, str) or not re.fullmatch(r"\d+\.\d+", expected):
                raise ValueError("gateway does not report Harbor compatibility")
            installed = version("harbor")
        except (ValueError, PackageNotFoundError) as error:
            self.logger.warning(f"Harbor version check unavailable: {error}; continuing")
            return
        match = re.match(r"^(\d+)\.(\d+)(?:\.|$)", installed)
        if match is None or tuple(map(int, match.groups())) != tuple(map(int, expected.split("."))):
            self.logger.warning(f"Harbor version mismatch: installed {installed}; gateway supports {expected}.x; continuing")

    async def _image(self, force_build: bool) -> str:
        """The task's published image, or one the gateway builds from environment/ (cached by content)."""
        image = self.task_env_config.docker_image
        dockerfile = (self.environment_dir / "Dockerfile").exists()
        if image and not (force_build and dockerfile):
            return image
        if not dockerfile:
            raise GatewayError("the task has neither [environment].docker_image nor environment/Dockerfile")
        context = await asyncio.to_thread(build_context, self.environment_dir)
        # The image id is the context's sha256 (the gateway's formula): asking first means a task already
        # built sends nothing but this GET. With build secrets the team is in the id, so the POST decides.
        image_id = hashlib.sha256(context).hexdigest()[:32]
        try:
            s = {} if self._build_secrets else await self._call("GET", f"/images/{image_id}")
            if s.get("state") == "ready":
                return s["image"]
        except GatewayNotFound:
            pass
        for _ in range(BUILD_RETRIES):
            build = await self._call("POST", "/images", content=context,
                                     headers=[("Content-Type", "application/gzip"), *self._build_secrets],
                                     timeout=httpx.Timeout(300, connect=10))
            try:
                # log=1: the build's output so far, into the trial log (a gateway before v0.6 ignores it).
                s = await self._wait(f"/images/{build['id']}?log=1", ("ready",))
            except GatewayNotFound:
                # A gateway before v0.6 kept builds per replica and another one answered: POST again.
                continue
            if s["state"] == "failed" and s.get("retry"):
                continue  # the context never reached the build Job (its replica restarted): POST again
            break
        else:
            raise GatewayError(f"image build: the gateway forgot the build {BUILD_RETRIES} times in a row")
        if s["state"] == "failed":
            raise GatewayError(f"image build failed:\n{s.get('log', '')[-4000:]}")
        return s["image"]

    @override
    async def start(self, force_build: bool) -> None:
        await self._check_harbor_version()
        spec = {
            "image": await self._image(force_build),
            "cpus": float(self.cpus),
            "memory_mb": int(self.memory_mb),
            "env": self._startup_env(),
            "name": f"{self.environment_name}/{self.session_id}"[:253],
        }
        if self.storage_mb:
            spec["storage_mb"] = int(self.storage_mb)
        if self._compose_mode:
            paths = [self.environment_dir / "docker-compose.yaml", *self.extra_docker_compose_paths]
            spec["compose"] = [Path(p).read_text() for p in paths if Path(p).exists()]
            # ${VAR} in compose files sees the task's env only, never this machine's.
            spec["compose_env"] = resolve_env_vars(self.task_env_config.env) if self.task_env_config.env else {}

        self._diagnostic_spec = {k: spec[k] for k in ("image", "cpus", "memory_mb", "storage_mb", "name") if k in spec}
        self.sandbox_id = (await self._call("POST", "/sandboxes", json=spec))["id"]
        if self._stopping:  # stop() ran while the POST was in flight
            await self._delete()
            raise GatewayError("sandbox creation was cancelled")
        s = await self._wait(f"/sandboxes/{self.sandbox_id}", ("ready",), grace_sec=CREATE_GRACE_SEC)
        if s["state"] == "failed":
            raise GatewayError(f"sandbox {self.sandbox_id} failed to start: {s['reason']}")

        r = await self.ensure_dirs(self._mount_targets(writable_only=True))
        if r is not None and r.return_code != 0:
            raise GatewayError(f"creating mounted directories failed: {r.stderr}")
        await self._upload_environment_dir_after_start()

    async def _delete(self) -> None:
        try:
            await self._report_reward()
            if self._diagnostics:
                await self._save_diagnostics()
        except OSError as e:
            self.logger.warning(f"could not save sandbox diagnostics: {type(e).__name__}")
        finally:
            try:
                await self._call("DELETE", f"/sandboxes/{self.sandbox_id}")
            except (GatewayError, httpx.HTTPError) as e:  # the gateway's lease deletes it anyway
                self.logger.warning(f"could not delete sandbox {self.sandbox_id}: {e}")

    async def _report_reward(self) -> None:
        try:
            r = await self._request("GET", f"/sandboxes/{self.sandbox_id}/files",
                                    params={"path": "/logs/verifier/reward.txt"})
        except GatewayNotFound:
            return
        except (GatewayError, httpx.HTTPError) as e:
            self.logger.warning(f"could not read verifier reward: {type(e).__name__}")
            return
        try:
            reward = float(r.text.strip())
        except ValueError:
            return
        if not math.isfinite(reward):
            return
        try:
            await self._request("POST", f"/sandboxes/{self.sandbox_id}/result",
                                json={"status": "ok" if reward > 0 else "failed", "reward": reward})
        except GatewayNotFound:
            self.logger.debug("gateway does not support verifier reward reporting")
        except GatewayError as e:
            log = self.logger.debug if ": 405 " in str(e) else self.logger.warning
            log(f"could not report verifier reward: {type(e).__name__}")
        except httpx.HTTPError as e:
            self.logger.warning(f"could not report verifier reward: {type(e).__name__}")

    async def _save_diagnostics(self) -> None:
        """A bounded, best-effort snapshot; Harbor already collects agent/verifier artifacts."""
        target = self.trial_paths.trial_dir / "sandbox-gateway"
        target.mkdir(parents=True, exist_ok=True)
        manifest = {"sandbox_id": self.sandbox_id, "spec": self._diagnostic_spec,
                    "last_state": self._sandbox_state, "collected_at": time.time(), "errors": {}}
        manifest_path = target / "manifest.json"
        manifest_path.write_text(json.dumps(manifest, indent=2) + "\n")

        async def collect(name: str, suffix: str, **kwargs) -> None:
            try:
                r = await asyncio.wait_for(
                    self._request("GET", f"/sandboxes/{self.sandbox_id}{suffix}", **kwargs),
                    timeout=DIAGNOSTICS_TIMEOUT_SEC)
                (target / name).write_bytes(r.content)
            except (GatewayError, httpx.HTTPError, TimeoutError, OSError) as e:
                # Response bodies and transport errors can contain credentials: record types only.
                manifest["errors"][name] = type(e).__name__
                self.logger.warning(f"sandbox diagnostics {name}: {type(e).__name__}")

        await asyncio.gather(collect("diagnostics.json", "/diagnostics"),
                             collect("main.log", "/logs", params={"container": "main", "tail": 200}))
        manifest_path.write_text(json.dumps(manifest, indent=2) + "\n")

    @override
    async def stop(self, delete: bool) -> None:
        # Without delete the sandbox stays until the gateway's lease expires (10 min idle by default).
        if delete:
            self._stopping = True
            if self.sandbox_id:
                await self._delete()

    # ── exec and files ───────────────────────────────────────────────────

    @override
    async def exec(self, command: str, cwd: str | None = None, env: dict[str, str] | None = None,
                   timeout_sec: int | None = None, user: str | int | None = None) -> ExecResult:
        if self.sandbox_id is None:
            return ExecResult(stdout=None, stderr="the sandbox was never created", return_code=1)
        user = self._resolve_user(user)
        env = self._merge_env(env)
        full = f"bash -c {shlex.quote(command)}"
        process_env, assignments = _exec_env(env, user)
        full = assignments + full
        workdir = cwd or self.task_env_config.workdir
        if workdir:
            full = f"cd {shlex.quote(workdir)} && {full}"
        if user is not None:
            if isinstance(user, int):
                # su needs a name: resolve the UID in the sandbox, fail if it has none.
                full = (f"exec_user=$(getent passwd {user}) && exec_user=${{exec_user%%:*}} && "
                        f'[ -n "$exec_user" ] && su "$exec_user" -s /bin/bash -c {shlex.quote(full)}')
            else:
                full = f"su {shlex.quote(user)} -s /bin/bash -c {shlex.quote(full)}"
        try:
            if timeout_sec is None or timeout_sec > DETACH_SEC:
                r = await self.run_detached(["sh", "-c", full], timeout_sec=timeout_sec, env=process_env)
            else:
                r = await self.run(["sh", "-c", full], timeout_sec=timeout_sec, env=process_env)
        except (httpx.HTTPError, GatewayError) as e:
            return ExecResult(stdout=None, stderr=f"sandbox {self.sandbox_id}: {e}", return_code=1)
        return exec_result(r, timeout_sec)

    @override
    async def upload_file(self, source_path: Path | str, target_path: str):
        await put_file(self.run, source_path, target_path)

    @override
    async def upload_dir(self, source_dir: Path | str, target_dir: str):
        if not Path(source_dir).is_dir():
            self.logger.warning(f"No files to upload from {source_dir}")
            return
        await put_dir(self.run, source_dir, target_dir)

    @override
    async def download_file(self, source_path: str, target_path: Path | str):
        await get_file(self.run, source_path, target_path)

    @override
    async def download_dir(self, source_dir: str, target_dir: Path | str):
        await get_dir(self.run, source_dir, target_dir)


class _Services:
    """Harbor's per-service operations for compose services other than main: exec in their container."""

    def __init__(self, env: SandboxGatewayEnvironment):
        self._env = env

    def _runner(self, service: str):
        async def run(argv: list[str], stdin: bytes = b"", timeout_sec: float | None = None, **kw) -> RunResult:
            return await self._env.run(argv, stdin, timeout_sec, container=service, **kw)
        return run

    async def service_exec(self, command: str, *, service: str, cwd: str | None = None,
                           env: dict[str, str] | None = None, timeout_sec: int | None = None,
                           user: str | int | None = None) -> ExecResult:
        # Service images may lack bash (alpine): plain sh, no main-specific defaults.
        full = f"sh -c {shlex.quote(command)}"
        process_env, assignments = _exec_env(env, user)
        full = assignments + full
        if cwd:
            full = f"cd {shlex.quote(cwd)} && {full}"
        if user is not None:
            full = f"su {shlex.quote(str(user))} -s /bin/sh -c {shlex.quote(full)}"
        return exec_result(await self._runner(service)(["sh", "-c", full], b"", timeout_sec, env=process_env), timeout_sec)

    async def service_download_file(self, source_path: str, target_path: Path | str, *, service: str) -> None:
        await get_file(self._runner(service), source_path, target_path)

    async def service_download_dir(self, source_dir: str, target_dir: Path | str, *, service: str) -> None:
        await get_dir(self._runner(service), source_dir, target_dir)

    async def stop_service(self, service: str) -> None:
        # One container of a Pod cannot be stopped from outside; its PID 1 can.
        await self._runner(service)(["sh", "-c", "kill 1"])


def _exec_env(env: dict[str, str] | None, user: str | int | None) -> tuple[dict[str, str], str]:
    """su resets HOME/USER/etc.; aliases carry explicit overrides into the target user's shell."""
    values, assignments = {}, ""
    for i, (key, value) in enumerate((env or {}).items()):
        if not _ENV_KEY.fullmatch(key):
            raise ValueError(f"bad environment variable name: {key!r}")
        if user is None:
            values[key] = value
        else:
            alias = f"SANDBOX_GATEWAY_EXEC_ENV_{i}"
            values[alias] = value
            assignments += f'{key}="${{{alias}}}" '
    return values, assignments


def _text(value: bytes) -> str:
    return value.decode("utf-8", "replace")


def exec_result(r: RunResult, timeout_sec: float | None) -> ExecResult:
    if r.timed_out:
        return ExecResult(stdout=None, stderr=f"Command timed out after {timeout_sec} seconds", return_code=124)
    return ExecResult(stdout=_text(r.stdout), stderr=_text(r.stderr), return_code=r.return_code)


# Tar transfer over any `run(argv, stdin, stdin_file=, stdout_file=) -> RunResult`: archives go
# through temp files and stream, so memory does not grow with their size; TRANSFERS at a time.

async def _checked(run, argv: list[str], **kw) -> RunResult:
    r = await run(argv, **kw)
    if r.return_code != 0:
        raise RuntimeError(f"{shlex.join(argv)} failed ({r.return_code}): {_text(r.stderr).strip()}")
    return r


async def _put_tar(run, target_dir: str, pack) -> None:
    d = shlex.quote(target_dir)
    async with _transfer_slot():
        with tempfile.TemporaryFile() as f:
            await asyncio.to_thread(pack, f)
            f.seek(0)
            await _checked(run, ["sh", "-c", f"mkdir -p {d} && tar xf - -C {d}"], stdin_file=f)


async def put_file(run, source_path: Path | str, target_path: str) -> None:
    def pack(f):
        with tarfile.open(fileobj=f, mode="w") as tar:
            tar.add(str(source_path), arcname=Path(target_path).name)
    await _put_tar(run, str(Path(target_path).parent), pack)


async def put_dir(run, source_dir: Path | str, target_dir: str) -> None:
    def pack(f):
        with tarfile.open(fileobj=f, mode="w") as tar:
            tar.add(Path(source_dir), arcname=".")
    await _put_tar(run, target_dir, pack)


def _extract_file(tar_path: str, name: str, target_path: Path) -> None:
    target_path.parent.mkdir(parents=True, exist_ok=True)
    with tarfile.open(tar_path, mode="r") as tar:
        for member in tar:
            if member.name == name and member.isfile():
                member.name = target_path.name
                tar.extract(member, path=str(target_path.parent), filter="data")
                return
    raise RuntimeError(f"archive did not contain file {name}")


async def get_file(run, source_path: str, target_path: Path | str) -> None:
    source = Path(source_path)
    async with _transfer_slot():
        with tempfile.NamedTemporaryFile() as f:
            # -h: a symlink is downloaded as the file it points to.
            await _checked(run, ["tar", "chf", "-", "-C", str(source.parent), "--", source.name], stdout_file=f)
            f.flush()
            await asyncio.to_thread(_extract_file, f.name, source.name, Path(target_path))


async def get_dir(run, source_dir: str, target_dir: Path | str) -> None:
    async with _transfer_slot():
        with tempfile.NamedTemporaryFile() as f:
            r = await run(["sh", "-c", f"cd {shlex.quote(source_dir)} && tar cf - ."], stdout_file=f)
            # tar exits 1 on "file changed as we read it": the archive is complete. Anything else
            # (2: unreadable files, missing directory) means a partial archive, so it fails.
            if r.return_code != 0 and not (r.return_code == 1 and f.tell() > 0):
                raise RuntimeError(f"cannot read directory {source_dir} (tar exit {r.return_code}): "
                                   f"{_text(r.stderr).strip()}")
            f.flush()
            target = Path(target_dir)
            target.mkdir(parents=True, exist_ok=True)
            try:
                await asyncio.to_thread(extract_dir_from_file, f.name, target)
            except tarfile.TarError as e:
                raise RuntimeError(f"cannot extract directory {source_dir}: {e}") from e