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"""8Python Code Action Environment.9 10This module provides a server-side environment implementation for executing11Python code actions using PyExecutor.12"""13 14import uuid15 16from openenv.core.env_server.interfaces import Action, Environment, Observation17 18from ..models import CodeAction, CodeObservation, CodeState19from .python_executor import PyExecutor20from .transforms import create_safe_coding_transform21 22 23class PythonCodeActEnv(Environment):24    """25    Python Code Action Environment for executing code and tracking state.26 27    This environment executes Python code submitted as CodeAction during step,28    maintains the last exit code in its state, and returns results wrapped29    in CodeObservation.30 31    Args:32        transform: Optional transform to apply to observations33        additional_imports: List of additional module imports to authorize34                          (e.g., ["numpy", "pandas", "matplotlib"])35 36    Example:37        >>> env = PythonCodeActEnv()38        >>> obs = env.reset()39        >>> action = CodeAction(code="print('Hello, World!')")40        >>> obs = env.step(action)41        >>> print(obs.stdout)  # "Hello, World!\n"42        >>> print(obs.exit_code)  # 043        >>> print(env.state.last_exit_code)  # 044    """45 46    def __init__(47        self,48    ):49        self.transform = create_safe_coding_transform()50        self._executor = PyExecutor()51        self._state = CodeState()52 53    def reset(self) -> Observation:54        """55        Reset environment and start fresh execution session.56 57        Returns:58            Initial observation with empty stdout/stderr and exit_code=059        """60        # Initialize fresh state61        self._state = CodeState(episode_id=str(uuid.uuid4()), step_count=0)62        # Add last_exit_code to state63        self._state.last_exit_code = 064 65        # Reset executor to clear any previously defined variables/functions66        self._executor = PyExecutor()67 68        # Reset transform to clear any accumulated state69        self.transform = create_safe_coding_transform()70 71        # Return initial observation72        observation = CodeObservation(73            stdout="",74            stderr="",75            exit_code=0,76        )77 78        return self._apply_transform(observation)79 80    def step(self, action: Action) -> Observation:81        """82        Execute code action and return observation.83 84        Args:85            action: CodeAction containing the code to execute86 87        Returns:88            CodeObservation with execution results (stdout, stderr, exit_code)89 90        Raises:91            ValueError: If action is not a CodeAction instance92        """93        if not isinstance(action, CodeAction):94            raise ValueError(f"Expected CodeAction, got {type(action)}")95 96        # Execute the code using PyExecutor97        result = self._executor.run(action.code)98 99        # Update state100        self._state.step_count += 1101        self._state.last_exit_code = result.exit_code102 103        # Create observation from execution result104        # Include code in metadata for transform reward calculation105        observation = CodeObservation(106            stdout=result.stdout,107            stderr=result.stderr,108            exit_code=result.exit_code,109            metadata={"last_code": action.code},110        )111 112        return self._apply_transform(observation)113 114    @property115    def state(self) -> CodeState:116        """Get current environment state including last exit code."""117        return self._state118