DEVessi/devops_sandbox

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"""Self-Healing DevOps Sandbox Environment Client."""8 9from typing import Dict10 11from openenv.core import EnvClient12from openenv.core.client_types import StepResult13from openenv.core.env_server.types import State14 15from models import BashAction, TerminalObservation16 17 18class DevopsSandboxEnv(19    EnvClient[BashAction, TerminalObservation, State]20):21    """22    Client for the Self-Healing DevOps Sandbox Environment.23 24    Example:25        >>> with DevopsSandboxEnv(base_url="http://localhost:8000") as client:26        ...     result = client.reset()27        ...     print(result.observation.stdout)28        ...29        ...     result = client.step(BashAction(command="ls -la"))30        ...     print(result.observation.stdout)31    """32 33    def _step_payload(self, action: BashAction) -> Dict:34        """Convert BashAction to JSON payload for step message."""35        return {36            "command": action.command,37        }38 39    def _parse_result(self, payload: Dict) -> StepResult[TerminalObservation]:40        """Parse server response into StepResult[TerminalObservation]."""41        obs_data = payload.get("observation", {})42        observation = TerminalObservation(43            stdout=obs_data.get("stdout", ""),44            stderr=obs_data.get("stderr", ""),45            current_dir=obs_data.get("current_dir", "/app"),46            task_id=obs_data.get("task_id", "devops_sandbox"),47            grader_score=obs_data.get("grader_score", 0.0),48            grader_feedback=obs_data.get("grader_feedback", ""),49            done=payload.get("done", False),50            reward=payload.get("reward"),51            metadata=obs_data.get("metadata", {}),52        )53 54        return StepResult(55            observation=observation,56            reward=payload.get("reward"),57            done=payload.get("done", False),58        )59 60    def _parse_state(self, payload: Dict) -> State:61        """Parse server response into State object."""62        return State(63            episode_id=payload.get("episode_id"),64            step_count=payload.get("step_count", 0),65        )66