[FEATURE] uv provider

src/openenv/core/containers/runtime/__init__.py

@@ -6,10 +6,13 @@
 
 """Container runtime providers."""
 
-from .providers import ContainerProvider, KubernetesProvider, LocalDockerProvider
+from .providers import ContainerProvider, KubernetesProvider, LocalDockerProvider, RuntimeProvider
+from .uv_provider import UVProvider
 
 __all__ = [
     "ContainerProvider",
     "LocalDockerProvider",
     "KubernetesProvider",
+    "RuntimeProvider",
+    "UVProvider",
 ]

src/openenv/core/containers/runtime/providers.py

@@ -118,10 +118,12 @@ def __init__(self):
                 capture_output=True,
                 timeout=5,
             )
-        except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired):
-            raise RuntimeError(
-                "Docker is not available. Please install Docker Desktop or Docker Engine."
-            )
+        except (
+            subprocess.CalledProcessError,
+            FileNotFoundError,
+            subprocess.TimeoutExpired,
+        ):
+            raise RuntimeError("Docker is not available. Please install Docker Desktop or Docker Engine.")
 
     def start_container(
         self,
@@ -154,10 +156,13 @@ def start_container(
 
         # Build docker run command
         cmd = [
-            "docker", "run",
+            "docker",
+            "run",
             "-d",  # Detached
-            "--name", self._container_name,
-            "-p", f"{port}:8000",  # Map port
+            "--name",
+            self._container_name,
+            "-p",
+            f"{port}:8000",  # Map port
         ]
 
         # Add environment variables
@@ -241,9 +246,7 @@ def wait_for_ready(self, base_url: str, timeout_s: float = 30.0) -> None:
 
             time.sleep(0.5)
 
-        raise TimeoutError(
-            f"Container at {base_url} did not become ready within {timeout_s}s"
-        )
+        raise TimeoutError(f"Container at {base_url} did not become ready within {timeout_s}s")
 
     def _find_available_port(self) -> int:
         """
@@ -290,4 +293,67 @@ class KubernetesProvider(ContainerProvider):
         >>> # Pod running in k8s, accessible via service or port-forward
         >>> provider.stop_container()
     """
+
     pass
+
+
+class RuntimeProvider(ABC):
+    """
+    Abstract base class for runtime providers that are not container providers.
+    Providers implement this interface to support different runtime platforms:
+    - UVProvider: Runs environments via `uv run`
+
+    The provider manages a single runtime lifecycle and provides the base URL
+    for connecting to it.
+
+    Example:
+        >>> provider = UVProvider(project_path="/path/to/env")
+        >>> base_url = provider.start()
+        >>> print(base_url)  # http://localhost:8000
+        >>> provider.stop()
+    """
+
+    @abstractmethod
+    def start(
+        self,
+        port: Optional[int] = None,
+        env_vars: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> str:
+        """
+        Start a runtime from the specified image.
+
+        Args:
+            image: Runtime image name
+            port: Port to expose (if None, provider chooses)
+            env_vars: Environment variables for the runtime
+            **kwargs: Additional runtime options
+        """
+
+    @abstractmethod
+    def stop(self) -> None:
+        """
+        Stop the runtime.
+        """
+        pass
+
+    @abstractmethod
+    def wait_for_ready(self, timeout_s: float = 30.0) -> None:
+        """
+        Wait for the runtime to be ready to accept requests.
+        """
+        pass
+
+    def __enter__(self) -> "RuntimeProvider":
+        """
+        Enter the runtime provider.
+        """
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        """
+        Exit the runtime provider.
+        """
+        self.stop()
+        return False

src/openenv/core/containers/runtime/uv_provider.py

@@ -0,0 +1,224 @@
+"""Providers for launching ASGI applications via ``uv run``."""
+
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import time
+from typing import Dict, Optional
+
+import requests
+
+from .providers import RuntimeProvider
+
+
+def _check_uv_installed() -> None:
+    try:
+        subprocess.check_output(["uv", "--version"])
+    except FileNotFoundError as exc:
+        raise RuntimeError(
+            "`uv` executable not found. Install uv from https://docs.astral.sh and ensure it is on PATH."
+        ) from exc
+
+
+def _find_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("", 0))
+        sock.listen(1)
+        return sock.getsockname()[1]
+
+
+def _create_uv_command(
+    *,
+    host: str,
+    port: int,
+    reload: bool,
+    workers: int,
+    app: str,
+    project_path: str,
+) -> list[str]:
+    command: list[str] = ["uv", "run", "--isolated", "--project", project_path]
+
+    command.append("--")
+    command.extend(
+        [
+            "uvicorn",
+            app,
+            "--host",
+            host,
+            "--port",
+            str(port),
+            "--workers",
+            str(workers),
+        ]
+    )
+
+    if reload:
+        command.append("--reload")
+
+    return command
+
+
+def _poll_health(health_url: str, timeout_s: float) -> None:
+    """Poll a health endpoint until it returns HTTP 200 or times out."""
+
+    deadline = time.time() + timeout_s
+    while time.time() < deadline:
+        try:
+            timeout = max(0.0001, min(deadline - time.time(), 2.0))
+            response = requests.get(health_url, timeout=timeout)
+            if response.status_code == 200:
+                return
+        except requests.RequestException:
+            continue
+
+        time.sleep(0.5)
+
+    raise TimeoutError(f"Server did not become ready within {timeout_s:.1f} seconds")
+
+
+class UVProvider(RuntimeProvider):
+    """
+    RuntimeProvider implementation backed by ``uv run``.
+
+    Args:
+        project_path: Local path to a uv project (passed to ``uv run --project``)
+        app: ASGI application path for uvicorn (defaults to ``server.app:app``)
+        host: Host interface to bind to (defaults to ``0.0.0.0``)
+        reload: Whether to enable uvicorn's reload mode
+        env_vars: Environment variables to pass through to the spawned process
+        context_timeout_s: How long to wait for the environment to become ready
+
+    Example:
+        >>> provider = UVProvider(project_path="/path/to/env")
+        >>> base_url = provider.start()
+        >>> print(base_url)  # http://localhost:8000
+        >>> # Use the environment via base_url
+        >>> provider.stop()
+    """
+
+    def __init__(
+        self,
+        *,
+        project_path: str,
+        app: str = "server.app:app",
+        host: str = "0.0.0.0",
+        reload: bool = False,
+        env_vars: Optional[Dict[str, str]] = None,
+        context_timeout_s: float = 60.0,
+    ):
+        """Initialize the UVProvider."""
+        self.project_path = os.path.abspath(project_path)
+        self.app = app
+        self.host = host
+        self.reload = reload
+        self.env_vars = env_vars
+        self.context_timeout_s = context_timeout_s
+        _check_uv_installed()
+        self._process = None
+        self._base_url = None
+
+    def start(
+        self,
+        port: Optional[int] = None,
+        env_vars: Optional[Dict[str, str]] = None,
+        workers: int = 1,
+        **_: Dict[str, str],
+    ) -> str:
+        """
+        Start the environment via `uv run`.
+
+        Args:
+            port: The port to bind the environment to
+            env_vars: Environment variables to pass to the environment
+            workers: The number of workers to use
+
+        Returns:
+            The base URL of the environment
+
+        Raises:
+            RuntimeError: If the environment is already running
+        """
+        if self._process is not None and self._process.poll() is None:
+            raise RuntimeError("UVProvider is already running")
+
+        bind_port = port or _find_free_port()
+
+        command = _create_uv_command(
+            host=self.host,
+            port=bind_port,
+            reload=self.reload,
+            workers=workers,
+            app=self.app,
+            project_path=self.project_path,
+        )
+
+        env = os.environ.copy()
+
+        if self.env_vars:
+            env.update(self.env_vars)
+        if env_vars:
+            env.update(env_vars)
+
+        try:
+            self._process = subprocess.Popen(command, env=env)
+        except OSError as exc:
+            raise RuntimeError(f"Failed to launch `uv run`: {exc}") from exc
+
+        client_host = "127.0.0.1" if self.host in {"0.0.0.0", "::"} else self.host
+        self._base_url = f"http://{client_host}:{bind_port}"
+        return self._base_url
+
+    def wait_for_ready(self, timeout_s: float = 60.0) -> None:
+        """
+        Wait for the environment to become ready.
+
+        Args:
+            timeout_s: The timeout to wait for the environment to become ready
+
+        Raises:
+            RuntimeError: If the environment is not running
+            TimeoutError: If the environment does not become ready within the timeout
+        """
+        if self._process and self._process.poll() is not None:
+            code = self._process.returncode
+            raise RuntimeError(f"uv process exited prematurely with code {code}")
+
+        _poll_health(f"{self._base_url}/health", timeout_s=timeout_s)
+
+    def stop(self) -> None:
+        """
+        Stop the environment.
+
+        Raises:
+            RuntimeError: If the environment is not running
+        """
+        if self._process is None:
+            return
+
+        if self._process.poll() is None:
+            self._process.terminate()
+            try:
+                self._process.wait(timeout=10.0)
+            except subprocess.TimeoutExpired:
+                self._process.kill()
+                self._process.wait(timeout=5.0)
+
+        self._process = None
+        self._base_url = None
+
+    @property
+    def base_url(self) -> str:
+        """
+        The base URL of the environment.
+
+        Returns:
+            The base URL of the environment
+
+        Raises:
+            RuntimeError: If the environment is not running
+        """
+        if self._base_url is None:
+            raise RuntimeError("UVProvider has not been started")
+        return self._base_url