openenv/coding_env

1# Copyright (c) Meta Platforms, Inc. and affiliates.2# All rights reserved.3#4# This source code is licensed under the BSD-style license found in the5# LICENSE file in the root directory of this source tree.6 7"""8Generic environment client that works with raw dictionaries.9 10This module provides a GenericEnvClient that doesn't require installing11environment-specific packages. It's useful for connecting to remote servers12without running any untrusted code locally.13"""14 15from typing import Any, Dict16 17from .client_types import StepResult18from .env_client import EnvClient19 20 21class GenericEnvClient(EnvClient[Dict[str, Any], Dict[str, Any], Dict[str, Any]]):22    """23    Environment client that works with raw dictionaries instead of typed classes.24 25    This client doesn't require installing environment-specific packages, making it26    ideal for:27    - Connecting to remote servers without installing their packages28    - Quick prototyping and testing29    - Environments where type safety isn't needed30    - Security-conscious scenarios where you don't want to run remote code31 32    The trade-off is that you lose type safety and IDE autocomplete for actions33    and observations. Instead of typed objects, you work with plain dictionaries.34 35    Example:36        >>> # Direct connection to a running server (no installation needed)37        >>> with GenericEnvClient(base_url="http://localhost:8000") as env:38        ...     result = env.reset()39        ...     result = env.step({"code": "print('hello')"})40        ...     print(result.observation)  # Dict[str, Any]41        ...     print(result.observation.get("output"))42 43        >>> # From local Docker image44        >>> env = GenericEnvClient.from_docker_image("coding-env:latest")45        >>> result = env.reset()46        >>> result = env.step({"code": "x = 1 + 2"})47        >>> env.close()48 49        >>> # From HuggingFace Hub (pulls Docker image, no pip install)50        >>> env = GenericEnvClient.from_env("user/my-env", use_docker=True)51        >>> result = env.reset()52        >>> env.close()53 54    Note:55        GenericEnvClient inherits `from_docker_image()` and `from_env()` from56        EnvClient, so you can use it with Docker containers and HuggingFace57        Spaces without any package installation.58    """59 60    def _step_payload(self, action: Dict[str, Any]) -> Dict[str, Any]:61        """62        Convert action to payload for the server.63 64        For GenericEnvClient, this handles both raw dictionaries and65        typed Action objects (Pydantic models). If a Pydantic model is66        passed, it will be converted to a dictionary using model_dump().67 68        Args:69            action: Action as a dictionary or Pydantic BaseModel70 71        Returns:72            The action as a dictionary for the server73        """74        # If it's already a dict, return as-is75        if isinstance(action, dict):76            return action77 78        # If it's a Pydantic model (Action subclass), convert to dict79        if hasattr(action, "model_dump"):80            return action.model_dump()81 82        # Fallback for other objects with __dict__83        if hasattr(action, "__dict__"):84            return vars(action)85 86        # Last resort: try to convert to dict87        return dict(action)88 89    def _parse_result(self, payload: Dict[str, Any]) -> StepResult[Dict[str, Any]]:90        """91        Parse server response into a StepResult.92 93        Extracts the observation, reward, and done fields from the94        server response.95 96        Args:97            payload: Response payload from the server98 99        Returns:100            StepResult with observation as a dictionary101        """102        return StepResult(103            observation=payload.get("observation", {}),104            reward=payload.get("reward"),105            done=payload.get("done", False),106        )107 108    def _parse_state(self, payload: Dict[str, Any]) -> Dict[str, Any]:109        """110        Parse state response from the server.111 112        For GenericEnvClient, this returns the payload as-is since113        we're working with dictionaries.114 115        Args:116            payload: State payload from the server117 118        Returns:119            The state as a dictionary120        """121        return payload122 123 124class GenericAction(Dict[str, Any]):125    """126    A dictionary subclass for creating actions when using GenericEnvClient.127 128    This provides a semantic wrapper around dictionaries to make code more129    readable when working with GenericEnvClient. It behaves exactly like a130    dict but signals intent that this is an action for an environment.131 132    Example:133        >>> # Without GenericAction (works fine)134        >>> env.step({"code": "print('hello')"})135 136        >>> # With GenericAction (more explicit)137        >>> action = GenericAction(code="print('hello')")138        >>> env.step(action)139 140        >>> # With multiple fields141        >>> action = GenericAction(code="x = 1", timeout=30, metadata={"tag": "test"})142        >>> env.step(action)143 144    Note:145        GenericAction is just a dict with a constructor that accepts keyword146        arguments. It's provided for symmetry with typed Action classes and147        to make code more readable.148    """149 150    def __init__(self, **kwargs: Any) -> None:151        """152        Create a GenericAction from keyword arguments.153 154        Args:155            **kwargs: Action fields as keyword arguments156 157        Example:158            >>> action = GenericAction(code="print(1)", timeout=30)159            >>> action["code"]160            'print(1)'161        """162        super().__init__(kwargs)163 164    def __repr__(self) -> str:165        """Return a readable representation."""166        items = ", ".join(f"{k}={v!r}" for k, v in self.items())167        return f"GenericAction({items})"168