Spaces:
Running
Running
| from __future__ import annotations | |
| import asyncio | |
| import os | |
| import uuid | |
| from pathlib import Path | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.responses import FileResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from .demo_fixture import DEMO_REQUEST | |
| from .models import RunRecord, RunRequest, RunStatus | |
| from .orchestrator import Orchestrator | |
| ROOT = Path(__file__).resolve().parent | |
| STATIC = ROOT / "static" | |
| app = FastAPI( | |
| title="BranchForge", | |
| description="Evidence-first parallel coding agent powered by NVIDIA Nemotron and Nebius.", | |
| version="1.0.0", | |
| ) | |
| app.mount("/static", StaticFiles(directory=STATIC), name="static") | |
| runs: dict[str, RunRecord] = {} | |
| orchestrator = Orchestrator() | |
| async def health() -> dict[str, object]: | |
| return { | |
| "ok": True, | |
| "model": orchestrator.gateway.model, | |
| "judge_model": orchestrator.gateway.judge_model, | |
| "provider": orchestrator.gateway.provider, | |
| "model_configured": orchestrator.gateway.enabled, | |
| "sandbox_configured": orchestrator.sandboxes.can_use_nebius, | |
| } | |
| async def sample() -> RunRequest: | |
| return DEMO_REQUEST | |
| async def create_run(request: RunRequest) -> RunRecord: | |
| run_id = uuid.uuid4().hex[:12] | |
| model_chain = orchestrator.gateway.model | |
| if orchestrator.gateway.judge_model != orchestrator.gateway.model: | |
| model_chain = f"{model_chain} → {orchestrator.gateway.judge_model}" | |
| record = RunRecord( | |
| id=run_id, | |
| status=RunStatus.queued, | |
| request=request, | |
| model=model_chain, | |
| provider=orchestrator.gateway.provider, | |
| ) | |
| runs[run_id] = record | |
| asyncio.create_task(orchestrator.run(record)) | |
| return record | |
| async def get_run(run_id: str) -> RunRecord: | |
| record = runs.get(run_id) | |
| if not record: | |
| raise HTTPException(status_code=404, detail="Run not found") | |
| return record | |
| async def index() -> FileResponse: | |
| return FileResponse(STATIC / "index.html") | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run( | |
| "app.main:app", | |
| host="0.0.0.0", | |
| port=int(os.getenv("PORT", "7860")), | |
| reload=False, | |
| ) | |