Debdeep123/capability_forge_codeops

1from typing import Optional, List, Dict, Any2from pydantic import Field3from openenv.core.env_server.types import Action as BaseAction, Observation as BaseObservation, State as BaseState4 5class Action(BaseAction):6    """The action taken by the model."""7    reasoning: str = Field(description="Step-by-step reasoning trace explaining the bug and the fix.")8    corrected_code: str = Field(description="The complete, corrected Python code. Do not include markdown formatting like ```python, just the raw code.")9 10class Observation(BaseObservation):11    """What the agent sees."""12    task_prompt: str = Field(default="", description="The instructions for the current debugging task.")13    buggy_code: str = Field(default="", description="The Python code containing the bug.")14    execution_feedback: Optional[str] = Field(default=None, description="Feedback or error trace from running the previous attempt's code.")15    hint: Optional[str] = Field(default=None, description="A hint provided after multiple failed attempts.")16    formatting_instructions: str = Field(17        default="Please provide your response using the Action schema: 'reasoning' for your thought process and 'corrected_code' for the raw, executable Python code."18    )19 20class State(BaseState):21    """Internal environment state."""22    episode_id: str23    difficulty_level: int = 124    current_task: Dict[str, Any] = Field(default_factory=dict)25    attempt_count: int = 026    max_attempts: int = 327    rolling_successes: List[int] = Field(default_factory=list) # 1 for success, 0 for failure28    29    @property30    def rolling_success_rate(self) -> float:31        if not self.rolling_successes:32            return 0.5 # Default to middle if no history33        return sum(self.rolling_successes[-10:]) / min(len(self.rolling_successes), 10)34