diff --git a/src/openenv/core/env_server/http_server.py b/src/openenv/core/env_server/http_server.py index 50dd5b174..ecc2c5c2d 100644 --- a/src/openenv/core/env_server/http_server.py +++ b/src/openenv/core/env_server/http_server.py @@ -640,6 +640,212 @@ def concurrency_config(self) -> ConcurrencyConfig: """Return the concurrency configuration.""" return self._concurrency_config + def _factory_produces_harness_env(self) -> bool: + """Return whether the env factory produces a HarnessEnvironment.""" + import inspect + + # Lazy import: openenv.core.harness imports env_server modules, so a + # top-level import here would be circular. + from ..harness.environment import HarnessEnvironment + + if inspect.isclass(self._env_factory): + return issubclass(self._env_factory, HarnessEnvironment) + _temp_env = self._env_factory() + try: + return isinstance(_temp_env, HarnessEnvironment) + finally: + _temp_env.close() + + def _register_harness_route(self, app: FastAPI) -> None: + """ + Register the production `/harness` WebSocket route (RFC 005). + + Each connection gets its own environment session: connecting resets + the environment (which starts the harness process and injects tools), + and each `{"type": "message", "content": ...}` frame runs one + conversational turn, streamed back as `HarnessEvent` JSON frames + ending with a `turn_complete` event. Malformed client frames receive a + recoverable `protocol_error` response without starting a turn; the + connection remains usable. Terminal failures use `error`. + """ + # Lazy import to avoid a circular import with openenv.core.harness. + from ..harness.adapter import HarnessNotRunningError + from ..harness.events import ( + HarnessClientMessage, + HarnessEvent, + HarnessEventType, + HarnessProtocolError, + ) + + @app.websocket("/harness") + async def harness_websocket_endpoint(websocket: WebSocket): + await websocket.accept() + + session_id = None + session_env = None + + async def send_error(message: str, code: WSErrorCode) -> None: + error_response = WSErrorResponse( + data={"message": message, "code": code} + ) + await websocket.send_text(error_response.model_dump_json()) + + async def send_protocol_error(message: str, code: WSErrorCode) -> None: + error_response = HarnessProtocolError( + data={"message": message, "code": code} + ) + await websocket.send_text(error_response.model_dump_json()) + + async def send_harness_error(message: str) -> None: + """Emit a terminal ERROR event in the harness event stream.""" + error_event = HarnessEvent( + type=HarnessEventType.ERROR, + data={"message": message, "recoverable": False}, + ) + await websocket.send_text(error_event.model_dump_json()) + + try: + session_id, session_env = await self._create_session() + # Protect the live harness from idle reaping and HTTP session + # close, including startup and turns that emit no events. + self._session_websocket_attachments.add(session_id) + + async with AsyncExitStack() as stack: + mcp_session_factory = getattr(session_env, "mcp_session", None) + if callable(mcp_session_factory): + mcp_session_cm = cast( + AsyncContextManager[Any], mcp_session_factory() + ) + await stack.enter_async_context(mcp_session_cm) + + # Starts the harness process and injects environment tools + await session_env.reset_async() + await websocket.send_text( + json.dumps( + { + "type": "session_started", + "data": { + "session_id": session_id, + "harness": session_env.adapter.config.name, + }, + } + ) + ) + + while True: + raw_message = await websocket.receive_text() + + try: + message_dict = json.loads(raw_message) + except json.JSONDecodeError as e: + await send_protocol_error( + f"Invalid JSON: {e}", WSErrorCode.INVALID_JSON + ) + continue + try: + client_message = HarnessClientMessage(**message_dict) + except (ValidationError, TypeError) as e: + await send_protocol_error( + f"Invalid message: {e}", + WSErrorCode.VALIDATION_ERROR, + ) + continue + + self._update_session_activity(session_id, increment_step=True) + + async def stream_turn(content: str) -> bool: + """Stream one turn; True if it ended with TURN_COMPLETE.""" + saw_terminal = False + adapter = session_env.adapter + if not await adapter.is_alive(): + raise HarnessNotRunningError( + "harness process is not running" + ) + async for event in adapter.send_message_streaming(content): + await websocket.send_text(event.model_dump_json()) + # Record progress throughout the turn. + self._update_session_activity(session_id) + saw_terminal = ( + event.type is HarnessEventType.TURN_COMPLETE + ) + return saw_terminal + + # Bound the turn in wall-clock time, matching what + # simulation mode does in HarnessEnvironment._run_turn. + # Without this a hung harness holds the session open + # forever, and the server sits at capacity. + turn_timeout_s = session_env.adapter.config.session_timeout_s + try: + completed = await asyncio.wait_for( + stream_turn(client_message.content), + turn_timeout_s, + ) + except asyncio.TimeoutError: + await send_harness_error( + f"harness turn exceeded {turn_timeout_s} seconds" + ) + break + except HarnessNotRunningError: + await send_harness_error("harness process is not running") + break + except Exception: + # Harness state after a crash is undefined; end + # the session so a reconnect gets a fresh one. + # Adapter exceptions can contain credentials or + # subprocess output; do not expose them to clients. + await send_harness_error("harness turn failed") + break + + if not completed: + # send_message() raises HarnessError here; the + # socket equivalent is to say so and end the + # session, rather than leaving a client that + # blocks on the terminal event waiting forever. + await send_harness_error( + "harness event stream ended without a " + "TURN_COMPLETE event" + ) + break + + except WebSocketDisconnect: + pass + except SessionCapacityError as e: + await send_error(str(e), WSErrorCode.CAPACITY_REACHED) + except EnvironmentFactoryError as e: + await send_error(str(e), WSErrorCode.FACTORY_ERROR) + except Exception: + try: + await send_error( + "harness session failed", WSErrorCode.SESSION_ERROR + ) + except (RuntimeError, WebSocketDisconnect): + pass + finally: + if session_id: + # Release ownership without an await so cancellation cannot + # leave a session permanently exempt from idle reaping. + self._session_websocket_attachments.discard(session_id) + cleanup = asyncio.create_task(self._destroy_session(session_id)) + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + # ASGI cancellation must not orphan a running harness. + # Finish teardown before propagating cancellation, even + # when the request's cancel scope cancels us repeatedly. + while not cleanup.done(): + try: + await asyncio.shield(cleanup) + except asyncio.CancelledError: + pass + cleanup.result() + raise + try: + await websocket.close() + except (RuntimeError, WebSocketDisconnect): + # TestClient raises RuntimeError, real ASGI servers raise + # WebSocketDisconnect when the client is already gone. + pass + def register_routes( self, app: FastAPI, mode: ServerMode | str = ServerMode.SIMULATION ) -> None: @@ -1290,6 +1496,11 @@ async def mcp_websocket_endpoint(websocket: WebSocket): except (RuntimeError, WebSocketDisconnect): pass + # In production mode, a harness environment is exposed directly to + # clients via a streaming WebSocket (RFC 005). + if mode == ServerMode.PRODUCTION and self._factory_produces_harness_env(): + self._register_harness_route(app) + # Register simulation control routes only in simulation mode if mode == ServerMode.SIMULATION: @@ -1781,6 +1992,8 @@ def create_app( show_default_tab: bool = True, title_override: Optional[str] = None, state_cls: Type[State] = State, + *, + mode: Optional[ServerMode | str] = None, ) -> FastAPI: """ Create a FastAPI application with or without web interface. @@ -1822,6 +2035,9 @@ def create_app( state_cls (`Type[State]`, *optional*, defaults to `State`): The `State` subclass this environment reports, used for the `/state` response model and the `state` entry of `/schema`. + mode (`ServerMode` or `str`, *optional*): + Server mode. When `None`, resolved from the `OPENENV_MODE` + environment variable, defaulting to simulation. Returns: `FastAPI` application instance with or without web interface and README integration. @@ -1851,6 +2067,7 @@ def create_app( custom_tab_primary=custom_tab_primary, show_default_tab=show_default_tab, title_override=title_override, + mode=mode, ) else: # Use standard FastAPI app without web interface @@ -1862,6 +2079,7 @@ def create_app( concurrency_config, env_name=env_name, state_cls=state_cls, + mode=mode, ) @@ -1873,6 +2091,8 @@ def create_fastapi_app( concurrency_config: Optional[ConcurrencyConfig] = None, env_name: Optional[str] = None, state_cls: Type[State] = State, + *, + mode: Optional[ServerMode | str] = None, ) -> FastAPI: """ Create a FastAPI application with comprehensive documentation. @@ -1895,6 +2115,9 @@ def create_fastapi_app( state_cls (`Type[State]`, *optional*, defaults to `State`): The `State` subclass this environment reports, used for the `/state` response model and the `state` entry of `/schema`. + mode (`ServerMode` or `str`, *optional*): + Server mode. When `None`, resolved from the `OPENENV_MODE` + environment variable, defaulting to simulation. Returns: `FastAPI` application instance. @@ -1975,5 +2198,7 @@ def create_fastapi_app( env_name=env_name, state_cls=state_cls, ) - server.register_routes(app) + if mode is None: + mode = os.environ.get("OPENENV_MODE", ServerMode.SIMULATION.value) + server.register_routes(app, mode=mode) return app diff --git a/src/openenv/core/env_server/web_interface.py b/src/openenv/core/env_server/web_interface.py index ddba90fa3..1c0ef4e5f 100644 --- a/src/openenv/core/env_server/web_interface.py +++ b/src/openenv/core/env_server/web_interface.py @@ -435,6 +435,8 @@ def create_web_interface_app( show_default_tab: bool = True, title_override: Optional[str] = None, state_cls: Type[State] = State, + *, + mode: Optional[Any] = None, ) -> FastAPI: """ Create a FastAPI application with web interface for the given environment. @@ -467,6 +469,9 @@ def create_web_interface_app( title instead of the default ``"OpenEnv Agentic Environment: {name}"``. state_cls: The State subclass this environment reports. Used for the /state response model and the state entry of /schema. Defaults to State. + mode: Server mode (``ServerMode`` or string). When ``None``, resolved + from the ``OPENENV_MODE`` environment variable, defaulting to + simulation. Returns: FastAPI application instance with web interface @@ -482,6 +487,7 @@ def create_web_interface_app( concurrency_config, env_name=env_name, state_cls=state_cls, + mode=mode, ) # Load environment metadata diff --git a/src/openenv/core/harness/__init__.py b/src/openenv/core/harness/__init__.py index 579d588a2..0b8f138bc 100644 --- a/src/openenv/core/harness/__init__.py +++ b/src/openenv/core/harness/__init__.py @@ -1,6 +1,6 @@ # SPDX-License-Identifier: BSD-3-Clause -"""Harness helpers for training, evaluation, and wrapping external agents. +"""Harness integration helpers for training, evaluation, and wrapping agents. This package hosts two complementary layers: @@ -8,9 +8,10 @@ drives an entire episode in one call (``run_white_box``/``run_black_box``) against a resource session. Used by ``openenv collect`` and the training tutorials. -2. **Turn-based agentic harness API** (RFC 005): the types describing an - external harness such as OpenClaw or Claude Code, where each ``step()`` is - one conversational turn. See +2. **Turn-based agentic harness API** (RFC 005): an external harness such as + OpenClaw or Claude Code runs inside the environment container, and each + ``step()`` is one conversational turn. See + [`~openenv.core.harness.environment.HarnessEnvironment`] and [`~openenv.core.harness.adapter.AgenticHarnessAdapter`]. Both layers are importable from ``openenv.core.harness``. @@ -23,7 +24,9 @@ HarnessStartupError, HarnessTurnTimeoutError, ) +from .bridge import build_bridge_server, HarnessMCPBridge from .config import HarnessConfig, HarnessTransport +from .environment import HarnessAction, HarnessEnvironment from .events import ( events_to_metadata, HarnessClientMessage, @@ -31,6 +34,7 @@ HarnessEventType, HarnessResponse, ) +from .process import HarnessProcess from .rollout import ( # noqa: F401 (_resolve_env_reward: private back-compat re-export) _resolve_env_reward, build_harness_rollout_func, @@ -80,16 +84,21 @@ "build_harness_rollout_func", # Turn-based agentic harness API (RFC 005) "AgenticHarnessAdapter", + "HarnessAction", "HarnessClientMessage", "HarnessConfig", + "HarnessEnvironment", "HarnessError", "HarnessEvent", "HarnessEventType", + "HarnessMCPBridge", "HarnessNotRunningError", + "HarnessProcess", "HarnessResponse", "HarnessStartupError", "HarnessTransport", "HarnessTurnTimeoutError", + "build_bridge_server", "events_to_metadata", "resolve_tool_conflicts", ] diff --git a/src/openenv/core/harness/bridge.py b/src/openenv/core/harness/bridge.py new file mode 100644 index 000000000..5bef30739 --- /dev/null +++ b/src/openenv/core/harness/bridge.py @@ -0,0 +1,220 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Loopback MCP bridge exposing environment tools to a harness (RFC 005).""" + +from __future__ import annotations + +import socket +import threading +import time +from contextlib import asynccontextmanager +from typing import Any, Optional + +from .adapter import HarnessError + + +async def _source_tools(mcp_server: Any) -> dict[str, Any]: + """Return {name: FastMCP tool}, compatible with FastMCP 2.x and 3.x. + + Async twin of [`~openenv.core.env_server.mcp_environment.get_server_tools`], + which wraps the same calls in `run_async_safely`; awaiting directly avoids + spawning a thread when we are already on a loop. + """ + if hasattr(mcp_server, "get_tools"): + result = await mcp_server.get_tools() + if isinstance(result, dict): + return result + if hasattr(mcp_server, "list_tools"): + return {tool.name: tool for tool in await mcp_server.list_tools()} + return {} + + +async def build_bridge_server(mcp_server: Any, renames: dict[str, str]) -> Any: + """ + Return a FastMCP server exposing the env's tools under their injected names. + + Tool-name conflict resolution can rename an environment tool before it is + injected into a harness (`read_file` -> `env_read_file`). The harness then + calls the new name, so the bridge has to answer to it: serving the source + server unchanged would advertise a name that does not resolve. + + Args: + mcp_server (`FastMCP`): + The environment's own tool server. + renames (`dict[str, str]`): + Mapping of injected name to source name. Empty means no renames. + + Returns: + `FastMCP`: `mcp_server` itself when there is nothing to rename, + otherwise a view of it whose tools carry the injected names. + """ + if not renames: + return mcp_server + + from fastmcp import Client + from fastmcp.server.providers.proxy import FastMCPProxy, StatefulProxyClient + from fastmcp.server.transforms import ToolTransform + from fastmcp.tools.tool_transform import ToolTransformConfig + + source_tools = await _source_tools(mcp_server) + missing = sorted(set(renames.values()) - set(source_tools)) + if missing: + raise HarnessError( + "Cannot rename tools that the MCP server does not expose: " + + ", ".join(missing) + ) + + @asynccontextmanager + async def lifespan(server): + # Preserve source resources across harness MCP reconnects until this + # bridge stops, just as when serving the source server directly. + async with Client(mcp_server): + yield {} + + # Forward callbacks in the current request context. Each harness MCP + # connection gets its own source session, closed on disconnect. + client = StatefulProxyClient(mcp_server) + view = FastMCPProxy( + name=f"{getattr(mcp_server, 'name', 'openenv')}-harness-view", + client_factory=client.new_stateful, + lifespan=lifespan, + ) + view.add_transform( + ToolTransform( + { + source: ToolTransformConfig(name=served) + for served, source in renames.items() + } + ) + ) + return view + + +class HarnessMCPBridge: + """ + Serve an in-process FastMCP server over loopback HTTP for harness use. + + The bridge runs the environment's FastMCP tool server as a + streamable-HTTP ASGI app on `127.0.0.1` with an ephemeral port, inside a + daemon thread with its own event loop. It carries only the tool surface: + the harness subprocess can reach the environment's domain tools but never + OpenEnv's orchestration API (`reset`/`step`/`state`), which lives on a + different server entirely. This makes the RFC 005 security boundary + structural rather than filter-based. + + Args: + mcp_server (`FastMCP`): + The environment's FastMCP server to expose. + host (`str`, *optional*, defaults to `"127.0.0.1"`): + Interface to bind. Keep this loopback-only. + + Examples: + + ```python + bridge = HarnessMCPBridge(env.mcp_server) + url = bridge.start() + # pass url to the harness adapter's inject_tools() + bridge.stop() + ``` + """ + + def __init__(self, mcp_server: Any, host: str = "127.0.0.1"): + self._mcp_server = mcp_server + self._host = host + self._url: Optional[str] = None + self._thread: Optional[threading.Thread] = None + self._uvicorn_server: Optional[Any] = None + self._startup_error: Optional[BaseException] = None + + @property + def url(self) -> Optional[str]: + """The bridge's MCP endpoint URL, or `None` when not running.""" + return self._url + + def start(self, timeout_s: float = 10.0) -> str: + """ + Start serving and return the MCP endpoint URL. + + Idempotent: returns the existing URL if already running. + + Args: + timeout_s (`float`, *optional*, defaults to `10.0`): + Maximum time to wait for the server to come up. + + Returns: + `str` URL of the MCP endpoint, e.g. `"http://127.0.0.1:54321/mcp"`. + + Raises: + [`~openenv.core.harness.adapter.HarnessError`]: + If the server fails to start within the timeout. + """ + if self._thread is not None and self._thread.is_alive(): + assert self._url is not None + return self._url + + import uvicorn + + app = self._mcp_server.http_app() + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind((self._host, 0)) + port = sock.getsockname()[1] + + config = uvicorn.Config(app, log_level="warning", lifespan="on") + server = uvicorn.Server(config) + self._uvicorn_server = server + self._startup_error = None + + def _serve() -> None: + try: + server.run(sockets=[sock]) + except BaseException as exc: # surfaced to start() below + self._startup_error = exc + finally: + try: + sock.close() + except OSError: + pass + + thread = threading.Thread(target=_serve, daemon=True) + thread.start() + self._thread = thread + + deadline = time.monotonic() + timeout_s + while not server.started: + if self._startup_error is not None or not thread.is_alive(): + self._teardown() + raise HarnessError(f"MCP bridge failed to start: {self._startup_error}") + if time.monotonic() > deadline: + self.stop() + raise HarnessError(f"MCP bridge did not start within {timeout_s}s") + time.sleep(0.01) + + self._url = f"http://{self._host}:{port}/mcp" + return self._url + + def stop(self, timeout_s: float = 5.0) -> None: + """ + Stop the bridge server. + + Idempotent: safe to call when the bridge was never started. + + Args: + timeout_s (`float`, *optional*, defaults to `5.0`): + Maximum time to wait for the server thread to exit. + """ + server = self._uvicorn_server + thread = self._thread + if server is not None: + server.should_exit = True + if thread is not None and thread.is_alive(): + thread.join(timeout=timeout_s) + self._teardown() + + def _teardown(self) -> None: + self._url = None + self._thread = None + self._uvicorn_server = None + + +__all__ = ["build_bridge_server", "HarnessMCPBridge"] diff --git a/src/openenv/core/harness/environment.py b/src/openenv/core/harness/environment.py new file mode 100644 index 000000000..6ce2f3850 --- /dev/null +++ b/src/openenv/core/harness/environment.py @@ -0,0 +1,423 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Environment wrapper for turn-based external agentic harnesses (RFC 005).""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Optional +from uuid import uuid4 + +from ..env_server.mcp_environment import MCPEnvironment +from ..env_server.mcp_types import CallToolAction, ListToolsAction, Tool +from ..env_server.types import Action, Observation, State +from ..utils import run_async_safely +from .adapter import ( + AgenticHarnessAdapter, + HarnessError, + HarnessNotRunningError, + HarnessTurnTimeoutError, +) +from .bridge import build_bridge_server, HarnessMCPBridge +from .events import events_to_metadata, HarnessEvent, HarnessEventType +from .tools import resolve_tool_conflicts + +try: + from fastmcp import FastMCP +except ModuleNotFoundError: # pragma: no cover - fastmcp is a core dependency + FastMCP = None + +logger = logging.getLogger(__name__) + + +class HarnessAction(Action): + """ + Action carrying one user message for a conversational harness turn. + + Args: + message (`str`): + The user message for this turn. + """ + + message: str + + +class HarnessEnvironment(MCPEnvironment): + """ + Environment that wraps an external turn-based agentic harness. + + In simulation mode, `reset()` starts a fresh harness process and + conversation, and each `step()` with a [`~openenv.core.harness.environment.HarnessAction`] + is one conversational turn: the harness runs its internal ReAct loop and + the response is returned as an observation. The harness keeps conversation + context across turns; the training loop controls episode boundaries. + MCP actions (`ListToolsAction`, `CallToolAction`) keep their standard + routing so orchestrators can inspect and invoke domain tools directly. + + Environment MCP tools are conflict-resolved against the adapter's built-in + tool names and injected into the harness before it starts. Rubrics run + after each turn completes, outside the harness's control loop. + + Factories used with `HTTPEnvServer` must construct a fresh adapter per + environment instance — adapters own a single harness process and must not + be shared. Constructing the environment starts nothing; the harness + process and any tool bridge start on `reset()`. + + Args: + adapter ([`~openenv.core.harness.adapter.AgenticHarnessAdapter`]): + Adapter owning the harness process lifecycle. + mcp (`FastMCP`, *optional*): + Environment-specific MCP tools to inject into the harness. When + `None`, an empty internal server is used and nothing is injected. + rubric ([`~openenv.core.rubrics.Rubric`], *optional*): + Reward rubric applied to each turn's observation. + transform (`Transform`, *optional*): + Optional observation transform (inherited from `Environment`). + """ + + SUPPORTS_CONCURRENT_SESSIONS = False + + def __init__( + self, + adapter: AgenticHarnessAdapter, + mcp: Optional[Any] = None, + rubric: Optional[Any] = None, + transform: Optional[Any] = None, + ): + if mcp is None: + if FastMCP is None: # pragma: no cover - fastmcp is a core dependency + raise ModuleNotFoundError( + "fastmcp is required to construct a HarnessEnvironment" + ) + mcp = FastMCP("harness-env-tools") + super().__init__(mcp, transform=transform) + self.rubric = rubric + self.adapter = adapter + self._state = State(episode_id=None, step_count=0) + self._trajectory: list[HarnessEvent] = [] + self._episode_active = False + self._closed = False + self._bridge: Optional[HarnessMCPBridge] = None + + @property + def state(self) -> State: + """Current episode state.""" + return self._state + + @property + def trajectory(self) -> list[HarnessEvent]: + """Full event trajectory across all turns in the current episode.""" + return list(self._trajectory) + + async def reset_async( + self, + seed: Optional[int] = None, + episode_id: Optional[str] = None, + **kwargs: Any, + ) -> Observation: + """ + Start a fresh harness process and conversation. + + Stops any running harness, injects the environment's conflict-resolved + MCP tools, and starts the harness in its configured working directory. + + Args: + seed (`int`, *optional*): + Unused; accepted for interface compatibility. + episode_id (`str`, *optional*): + Episode identifier. A UUID is generated when `None`. + + Returns: + `Observation` with `done=False` and metadata listing the names of + the injected tools. + """ + self._episode_active = False + # Unconditional: stop() is contractually idempotent, and a harness that + # died on its own still holds reapable resources (pipes, reader threads, + # an unwaited process) that is_alive() reports nothing about. + bridge_start: Optional[asyncio.Task[str]] = None + try: + await self.adapter.stop() + await self._stop_bridge() + + tools = await self._collect_injectable_tools() + resolved = resolve_tool_conflicts(tools, self.adapter.BUILTIN_TOOL_NAMES) + + bridge_url: Optional[str] = None + if resolved: + # The bridge must serve the tools under the names we inject, so a + # tool renamed by conflict resolution stays callable. + renames = { + new.name: old.name + for new, old in zip(resolved, tools) + if new.name != old.name + } + served = await build_bridge_server(self._require_mcp_server(), renames) + self._bridge = HarnessMCPBridge(served) + bridge_start = asyncio.create_task( + asyncio.to_thread(self._bridge.start) + ) + bridge_url = await asyncio.shield(bridge_start) + + await self.adapter.inject_tools(resolved, bridge_url) + await self.adapter.start(self.adapter.config.working_directory) + + self._state = State( + episode_id=episode_id or str(uuid4()), + step_count=0, + ) + self._trajectory = [] + if self.rubric is not None: + await self._reset_rubric_async() + self._episode_active = True + + observation = Observation( + done=False, + reward=0.0, + metadata={ + "episode_id": self._state.episode_id, + "injected_tools": [tool.name for tool in resolved], + }, + ) + return self._apply_transform(observation) + except BaseException: + await self._cleanup_episode(bridge_start) + raise + + def reset( + self, + seed: Optional[int] = None, + episode_id: Optional[str] = None, + **kwargs: Any, + ) -> Observation: + """Sync facade over `reset_async` for non-async callers.""" + return run_async_safely( + self.reset_async(seed=seed, episode_id=episode_id, **kwargs) + ) + + async def step_async( + self, + action: Action, + timeout_s: Optional[float] = None, + **kwargs: Any, + ) -> Observation: + """ + Execute one action: a conversational turn or an MCP action. + + Args: + action (`Action`): + A [`~openenv.core.harness.environment.HarnessAction`] runs one + conversational turn; `ListToolsAction`/`CallToolAction` keep + their standard MCP routing. + timeout_s (`float`, *optional*): + Wall-clock budget for this turn. Defaults to the adapter + config's `session_timeout_s`. + + Returns: + `Observation` with the harness response and turn events in + metadata. + """ + if isinstance(action, (ListToolsAction, CallToolAction)): + return await super().step_async(action, timeout_s=timeout_s, **kwargs) + if isinstance(action, HarnessAction): + return await self._run_turn(action, timeout_s=timeout_s) + raise TypeError( + "HarnessEnvironment only accepts HarnessAction, ListToolsAction, " + f"or CallToolAction; got {type(action).__name__}" + ) + + def _step_impl( + self, + action: Action, + timeout_s: Optional[float] = None, + **kwargs: Any, + ) -> Observation: + """Sync facade for non-MCP actions; runs one conversational turn.""" + if isinstance(action, HarnessAction): + return run_async_safely(self._run_turn(action, timeout_s=timeout_s)) + raise TypeError( + "HarnessEnvironment only accepts HarnessAction, ListToolsAction, " + f"or CallToolAction; got {type(action).__name__}" + ) + + async def _run_turn( + self, + action: HarnessAction, + timeout_s: Optional[float] = None, + ) -> Observation: + """Send one message to the harness and build the turn's observation.""" + if not self._episode_active: + raise HarnessNotRunningError( + "No active episode; call reset() before step()" + ) + try: + if not await self.adapter.is_alive(): + return await self._terminal_error_observation( + "harness process is not running", + error_type="harness_crashed", + ) + + timeout = ( + timeout_s + if timeout_s is not None + else (self.adapter.config.session_timeout_s) + ) + try: + harness_response = await asyncio.wait_for( + self.adapter.send_message(action.message), timeout + ) + except asyncio.TimeoutError: + return await self._terminal_error_observation( + f"harness turn exceeded {timeout} seconds", + error_type="turn_timeout", + ) + except HarnessTurnTimeoutError as exc: + # Must precede HarnessError: an adapter that raises the dedicated + # timeout exception means a timeout, not a crash. + return await self._terminal_error_observation( + str(exc), + error_type="turn_timeout", + ) + except HarnessError as exc: + return await self._terminal_error_observation( + str(exc), + error_type="harness_crashed", + ) + + self._trajectory.extend(harness_response.events) + self._state.step_count += 1 + + observation = Observation( + done=harness_response.done, + reward=0.0, + metadata={ + "response": harness_response.response, + "turn_events": events_to_metadata(harness_response.events), + "turn_number": self._state.step_count, + }, + ) + if self.rubric is not None: + observation.reward = await self._apply_rubric_async(action, observation) + return self._apply_transform(observation) + except BaseException: + # A failed or interrupted turn cannot safely resume. Preserve the + # original failure for the caller after releasing resources. + await self._cleanup_episode() + raise + + async def _cleanup_episode( + self, bridge_start: Optional[asyncio.Task[str]] = None + ) -> None: + """Finish teardown even when the caller is cancelled repeatedly.""" + self._episode_active = False + + async def cleanup() -> None: + try: + if bridge_start is not None: + # Cancelling to_thread does not stop its worker. Startup + # must finish before teardown can safely stop the bridge. + try: + await bridge_start + except Exception: + pass + finally: + try: + await self.adapter.stop() + except Exception: + pass + finally: + await self._stop_bridge() + + cleanup_task = asyncio.create_task(cleanup()) + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + # ASGI cancel scopes can cancel every await. Keep the cleanup in + # its own task and join it before allowing cancellation to escape. + while not cleanup_task.done(): + try: + await asyncio.shield(cleanup_task) + except asyncio.CancelledError: + pass + cleanup_task.result() + raise + + async def _terminal_error_observation( + self, message: str, error_type: str + ) -> Observation: + """Stop the harness and return a terminal observation for a failed turn.""" + await self._cleanup_episode() + error_event = HarnessEvent( + type=HarnessEventType.ERROR, + data={"message": message, "recoverable": False}, + ) + self._trajectory.append(error_event) + observation = Observation( + done=True, + reward=0.0, + metadata={ + "error": message, + "error_type": error_type, + "turn_events": events_to_metadata([error_event]), + }, + ) + return self._apply_transform(observation) + + async def _collect_injectable_tools(self) -> list[Tool]: + """ + Enumerate the environment's MCP tools for injection into the harness. + + Mode-specific tools registered with `MCPEnvironment.tool(mode=...)` are + excluded: they are tracked by the environment rather than registered on + the FastMCP server, so the bridge cannot serve them. Advertising a tool + the harness then cannot call is worse than not advertising it, so they + are dropped with a warning. Supporting them needs a design decision + about mode semantics inside a harness turn (follow-up). + """ + list_tools_observation = await self._async_handle_list_tools() + error = list_tools_observation.metadata.get("error") + if error: + raise HarnessError(f"Failed to enumerate environment tools: {error}") + + tools = list(list_tools_observation.tools) + mode_specific = [t for t in tools if t.name in self._mode_tool_schemas] + if mode_specific: + logger.warning( + "Not injecting mode-specific tools into harness %r (the tool " + "bridge cannot serve them): %s", + self.adapter.config.name, + ", ".join(sorted(t.name for t in mode_specific)), + ) + tools = [t for t in tools if t.name not in self._mode_tool_schemas] + return tools + + async def _stop_bridge(self) -> None: + """Best-effort stop of the tool bridge, off the event loop.""" + if self._bridge is None: + return + try: + await asyncio.to_thread(self._bridge.stop) + except Exception: + pass + + def close(self) -> None: + """Stop the harness process and release environment resources.""" + if self._closed: + return + self._closed = True + self._episode_active = False + try: + run_async_safely(self.adapter.stop()) + except Exception: + pass + if self._bridge is not None: + try: + self._bridge.stop() + except Exception: + pass + self._bridge = None + super().close() + + +__all__ = ["HarnessAction", "HarnessEnvironment"] diff --git a/src/openenv/core/harness/events.py b/src/openenv/core/harness/events.py index 82cf2189a..1aeed655d 100644 --- a/src/openenv/core/harness/events.py +++ b/src/openenv/core/harness/events.py @@ -84,6 +84,23 @@ class HarnessClientMessage(BaseModel): content: str +class HarnessProtocolError(BaseModel): + """ + A recoverable rejection of a client frame on `/harness`. + + No turn was started. The client can correct its message and retry on the + same connection. The distinct `protocol_error` discriminator keeps this + response separate from terminal harness `error` events. + + Args: + data (`dict[str, Any]`): + Error details including `message` and `code`. + """ + + type: Literal["protocol_error"] = "protocol_error" + data: dict[str, Any] + + def events_to_metadata(events: list[HarnessEvent]) -> list[dict[str, Any]]: """ Convert harness events to JSON-serializable dicts for observation metadata. @@ -107,5 +124,6 @@ def events_to_metadata(events: list[HarnessEvent]) -> list[dict[str, Any]]: "HarnessEvent", "HarnessEventType", "HarnessResponse", + "HarnessProtocolError", "events_to_metadata", ] diff --git a/src/openenv/core/harness/process.py b/src/openenv/core/harness/process.py new file mode 100644 index 000000000..756a20c85 --- /dev/null +++ b/src/openenv/core/harness/process.py @@ -0,0 +1,304 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Loop-agnostic subprocess helper for CLI harnesses (RFC 005).""" + +from __future__ import annotations + +import asyncio +import os +import queue +import signal +import subprocess +import threading +import time +from collections import deque +from typing import Callable, IO, Optional + +from .adapter import HarnessNotRunningError, HarnessStartupError + +_STDERR_TAIL_LINES = 50 + + +class HarnessProcess: + """ + Manage a long-lived CLI harness subprocess with line-based stdio. + + Built on `subprocess.Popen` with daemon reader threads rather than asyncio + subprocess transports so the same instance works across event loops: the + sync facades in [`~openenv.core.harness.environment.HarnessEnvironment`] + spin a fresh loop per call, while the HTTP server keeps one long-lived + loop. All blocking pipe I/O is offloaded via `asyncio.to_thread`. + + Args: + command (`list[str]`): + Command line to execute. + cwd (`str`): + Working directory for the process. + env_vars (`dict[str, str]`, *optional*): + Extra environment variables, merged over `os.environ`. + startup_timeout_s (`float`, *optional*, defaults to `30.0`): + Maximum time for the process to become ready in `start()`. + terminate_grace_s (`float`, *optional*, defaults to `5.0`): + Time to wait after SIGTERM before escalating to SIGKILL. + + Examples: + + ```python + process = HarnessProcess(["openclaw", "run"], cwd="/workspace") + await process.start(ready_check=lambda line: line.startswith("ready")) + await process.write_line("hello") + reply = await process.read_line(timeout_s=10.0) + await process.stop() + ``` + """ + + def __init__( + self, + command: list[str], + cwd: str, + env_vars: Optional[dict[str, str]] = None, + startup_timeout_s: float = 30.0, + terminate_grace_s: float = 5.0, + ): + self.command = list(command) + self.cwd = cwd + self.env_vars = dict(env_vars or {}) + self.startup_timeout_s = startup_timeout_s + self.terminate_grace_s = terminate_grace_s + + self._proc: Optional[subprocess.Popen] = None + self._stdout_queue: queue.Queue[Optional[str]] = queue.Queue() + self._stderr_tail: deque[str] = deque(maxlen=_STDERR_TAIL_LINES) + self._reader_threads: list[threading.Thread] = [] + + def is_running(self) -> bool: + """Whether the subprocess is alive.""" + return self._proc is not None and self._proc.poll() is None + + def drain_stderr(self) -> str: + """Return the most recent stderr output for diagnostics.""" + return "".join(self._stderr_tail) + + async def start(self, ready_check: Optional[Callable[[str], bool]] = None) -> None: + """ + Start the subprocess and optionally wait for a readiness line. + + Args: + ready_check (`Callable[[str], bool]`, *optional*): + Predicate applied to each stdout line. `start()` returns once + a line matches. When `None`, `start()` returns as soon as the + process is spawned. + + Raises: + [`~openenv.core.harness.adapter.HarnessStartupError`]: + If the process cannot be spawned, exits before becoming + ready, or does not become ready within `startup_timeout_s`. + """ + if self.is_running(): + raise HarnessStartupError("harness process is already running") + + # An exited process can still own open pipes and reader threads. + # Release them before replacing any of its lifecycle state. + await self.stop() + + try: + self._proc = subprocess.Popen( + self.command, + cwd=self.cwd, + env={**os.environ, **self.env_vars}, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + # Explicit UTF-8: text=True alone decodes with the locale + # encoding, which is often ASCII in a container, and harness + # output is routinely non-ASCII. errors="replace" keeps a + # stray byte from killing the reader thread mid-turn. + encoding="utf-8", + errors="replace", + bufsize=1, + start_new_session=True, + ) + except OSError as exc: + self._proc = None + raise HarnessStartupError("failed to spawn harness process") from exc + + try: + stdout_queue: queue.Queue[Optional[str]] = queue.Queue() + self._stdout_queue = stdout_queue + self._stderr_tail = deque(maxlen=_STDERR_TAIL_LINES) + self._reader_threads = [ + self._spawn_reader( + self._proc.stdout, + lambda line: stdout_queue.put(line.rstrip("\n")), + on_eof=lambda: stdout_queue.put(None), + ), + self._spawn_reader(self._proc.stderr, self._stderr_tail.append), + ] + + if ready_check is None: + return + + deadline = time.monotonic() + self.startup_timeout_s + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + await self.stop() + raise HarnessStartupError( + f"harness did not become ready within {self.startup_timeout_s}s" + ) + line = await self.read_line(timeout_s=min(remaining, 0.2)) + if line is None: + if not self.is_running(): + exit_code = self._proc.poll() + await self.stop() + raise HarnessStartupError( + f"harness exited with code {exit_code} during startup" + ) + continue + if ready_check(line): + return + except BaseException: + # CancelledError and readiness predicate failures must also reap + # the process and close its pipes before propagating to callers. + await self.stop() + raise + + async def stop(self) -> None: + """ + Stop the subprocess, escalating SIGTERM to SIGKILL after the grace period. + + Idempotent: safe to call when the process was never started or has + already exited. + """ + proc = self._proc + if proc is None: + return + await asyncio.to_thread(self._stop_blocking, proc) + self._proc = None + + def _stop_blocking(self, proc: subprocess.Popen) -> None: + # start_new_session=True makes proc.pid the process-group ID. The + # group can outlive its leader, including after proc.poll() reaps it. + self._signal_group(proc, signal.SIGTERM) + deadline = time.monotonic() + self.terminate_grace_s + while True: + proc.poll() + try: + os.killpg(proc.pid, 0) + except ProcessLookupError: + break + if time.monotonic() >= deadline: + self._signal_group(proc, signal.SIGKILL) + break + time.sleep(0.01) + proc.wait() + + # Readers close their own streams after EOF. Closing a TextIOWrapper + # from here can block on its reader lock while a descendant owns the + # other end of the pipe. Only join readers with a bounded wait. + for thread in self._reader_threads: + thread.join(timeout=2.0) + if proc.stdin is not None: + try: + proc.stdin.close() + except OSError: + pass + self._reader_threads = [] + + @staticmethod + def _signal_group(proc: subprocess.Popen, sig: signal.Signals) -> None: + """Signal the process group (start_new_session=True) with a fallback.""" + try: + os.killpg(proc.pid, sig) + except (ProcessLookupError, PermissionError, OSError): + try: + proc.send_signal(sig) + except ProcessLookupError: + pass + + async def write_line(self, text: str) -> None: + """ + Write one line to the subprocess's stdin. + + Args: + text (`str`): + Line content; a trailing newline is appended if missing. + + Raises: + [`~openenv.core.harness.adapter.HarnessNotRunningError`]: + If the process is not running or stdin is closed. + """ + if not self.is_running() or self._proc.stdin is None: + raise HarnessNotRunningError( + "cannot write to harness process: it is not running" + ) + if not text.endswith("\n"): + text += "\n" + + def _write() -> None: + try: + self._proc.stdin.write(text) + self._proc.stdin.flush() + except (BrokenPipeError, ValueError, OSError) as exc: + raise HarnessNotRunningError( + f"harness process stdin is closed: {exc}" + ) from exc + + await asyncio.to_thread(_write) + + async def read_line(self, timeout_s: Optional[float] = None) -> Optional[str]: + """ + Read one stdout line from the subprocess. + + Args: + timeout_s (`float`, *optional*): + Maximum time to wait. `None` blocks until a line or EOF. + + Returns: + `str` line without its trailing newline, or `None` on EOF (the + process exited or closed stdout) or timeout. + """ + stdout_queue = self._stdout_queue + + def _read() -> Optional[str]: + try: + item = stdout_queue.get(timeout=timeout_s) + except queue.Empty: + return None + if item is None: + # Leave EOF available for subsequent or concurrent readers. + stdout_queue.put(None) + return item + + return await asyncio.to_thread(_read) + + def _spawn_reader( + self, + stream: Optional[IO[str]], + sink: Callable[[str], None], + on_eof: Optional[Callable[[], None]] = None, + ) -> threading.Thread: + def _pump() -> None: + try: + if stream is None: + return + for line in stream: + sink(line) + except (ValueError, OSError): + # Raised when the pipe is closed during shutdown. Decoding + # cannot raise here: the stream is opened with + # errors="replace". + pass + finally: + if stream is not None: + stream.close() + if on_eof is not None: + on_eof() + + thread = threading.Thread(target=_pump, daemon=True) + thread.start() + return thread + + +__all__ = ["HarnessProcess"] diff --git a/tests/core/scripted_harness.py b/tests/core/scripted_harness.py new file mode 100644 index 000000000..6248a3b98 --- /dev/null +++ b/tests/core/scripted_harness.py @@ -0,0 +1,161 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Scripted subprocess behaviors for the agentic harness tests. + +This module is not imported by the tests — it is *spawned* by them +(``python scripted_harness.py ``) to play the role of an external CLI +harness, so the subprocess-facing code lives in a real, lintable file instead +of inline strings. + +Modes: + mcp - full fake harness: announces readiness, then for each + JSON stdin line calls the env tool over the MCP bridge + (from the config file at ``$HARNESS_MCP_CONFIG``) and + emits JSON-line events ending in ``turn_complete``. + echo - print ``ready``, then echo each stdin line. + slow-start - sleep past any startup timeout before printing ``ready``. + exit-now - print to stderr and exit with code 3. + exit-with-secret - print a fake credential to stderr, then exit with code 3. + spawn-descendant - spawn a SIGTERM-ignoring child with inherited pipes. + descendant - print its PID, ignore SIGTERM, and sleep. + exit-with-output - print multiple lines, including an unterminated final line. + close-stdout - close stdout on request, then wait on stdin without exiting. + crash-after-echo - ``ready``, echo one line, then exit with code 1. + ignore-sigterm - ignore SIGTERM, print ``ready``, then sleep. + unicode-echo - like echo, but the reply carries non-ASCII characters. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import signal +import subprocess +import sys +import time + + +def mode_mcp() -> None: + """Fake agentic harness speaking JSON lines over stdio.""" + with open(os.environ["HARNESS_MCP_CONFIG"]) as f: + config = json.load(f) + bridge_url = config.get("mcp_url") + tool_names = config.get("tools", []) + + def emit(payload: dict) -> None: + print(json.dumps(payload), flush=True) + + emit({"event": "ready", "tools": tool_names}) + + async def run_turn(message: str) -> None: + if bridge_url and "add" in tool_names: + from fastmcp import Client + + arguments = {"a": 2, "b": 3} + emit({"event": "tool_call", "tool_name": "add", "arguments": arguments}) + async with Client(bridge_url) as client: + result = await client.call_tool("add", arguments) + value = result.content[0].text + emit({"event": "tool_result", "tool_name": "add", "result": value}) + response = f"sum={value} for: {message}" + else: + response = f"no tools for: {message}" + emit( + { + "event": "turn_complete", + "response": response, + "done": "finish" in message, + } + ) + + for line in sys.stdin: + request = json.loads(line) + asyncio.run(run_turn(request["message"])) + + +def mode_echo() -> None: + print("ready", flush=True) + for line in sys.stdin: + print("echo:" + line.strip(), flush=True) + + +def mode_unicode_echo() -> None: + """Echo with non-ASCII output, which locale-encoded pipes would fail on.""" + print("ready", flush=True) + for line in sys.stdin: + print("echo:\u2713 " + line.strip() + " \u4e16\u754c \U0001f600", flush=True) + + +def mode_slow_start() -> None: + time.sleep(60) + print("ready", flush=True) + + +def mode_exit_now() -> None: + print("dying", file=sys.stderr, flush=True) + sys.exit(3) + + +def mode_exit_with_secret() -> None: + print("FAKE_HARNESS_SECRET_FOR_TEST", file=sys.stderr, flush=True) + sys.exit(3) + + +def mode_spawn_descendant() -> None: + print("ready", flush=True) + subprocess.Popen([sys.executable, "-u", __file__, "descendant"]) + sys.stdin.readline() + sys.exit(1) + + +def mode_descendant() -> None: + signal.signal(signal.SIGTERM, signal.SIG_IGN) + print(os.getpid(), flush=True) + time.sleep(60) + + +def mode_crash_after_echo() -> None: + print("ready", flush=True) + line = sys.stdin.readline() + print("echo:" + line.strip(), flush=True) + sys.exit(1) + + +def mode_exit_with_output() -> None: + print("first") + print("second") + print("last", end="", flush=True) + + +def mode_close_stdout() -> None: + print("ready", flush=True) + sys.stdin.readline() + os.close(sys.stdout.fileno()) + sys.stdin.readline() + + +def mode_ignore_sigterm() -> None: + signal.signal(signal.SIGTERM, signal.SIG_IGN) + print("ready", flush=True) + time.sleep(60) + + +MODES = { + "mcp": mode_mcp, + "echo": mode_echo, + "unicode-echo": mode_unicode_echo, + "slow-start": mode_slow_start, + "exit-now": mode_exit_now, + "exit-with-secret": mode_exit_with_secret, + "spawn-descendant": mode_spawn_descendant, + "descendant": mode_descendant, + "exit-with-output": mode_exit_with_output, + "close-stdout": mode_close_stdout, + "crash-after-echo": mode_crash_after_echo, + "ignore-sigterm": mode_ignore_sigterm, +} + + +if __name__ == "__main__": + MODES[sys.argv[1]]() diff --git a/tests/core/test_agentic_harness_bridge.py b/tests/core/test_agentic_harness_bridge.py new file mode 100644 index 000000000..155ff066d --- /dev/null +++ b/tests/core/test_agentic_harness_bridge.py @@ -0,0 +1,407 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Integration tests for the loopback MCP bridge (real HTTP server).""" + +from __future__ import annotations + +import asyncio +import threading +from contextlib import asynccontextmanager +from typing import AsyncIterator, Optional + +import anyio +import pytest +from fastmcp import Client, Context, FastMCP +from fastmcp.exceptions import ToolError +from fastmcp.server.middleware import Middleware +from mcp.types import CreateMessageResult, TextContent +from openenv.core.harness import ( + AgenticHarnessAdapter, + HarnessConfig, + HarnessEnvironment, + HarnessEvent, + HarnessEventType, + HarnessMCPBridge, +) + + +def make_mcp() -> FastMCP: + mcp = FastMCP("domain") + + @mcp.tool + def add(a: int, b: int) -> int: + """Add two numbers.""" + return a + b + + return mcp + + +class RecordingAdapter(AgenticHarnessAdapter): + BUILTIN_TOOL_NAMES = frozenset({"read_file"}) + + def __init__(self): + super().__init__(HarnessConfig(name="fake", command=["fake"])) + self.bridge_url: Optional[str] = "unset" + self.alive = False + + async def start(self, working_directory: str) -> None: + self.alive = True + + async def stop(self) -> None: + self.alive = False + + async def inject_tools(self, tools, bridge_url: Optional[str] = None) -> None: + self.bridge_url = bridge_url + + async def send_message_streaming(self, message: str) -> AsyncIterator[HarnessEvent]: + yield HarnessEvent(type=HarnessEventType.TURN_COMPLETE, data={"response": "ok"}) + + async def is_alive(self) -> bool: + return self.alive + + +class TestBridgeStandalone: + async def test_serves_tools_over_http(self): + bridge = HarnessMCPBridge(make_mcp()) + url = bridge.start() + try: + assert url.startswith("http://127.0.0.1:") + assert url.endswith("/mcp") + async with Client(url) as client: + tools = await client.list_tools() + assert [tool.name for tool in tools] == ["add"] + result = await client.call_tool("add", {"a": 2, "b": 3}) + assert result.content[0].text == "5" + finally: + bridge.stop() + + def test_two_bridges_get_distinct_ports(self): + bridge_a = HarnessMCPBridge(make_mcp()) + bridge_b = HarnessMCPBridge(make_mcp()) + try: + url_a = bridge_a.start() + url_b = bridge_b.start() + assert url_a != url_b + finally: + bridge_a.stop() + bridge_b.stop() + + def test_stop_is_idempotent(self): + bridge = HarnessMCPBridge(make_mcp()) + bridge.start() + bridge.stop() + bridge.stop() + assert bridge.url is None + + def test_stop_before_start_is_noop(self): + bridge = HarnessMCPBridge(make_mcp()) + bridge.stop() + assert bridge.url is None + + def test_start_is_idempotent_while_running(self): + bridge = HarnessMCPBridge(make_mcp()) + try: + assert bridge.start() == bridge.start() + finally: + bridge.stop() + + +class TestBridgeEnvironmentIntegration: + @pytest.mark.parametrize("cancellation", ["asyncio", "repeated_asyncio", "anyio"]) + async def test_cancelled_reset_waits_for_bridge_start_before_stopping( + self, monkeypatch, cancellation + ): + entered = threading.Event() + release = threading.Event() + finished = threading.Event() + stop_entered = asyncio.Event() + stop_release = asyncio.Event() + stopped = asyncio.Event() + bridges = [] + server_threads = [] + original_start = HarnessMCPBridge.start + + def delayed_start(bridge): + bridges.append(bridge) + entered.set() + try: + assert release.wait(timeout=10.0) + url = original_start(bridge) + server_threads.append(bridge._thread) + return url + finally: + finished.set() + + monkeypatch.setattr(HarnessMCPBridge, "start", delayed_start) + adapter = RecordingAdapter() + original_stop = adapter.stop + stop_calls = 0 + + async def delayed_stop(): + nonlocal stop_calls + stop_calls += 1 + if stop_calls == 2: + stop_entered.set() + await stop_release.wait() + await original_stop() + if stop_calls == 2: + stopped.set() + + monkeypatch.setattr(adapter, "stop", delayed_stop) + env = HarnessEnvironment(adapter=adapter, mcp=make_mcp()) + cancel_scope = anyio.CancelScope() + + async def reset(): + with cancel_scope: + await env.reset_async() + + reset_task = asyncio.create_task(reset()) + try: + assert await asyncio.to_thread(entered.wait, 10.0) + if cancellation == "anyio": + cancel_scope.cancel() + else: + reset_task.cancel() + await asyncio.sleep(0) + if cancellation == "repeated_asyncio": + reset_task.cancel() + + done, _ = await asyncio.wait([reset_task], timeout=0.05) + assert not done, "cancelled reset returned before bridge startup finished" + release.set() + await asyncio.wait_for(stop_entered.wait(), timeout=10.0) + if cancellation == "repeated_asyncio": + reset_task.cancel() + done, _ = await asyncio.wait([reset_task], timeout=0.05) + assert not done, "cancelled reset returned before adapter teardown finished" + stop_release.set() + + results = await asyncio.wait_for( + asyncio.gather(reset_task, return_exceptions=True), timeout=15.0 + ) + if cancellation == "anyio": + assert cancel_scope.cancelled_caught + else: + assert isinstance(results[0], asyncio.CancelledError) + assert stopped.is_set() + assert finished.is_set() + assert server_threads + assert not server_threads[0].is_alive() + assert env._bridge.url is None + assert adapter.alive is False + assert env._episode_active is False + finally: + release.set() + stop_release.set() + if not reset_task.done(): + reset_task.cancel() + await asyncio.wait_for( + asyncio.gather(reset_task, return_exceptions=True), timeout=15.0 + ) + await asyncio.to_thread(finished.wait, 10.0) + # Keep direct references in case a failed reset/close loses the + # bridge while its startup worker is still running. + for bridge in bridges: + await asyncio.to_thread(bridge.stop) + env.close() + assert all(not thread.is_alive() for thread in server_threads) + + async def test_reset_passes_live_bridge_url(self): + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=make_mcp()) + try: + await env.reset_async() + assert adapter.bridge_url is not None + async with Client(adapter.bridge_url) as client: + tools = await client.list_tools() + assert [tool.name for tool in tools] == ["add"] + finally: + env.close() + assert env._bridge is None + + async def test_no_tools_no_bridge(self): + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=None) + try: + await env.reset_async() + assert adapter.bridge_url is None + finally: + env.close() + + +class TestRenamedToolsAreServed: + """Conflict resolution renames tools before injection; the bridge must + answer to the injected names or the harness calls a name that 404s.""" + + @staticmethod + def make_colliding_mcp() -> FastMCP: + mcp = FastMCP("domain") + + @mcp.tool + def add(a: int, b: int) -> int: + """Add two numbers.""" + return a + b + + @mcp.tool + def read_file(path: str) -> str: + """Collides with the adapter's built-in read_file.""" + return f"contents of {path}" + + return mcp + + async def test_renamed_tool_is_listed_and_callable(self): + adapter = RecordingAdapter() # BUILTIN_TOOL_NAMES == {"read_file"} + mcp = self.make_colliding_mcp() + async with Client(mcp) as client: + source_tools = {tool.name: tool for tool in await client.list_tools()} + env = HarnessEnvironment(adapter=adapter, mcp=mcp) + try: + obs = await env.reset_async() + assert sorted(obs.metadata["injected_tools"]) == ["add", "env_read_file"] + + async with Client(adapter.bridge_url) as client: + served = {tool.name: tool for tool in await client.list_tools()} + assert sorted(served) == ["add", "env_read_file"] + for name, source_name in { + "add": "add", + "env_read_file": "read_file", + }.items(): + assert ( + served[name].inputSchema + == source_tools[source_name].inputSchema + ) + assert ( + served[name].outputSchema + == source_tools[source_name].outputSchema + ) + + # The renamed tool must actually resolve, not just be listed. + result = await client.call_tool("env_read_file", {"path": "a.py"}) + assert result.content[0].text == "contents of a.py" + with pytest.raises(ToolError, match="Unknown tool"): + await client.call_tool("read_file", {"path": "a.py"}) + + # The un-renamed one is untouched. + assert (await client.call_tool("add", {"a": 2, "b": 3})).content[ + 0 + ].text == "5" + finally: + env.close() + async with Client(mcp) as client: + assert sorted(tool.name for tool in await client.list_tools()) == [ + "add", + "read_file", + ] + + @pytest.mark.parametrize( + ("name", "arguments"), + [("env_read_file", {"path": "secret"}), ("add", {"a": 2, "b": 3})], + ) + async def test_renamed_bridge_preserves_source_middleware(self, name, arguments): + calls = [] + + class DenyCalls(Middleware): + async def on_call_tool(self, context, call_next): + calls.append(context.message.name) + raise ToolError("blocked by source middleware") + + mcp = self.make_colliding_mcp() + mcp.add_middleware(DenyCalls()) + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=mcp) + try: + await env.reset_async() + async with Client(adapter.bridge_url) as client: + with pytest.raises(ToolError, match="blocked by source middleware"): + await client.call_tool(name, arguments) + assert calls == ["read_file" if name == "env_read_file" else name] + finally: + env.close() + + async def test_renamed_bridge_preserves_source_lifespan_context(self): + resources = [] + + @asynccontextmanager + async def lifespan(server): + resource = {"open": True, "calls": 0} + resources.append(resource) + try: + yield {"resource": resource} + finally: + resource["open"] = False + + mcp = FastMCP("domain", lifespan=lifespan) + + @mcp.tool + def read_file(ctx: Context) -> int: + resource = ctx.lifespan_context["resource"] + if not resource["open"]: + raise RuntimeError("resource already closed") + resource["calls"] += 1 + return resource["calls"] + + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=mcp) + try: + await env.reset_async() + for expected in (1, 2): + async with Client(adapter.bridge_url) as client: + result = await client.call_tool("env_read_file", {}) + assert result.content[0].text == str(expected) + assert resources[-1]["open"] + assert sum(resource["calls"] for resource in resources) == 2 + finally: + env.close() + assert resources + assert all(not resource["open"] for resource in resources) + + async def test_renamed_bridge_forwards_sampling_and_progress(self): + sampled = [] + progress = [] + mcp = FastMCP("domain") + + @mcp.tool + async def read_file(path: str, ctx: Context) -> str: + response = await ctx.sample(path) + await ctx.report_progress(1, 1, path) + return response.text + + async def sample(messages, params, context): + text = messages[0].content.text + sampled.append(text) + return CreateMessageResult( + role="assistant", + content=TextContent(type="text", text=f"sampled {text}"), + model="test", + ) + + async def record_progress(completed, total, message): + progress.append((completed, total, message)) + + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=mcp) + try: + await env.reset_async() + async with Client(adapter.bridge_url, sampling_handler=sample) as client: + for path in ("first", "second"): + result = await client.call_tool( + "env_read_file", + {"path": path}, + progress_handler=record_progress, + timeout=5.0, + ) + assert result.content[0].text == f"sampled {path}" + assert sampled == ["first", "second"] + assert progress == [(1, 1, "first"), (1, 1, "second")] + finally: + env.close() + + async def test_no_renames_serves_the_source_server_directly(self): + adapter = RecordingAdapter() + env = HarnessEnvironment(adapter=adapter, mcp=make_mcp()) # only "add" + try: + await env.reset_async() + async with Client(adapter.bridge_url) as client: + assert [t.name for t in await client.list_tools()] == ["add"] + finally: + env.close() diff --git a/tests/core/test_agentic_harness_e2e.py b/tests/core/test_agentic_harness_e2e.py new file mode 100644 index 000000000..015466715 --- /dev/null +++ b/tests/core/test_agentic_harness_e2e.py @@ -0,0 +1,251 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""End-to-end tests for the RFC 005 agentic harness stack. + +Unlike the unit tests, nothing on the critical path is faked here: + +- A real CLI "harness" subprocess is spawned via ``HarnessProcess``. +- ``inject_tools`` writes a real MCP config file, which the subprocess reads. +- The subprocess calls the environment's domain tool over the live + ``HarnessMCPBridge`` using a real MCP client. +- Turn events stream back over stdio and are mapped to ``HarnessEvent``. + +The stack is validated in both modes: + +- Simulation: ``HarnessEnvironment.reset()``/``step()`` multi-turn with a + rubric, through the sync facade a training loop would use. +- Production: a real uvicorn server registered with ``mode="production"``, + driven by a real WebSocket client on ``/harness``. +""" + +from __future__ import annotations + +import json +import socket +import sys +import threading +import time +from pathlib import Path +from typing import AsyncIterator, Optional + +import httpx +import pytest +import uvicorn +from fastmcp import FastMCP +from openenv.core.env_server.http_server import create_fastapi_app +from openenv.core.env_server.types import Observation +from openenv.core.harness import ( + AgenticHarnessAdapter, + HarnessAction, + HarnessConfig, + HarnessEnvironment, + HarnessEvent, + HarnessEventType, + HarnessProcess, +) +from openenv.core.rubrics import Rubric +from websockets.sync.client import connect as ws_connect + +# The harness role is played by a real, lintable module spawned as a +# subprocess in "mcp" mode: it reads the injected MCP config, connects to the +# bridge with a real MCP client each turn, and emits JSON-line events. +SCRIPTED_HARNESS = Path(__file__).parent / "scripted_harness.py" + + +class ScriptedCLIAdapter(AgenticHarnessAdapter): + """Adapter for the scripted harness: config-file injection + stdio JSON.""" + + BUILTIN_TOOL_NAMES = frozenset({"read_file"}) + + def __init__(self, config: HarnessConfig, config_dir: Path): + super().__init__(config) + self._config_path = config_dir / "mcp.json" + self._process: Optional[HarnessProcess] = None + + async def inject_tools(self, tools, bridge_url: Optional[str] = None) -> None: + self._config_path.write_text( + json.dumps({"mcp_url": bridge_url, "tools": [t.name for t in tools]}) + ) + + async def start(self, working_directory: str) -> None: + self._process = HarnessProcess( + self.config.command, + cwd=working_directory, + env_vars={"HARNESS_MCP_CONFIG": str(self._config_path)}, + startup_timeout_s=self.config.startup_timeout_s, + ) + + def is_ready(line: str) -> bool: + try: + return json.loads(line).get("event") == "ready" + except json.JSONDecodeError: + return False + + await self._process.start(ready_check=is_ready) + + async def stop(self) -> None: + if self._process is not None: + await self._process.stop() + + async def is_alive(self) -> bool: + return self._process is not None and self._process.is_running() + + async def send_message_streaming(self, message: str) -> AsyncIterator[HarnessEvent]: + assert self._process is not None + await self._process.write_line(json.dumps({"message": message})) + while True: + line = await self._process.read_line(timeout_s=30.0) + if line is None: + raise RuntimeError("harness died mid-turn") + native = json.loads(line) + event_type = native.pop("event") + if event_type == "turn_complete": + yield HarnessEvent(type=HarnessEventType.TURN_COMPLETE, data=native) + return + yield HarnessEvent(type=HarnessEventType(event_type), data=native) + + +class SumRubric(Rubric): + def forward(self, action, observation) -> float: + return 1.0 if "sum=5" in observation.metadata.get("response", "") else 0.0 + + +def make_mcp() -> FastMCP: + mcp = FastMCP("domain") + + @mcp.tool + def add(a: int, b: int) -> int: + """Add two numbers.""" + return a + b + + @mcp.tool + def read_file(path: str) -> str: + """Read a file (collides with a harness builtin -> env_ prefix).""" + return path + + return mcp + + +@pytest.fixture +def env_factory(tmp_path_factory, request): + scripted_mode = getattr(request, "param", "mcp") + + def factory() -> HarnessEnvironment: + config = HarnessConfig( + name="scripted-cli", + command=[sys.executable, "-u", str(SCRIPTED_HARNESS), scripted_mode], + working_directory=".", + startup_timeout_s=20.0, + session_timeout_s=60.0, + ) + adapter = ScriptedCLIAdapter( + config, config_dir=tmp_path_factory.mktemp("harness-config") + ) + return HarnessEnvironment(adapter=adapter, mcp=make_mcp(), rubric=SumRubric()) + + return factory + + +class TestSimulationEndToEnd: + def test_multi_turn_episode_through_real_stack(self, env_factory): + env = env_factory() + try: + obs = env.reset(episode_id="e2e-episode") + assert obs.done is False + # Conflict resolution visible on the wire: the colliding + # read_file was prefixed, add passed through untouched. + assert sorted(obs.metadata["injected_tools"]) == [ + "add", + "env_read_file", + ] + + # Turn 1: the subprocess calls the env tool through the bridge + obs1 = env.step(HarnessAction(message="please compute")) + assert "sum=5" in obs1.metadata["response"] + assert obs1.reward == 1.0 # rubric saw the turn's observation + assert obs1.done is False + assert [e["type"] for e in obs1.metadata["turn_events"]] == [ + "tool_call", + "tool_result", + "turn_complete", + ] + + # Turn 2: the harness's done signal ends the episode + obs2 = env.step(HarnessAction(message="finish up")) + assert obs2.done is True + assert env.state.step_count == 2 + assert len(env.trajectory) == 6 # both turns' events accumulated + finally: + env.close() + + +class TestProductionEndToEnd: + @pytest.fixture + def live_server(self, env_factory): + app = create_fastapi_app( + env_factory, HarnessAction, Observation, mode="production" + ) + + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + server = uvicorn.Server(uvicorn.Config(app, log_level="warning")) + thread = threading.Thread( + target=lambda: server.run(sockets=[sock]), daemon=True + ) + thread.start() + deadline = time.monotonic() + 15 + while not server.started and time.monotonic() < deadline: + time.sleep(0.05) + assert server.started + try: + yield port + finally: + server.should_exit = True + thread.join(timeout=10) + + def test_harness_websocket_over_real_server(self, live_server): + port = live_server + base = f"http://127.0.0.1:{port}" + + assert httpx.get(f"{base}/health").status_code == 200 + assert httpx.post(f"{base}/reset", json={}).status_code in (404, 405) + + with ws_connect(f"ws://127.0.0.1:{port}/harness") as websocket: + started = json.loads(websocket.recv(timeout=30)) + assert started["type"] == "session_started" + assert started["data"]["harness"] == "scripted-cli" + assert started["data"]["session_id"] + + websocket.send(json.dumps({"type": "message", "content": "compute"})) + frames = [] + while True: + frame = json.loads(websocket.recv(timeout=30)) + frames.append(frame) + if frame["type"] == "turn_complete": + break + assert [f["type"] for f in frames] == [ + "tool_call", + "tool_result", + "turn_complete", + ] + assert "sum=5" in frames[-1]["data"]["response"] + + # Second turn on the same connection reuses the same harness + websocket.send(json.dumps({"type": "message", "content": "finish"})) + while True: + frame = json.loads(websocket.recv(timeout=30)) + if frame["type"] == "turn_complete": + break + assert frame["data"]["done"] is True + + @pytest.mark.parametrize("env_factory", ["exit-with-secret"], indirect=True) + def test_startup_stderr_is_not_sent_to_websocket_client(self, live_server): + with ws_connect(f"ws://127.0.0.1:{live_server}/harness") as websocket: + frame = json.loads(websocket.recv(timeout=30)) + assert frame["type"] == "error" + assert frame["data"]["code"] == "SESSION_ERROR" + assert frame["data"]["message"] == "harness session failed" + assert "FAKE_HARNESS_SECRET_FOR_TEST" not in json.dumps(frame) + assert "stderr tail" not in json.dumps(frame) diff --git a/tests/core/test_agentic_harness_environment.py b/tests/core/test_agentic_harness_environment.py new file mode 100644 index 000000000..c66d6217a --- /dev/null +++ b/tests/core/test_agentic_harness_environment.py @@ -0,0 +1,621 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for HarnessEnvironment (RFC 005 turn-based wrapper).""" + +from __future__ import annotations + +import asyncio +import json +from typing import AsyncIterator, Optional + +import pytest +from fastmcp import FastMCP +from openenv.core.env_server._utils import overrides_method +from openenv.core.env_server.interfaces import Environment +from openenv.core.env_server.mcp_types import ListToolsAction, Tool +from openenv.core.env_server.serialization import serialize_observation +from openenv.core.env_server.types import Action +from openenv.core.harness import ( + AgenticHarnessAdapter, + HarnessAction, + HarnessConfig, + HarnessEnvironment, + HarnessError, + HarnessEvent, + HarnessEventType, + HarnessNotRunningError, + HarnessTurnTimeoutError, +) +from openenv.core.rubrics import Rubric + + +def turn_events(response: str, done: bool = False) -> list[HarnessEvent]: + return [ + HarnessEvent(type=HarnessEventType.TOOL_CALL, data={"tool_name": "shell"}), + HarnessEvent( + type=HarnessEventType.TURN_COMPLETE, + data={"response": response, "done": done}, + ), + ] + + +class FakeAdapter(AgenticHarnessAdapter): + """Recording fake with failure knobs.""" + + BUILTIN_TOOL_NAMES = frozenset({"read_file"}) + + def __init__(self, config: Optional[HarnessConfig] = None): + super().__init__(config or HarnessConfig(name="fake", command=["fake"])) + self.calls: list[str] = [] + self.injected_tools: Optional[list[Tool]] = None + self.injected_bridge_url: Optional[str] = "unset" + self.alive = False + self.fail_on_start = False + self.fail_on_send = False + self.raise_turn_timeout = False + self.send_delay_s = 0.0 + self.scripted_turns: list[list[HarnessEvent]] = [] + + async def start(self, working_directory: str) -> None: + self.calls.append("start") + if self.fail_on_start: + raise HarnessError("boom on start") + self.alive = True + + async def stop(self) -> None: + self.calls.append("stop") + self.alive = False + + async def inject_tools(self, tools, bridge_url: Optional[str] = None) -> None: + self.calls.append("inject_tools") + self.injected_tools = list(tools) + self.injected_bridge_url = bridge_url + + async def send_message_streaming(self, message: str) -> AsyncIterator[HarnessEvent]: + self.calls.append("send") + if self.send_delay_s: + await asyncio.sleep(self.send_delay_s) + if self.raise_turn_timeout: + raise HarnessTurnTimeoutError("adapter timed the turn out itself") + if self.fail_on_send: + raise HarnessError("harness exploded mid-turn") + events = ( + self.scripted_turns.pop(0) + if self.scripted_turns + else turn_events("default response") + ) + for event in events: + yield event + + async def is_alive(self) -> bool: + return self.alive + + +class FakeBridge: + """In-process stand-in for HarnessMCPBridge (real one tested separately).""" + + instances: list["FakeBridge"] = [] + + def __init__(self, mcp_server): + self.mcp_server = mcp_server + self.started = 0 + self.stopped = 0 + FakeBridge.instances.append(self) + + def start(self, timeout_s: float = 10.0) -> str: + self.started += 1 + return "http://127.0.0.1:9/mcp" + + def stop(self, timeout_s: float = 5.0) -> None: + self.stopped += 1 + + +@pytest.fixture(autouse=True) +def fake_bridge(monkeypatch): + FakeBridge.instances = [] + monkeypatch.setattr("openenv.core.harness.environment.HarnessMCPBridge", FakeBridge) + return FakeBridge + + +class SpyRubric(Rubric): + def __init__(self): + super().__init__() + self.seen: list[tuple] = [] + self.resets = 0 + + def forward(self, action, observation) -> float: + self.seen.append((action, observation)) + return 0.75 + + def reset(self) -> None: + self.resets += 1 + + +def make_env(**kwargs) -> tuple[HarnessEnvironment, FakeAdapter]: + adapter = FakeAdapter() + mcp = kwargs.pop("mcp", "default") + if mcp == "default": + mcp = FastMCP("domain") + + @mcp.tool + def query_db(sql: str) -> str: + """Run a SQL query.""" + return sql + + @mcp.tool + def read_file(path: str) -> str: + """Read a task file.""" + return path + + env = HarnessEnvironment(adapter=adapter, mcp=mcp, **kwargs) + return env, adapter + + +class TestReset: + async def test_inject_before_start_with_conflict_resolution(self): + env, adapter = make_env() + obs = await env.reset_async() + + assert adapter.calls == ["stop", "inject_tools", "start"] + names = sorted(t.name for t in adapter.injected_tools) + assert names == ["env_read_file", "query_db"] + assert sorted(obs.metadata["injected_tools"]) == names + assert obs.done is False + assert env.state.step_count == 0 + assert env.state.episode_id + assert adapter.injected_bridge_url == "http://127.0.0.1:9/mcp" + assert len(FakeBridge.instances) == 1 + assert FakeBridge.instances[0].started == 1 + + async def test_reset_stops_live_adapter_first(self): + env, adapter = make_env() + await env.reset_async() + adapter.calls.clear() + await env.reset_async() + assert adapter.calls == ["stop", "inject_tools", "start"] + + async def test_reset_clears_trajectory_and_resets_rubric(self): + rubric = SpyRubric() + env, adapter = make_env(rubric=rubric) + await env.reset_async() + await env.step_async(HarnessAction(message="turn 1")) + assert env.trajectory + await env.reset_async(episode_id="ep-2") + assert env.trajectory == [] + assert env.state.episode_id == "ep-2" + assert env.state.step_count == 0 + assert rubric.resets == 2 + + async def test_no_mcp_tools_injects_empty_list_and_no_bridge(self): + env, adapter = make_env(mcp=None) + await env.reset_async() + assert adapter.injected_tools == [] + assert adapter.injected_bridge_url is None + assert FakeBridge.instances == [] + + @pytest.mark.parametrize("async_rubric", [False, True]) + async def test_rubric_reset_failure_stops_resources_and_allows_retry( + self, async_rubric + ): + failure = RuntimeError("rubric reset failed") + + class FailingRubric(SpyRubric): + fail = True + + def reset(self): + assert adapter.alive + assert FakeBridge.instances[-1].started == 1 + if self.fail: + raise failure + super().reset() + + class AsyncFailingRubric(FailingRubric): + async def reset_async(self): + self.reset() + + rubric = AsyncFailingRubric() if async_rubric else FailingRubric() + env, adapter = make_env(rubric=rubric) + with pytest.raises(RuntimeError) as caught: + await env.reset_async() + + assert caught.value is failure + assert adapter.alive is False + assert adapter.calls[-1] == "stop" + assert FakeBridge.instances[0].stopped >= 1 + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="after failed reset")) + + rubric.fail = False + try: + obs = await env.reset_async() + assert obs.done is False + assert adapter.alive + assert rubric.resets == 1 + assert len(FakeBridge.instances) == 2 + obs = await env.step_async(HarnessAction(message="retry")) + assert obs.metadata["response"] == "default response" + finally: + env.close() + + async def test_start_failure_leaves_episode_inactive(self): + env, adapter = make_env() + adapter.fail_on_start = True + with pytest.raises(HarnessError): + await env.reset_async() + assert FakeBridge.instances[0].stopped >= 1 + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="hi")) + + +class TestStep: + async def test_turn_updates_state_and_trajectory(self): + env, adapter = make_env() + adapter.scripted_turns = [turn_events("first"), turn_events("second", True)] + await env.reset_async() + + obs1 = await env.step_async(HarnessAction(message="fix the bug")) + assert obs1.done is False + assert obs1.metadata["response"] == "first" + assert obs1.metadata["turn_number"] == 1 + assert env.state.step_count == 1 + + obs2 = await env.step_async(HarnessAction(message="tests still fail")) + assert obs2.done is True + assert obs2.metadata["response"] == "second" + assert env.state.step_count == 2 + assert len(env.trajectory) == 4 + + async def test_observation_serializes_to_json(self): + env, adapter = make_env() + await env.reset_async() + obs = await env.step_async(HarnessAction(message="go")) + payload = serialize_observation(obs) + assert json.loads(json.dumps(payload))["metadata"]["turn_events"] + + async def test_rubric_applied_after_turn(self): + rubric = SpyRubric() + env, adapter = make_env(rubric=rubric) + await env.reset_async() + action = HarnessAction(message="go") + obs = await env.step_async(action) + assert obs.reward == 0.75 + seen_action, seen_obs = rubric.seen[0] + assert seen_action is action + assert seen_obs.metadata["response"] == "default response" + + async def test_step_before_reset_raises(self): + env, _ = make_env() + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="hi")) + + async def test_dead_adapter_returns_terminal_error_observation(self): + env, adapter = make_env() + await env.reset_async() + adapter.alive = False + obs = await env.step_async(HarnessAction(message="hi")) + assert obs.done is True + assert obs.metadata["error_type"] == "harness_crashed" + + async def test_mid_turn_crash_stops_adapter(self): + env, adapter = make_env() + await env.reset_async() + adapter.fail_on_send = True + obs = await env.step_async(HarnessAction(message="hi")) + assert obs.done is True + assert obs.metadata["error_type"] == "harness_crashed" + assert adapter.calls[-1] == "stop" + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="again")) + + async def test_turn_timeout_returns_terminal_error_observation(self): + env, adapter = make_env() + adapter.config.session_timeout_s = 0.05 + adapter.send_delay_s = 5.0 + await env.reset_async() + obs = await env.step_async(HarnessAction(message="hi")) + assert obs.done is True + assert obs.metadata["error_type"] == "turn_timeout" + assert adapter.calls[-1] == "stop" + + async def test_unknown_action_type_raises(self): + env, _ = make_env() + await env.reset_async() + with pytest.raises(TypeError): + await env.step_async(Action()) + + async def test_list_tools_routing_preserved(self): + env, _ = make_env() + obs = await env.step_async(ListToolsAction()) + assert sorted(t.name for t in obs.tools) == ["query_db", "read_file"] + + +class TestSyncFacadeAndClose: + def test_sync_reset_and_step(self): + env, adapter = make_env() + adapter.scripted_turns = [turn_events("sync response", True)] + obs = env.reset() + assert obs.done is False + obs = env.step(HarnessAction(message="go")) + assert obs.metadata["response"] == "sync response" + assert obs.done is True + assert env.state.step_count == 1 + + def test_close_is_idempotent_and_stops_adapter_and_bridge(self): + env, adapter = make_env() + env.reset() + env.close() + assert "stop" in adapter.calls + assert FakeBridge.instances[0].stopped >= 1 + stop_count = adapter.calls.count("stop") + env.close() + assert adapter.calls.count("stop") == stop_count + + def test_constructor_starts_nothing(self): + env, adapter = make_env() + assert adapter.calls == [] + + def test_server_selects_async_paths(self): + env, _ = make_env() + assert overrides_method(env.reset_async, Environment.reset_async) + assert overrides_method(env.step_async, Environment.step_async) + assert env.SUPPORTS_CONCURRENT_SESSIONS is False + + +class TestTransforms: + @pytest.mark.parametrize("sync", [False, True]) + async def test_reset_and_turn_apply_transform_once_after_rubric(self, sync): + seen = [] + + def transform(observation): + seen.append(observation) + return observation.model_copy( + update={ + "reward": observation.reward + 1.0, + "metadata": {**observation.metadata, "transformed": True}, + } + ) + + rubric = SpyRubric() + env, adapter = make_env(rubric=rubric, transform=transform) + adapter.scripted_turns = [turn_events("first"), turn_events("last", True)] + try: + reset_obs = env.reset() if sync else await env.reset_async() + assert reset_obs.metadata["transformed"] is True + assert reset_obs.reward == 1.0 + assert len(seen) == 1 + + for message, done in [("first", False), ("last", True)]: + action = HarnessAction(message=message) + obs = env.step(action) if sync else await env.step_async(action) + assert obs.metadata["transformed"] is True + assert obs.metadata["response"] == message + assert obs.done is done + assert obs.reward == 1.75 + assert rubric.seen[-1][1] is seen[-1] + assert "transformed" not in rubric.seen[-1][1].metadata + assert len(seen) == 3 + finally: + env.close() + + @pytest.mark.parametrize( + "failure, value, error_type", + [ + ("alive", False, "harness_crashed"), + ("fail_on_send", True, "harness_crashed"), + ("raise_turn_timeout", True, "turn_timeout"), + ("send_delay_s", 5.0, "turn_timeout"), + ], + ) + async def test_terminal_errors_apply_transform(self, failure, value, error_type): + seen = [] + + def transform(observation): + seen.append(observation) + return observation.model_copy( + update={"metadata": {**observation.metadata, "transformed": True}} + ) + + env, adapter = make_env(transform=transform) + try: + await env.reset_async() + seen.clear() + setattr(adapter, failure, value) + obs = await env.step_async(HarnessAction(message="go"), timeout_s=0.05) + assert obs.done is True + assert obs.metadata["error_type"] == error_type + assert obs.metadata["transformed"] is True + assert len(seen) == 1 + assert seen[0].metadata["turn_events"] == obs.metadata["turn_events"] + finally: + env.close() + + +class TestCleanupAndErrorClassification: + @pytest.mark.parametrize( + "stage, failure", + [ + ("is_alive", RuntimeError("adapter liveness check failed")), + ("send_message", RuntimeError("unexpected adapter failure")), + ("send_message", json.JSONDecodeError("invalid harness JSON", "{", 1)), + ("rubric", RuntimeError("turn evaluation failed")), + ], + ) + @pytest.mark.parametrize("stop_fails", [False, True]) + async def test_unexpected_turn_failure_stops_resources_and_allows_retry( + self, stage, failure, stop_fails, monkeypatch + ): + env, adapter = make_env(rubric=SpyRubric()) + await env.reset_async() + target = env if stage == "rubric" else adapter + method = "_apply_rubric_async" if stage == "rubric" else stage + original = getattr(target, method) + original_stop = adapter.stop + + async def fail(*args, **kwargs): + raise failure + + async def failing_stop(): + await original_stop() + raise RuntimeError("cleanup failed") + + monkeypatch.setattr(target, method, fail) + if stop_fails: + monkeypatch.setattr(adapter, "stop", failing_stop) + try: + with pytest.raises(type(failure)) as caught: + await env.step_async(HarnessAction(message="go")) + + assert caught.value is failure + assert adapter.alive is False + assert adapter.calls[-1] == "stop" + assert FakeBridge.instances[0].stopped >= 1 + with pytest.raises(HarnessNotRunningError, match="No active episode"): + await env.step_async(HarnessAction(message="again")) + + monkeypatch.setattr(target, method, original) + monkeypatch.setattr(adapter, "stop", original_stop) + await env.reset_async() + assert adapter.alive + assert len(FakeBridge.instances) == 2 + obs = await env.step_async(HarnessAction(message="fresh turn")) + assert obs.metadata["response"] == "default response" + finally: + env.close() + + @pytest.mark.parametrize("stage", ["is_alive", "send_message", "rubric"]) + @pytest.mark.parametrize("stop_fails", [False, True]) + async def test_cancelled_turn_cleans_up_resources( + self, stage, stop_fails, monkeypatch + ): + env, adapter = make_env(rubric=SpyRubric()) + await env.reset_async() + entered = asyncio.Event() + target = env if stage == "rubric" else adapter + method = "_apply_rubric_async" if stage == "rubric" else stage + original = getattr(target, method) + original_stop = adapter.stop + + async def stall(*args, **kwargs): + await original(*args, **kwargs) + entered.set() + await asyncio.Event().wait() + + async def failing_stop(): + await original_stop() + raise HarnessError("cleanup failed") + + monkeypatch.setattr(target, method, stall) + if stop_fails: + monkeypatch.setattr(adapter, "stop", failing_stop) + turn_task = asyncio.create_task(env.step_async(HarnessAction(message="go"))) + try: + await asyncio.wait_for(entered.wait(), timeout=10.0) + turn_task.cancel() + with pytest.raises(asyncio.CancelledError): + await turn_task + + assert adapter.alive is False + assert adapter.calls[-1] == "stop" + assert FakeBridge.instances[0].stopped >= 1 + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="again")) + + monkeypatch.setattr(target, method, original) + monkeypatch.setattr(adapter, "stop", original_stop) + await env.reset_async() + obs = await env.step_async(HarnessAction(message="fresh turn")) + assert obs.metadata["response"] == "default response" + finally: + if not turn_task.done(): + turn_task.cancel() + await asyncio.gather(turn_task, return_exceptions=True) + env.close() + + @pytest.mark.parametrize("stage", ["inject_tools", "start", "rubric"]) + async def test_cancelled_reset_cleans_up_resources(self, stage, monkeypatch): + env, adapter = make_env(rubric=SpyRubric()) + entered = asyncio.Event() + target = env if stage == "rubric" else adapter + method = "_reset_rubric_async" if stage == "rubric" else stage + original = getattr(target, method) + + async def stall(*args, **kwargs): + await original(*args, **kwargs) + entered.set() + await asyncio.Event().wait() + + monkeypatch.setattr(target, method, stall) + reset_task = asyncio.create_task(env.reset_async()) + try: + await asyncio.wait_for(entered.wait(), timeout=10.0) + reset_task.cancel() + with pytest.raises(asyncio.CancelledError): + await reset_task + + assert adapter.alive is False + assert adapter.calls[-1] == "stop" + assert FakeBridge.instances[0].stopped >= 1 + with pytest.raises(HarnessNotRunningError): + await env.step_async(HarnessAction(message="go")) + finally: + if not reset_task.done(): + reset_task.cancel() + await asyncio.gather(reset_task, return_exceptions=True) + env.close() + + async def test_reset_stops_adapter_even_when_not_alive(self): + # A harness that died on its own reports is_alive() False while still + # holding reapable resources; skipping stop() leaked them. + env, adapter = make_env() + await env.reset_async() + adapter.alive = False + adapter.calls.clear() + + await env.reset_async() + + assert adapter.calls[0] == "stop" + + async def test_start_failure_stops_the_adapter(self): + env, adapter = make_env() + adapter.fail_on_start = True + with pytest.raises(HarnessError): + await env.reset_async() + # inject -> start (raises) -> stop + assert adapter.calls[-1] == "stop" + + async def test_adapter_raised_timeout_is_not_reported_as_a_crash(self): + env, adapter = make_env() + await env.reset_async() + adapter.raise_turn_timeout = True + + obs = await env.step_async(HarnessAction(message="hi")) + + assert obs.done is True + assert obs.metadata["error_type"] == "turn_timeout" + + +class TestModeSpecificTools: + async def test_mode_tools_are_not_injected(self, caplog): + # Tools registered with tool(mode=...) are tracked by the environment + # rather than the FastMCP server, so the bridge cannot serve them. + # Advertising them would promise the harness a tool it cannot call. + env, adapter = make_env() + + @env.tool(mode="production") + def prod_only(x: int) -> int: + """Production-only tool.""" + return x + + # A mode tool is only advertised when the env is in that mode; without + # this the tool never reaches the injectable list and the test would + # pass vacuously. + env._mode = "production" + listed = await env._async_handle_list_tools() + assert "prod_only" in [t.name for t in listed.tools] + + with caplog.at_level("WARNING"): + obs = await env.reset_async() + + assert "prod_only" not in obs.metadata["injected_tools"] + assert all(t.name != "prod_only" for t in adapter.injected_tools) + assert "prod_only" in caplog.text diff --git a/tests/core/test_agentic_harness_process.py b/tests/core/test_agentic_harness_process.py new file mode 100644 index 000000000..0d1f364f8 --- /dev/null +++ b/tests/core/test_agentic_harness_process.py @@ -0,0 +1,329 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Subprocess lifecycle tests for HarnessProcess (real processes).""" + +from __future__ import annotations + +import asyncio +import os +import signal +import sys +import time +from pathlib import Path + +import pytest +from openenv.core.harness import ( + HarnessNotRunningError, + HarnessProcess, + HarnessStartupError, +) + +# Subprocess behaviors live in a real, lintable module (see its docstring +# for the available modes) instead of inline code strings. +SCRIPTED_HARNESS = Path(__file__).parent / "scripted_harness.py" + + +def make_process(mode: str, **kwargs) -> HarnessProcess: + defaults = {"startup_timeout_s": 10.0, "terminate_grace_s": 1.0} + defaults.update(kwargs) + return HarnessProcess( + [sys.executable, "-u", str(SCRIPTED_HARNESS), mode], cwd=".", **defaults + ) + + +def ready(line: str) -> bool: + return line == "ready" + + +class TestLifecycle: + async def test_start_echo_read_stop(self): + process = make_process("echo") + assert process.is_running() is False + await process.start(ready_check=ready) + assert process.is_running() is True + + await process.write_line("hello") + assert await process.read_line(timeout_s=10.0) == "echo:hello" + + await process.stop() + assert process.is_running() is False + + async def test_stop_is_idempotent(self): + process = make_process("echo") + await process.start(ready_check=ready) + await process.stop() + await process.stop() + assert process.is_running() is False + assert process._proc is None + assert process._reader_threads == [] + + async def test_stop_before_start_is_noop(self): + process = make_process("echo") + await process.stop() + assert process.is_running() is False + + async def test_double_start_rejected(self): + process = make_process("echo") + await process.start(ready_check=ready) + try: + with pytest.raises(HarnessStartupError, match="already running"): + await process.start(ready_check=ready) + finally: + await process.stop() + + +class TestStartupFailures: + @pytest.mark.parametrize("cancel", [False, True]) + async def test_interrupted_readiness_cleans_up_process(self, cancel): + process = make_process("echo") + checking = asyncio.Event() + resources = [] + + def check(line): + resources.append((process._proc, list(process._reader_threads))) + checking.set() + if not cancel: + raise ValueError("readiness check failed") + return False + + start_task = asyncio.create_task(process.start(ready_check=check)) + try: + await asyncio.wait_for(checking.wait(), timeout=10.0) + proc, readers = resources[0] + if cancel: + start_task.cancel() + with pytest.raises(asyncio.CancelledError): + await start_task + else: + with pytest.raises(ValueError, match="readiness check failed"): + await start_task + + assert proc.poll() is not None + assert all(s.closed for s in (proc.stdin, proc.stdout, proc.stderr)) + assert all(not thread.is_alive() for thread in readers) + assert process._proc is None + finally: + if not start_task.done(): + start_task.cancel() + await asyncio.gather(start_task, return_exceptions=True) + await process.stop() + + async def test_startup_timeout_kills_process(self): + process = make_process("slow-start", startup_timeout_s=0.5) + with pytest.raises(HarnessStartupError, match="did not become ready"): + await process.start(ready_check=ready) + assert process.is_running() is False + + async def test_immediate_exit_reports_code_and_retains_diagnostics(self): + process = make_process("exit-now") + with pytest.raises(HarnessStartupError) as exc_info: + await process.start(ready_check=ready) + assert "exited with code 3" in str(exc_info.value) + assert "dying" in process.drain_stderr() + assert "dying" not in str(exc_info.value) + + async def test_startup_error_does_not_expose_stderr(self): + process = make_process("exit-with-secret") + with pytest.raises(HarnessStartupError) as exc_info: + await process.start(ready_check=ready) + assert "exited with code 3" in str(exc_info.value) + assert "FAKE_HARNESS_SECRET_FOR_TEST" not in str(exc_info.value) + assert "stderr tail" not in str(exc_info.value) + assert "FAKE_HARNESS_SECRET_FOR_TEST" in process.drain_stderr() + + async def test_unspawnable_command(self): + process = HarnessProcess( + ["/nonexistent/definitely-not-a-binary", "FAKE_HARNESS_SECRET_FOR_TEST"], + cwd=".", + ) + with pytest.raises(HarnessStartupError, match="failed to spawn") as exc_info: + await process.start() + assert "FAKE_HARNESS_SECRET_FOR_TEST" not in str(exc_info.value) + assert "/nonexistent" not in str(exc_info.value) + + +class TestCrashDetection: + @pytest.mark.parametrize("restart_fails", [False, True]) + async def test_restart_cleans_up_exited_process(self, restart_fails): + process = make_process("crash-after-echo") + await process.start(ready_check=ready) + old_proc = process._proc + old_readers = list(process._reader_threads) + try: + await process.write_line("old turn") + assert await asyncio.to_thread(old_proc.wait, timeout=10.0) == 1 + + if restart_fails: + process.command = ["/nonexistent/definitely-not-a-binary"] + with pytest.raises(HarnessStartupError, match="failed to spawn"): + await process.start() + else: + await process.start(ready_check=ready) + + assert all( + stream.closed + for stream in (old_proc.stdin, old_proc.stdout, old_proc.stderr) + ) + assert all(not thread.is_alive() for thread in old_readers) + + if not restart_fails: + await process.write_line("new turn") + assert await process.read_line(timeout_s=10.0) == "echo:new turn" + finally: + await process.stop() + # Also release the original handles if a regression loses them. + for stream in (old_proc.stdin, old_proc.stdout, old_proc.stderr): + stream.close() + for thread in old_readers: + thread.join(timeout=2.0) + + async def test_crash_mid_session(self): + process = make_process("crash-after-echo") + await process.start(ready_check=ready) + await process.write_line("one") + assert await process.read_line(timeout_s=10.0) == "echo:one" + + # Process exits after the first echo; wait for it to die + deadline = time.monotonic() + 10.0 + while process.is_running() and time.monotonic() < deadline: + time.sleep(0.02) + assert process.is_running() is False + + with pytest.raises(HarnessNotRunningError): + await process.write_line("two") + await process.stop() + + async def test_read_line_returns_none_on_timeout(self): + process = make_process("echo") + await process.start(ready_check=ready) + try: + assert await process.read_line(timeout_s=0.1) is None + finally: + await process.stop() + + +class TestEOF: + @pytest.mark.parametrize("timeout_s", [None, 10.0]) + async def test_exit_preserves_buffered_lines_before_repeated_eof(self, timeout_s): + process = make_process("exit-with-output") + await process.start() + try: + deadline = time.monotonic() + 10.0 + while process.is_running() and time.monotonic() < deadline: + await asyncio.sleep(0.02) + assert process.is_running() is False + + for expected in ("first", "second", "last"): + assert ( + await asyncio.wait_for(process.read_line(timeout_s), timeout=2.0) + == expected + ) + for _ in range(2): + assert ( + await asyncio.wait_for(process.read_line(timeout_s), timeout=2.0) + is None + ) + finally: + await process.stop() + + @pytest.mark.parametrize("timeout_s", [None, 10.0]) + async def test_read_line_returns_none_when_stdout_closes(self, timeout_s): + process = make_process("close-stdout") + await process.start(ready_check=ready) + try: + await process.write_line("close") + for _ in range(2): + assert ( + await asyncio.wait_for(process.read_line(timeout_s), timeout=2.0) + is None + ) + assert process.is_running() is True + finally: + await process.stop() + + async def test_stop_wakes_pending_read_and_restart_can_read_output(self): + process = make_process("echo") + await process.start(ready_check=ready) + try: + pending_read = asyncio.create_task(process.read_line()) + await asyncio.sleep(0) + await process.stop() + assert await asyncio.wait_for(pending_read, timeout=2.0) is None + assert await asyncio.wait_for(process.read_line(), timeout=2.0) is None + + await process.start(ready_check=ready) + await process.write_line("again") + assert ( + await asyncio.wait_for(process.read_line(), timeout=2.0) == "echo:again" + ) + finally: + await process.stop() + + +class TestTerminateEscalation: + @pytest.mark.parametrize("leader_exits_first", [False, True]) + async def test_stop_kills_descendants_after_leader_exits(self, leader_exits_first): + process = make_process("spawn-descendant", terminate_grace_s=0.1) + await process.start(ready_check=ready) + proc = process._proc + readers = list(process._reader_threads) + stop_task = None + try: + descendant_pid = int(await process.read_line(timeout_s=5.0)) + if leader_exits_first: + await process.write_line("exit") + assert await asyncio.to_thread(proc.wait, timeout=5.0) == 1 + + stop_task = asyncio.create_task(process.stop()) + done, _ = await asyncio.wait([stop_task], timeout=2.0) + assert stop_task in done, "stop() blocked on a surviving descendant's pipes" + await stop_task + assert all(not reader.is_alive() for reader in readers) + assert all( + stream.closed for stream in (proc.stdin, proc.stdout, proc.stderr) + ) + assert process._proc is None + + # This child keeps both output pipes open until it exits. EOF + # therefore proves teardown without depending on init reaping it. + assert descendant_pid != proc.pid + assert await process.read_line(timeout_s=1.0) is None + finally: + # A regression must fail promptly without leaking the sleeping child. + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + if stop_task is not None: + await asyncio.wait_for(stop_task, timeout=5.0) + await process.stop() + + async def test_sigterm_ignorer_is_killed_within_grace(self): + process = make_process("ignore-sigterm", terminate_grace_s=0.5) + await process.start(ready_check=ready) + + started = time.monotonic() + await process.stop() + elapsed = time.monotonic() - started + + assert process.is_running() is False + assert elapsed < 5.0 # grace (0.5s) + margin, well under the 60s sleep + + +class TestEncoding: + async def test_non_ascii_output_survives_the_reader(self): + # text=True alone decodes with the locale encoding (often ASCII in a + # container), and UnicodeDecodeError is a ValueError, which the reader + # thread used to swallow -- killing stdout pumping for the whole turn. + process = make_process("unicode-echo") + await process.start(ready_check=ready) + try: + await process.write_line("hello") + line = await process.read_line(timeout_s=10.0) + assert line == "echo:✓ hello 世界 \U0001f600" + + # The pump is still alive for subsequent turns. + await process.write_line("again") + assert await process.read_line(timeout_s=10.0) is not None + finally: + await process.stop() diff --git a/tests/core/test_harness_ws_route.py b/tests/core/test_harness_ws_route.py new file mode 100644 index 000000000..540d785c5 --- /dev/null +++ b/tests/core/test_harness_ws_route.py @@ -0,0 +1,461 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for the production /harness WebSocket route and mode wiring.""" + +from __future__ import annotations + +import asyncio +import json +from typing import AsyncIterator, Optional + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from openenv.core.env_server.http_server import create_fastapi_app, HTTPEnvServer +from openenv.core.env_server.types import ConcurrencyConfig, Observation, ServerMode +from openenv.core.harness import ( + AgenticHarnessAdapter, + HarnessAction, + HarnessConfig, + HarnessEnvironment, + HarnessEvent, + HarnessEventType, +) + + +class FakeAdapter(AgenticHarnessAdapter): + """Scripted adapter recording lifecycle calls; no real subprocess.""" + + created: list["FakeAdapter"] = [] + + def __init__(self): + super().__init__(HarnessConfig(name="fake-harness", command=["fake"])) + self.calls: list[str] = [] + self.alive = False + self.fail_on_send = False + self.hang_on_send = False + self.omit_turn_complete = False + FakeAdapter.created.append(self) + + async def start(self, working_directory: str) -> None: + self.calls.append("start") + self.alive = True + + async def stop(self) -> None: + self.calls.append("stop") + self.alive = False + + async def inject_tools(self, tools, bridge_url: Optional[str] = None) -> None: + self.calls.append("inject_tools") + + async def send_message_streaming(self, message: str) -> AsyncIterator[HarnessEvent]: + self.calls.append(f"send:{message}") + if self.hang_on_send: + await asyncio.sleep(3600) + if self.fail_on_send: + raise RuntimeError("adapter blew up") + if self.omit_turn_complete: + yield HarnessEvent( + type=HarnessEventType.TEXT_OUTPUT, data={"text": "no terminal event"} + ) + return + yield HarnessEvent(type=HarnessEventType.TOOL_CALL, data={"tool_name": "shell"}) + yield HarnessEvent( + type=HarnessEventType.TURN_COMPLETE, + data={"response": f"handled: {message}", "done": False}, + ) + + async def is_alive(self) -> bool: + return self.alive + + +def harness_env_factory() -> HarnessEnvironment: + # mcp=None: no domain tools, so no bridge server is started in tests + return HarnessEnvironment(adapter=FakeAdapter(), mcp=None) + + +def make_app( + mode: ServerMode, env_factory=harness_env_factory, max_concurrent_envs: int = 1 +) -> tuple[FastAPI, HTTPEnvServer]: + # HarnessEnvironment is SUPPORTS_CONCURRENT_SESSIONS=False (one env = one + # trajectory), so the server only permits max_concurrent_envs=1. + app = FastAPI() + server = HTTPEnvServer( + env_factory, + HarnessAction, + Observation, + max_concurrent_envs=max_concurrent_envs, + ) + server.register_routes(app, mode=mode) + return app, server + + +@pytest.fixture(autouse=True) +def clear_created_adapters(): + FakeAdapter.created = [] + yield + FakeAdapter.created = [] + + +class TestRouteRegistration: + def test_registered_in_production_with_harness_factory(self): + app, _ = make_app(ServerMode.PRODUCTION) + assert any(getattr(r, "path", None) == "/harness" for r in app.routes) + + def test_absent_in_simulation(self): + app, _ = make_app(ServerMode.SIMULATION) + assert not any(getattr(r, "path", None) == "/harness" for r in app.routes) + + def test_absent_for_non_harness_env(self): + from openenv.core.env_server.interfaces import Environment + from openenv.core.env_server.types import State + + class PlainEnv(Environment): + def reset(self, **kwargs): + return Observation() + + def step(self, action, timeout_s=None, **kwargs): + return Observation() + + @property + def state(self): + return State() + + app, _ = make_app(ServerMode.PRODUCTION, env_factory=PlainEnv) + assert not any(getattr(r, "path", None) == "/harness" for r in app.routes) + + def test_production_still_excludes_orchestration_routes(self): + app, _ = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + assert client.post("/reset", json={}).status_code in (404, 405) + assert client.post("/step", json={"action": {}}).status_code in (404, 405) + assert client.get("/state").status_code in (404, 405) + assert client.get("/health").status_code == 200 + + def test_detection_probe_is_side_effect_free(self): + make_app(ServerMode.PRODUCTION) + # The factory-detection probe instantiates one env; it must not + # start the harness. + for adapter in FakeAdapter.created: + assert "start" not in adapter.calls + + +class TestHarnessWebSocket: + @pytest.mark.parametrize("phase", ["factory", "startup", "turn"]) + def test_internal_errors_do_not_expose_credentials(self, phase): + secret = "FAKE_HARNESS_SECRET_FOR_TEST" + failing = False + + class SensitiveAdapter(FakeAdapter): + async def start(self, working_directory): + if failing and phase == "startup": + raise RuntimeError(f"Authorization: Bearer {secret}") + await super().start(working_directory) + + async def send_message_streaming(self, message): + if failing and phase == "turn": + raise RuntimeError(f"Authorization: Bearer {secret}") + async for event in super().send_message_streaming(message): + yield event + + def factory(): + if failing and phase == "factory": + raise RuntimeError(f"Authorization: Bearer {secret}") + return HarnessEnvironment(adapter=SensitiveAdapter(), mcp=None) + + app, server = make_app(ServerMode.PRODUCTION, env_factory=factory) + failing = True + with TestClient(app) as client: + with client.websocket_connect("/harness") as websocket: + if phase == "turn": + assert websocket.receive_json()["type"] == "session_started" + websocket.send_json({"type": "message", "content": "hello"}) + error = websocket.receive_json() + assert error["type"] == "error" + assert secret not in json.dumps(error) + if phase == "turn": + assert error["data"]["recoverable"] is False + else: + assert error["data"]["code"] == ( + "FACTORY_ERROR" if phase == "factory" else "SESSION_ERROR" + ) + + assert server.active_sessions == 0 + + def test_session_started_then_streamed_turns(self): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + started = json.loads(websocket.receive_text()) + assert started["type"] == "session_started" + assert started["data"]["harness"] == "fake-harness" + assert started["data"]["session_id"] + + websocket.send_text(json.dumps({"type": "message", "content": "hi"})) + first = json.loads(websocket.receive_text()) + assert first["type"] == "tool_call" + second = json.loads(websocket.receive_text()) + assert second["type"] == "turn_complete" + assert second["data"]["response"] == "handled: hi" + + def test_two_turns_reuse_one_adapter(self): + app, _ = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_text() # session_started + for content in ("one", "two"): + websocket.send_text(json.dumps({"type": "message", "content": content})) + websocket.receive_text() # tool_call + websocket.receive_text() # turn_complete + + # One adapter served the whole connection (probe adapters saw no start) + session_adapters = [a for a in FakeAdapter.created if "start" in a.calls] + assert len(session_adapters) == 1 + adapter = session_adapters[0] + assert adapter.calls.count("start") == 1 + assert [c for c in adapter.calls if c.startswith("send:")] == [ + "send:one", + "send:two", + ] + + @pytest.mark.parametrize( + ("frame", "code"), + [ + ("this is not json", "INVALID_JSON"), + ('{"type": "bogus"}', "VALIDATION_ERROR"), + ('{"type": "message"}', "VALIDATION_ERROR"), + ("null", "VALIDATION_ERROR"), + ("[]", "VALIDATION_ERROR"), + ], + ) + def test_malformed_frame_keeps_connection_usable(self, frame, code): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_json() # session_started + adapter = [a for a in FakeAdapter.created if a.alive][0] + websocket.send_text(frame) + error = websocket.receive_json() + assert error["type"] == "protocol_error" + assert error["type"] not in {event.value for event in HarnessEventType} + assert error["data"]["code"] == code + assert error["data"]["message"] + assert not any(call.startswith("send:") for call in adapter.calls) + assert server.active_sessions == 1 + + websocket.send_json({"type": "message", "content": "ok"}) + assert websocket.receive_json()["type"] == "tool_call" + completed = websocket.receive_json() + assert completed["type"] == "turn_complete" + assert completed["data"]["response"] == "handled: ok" + + assert server.active_sessions == 0 + + def test_adapter_crash_streams_error_event_then_closes(self): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_text() # session_started + adapter = [a for a in FakeAdapter.created if a.alive][0] + adapter.fail_on_send = True + + websocket.send_text(json.dumps({"type": "message", "content": "boom"})) + error_event = json.loads(websocket.receive_text()) + assert error_event["type"] == "error" + assert error_event["data"]["recoverable"] is False + assert error_event["data"]["message"] == "harness turn failed" + + assert server.active_sessions == 0 + + def test_disconnect_stops_adapter_and_frees_session(self): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_text() # session_started + session_adapter = [a for a in FakeAdapter.created if a.alive][0] + assert server.active_sessions == 1 + + assert server.active_sessions == 0 + assert "stop" in session_adapter.calls + assert session_adapter.alive is False + + def test_silent_turn_survives_idle_session_reaper(self): + class SilentAdapter(FakeAdapter): + async def send_message_streaming(self, message): + # The real reaper has a five-second minimum polling interval. + # Do not emit events while it checks this otherwise idle session. + await asyncio.sleep(5.2) + yield HarnessEvent( + type=HarnessEventType.TURN_COMPLETE, + data={"response": message, "alive": self.alive}, + ) + + app = FastAPI() + server = HTTPEnvServer( + lambda: HarnessEnvironment(adapter=SilentAdapter(), mcp=None), + HarnessAction, + Observation, + concurrency_config=ConcurrencyConfig(session_timeout=0.05), + ) + server.register_routes(app, mode=ServerMode.PRODUCTION) + + with TestClient(app) as client: + with client.websocket_connect("/harness") as websocket: + websocket.receive_json() # session_started + websocket.send_json({"type": "message", "content": "silent turn"}) + event = websocket.receive_json() + assert event["type"] == "turn_complete" + assert event["data"]["alive"] is True + assert server.active_sessions == 1 + + assert server.active_sessions == 0 + assert not server._session_websocket_attachments + + def test_http_close_defers_harness_cleanup_until_disconnect(self): + app, server = make_app(ServerMode.PRODUCTION) + with TestClient(app) as client: + with client.websocket_connect("/harness") as websocket: + started = websocket.receive_json() + session_id = started["data"]["session_id"] + adapter = [a for a in FakeAdapter.created if a.alive][0] + + response = client.post( + "/mcp", + json={ + "jsonrpc": "2.0", + "method": "openenv/session/close", + "params": {"session_id": session_id}, + "id": 1, + }, + ) + assert response.json()["result"] == { + "session_id": session_id, + "closed": False, + "closing": True, + } + assert adapter.alive is True + assert server.active_sessions == 1 + + websocket.send_json({"type": "message", "content": "finish"}) + assert websocket.receive_json()["type"] == "tool_call" + assert websocket.receive_json()["type"] == "turn_complete" + + assert server.active_sessions == 0 + assert adapter.alive is False + assert not server._session_websocket_attachments + assert not server._session_pending_closes + + def test_capacity_limit_rejects_second_connection(self): + app, _ = make_app(ServerMode.PRODUCTION, max_concurrent_envs=1) + client = TestClient(app) + + with client.websocket_connect("/harness") as first: + first.receive_text() # session_started + with client.websocket_connect("/harness") as second: + error = json.loads(second.receive_text()) + assert error["type"] == "error" + assert error["data"]["code"] == "CAPACITY_REACHED" + + +class TestModeWiring: + def test_create_fastapi_app_mode_production(self): + app = create_fastapi_app( + harness_env_factory, HarnessAction, Observation, mode="production" + ) + client = TestClient(app) + assert client.post("/reset", json={}).status_code in (404, 405) + assert any(getattr(r, "path", None) == "/harness" for r in app.routes) + + def test_openenv_mode_env_var_honored(self, monkeypatch): + monkeypatch.setenv("OPENENV_MODE", "production") + app = create_fastapi_app(harness_env_factory, HarnessAction, Observation) + client = TestClient(app) + assert client.post("/reset", json={}).status_code in (404, 405) + + def test_default_stays_simulation(self, monkeypatch): + monkeypatch.delenv("OPENENV_MODE", raising=False) + app = create_fastapi_app(harness_env_factory, HarnessAction, Observation) + assert not any(getattr(r, "path", None) == "/harness" for r in app.routes) + route_paths = {getattr(r, "path", None) for r in app.routes} + assert "/reset" in route_paths + + def test_explicit_mode_overrides_env_var(self, monkeypatch): + monkeypatch.setenv("OPENENV_MODE", "production") + app = create_fastapi_app( + harness_env_factory, HarnessAction, Observation, mode="simulation" + ) + route_paths = {getattr(r, "path", None) for r in app.routes} + assert "/reset" in route_paths + + +class TestTurnBoundaries: + """A production turn must end: on its own, on timeout, or with an error.""" + + @pytest.mark.parametrize("completed_turns", [0, 1]) + def test_dead_harness_is_reported_before_sending(self, completed_turns): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_json() # session_started + adapter = [a for a in FakeAdapter.created if a.alive][0] + for _ in range(completed_turns): + websocket.send_json({"type": "message", "content": "first"}) + assert websocket.receive_json()["type"] == "tool_call" + assert websocket.receive_json()["type"] == "turn_complete" + + adapter.alive = False + adapter.hang_on_send = True + adapter.config.session_timeout_s = 0.2 + + websocket.send_json({"type": "message", "content": "after crash"}) + event = websocket.receive_json() + assert event["type"] == "error" + assert event["data"]["recoverable"] is False + assert event["data"]["message"] == "harness process is not running" + assert "send:after crash" not in adapter.calls + + assert server.active_sessions == 0 + assert "stop" in adapter.calls + assert not server._session_websocket_attachments + + def test_hung_turn_is_bounded_by_session_timeout(self): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_text() # session_started + adapter = [a for a in FakeAdapter.created if a.alive][0] + adapter.hang_on_send = True + adapter.config.session_timeout_s = 0.2 + + websocket.send_text(json.dumps({"type": "message", "content": "hang"})) + event = json.loads(websocket.receive_text()) + assert event["type"] == "error" + assert "exceeded 0.2 seconds" in event["data"]["message"] + + # The session is released rather than pinned at capacity forever. + assert server.active_sessions == 0 + + def test_stream_without_turn_complete_is_reported(self): + app, server = make_app(ServerMode.PRODUCTION) + client = TestClient(app) + + with client.websocket_connect("/harness") as websocket: + websocket.receive_text() # session_started + adapter = [a for a in FakeAdapter.created if a.alive][0] + adapter.omit_turn_complete = True + + websocket.send_text(json.dumps({"type": "message", "content": "hi"})) + assert json.loads(websocket.receive_text())["type"] == "text_output" + # A client blocking on the terminal event would otherwise hang. + event = json.loads(websocket.receive_text()) + assert event["type"] == "error" + assert "TURN_COMPLETE" in event["data"]["message"] + + assert server.active_sessions == 0