Spaces:
Sleeping
Sleeping
File size: 3,949 Bytes
45c94ac f974243 45c94ac 7a06d56 45c94ac 7a06d56 f974243 45c94ac 7a06d56 45c94ac f974243 45c94ac 7a06d56 f974243 45c94ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
FastAPI application for the CI/CD Doctor environment.
Wraps PipelineEnvironment in an OpenEnv-compatible interface so it can be
served via openenv's create_app() infrastructure.
"""
try:
from openenv.core.env_server.http_server import create_app
from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import Action, Observation, State, EnvironmentMetadata
except Exception as e: # pragma: no cover
raise ImportError(
"openenv is required for the web interface. Install dependencies with '\n uv sync\n'"
) from e
from uuid import uuid4
import random
from pydantic import Field
from models import PipelineObservation
from .environment import PipelineEnvironment
from fastapi.responses import PlainTextResponse
from pathlib import Path
import os
class CiCdDoctorAction(Action):
command: str = Field(..., description="Shell-like command the agent issues")
class CiCdDoctorObservation(Observation):
stdout: str = Field(default="", description="Command output / logs seen by agent")
exit_code: int = Field(default=0, description="0 = success, 1 = error")
pipeline_status: str = Field(default="not_run", description="not_run | running | passed | failed")
steps_remaining: int = Field(default=15, description="Steps left in episode")
def _to_obs(obs: PipelineObservation) -> CiCdDoctorObservation:
return CiCdDoctorObservation(
stdout=obs.stdout,
exit_code=obs.exit_code,
pipeline_status=obs.pipeline_status,
steps_remaining=obs.steps_remaining,
done=obs.done,
reward=obs.reward,
)
class CiCdDoctorEnvironment(Environment):
"""OpenEnv adapter that wraps PipelineEnvironment."""
SUPPORTS_CONCURRENT_SESSIONS: bool = True
def __init__(self):
self._env = PipelineEnvironment()
self._state_obj = State(episode_id=str(uuid4()), step_count=0)
def reset(self, task: str = "default", seed: int = 42) -> CiCdDoctorObservation:
if task == "default":
task = random.choice(["easy", "medium", "hard"])
obs = self._env.reset(task=task, seed=seed)
s = self._env.state()
self._state_obj = State(episode_id=s.episode_id, step_count=s.step_count)
return _to_obs(obs)
def step(self, action: CiCdDoctorAction) -> CiCdDoctorObservation: # type: ignore[override]
from models import PipelineAction
obs = self._env.step(PipelineAction(command=action.command))
s = self._env.state()
self._state_obj = State(episode_id=s.episode_id, step_count=s.step_count)
return _to_obs(obs)
@property
def state(self) -> State:
return self._state_obj
def get_metadata(self) -> EnvironmentMetadata:
return EnvironmentMetadata(
name="CI_CD_Doctor",
description="An interactive environment where agents diagnose and fix broken CI/CD pipelines.",
version="1.0.0",
readme_content=_load_readme() or None,
)
_README_DIR = Path(__file__).resolve().parent.parent / "docs"
_README_PATH = _README_DIR / "README.md"
if _README_PATH.exists() and not os.environ.get("ENV_README_PATH"):
os.environ["ENV_README_PATH"] = str(_README_PATH)
def _load_readme() -> str:
return _README_PATH.read_text() if _README_PATH.exists() else ""
app = create_app(
CiCdDoctorEnvironment,
CiCdDoctorAction,
CiCdDoctorObservation,
env_name="CI_CD_Doctor",
max_concurrent_envs=1,
)
@app.get("/instructions", response_class=PlainTextResponse)
def instructions():
return _load_readme()
def main():
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=args.port)
|