SidhaGarg/Cloud-DevOps-RLEnv

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"""Async entrypoint wrapper for external evaluators and custom graders."""8 9from __future__ import annotations10 11from typing import Any, Dict12 13from pydantic import BaseModel14 15try:16    from .models import CloudAction, CloudObservation, CloudState17    from .server.cloud_devops_env_environment import CloudDevopsEnvironment18except ImportError:19    from models import CloudAction, CloudObservation, CloudState20    from server.cloud_devops_env_environment import CloudDevopsEnvironment21 22 23class EnvResult(BaseModel):24    """Canonical environment result payload for async evaluator loops."""25 26    observation: CloudObservation27    reward: float28    done: bool29    info: Dict[str, Any]30 31 32class CloudDevOpsEnv:33    """Async-compatible facade over the OpenEnv server-side environment logic."""34 35    def __init__(self, task_name: str = "easy"):36        self._impl = CloudDevopsEnvironment(task_name=task_name)37 38    @property39    def achievements(self) -> set[str]:40        """Expose completed shaped-reward checkpoints for debugging/evaluation."""41        return set(self._impl._achievements)42 43    async def reset(self) -> EnvResult:44        """Reset the environment to the initial task state."""45        obs = self._impl.reset()46        return EnvResult(47            observation=obs,48            reward=float(obs.reward or 0.0),49            done=bool(obs.done),50            info=dict(obs.metadata or {}),51        )52 53    async def step(self, action: CloudAction) -> EnvResult:54        """Execute an action and return a structured async result."""55        obs = self._impl.step(action)56        return EnvResult(57            observation=obs,58            reward=float(obs.reward or 0.0),59            done=bool(obs.done),60            info=dict(obs.metadata or {}),61        )62 63    async def state(self) -> CloudState:64        """Return hidden environment state for deterministic evaluators."""65        state = self._impl.state66        assert isinstance(state, CloudState)67        return state68 69    async def close(self) -> None:70        """Async close hook for evaluator compatibility."""71        return None72