server/app.py

"""
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)