Spaces:
Sleeping
Sleeping
File size: 6,865 Bytes
30bf68a 2177064 30bf68a 2177064 30bf68a | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 | from __future__ import annotations
from fastapi.responses import HTMLResponse
from dataclasses import dataclass
import os
from typing import Any
from fastapi import FastAPI
from fastapi import HTTPException
from pydantic import BaseModel, Field
import uvicorn
from env.environment import CICDDebuggerEnvironment, MAX_STEPS
from env.models import Action, Observation, Reward
app = FastAPI(title="CI/CD Debugger OpenEnv Server")
class ResetRequest(BaseModel):
task_id: str | None = None
difficulty: str | None = None
max_steps: int = Field(default=MAX_STEPS, ge=1, le=100)
class StepRequest(BaseModel):
action: Action | str | dict[str, Any]
class StepResponse(BaseModel):
task_id: str
step_count: int
reward: float
reward_model: Reward
done: bool
observation: Observation
last_action: str | None = None
info: dict[str, Any] = Field(default_factory=dict)
class StateResponse(BaseModel):
initialized: bool
task_id: str | None = None
step_count: int = 0
done: bool = False
last_action: str | None = None
observation: Observation | None = None
internal_state: dict[str, Any] = Field(default_factory=dict)
@dataclass
class RuntimeSession:
env: CICDDebuggerEnvironment
task_id: str
step_count: int = 0
done: bool = False
last_action: str | None = None
last_reward: float = 0.0
last_observation: dict[str, Any] | None = None
last_info: dict[str, Any] | None = None
runtime_session: RuntimeSession | None = None
def _as_observation_model(observation: dict[str, Any] | Observation) -> Observation:
if isinstance(observation, Observation):
return observation
return Observation.model_validate(observation)
def _build_step_response(session: RuntimeSession) -> StepResponse:
observation = session.last_observation or {}
info_payload = session.last_info or {}
reward_payload = info_payload.get("reward_model")
if isinstance(reward_payload, dict):
reward_model = Reward.model_validate(reward_payload)
else:
reward_model = Reward(value=float(session.last_reward), components={"total": float(session.last_reward)})
return StepResponse(
task_id=session.task_id,
step_count=int(observation.get("step_count") or session.step_count),
reward=float(session.last_reward),
reward_model=reward_model,
done=bool(session.done),
observation=_as_observation_model(observation),
last_action=session.last_action,
info=info_payload,
)
@app.get("/", response_class=HTMLResponse)
async def home():
return """
<html>
<head>
<title>CI/CD Debugger</title>
</head>
<body style="font-family: Arial; padding: 40px; background:#111; color:white;">
<h1>CI/CD Debugger Environment</h1>
<button onclick="reset()">Reset</button>
<br><br>
<select id="action">
<option>READ_LOGS</option>
<option>READ_FILE</option>
<option>RUN_STAGE</option>
<option>VALIDATE</option>
</select>
<button onclick="step()">Step</button>
<pre id="output" style="background:#222; padding:20px;"></pre>
<script>
async function reset(){
const res = await fetch('/reset', {method:'POST'});
const data = await res.json();
document.getElementById('output').textContent =
JSON.stringify(data, null, 2);
}
async function step(){
const action = document.getElementById('action').value;
const res = await fetch('/step', {
method:'POST',
headers:{'Content-Type':'application/json'},
body: JSON.stringify({
action_type: action,
payload: {}
})
});
const data = await res.json();
document.getElementById('output').textContent =
JSON.stringify(data, null, 2);
}
</script>
</body>
</html>
"""
@app.get("/health")
def health() -> dict[str, str]:
return {"status": "ok"}
@app.post("/reset", response_model=StepResponse)
async def reset(payload: ResetRequest | None = None) -> StepResponse:
global runtime_session
request = payload or ResetRequest()
env = CICDDebuggerEnvironment(max_steps=int(request.max_steps))
observation = await env.reset(task_id=request.task_id, difficulty=request.difficulty)
runtime_session = RuntimeSession(
env=env,
task_id=str(observation.get("task_id", request.task_id or "cicd-debugger-task")),
step_count=0,
done=False,
last_action=None,
last_reward=0.0,
last_observation=observation,
last_info={
"message": "environment reset",
"tool": "reset",
"error": None,
"reward_model": Reward(value=0.0, components={"total": 0.0}).model_dump(),
},
)
return _build_step_response(runtime_session)
@app.post("/step", response_model=StepResponse)
async def step(payload: StepRequest) -> StepResponse:
global runtime_session
if runtime_session is None:
raise HTTPException(status_code=400, detail="Environment not initialized. Call /reset first.")
if runtime_session.done:
return _build_step_response(runtime_session)
observation, reward, done, info = await runtime_session.env.step(payload.action)
runtime_session.step_count = int(observation.get("step_count", runtime_session.step_count + 1))
runtime_session.done = bool(done)
runtime_session.last_action = payload.action if isinstance(payload.action, str) else str(payload.action)
runtime_session.last_reward = float(reward)
runtime_session.last_observation = observation
runtime_session.last_info = dict(info or {})
return _build_step_response(runtime_session)
@app.get("/state", response_model=StateResponse)
async def state() -> StateResponse:
if runtime_session is None:
return StateResponse(initialized=False)
observation = None
if runtime_session.last_observation is not None:
observation = _as_observation_model(runtime_session.last_observation)
return StateResponse(
initialized=True,
task_id=runtime_session.task_id,
step_count=runtime_session.step_count,
done=runtime_session.done,
last_action=runtime_session.last_action,
observation=observation,
internal_state=runtime_session.env.state(),
)
@app.post("/state", response_model=StateResponse)
async def state_post() -> StateResponse:
return await state()
def main() -> None:
port = int(os.getenv("PORT", "7860"))
uvicorn.run(app, host="0.0.0.0", port=port)
if __name__ == "__main__":
main()
|