[RFC 005] 4/4: production /harness WebSocket route and mode wiring

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

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

src/openenv/core/harness/__init__.py

@@ -1,16 +1,17 @@
 # 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:
 
 1. **Trainer-side rollout API** (``openenv.core.harness.rollout``): a harness
    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,14 +24,17 @@
     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,
     HarnessEvent,
     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",
 ]