From bdbbeea7b38b9a426fa07f71cd7ab1fdc2743676 Mon Sep 17 00:00:00 2001 From: OfficialAbhinavSingh Date: Mon, 13 Jul 2026 19:38:12 +0530 Subject: [PATCH 1/2] feat(env-client): add sync bootstrap constructors and public base_url (#935) --- src/openenv/core/env_client.py | 231 +++++++++++++++++++------ tests/test_core/test_generic_client.py | 112 ++++++++++++ 2 files changed, 287 insertions(+), 56 deletions(-) diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py index bd77b91e8..8d9f256b4 100644 --- a/src/openenv/core/env_client.py +++ b/src/openenv/core/env_client.py @@ -257,6 +257,20 @@ def __init__( if base_url is not None: self._set_base_url(base_url) + @property + def base_url(self) -> Optional[str]: + """Public read-only URL of the environment server this client targets. + + Returns the normalized `http(s)`/`ws(s)` base URL (no trailing slash), + or `None` when the client was created with a provider but has not yet + started its container (the URL is assigned lazily on `connect()`). + + This is the public counterpart to the private `self._base_url`; the + previously-public `base_url` attribute was dropped in the `core` + refactor, leaving sync consumers with no way to read the URL back. + """ + return self._base_url + def _set_base_url(self, base_url: str) -> None: self._base_url = base_url.rstrip("/") ws_url = convert_to_ws_url(base_url) @@ -502,6 +516,31 @@ async def _send_and_receive(self, message: Dict[str, Any]) -> Dict[str, Any]: return response + @classmethod + def _bootstrap_container( + cls: Type[EnvClientT], + image: str, + provider: Optional["ContainerProvider"] = None, + **kwargs: Any, + ) -> EnvClientT: + """Start a Docker container and build an *unconnected* client for it. + + Shared by the async `from_docker_image` and the sync + `from_docker_image_sync`: the container start / readiness wait is plain + blocking code, so the only thing that differs between the two entry + points is how the WebSocket is connected (awaited vs. run sync). + """ + if provider is None: + provider = LocalDockerProvider() + + # Start container + base_url = provider.start_container(image, **kwargs) + + # Wait for server to be ready + provider.wait_for_ready(base_url) + + return cls(base_url=base_url, provider=provider) + @classmethod async def from_docker_image( cls: Type[EnvClientT], @@ -512,6 +551,10 @@ async def from_docker_image( """ Create an environment client by spinning up a Docker container. + This is the async entry point. Synchronous callers (e.g. a TRL GRPO + rollout loop) that cannot `await` should use + [`~openenv.core.EnvClient.from_docker_image_sync`]. + Args: image (`str`): Docker image name to run (e.g., `"coding-env:latest"`). @@ -523,23 +566,45 @@ async def from_docker_image( Returns: Connected client instance """ - if provider is None: - provider = LocalDockerProvider() + client = cls._bootstrap_container(image, provider, **kwargs) + await client.connect() - # Start container - base_url = provider.start_container(image, **kwargs) + return client - # Wait for server to be ready - provider.wait_for_ready(base_url) + @classmethod + def from_docker_image_sync( + cls: Type[EnvClientT], + image: str, + provider: Optional["ContainerProvider"] = None, + **kwargs: Any, + ) -> EnvClientT: + """ + Synchronous counterpart of + [`~openenv.core.EnvClient.from_docker_image`]. - # Create and connect client - client = cls(base_url=base_url, provider=provider) - await client.connect() + Bootstraps the container and returns an already-connected client + without requiring an `await`, so synchronous consumers no longer have + to hand-roll `provider.start_container()` / `wait_for_ready()` and + duplicate this bootstrap sequence. The connection is driven on the same + dedicated background loop the sync API uses everywhere else, so the + returned client is safe to use from synchronous code immediately. + + Examples: + + ```python + env = MyEnv.from_docker_image_sync("coding-env:latest") + result = env.reset() + ``` + + Args and return value match [`~openenv.core.EnvClient.from_docker_image`]. + """ + client = cls._bootstrap_container(image, provider, **kwargs) + client._run_sync(client._connect_async) return client @classmethod - async def from_env( + def _bootstrap_env( cls: Type[EnvClientT], repo_id: str, *, @@ -547,43 +612,13 @@ async def from_env( provider: Optional["ContainerProvider | RuntimeProvider"] = None, **provider_kwargs: Any, ) -> EnvClientT: - """ - Create a client from a Hugging Face Space. + """Start a HF Space (docker or uv) and build an *unconnected* client. - Args: - repo_id (`str`): - Hugging Face space identifier `{org}/{space}`. - use_docker (`bool`, *optional*, defaults to `True`): - When `True`, pull from the HF registry and launch via `LocalDockerProvider`. - When `False`, run the space locally with `UVProvider`. - provider (`ContainerProvider` or `RuntimeProvider`, *optional*): - Provider instance to reuse. Must be a `ContainerProvider` when - `use_docker=True` and a `RuntimeProvider` otherwise. - **provider_kwargs: - Additional keyword arguments forwarded to either the container provider's - `start_container` (docker) or to the `UVProvider` constructor/start (uv). - When `use_docker=False`, the `project_path` argument can be used to override - the default git URL (`git+https://huggingface.co/spaces/{repo_id}`). - - Returns: - Connected client instance - - Examples: - - ```python - # Pull and run from HF Docker registry - env = await MyEnv.from_env("openenv/echo-env") - - # Run locally with UV (clones the space) - env = await MyEnv.from_env("openenv/echo-env", use_docker=False) - - # Run from a local checkout - env = await MyEnv.from_env( - "openenv/echo-env", - use_docker=False, - project_path="/path/to/local/checkout" - ) - ``` + Shared bootstrap for the async `from_env` and the sync `from_env_sync`; + see `_bootstrap_container` for the same split rationale. On a startup + failure the spawned process (and, for a git+ `project_path`, the temp + clone directory) is released before re-raising; a later *connection* + failure is cleaned up by `_connect_async` via `close()`. """ # Extract start args that apply to both providers start_args = {} @@ -601,9 +636,7 @@ async def from_env( ) docker_provider.wait_for_ready(base_url) - client = cls(base_url=base_url, provider=docker_provider) - await client.connect() - return client + return cls(base_url=base_url, provider=docker_provider) else: # UV mode: clone and run with uv if provider is None: @@ -633,16 +666,102 @@ async def from_env( provider.wait_for_ready( timeout_s=max(0.0, deadline - time.monotonic()) ) - - client = cls(base_url=base_url, provider=provider) - await client.connect() except Exception: - # No EnvClient may exist yet for the caller to close(), so - # this is the only chance to release the spawned process and - # (for a git+ project_path) the temp clone directory. + # No EnvClient exists yet for the caller to close(), so this is + # the only chance to release the spawned process and (for a + # git+ project_path) the temp clone directory. provider.stop() raise - return client + + return cls(base_url=base_url, provider=provider) + + @classmethod + async def from_env( + cls: Type[EnvClientT], + repo_id: str, + *, + use_docker: bool = True, + provider: Optional["ContainerProvider | RuntimeProvider"] = None, + **provider_kwargs: Any, + ) -> EnvClientT: + """ + Create a client from a Hugging Face Space. + + This is the async entry point. Synchronous callers that cannot `await` + should use [`~openenv.core.EnvClient.from_env_sync`]. + + Args: + repo_id (`str`): + Hugging Face space identifier `{org}/{space}`. + use_docker (`bool`, *optional*, defaults to `True`): + When `True`, pull from the HF registry and launch via `LocalDockerProvider`. + When `False`, run the space locally with `UVProvider`. + provider (`ContainerProvider` or `RuntimeProvider`, *optional*): + Provider instance to reuse. Must be a `ContainerProvider` when + `use_docker=True` and a `RuntimeProvider` otherwise. + **provider_kwargs: + Additional keyword arguments forwarded to either the container provider's + `start_container` (docker) or to the `UVProvider` constructor/start (uv). + When `use_docker=False`, the `project_path` argument can be used to override + the default git URL (`git+https://huggingface.co/spaces/{repo_id}`). + + Returns: + Connected client instance + + Examples: + + ```python + # Pull and run from HF Docker registry + env = await MyEnv.from_env("openenv/echo-env") + + # Run locally with UV (clones the space) + env = await MyEnv.from_env("openenv/echo-env", use_docker=False) + + # Run from a local checkout + env = await MyEnv.from_env( + "openenv/echo-env", + use_docker=False, + project_path="/path/to/local/checkout" + ) + ``` + """ + client = cls._bootstrap_env( + repo_id, + use_docker=use_docker, + provider=provider, + **provider_kwargs, + ) + await client.connect() + + return client + + @classmethod + def from_env_sync( + cls: Type[EnvClientT], + repo_id: str, + *, + use_docker: bool = True, + provider: Optional["ContainerProvider | RuntimeProvider"] = None, + **provider_kwargs: Any, + ) -> EnvClientT: + """ + Synchronous counterpart of [`~openenv.core.EnvClient.from_env`]. + + Bootstraps the Space and returns an already-connected client without + requiring an `await`. See + [`~openenv.core.EnvClient.from_docker_image_sync`] for the rationale. + + Args and return value match [`~openenv.core.EnvClient.from_env`]. + """ + client = cls._bootstrap_env( + repo_id, + use_docker=use_docker, + provider=provider, + **provider_kwargs, + ) + client._run_sync(client._connect_async) + + return client @abstractmethod def _step_payload(self, action: ActT) -> Dict[str, Any]: diff --git a/tests/test_core/test_generic_client.py b/tests/test_core/test_generic_client.py index 079c3b83b..5cfe3f564 100644 --- a/tests/test_core/test_generic_client.py +++ b/tests/test_core/test_generic_client.py @@ -368,6 +368,118 @@ async def test_from_env_uv_wait_uses_remaining_context_timeout(self): provider.wait_for_ready.assert_called_once_with(timeout_s=6.5) +class TestSyncBootstrapConstructors: + """Test the synchronous from_docker_image_sync / from_env_sync entry points. + + These let a synchronous consumer (e.g. a TRL GRPO rollout loop) bootstrap a + container without an event loop and without hand-rolling + provider.start_container()/wait_for_ready(). See issue #935. + """ + + @staticmethod + def _fake_ws_connect(): + """A ws_connect stand-in that records every socket it opens.""" + sockets = [] + + async def _connect(*args, **kwargs): + ws = AsyncMock() + sockets.append(ws) + return ws + + return _connect, sockets + + def test_from_docker_image_sync_returns_connected_client(self, mock_provider): + """from_docker_image_sync returns a concrete, connected client (no await).""" + fake_connect, sockets = self._fake_ws_connect() + + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + client = GenericEnvClient.from_docker_image_sync( + image="coding-env:latest", + provider=mock_provider, + ) + + # A concrete client, not a coroutine that a sync caller can't await. + assert isinstance(client, GenericEnvClient) + assert not asyncio.iscoroutine(client) + mock_provider.start_container.assert_called_once_with("coding-env:latest") + mock_provider.wait_for_ready.assert_called_once() + # Connection was actually established on the sync background loop. + assert len(sockets) == 1 + assert client.base_url == "http://localhost:8000" + + client.close() + + mock_provider.stop_container.assert_called_once_with() + + def test_from_docker_image_sync_forwards_start_kwargs(self, mock_provider): + """Extra kwargs reach provider.start_container().""" + fake_connect, _ = self._fake_ws_connect() + + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + client = GenericEnvClient.from_docker_image_sync( + image="coding-env:latest", + provider=mock_provider, + env_vars={"DEBUG": "1"}, + ) + mock_provider.start_container.assert_called_once_with( + "coding-env:latest", env_vars={"DEBUG": "1"} + ) + client.close() + + def test_from_env_sync_with_docker(self, mock_provider): + """from_env_sync(use_docker=True) pulls from the HF registry, connects, no await.""" + fake_connect, sockets = self._fake_ws_connect() + + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + client = GenericEnvClient.from_env_sync( + "user/my-env", + use_docker=True, + provider=mock_provider, + ) + + assert isinstance(client, GenericEnvClient) + call_args = mock_provider.start_container.call_args + assert "registry.hf.space/user-my-env" in call_args[0][0] + assert len(sockets) == 1 + + client.close() + + def test_from_env_sync_uv_stops_provider_on_start_failure(self): + """A UV start() failure releases the provider before re-raising (sync path).""" + provider = Mock() + provider.context_timeout_s = None + provider.start.side_effect = RuntimeError("boom") + + with pytest.raises(RuntimeError, match="boom"): + GenericEnvClient.from_env_sync( + "user/my-env", + use_docker=False, + provider=provider, + ) + + provider.stop.assert_called_once_with() + + +class TestBaseUrlProperty: + """Test the public read-only base_url property (issue #935).""" + + def test_base_url_returns_normalized_url(self): + """base_url exposes the normalized URL with any trailing slash stripped.""" + client = GenericEnvClient(base_url="http://localhost:8000/") + assert client.base_url == "http://localhost:8000" + + def test_base_url_is_none_before_provider_start(self, mock_provider): + """A provider-only client has no URL until it connects/starts its container.""" + client = GenericEnvClient(provider=mock_provider) + assert client.base_url is None + + def test_base_url_is_read_only(self): + """base_url is a read-only property; callers must not reassign it.""" + client = GenericEnvClient(base_url="http://localhost:8000") + with pytest.raises(AttributeError): + client.base_url = "http://evil:9999" # type: ignore[misc] + + # ============================================================================ # AutoEnv skip_install Integration Tests # ============================================================================ From a097307e2cae76887068f6b9f9c3b79ac6ce42df Mon Sep 17 00:00:00 2001 From: OfficialAbhinavSingh Date: Sat, 18 Jul 2026 10:28:04 +0530 Subject: [PATCH 2/2] refactor(env-client): rework sync bootstrap to chained .sync() (#935) --- .../cli/templates/openenv_env/README.md | 4 +- .../cli/templates/openenv_env/client.py | 4 +- src/openenv/core/env_client.py | 264 +++++++++++------- src/openenv/core/env_server/web_interface.py | 4 +- src/openenv/core/generic_client.py | 6 +- tests/test_core/test_generic_client.py | 171 ++++++++++-- 6 files changed, 325 insertions(+), 128 deletions(-) diff --git a/src/openenv/cli/templates/openenv_env/README.md b/src/openenv/cli/templates/openenv_env/README.md index 09419f75f..eea9f1942 100644 --- a/src/openenv/cli/templates/openenv_env/README.md +++ b/src/openenv/cli/templates/openenv_env/README.md @@ -23,8 +23,8 @@ The simplest way to use the __ENV_TITLE_NAME__ environment is through the `__ENV from __ENV_NAME__ import __ENV_CLASS_NAME__Action, __ENV_CLASS_NAME__Env try: - # Create environment from Docker image - __ENV_NAME__env = __ENV_CLASS_NAME__Env.from_docker_image("__ENV_NAME__-env:latest") + # Create environment from Docker image (.sync() for synchronous use) + __ENV_NAME__env = __ENV_CLASS_NAME__Env.from_docker_image("__ENV_NAME__-env:latest").sync() # Reset result = __ENV_NAME__env.reset() diff --git a/src/openenv/cli/templates/openenv_env/client.py b/src/openenv/cli/templates/openenv_env/client.py index da7fd3caf..23509d412 100644 --- a/src/openenv/cli/templates/openenv_env/client.py +++ b/src/openenv/cli/templates/openenv_env/client.py @@ -31,8 +31,8 @@ class __ENV_CLASS_NAME__Env( ... print(result.observation.echoed_message) Example with Docker: - >>> # Automatically start container and connect - >>> client = __ENV_CLASS_NAME__Env.from_docker_image("__ENV_NAME__-env:latest") + >>> # Automatically start container and connect (.sync() for sync use) + >>> client = __ENV_CLASS_NAME__Env.from_docker_image("__ENV_NAME__-env:latest").sync() >>> try: ... result = client.reset() ... result = client.step(__ENV_CLASS_NAME__Action(message="Test")) diff --git a/src/openenv/core/env_client.py b/src/openenv/core/env_client.py index 8d9f256b4..bbd61842a 100644 --- a/src/openenv/core/env_client.py +++ b/src/openenv/core/env_client.py @@ -39,6 +39,7 @@ import os import time from abc import ABC, abstractmethod +from collections.abc import Coroutine from contextlib import suppress from typing import Any, Callable, Dict, Generic, Optional, Type, TYPE_CHECKING, TypeVar from urllib.parse import urlsplit @@ -101,6 +102,87 @@ def __repr__(self) -> str: return f"<{type(self).__name__} pending>" +class _BootstrapResult(Coroutine, Generic[EnvClientT]): + """Bootstrap handle returned by the client factory methods. + + Returned by [`~openenv.core.EnvClient.from_docker_image`] and + [`~openenv.core.EnvClient.from_env`]. Resolving the handle is what actually + starts the container / Space and connects the WebSocket, so a single factory + call serves both execution modes: + + - `await handle` connects on the running event loop and returns the connected + async client (unchanged async behavior). + - `handle.sync()` connects on the sync background loop and returns a + `SyncEnvClient`, mirroring the instance-level [`~openenv.core.EnvClient.sync`]. + + The handle is a full coroutine (it implements `send` / `throw` / `close`), so + it is a drop-in for the previous `async def` factories: `asyncio.run(...)`, + `run_async_safely(...)`, and bare `await` all accept it, in addition to the + new `.sync()` chain. Bootstrap is lazy — the underlying coroutine (and the + container/Space start) is created only when the handle is driven or `.sync()` + is called. + """ + + def __init__(self, bootstrap: Callable[[], EnvClientT]): + self._bootstrap = bootstrap + self._coro: Optional[Coroutine[Any, Any, EnvClientT]] = None + self._used = False + + def _consume(self) -> EnvClientT: + """Run the bootstrap exactly once; a second resolve is a programming error. + + Mirrors native coroutine semantics (a coroutine cannot be awaited twice) + so that a handle re-driven via a second `.sync()`, or `await` followed by + `.sync()`, raises instead of silently starting a second container/Space. + """ + if self._used: + raise RuntimeError( + "This bootstrap handle has already been resolved; call the " + "factory again to start a new environment." + ) + self._used = True + return self._bootstrap() + + async def _resolve_async(self) -> EnvClientT: + client = self._consume() + await client.connect() + return client + + def _ensure_coro(self) -> "Coroutine[Any, Any, EnvClientT]": + if self._coro is None: + self._coro = self._resolve_async() + return self._coro + + def __await__(self): + return self._ensure_coro().__await__() + + def send(self, value: Any) -> Any: + return self._ensure_coro().send(value) + + def throw(self, *args: Any, **kwargs: Any) -> Any: + return self._ensure_coro().throw(*args, **kwargs) + + def close(self) -> None: + if self._coro is not None: + self._coro.close() + + def sync(self) -> "SyncEnvClient": + client = self._consume() + try: + client._run_sync(client._connect_async) + except Exception: + # _consume() already started the provider. On the async path + # _connect_async releases it, but here the failure may be the sync + # loop setup itself (before _connect_async runs), so stop the + # provider directly rather than routing through the broken loop. + client._stop_provider_best_effort() + raise + return client.sync() + + def __repr__(self) -> str: + return f"<{type(self).__name__} pending (await, run, or call .sync())>" + + def _normalize_mode(mode: Optional[str]) -> str: """Resolve and validate the client communication mode.""" raw_mode = ( @@ -525,35 +607,40 @@ def _bootstrap_container( ) -> EnvClientT: """Start a Docker container and build an *unconnected* client for it. - Shared by the async `from_docker_image` and the sync - `from_docker_image_sync`: the container start / readiness wait is plain - blocking code, so the only thing that differs between the two entry - points is how the WebSocket is connected (awaited vs. run sync). + Invoked lazily by the `from_docker_image` bootstrap handle when it is + awaited or `.sync()`'d: the container start / readiness wait is plain + blocking code, so the only thing that differs between the async and sync + resolution paths is how the WebSocket is connected (awaited vs. run sync). """ if provider is None: provider = LocalDockerProvider() - # Start container - base_url = provider.start_container(image, **kwargs) - - # Wait for server to be ready - provider.wait_for_ready(base_url) - - return cls(base_url=base_url, provider=provider) + try: + # Start container, wait for readiness, then build the client. + base_url = provider.start_container(image, **kwargs) + provider.wait_for_ready(base_url) + return cls(base_url=base_url, provider=provider) + except Exception: + # No EnvClient exists yet for the caller to close(), so release the + # container here if start / readiness / construction fails. + provider.stop_container() + raise @classmethod - async def from_docker_image( + def from_docker_image( cls: Type[EnvClientT], image: str, provider: Optional["ContainerProvider"] = None, **kwargs: Any, - ) -> EnvClientT: + ) -> "_BootstrapResult[EnvClientT]": """ Create an environment client by spinning up a Docker container. - This is the async entry point. Synchronous callers (e.g. a TRL GRPO - rollout loop) that cannot `await` should use - [`~openenv.core.EnvClient.from_docker_image_sync`]. + Returns a bootstrap handle so the same call works from both async and + synchronous code: `await` it for the connected async client, or chain + `.sync()` for a connected `SyncEnvClient` (e.g. a TRL GRPO rollout loop + that cannot `await`). Bootstrap is lazy — the container starts when the + handle is resolved. Args: image (`str`): @@ -564,44 +651,23 @@ async def from_docker_image( Additional arguments to pass to `provider.start_container()`. Returns: - Connected client instance - """ - client = cls._bootstrap_container(image, provider, **kwargs) - await client.connect() - - return client - - @classmethod - def from_docker_image_sync( - cls: Type[EnvClientT], - image: str, - provider: Optional["ContainerProvider"] = None, - **kwargs: Any, - ) -> EnvClientT: - """ - Synchronous counterpart of - [`~openenv.core.EnvClient.from_docker_image`]. - - Bootstraps the container and returns an already-connected client - without requiring an `await`, so synchronous consumers no longer have - to hand-roll `provider.start_container()` / `wait_for_ready()` and - duplicate this bootstrap sequence. The connection is driven on the same - dedicated background loop the sync API uses everywhere else, so the - returned client is safe to use from synchronous code immediately. + `_BootstrapResult`: `await` for a connected async client, or call + `.sync()` for a connected `SyncEnvClient`. Examples: ```python - env = MyEnv.from_docker_image_sync("coding-env:latest") + # Async + env = await MyEnv.from_docker_image("coding-env:latest") + + # Sync + env = MyEnv.from_docker_image("coding-env:latest").sync() result = env.reset() ``` - - Args and return value match [`~openenv.core.EnvClient.from_docker_image`]. """ - client = cls._bootstrap_container(image, provider, **kwargs) - client._run_sync(client._connect_async) - - return client + return _BootstrapResult( + lambda: cls._bootstrap_container(image, provider, **kwargs) + ) @classmethod def _bootstrap_env( @@ -614,8 +680,8 @@ def _bootstrap_env( ) -> EnvClientT: """Start a HF Space (docker or uv) and build an *unconnected* client. - Shared bootstrap for the async `from_env` and the sync `from_env_sync`; - see `_bootstrap_container` for the same split rationale. On a startup + Invoked lazily by the `from_env` bootstrap handle; see + `_bootstrap_container` for the same split rationale. On a startup failure the spawned process (and, for a git+ `project_path`, the temp clone directory) is released before re-raising; a later *connection* failure is cleaned up by `_connect_async` via `close()`. @@ -631,12 +697,17 @@ def _bootstrap_env( docker_provider = provider or LocalDockerProvider() tag = provider_kwargs.pop("tag", "latest") image = f"registry.hf.space/{repo_id.replace('/', '-')}:{tag}" - base_url = docker_provider.start_container( - image, **start_args, **provider_kwargs - ) - docker_provider.wait_for_ready(base_url) - - return cls(base_url=base_url, provider=docker_provider) + try: + base_url = docker_provider.start_container( + image, **start_args, **provider_kwargs + ) + docker_provider.wait_for_ready(base_url) + return cls(base_url=base_url, provider=docker_provider) + except Exception: + # No EnvClient exists yet for the caller to close(), so release + # the container here if start / readiness / construction fails. + docker_provider.stop_container() + raise else: # UV mode: clone and run with uv if provider is None: @@ -666,29 +737,30 @@ def _bootstrap_env( provider.wait_for_ready( timeout_s=max(0.0, deadline - time.monotonic()) ) + return cls(base_url=base_url, provider=provider) except Exception: # No EnvClient exists yet for the caller to close(), so this is # the only chance to release the spawned process and (for a - # git+ project_path) the temp clone directory. + # git+ project_path) the temp clone directory. Covers start, + # readiness, and client construction (e.g. an invalid mode). provider.stop() raise - return cls(base_url=base_url, provider=provider) - @classmethod - async def from_env( + def from_env( cls: Type[EnvClientT], repo_id: str, *, use_docker: bool = True, provider: Optional["ContainerProvider | RuntimeProvider"] = None, **provider_kwargs: Any, - ) -> EnvClientT: + ) -> "_BootstrapResult[EnvClientT]": """ Create a client from a Hugging Face Space. - This is the async entry point. Synchronous callers that cannot `await` - should use [`~openenv.core.EnvClient.from_env_sync`]. + Returns a bootstrap handle: `await` it for the connected async client, or + chain `.sync()` for a connected `SyncEnvClient`. Bootstrap is lazy — the + Space starts when the handle is resolved. Args: repo_id (`str`): @@ -706,14 +778,18 @@ async def from_env( the default git URL (`git+https://huggingface.co/spaces/{repo_id}`). Returns: - Connected client instance + `_BootstrapResult`: `await` for a connected async client, or call + `.sync()` for a connected `SyncEnvClient`. Examples: ```python - # Pull and run from HF Docker registry + # Async: pull and run from HF Docker registry env = await MyEnv.from_env("openenv/echo-env") + # Sync: chain .sync() + env = MyEnv.from_env("openenv/echo-env").sync() + # Run locally with UV (clones the space) env = await MyEnv.from_env("openenv/echo-env", use_docker=False) @@ -725,43 +801,14 @@ async def from_env( ) ``` """ - client = cls._bootstrap_env( - repo_id, - use_docker=use_docker, - provider=provider, - **provider_kwargs, - ) - await client.connect() - - return client - - @classmethod - def from_env_sync( - cls: Type[EnvClientT], - repo_id: str, - *, - use_docker: bool = True, - provider: Optional["ContainerProvider | RuntimeProvider"] = None, - **provider_kwargs: Any, - ) -> EnvClientT: - """ - Synchronous counterpart of [`~openenv.core.EnvClient.from_env`]. - - Bootstraps the Space and returns an already-connected client without - requiring an `await`. See - [`~openenv.core.EnvClient.from_docker_image_sync`] for the rationale. - - Args and return value match [`~openenv.core.EnvClient.from_env`]. - """ - client = cls._bootstrap_env( - repo_id, - use_docker=use_docker, - provider=provider, - **provider_kwargs, + return _BootstrapResult( + lambda: cls._bootstrap_env( + repo_id, + use_docker=use_docker, + provider=provider, + **provider_kwargs, + ) ) - client._run_sync(client._connect_async) - - return client @abstractmethod def _step_payload(self, action: ActT) -> Dict[str, Any]: @@ -866,6 +913,23 @@ async def _close_async(self) -> None: self._base_url = None self._ws_url = None + def _stop_provider_best_effort(self) -> None: + """Stop the underlying provider directly, ignoring any errors. + + Releases a started container/process when there is no connected client + to `close()` through the normal path — e.g. sync bootstrap setup fails + after the provider started but before the connection is established, so + routing cleanup through the (possibly broken) sync loop is not an option. + """ + provider = self._provider + if provider is None: + return + with suppress(Exception): + if hasattr(provider, "stop_container"): + provider.stop_container() + elif hasattr(provider, "stop"): + provider.stop() + async def __aenter__(self) -> "EnvClient": """Enter async context manager, ensuring connection is established.""" self._claim_execution_mode("async") diff --git a/src/openenv/core/env_server/web_interface.py b/src/openenv/core/env_server/web_interface.py index b20c08cbc..665ad47e9 100644 --- a/src/openenv/core/env_server/web_interface.py +++ b/src/openenv/core/env_server/web_interface.py @@ -39,8 +39,8 @@ ```python from __ENV_NAME__ import __ENV_CLASS_NAME__Action, __ENV_CLASS_NAME__Env -with __ENV_CLASS_NAME__Env.from_env("") as env: - result = await env.step(__ENV_CLASS_NAME__Action(message="...")) +with __ENV_CLASS_NAME__Env.from_env("").sync() as env: + result = env.step(__ENV_CLASS_NAME__Action(message="...")) ``` Or connect directly to a running server: diff --git a/src/openenv/core/generic_client.py b/src/openenv/core/generic_client.py index 1c6fd395c..b538bccc8 100644 --- a/src/openenv/core/generic_client.py +++ b/src/openenv/core/generic_client.py @@ -38,14 +38,14 @@ class GenericEnvClient(EnvClient[Dict[str, Any], Dict[str, Any], Dict[str, Any]] print(result.observation) # Dict[str, Any] print(result.observation.get("output")) - # From local Docker image - env = GenericEnvClient.from_docker_image("coding-env:latest") + # From local Docker image (chain .sync() for synchronous use) + env = GenericEnvClient.from_docker_image("coding-env:latest").sync() result = env.reset() result = env.step({"code": "x = 1 + 2"}) env.close() # From HuggingFace Hub (pulls Docker image, no pip install) - env = GenericEnvClient.from_env("user/my-env", use_docker=True) + env = GenericEnvClient.from_env("user/my-env", use_docker=True).sync() result = env.reset() env.close() ``` diff --git a/tests/test_core/test_generic_client.py b/tests/test_core/test_generic_client.py index 5cfe3f564..c7c031097 100644 --- a/tests/test_core/test_generic_client.py +++ b/tests/test_core/test_generic_client.py @@ -369,11 +369,13 @@ async def test_from_env_uv_wait_uses_remaining_context_timeout(self): class TestSyncBootstrapConstructors: - """Test the synchronous from_docker_image_sync / from_env_sync entry points. + """Test the synchronous `.sync()` bootstrap chain on the client factories. - These let a synchronous consumer (e.g. a TRL GRPO rollout loop) bootstrap a - container without an event loop and without hand-rolling - provider.start_container()/wait_for_ready(). See issue #935. + `from_docker_image(...).sync()` / `from_env(...).sync()` let a synchronous + consumer (e.g. a TRL GRPO rollout loop) bootstrap a container without an + event loop and without hand-rolling provider.start_container() / + wait_for_ready(), while `await from_docker_image(...)` keeps the async path. + See issue #935. """ @staticmethod @@ -388,23 +390,103 @@ async def _connect(*args, **kwargs): return _connect, sockets - def test_from_docker_image_sync_returns_connected_client(self, mock_provider): - """from_docker_image_sync returns a concrete, connected client (no await).""" - fake_connect, sockets = self._fake_ws_connect() + def test_from_docker_image_is_lazy_until_resolved(self, mock_provider): + """The factory returns a handle and starts nothing until driven/.sync()'d.""" + handle = GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ) + # Nothing is started at call time — the container starts only on resolve. + mock_provider.start_container.assert_not_called() + # Clean up the undriven coroutine to avoid a "never awaited" warning. + handle.close() + + def test_bootstrap_handle_is_single_use(self, mock_provider): + """Resolving a handle twice raises instead of silently double-starting.""" + fake_connect, _ = self._fake_ws_connect() with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): - client = GenericEnvClient.from_docker_image_sync( + handle = GenericEnvClient.from_docker_image( image="coding-env:latest", provider=mock_provider, ) + client = handle.sync() + with pytest.raises(RuntimeError, match="already been resolved"): + handle.sync() + client.close() + + # Only one container was ever started. + mock_provider.start_container.assert_called_once() + + def test_sync_bootstrap_stops_provider_if_loop_setup_fails(self, mock_provider): + """If sync setup fails after the container starts, the provider is released. + + Exercises the failure window in `.sync()` before `_connect_async` (and its + own cleanup) runs — here `_run_sync` itself raises. + """ + with patch.object( + GenericEnvClient, "_run_sync", side_effect=RuntimeError("loop boom") + ): + with pytest.raises(RuntimeError, match="loop boom"): + GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ).sync() + + mock_provider.start_container.assert_called_once() + mock_provider.stop_container.assert_called_once_with() + + def test_factory_handle_is_coroutine_compatible(self, mock_provider): + """The handle is a real coroutine, so asyncio.run / run_async_safely accept it. + + The previous async factories returned a coroutine; the handle must remain a + drop-in so `asyncio.run(from_docker_image(...))` and + `run_async_safely(from_docker_image(...))` keep working (regression guard). + """ + import asyncio as _asyncio + + from openenv.core.utils import run_async_safely - # A concrete client, not a coroutine that a sync caller can't await. + fake_connect, _ = self._fake_ws_connect() + + # asyncio.run(...) strictly requires a coroutine; a bare awaitable is rejected. + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + handle = GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ) + assert _asyncio.iscoroutine(handle) + client = _asyncio.run(handle) assert isinstance(client, GenericEnvClient) - assert not asyncio.iscoroutine(client) + + # run_async_safely (asyncio.run under the hood) must accept it too. + fake_connect, _ = self._fake_ws_connect() + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + client2 = run_async_safely( + GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ) + ) + assert isinstance(client2, GenericEnvClient) + + def test_from_docker_image_sync_returns_connected_client(self, mock_provider): + """from_docker_image(...).sync() returns a connected SyncEnvClient (no await).""" + fake_connect, sockets = self._fake_ws_connect() + + with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): + client = GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ).sync() + + # A concrete sync wrapper, consistent with the instance .sync(). + assert isinstance(client, SyncEnvClient) mock_provider.start_container.assert_called_once_with("coding-env:latest") mock_provider.wait_for_ready.assert_called_once() # Connection was actually established on the sync background loop. assert len(sockets) == 1 + # base_url proxies through to the wrapped async client. assert client.base_url == "http://localhost:8000" client.close() @@ -416,28 +498,28 @@ def test_from_docker_image_sync_forwards_start_kwargs(self, mock_provider): fake_connect, _ = self._fake_ws_connect() with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): - client = GenericEnvClient.from_docker_image_sync( + client = GenericEnvClient.from_docker_image( image="coding-env:latest", provider=mock_provider, env_vars={"DEBUG": "1"}, - ) + ).sync() mock_provider.start_container.assert_called_once_with( "coding-env:latest", env_vars={"DEBUG": "1"} ) client.close() def test_from_env_sync_with_docker(self, mock_provider): - """from_env_sync(use_docker=True) pulls from the HF registry, connects, no await.""" + """from_env(...).sync() with use_docker=True pulls from the HF registry, no await.""" fake_connect, sockets = self._fake_ws_connect() with patch("openenv.core.env_client.ws_connect", side_effect=fake_connect): - client = GenericEnvClient.from_env_sync( + client = GenericEnvClient.from_env( "user/my-env", use_docker=True, provider=mock_provider, - ) + ).sync() - assert isinstance(client, GenericEnvClient) + assert isinstance(client, SyncEnvClient) call_args = mock_provider.start_container.call_args assert "registry.hf.space/user-my-env" in call_args[0][0] assert len(sockets) == 1 @@ -451,14 +533,64 @@ def test_from_env_sync_uv_stops_provider_on_start_failure(self): provider.start.side_effect = RuntimeError("boom") with pytest.raises(RuntimeError, match="boom"): - GenericEnvClient.from_env_sync( + GenericEnvClient.from_env( "user/my-env", use_docker=False, provider=provider, - ) + ).sync() + + provider.stop.assert_called_once_with() + + def test_from_env_uv_stops_provider_on_construction_failure(self, monkeypatch): + """A client-construction failure after a successful start() releases the UV provider. + + The provider start()/wait succeed, but `EnvClient.__init__` rejects an + invalid OPENENV_CLIENT_MODE — the spawned process must still be stopped. + """ + monkeypatch.setenv("OPENENV_CLIENT_MODE", "not-a-valid-mode") + provider = Mock() + provider.context_timeout_s = None + provider.start.return_value = "http://localhost:8000" + provider.wait_for_ready.return_value = None + + with pytest.raises(ValueError, match="Invalid mode"): + GenericEnvClient.from_env( + "user/my-env", + use_docker=False, + provider=provider, + ).sync() provider.stop.assert_called_once_with() + def test_from_env_docker_stops_container_on_construction_failure( + self, mock_provider, monkeypatch + ): + """A client-construction failure after start_container releases the container.""" + monkeypatch.setenv("OPENENV_CLIENT_MODE", "not-a-valid-mode") + + with pytest.raises(ValueError, match="Invalid mode"): + GenericEnvClient.from_env( + "user/my-env", + use_docker=True, + provider=mock_provider, + ).sync() + + mock_provider.stop_container.assert_called_once_with() + + def test_from_docker_image_stops_container_on_construction_failure( + self, mock_provider, monkeypatch + ): + """from_docker_image releases the container if client construction fails.""" + monkeypatch.setenv("OPENENV_CLIENT_MODE", "not-a-valid-mode") + + with pytest.raises(ValueError, match="Invalid mode"): + GenericEnvClient.from_docker_image( + image="coding-env:latest", + provider=mock_provider, + ).sync() + + mock_provider.stop_container.assert_called_once_with() + class TestBaseUrlProperty: """Test the public read-only base_url property (issue #935).""" @@ -538,7 +670,8 @@ def test_skip_install_with_hub_url_and_docker(self, mock_provider): """Test skip_install=True with HF Space not running uses Docker.""" from openenv.auto.auto_env import AutoEnv - # Create an async mock for from_env (since GenericEnvClient.from_env is now async) + # AutoEnv resolves the from_env bootstrap handle via run_async_safely, so + # an awaitable stand-in returning a connected client is a valid double. async def mock_from_env_async(*args, **kwargs): return GenericEnvClient(base_url="http://localhost:8000")