| """Typed Pydantic models for API and environment state.""" |
|
|
| from __future__ import annotations |
|
|
| from enum import Enum |
| from typing import Any, Dict, List, Optional, Union |
|
|
| from pydantic import BaseModel, Field |
|
|
|
|
| class TaskDifficulty(str, Enum): |
| EASY = "easy" |
| MEDIUM = "medium" |
| HARD = "hard" |
|
|
|
|
| class ActionType(str, Enum): |
| EDIT_FILE = "edit_file" |
| ADD_LINE = "add_line" |
| DELETE_LINE = "delete_line" |
| REPLACE_LINE = "replace_line" |
| ADD_BLOCK = "add_block" |
| DELETE_BLOCK = "delete_block" |
| SUBMIT = "submit" |
| REQUEST_HINT = "request_hint" |
|
|
|
|
| class FileType(str, Enum): |
| DOCKERFILE = "dockerfile" |
| WORKFLOW = "workflow" |
| DOCKER_COMPOSE = "docker_compose" |
| REQUIREMENTS = "requirements" |
| OTHER = "other" |
|
|
|
|
| class ErrorPhase(str, Enum): |
| WORKFLOW_PARSE = "workflow_parse" |
| DOCKER_BUILD = "docker_build" |
| DOCKER_RUN = "docker_run" |
| TEST = "test" |
| PUSH = "push" |
| DEPLOY = "deploy" |
|
|
|
|
| class FileContent(BaseModel): |
| path: str = Field(..., description="File path") |
| content: str = Field(..., description="Current file content") |
| file_type: FileType = Field(..., description="Type of file") |
| line_count: int = Field(..., description="Number of lines") |
|
|
|
|
| class ErrorInfo(BaseModel): |
| phase: ErrorPhase = Field(..., description="Failed phase") |
| error_message: str = Field(..., description="Error output") |
| exit_code: Optional[int] = Field(None, description="Exit code") |
| failed_step: Optional[str] = Field(None, description="Failed step") |
| line_hint: Optional[int] = Field(None, description="Suggested line") |
|
|
|
|
| class Observation(BaseModel): |
| task_id: str |
| task_description: str |
| difficulty: TaskDifficulty |
| files: List[FileContent] |
| error: ErrorInfo |
| available_secrets: List[str] = Field(default_factory=list) |
| runner_os: str = "ubuntu-latest" |
| step_number: int |
| max_steps: int |
| hints_used: int = 0 |
| hints_available: int = 3 |
| last_action_success: Optional[bool] = None |
| last_action_feedback: Optional[str] = None |
| issues_found: int = 0 |
| issues_fixed: int = 0 |
| total_issues: int |
|
|
|
|
| class FileEdit(BaseModel): |
| file_path: str |
| line_number: Optional[int] = None |
| old_content: Optional[str] = None |
| new_content: Optional[str] = None |
|
|
|
|
| class Action(BaseModel): |
| action_type: ActionType |
| edits: Optional[List[FileEdit]] = None |
| reasoning: Optional[str] = None |
|
|
|
|
| class StepResult(BaseModel): |
| observation: Observation |
| reward: float = Field(..., ge=-1.0, le=2.0) |
| done: bool |
| info: Dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class TaskInfo(BaseModel): |
| id: str |
| name: str |
| description: str |
| difficulty: TaskDifficulty |
| num_scenarios: int |
|
|
|
|
| class EnvironmentInfo(BaseModel): |
| name: str = "cicd-debug-env" |
| version: str = "1.0.0" |
| description: str = "Debug CI/CD infrastructure issues" |
| tasks: List[TaskInfo] |
| max_steps: int = 10 |
| action_space: Dict[str, Any] |
| observation_space: Dict[str, Any] |
|
|
|
|
| class GraderResult(BaseModel): |
| task_id: str |
| score: float = Field(..., ge=0.0, le=1.0) |
| max_score: float = 1.0 |
| breakdown: Dict[str, float] = Field(default_factory=dict) |
| feedback: str = "" |
| steps_taken: int |
| hints_used: int = 0 |
|
|
|
|
| class ResetRequest(BaseModel): |
| task_id: Optional[Union[str, int]] = None |
| scenario_id: Optional[str] = None |
| seed: Optional[int] = None |
|
|
|
|
| class ResetResponse(BaseModel): |
| observation: Observation |
| info: Dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class StepRequest(BaseModel): |
| action: Action |
|
|
|
|
| class StepResponse(BaseModel): |
| observation: Observation |
| reward: float |
| done: bool |
| info: Dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| class StateResponse(BaseModel): |
| observation: Observation |
| episode_reward: float |
| steps_taken: int |
| done: bool |
|
|
|
|
| class GraderRequest(BaseModel): |
| task_id: str |
| trajectory: List[Dict[str, Any]] |
|
|
|
|
| class GraderResponse(BaseModel): |
| result: GraderResult |
|
|
|
|
| class BaselineRequest(BaseModel): |
| task_id: Optional[str] = None |
| num_episodes: int = 1 |
|
|
|
|
| class BaselineResponse(BaseModel): |
| results: List[GraderResult] |
| aggregate_score: float |
|
|
