Spaces:

Nithin1026
/

synapse-x

Sleeping

File size: 9,150 Bytes

"""
SYNAPSE-X API server - FastAPI application for HF Spaces deployment.

Endpoints required by OpenEnv spec and hackathon pre-submission checklist:
  GET  /health   - liveness check
  POST /reset    - start new episode
  GET  /reset    - same, query-param form
  POST /step     - apply one action
  GET  /state    - inspect current env state
  GET  /tasks    - enumerate task presets
  GET  /validate - submission readiness check
  POST /grade    - score an action trace
"""

import json
import os
import sys
from pathlib import Path
from typing import Any, Optional
from urllib.parse import parse_qs

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, ValidationError

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from env.environment import SynapseXEnvironment
from env.grader import TASK_REGISTRY, TASK_SEEDS, grade as run_grade
from env.models import Action, GradeResult, Observation, StepResult

app = FastAPI(
    title="SYNAPSE-X",
    description="Predictive decision intelligence environment - OpenEnv Hackathon submission",
    version="1.0.0",
)

LOCAL_IMAGE_NAME = os.environ.get("LOCAL_IMAGE_NAME", "synapse-x")
APP_VERSION = "1.0.0"
APP_DESCRIPTION = "Predictive decision intelligence environment - OpenEnv Hackathon submission"
API_ENDPOINTS = [
    "/health",
    "/metadata",
    "/schema",
    "/mcp",
    "/reset",
    "/step",
    "/state",
    "/tasks",
    "/grade",
    "/validate",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/", response_class=HTMLResponse, include_in_schema=False)
def root():
    return HTMLResponse(
        content="""
<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <title>SYNAPSE-X</title>
  <meta http-equiv="refresh" content="0; url=/docs">
  <style>
    body { font-family: sans-serif; background: #0f172a; color: #e2e8f0;
           display: flex; align-items: center; justify-content: center;
           height: 100vh; margin: 0; }
    a { color: #818cf8; }
  </style>
</head>
<body>
  <p>Redirecting to <a href="/docs">SYNAPSE-X API docs</a>...</p>
</body>
</html>
"""
    )


_env: Optional[SynapseXEnvironment] = None


def get_env() -> SynapseXEnvironment:
    global _env
    if _env is None:
        _env = SynapseXEnvironment(task_config=TASK_REGISTRY["easy"], seed=42)
        _env.reset()
    return _env


def to_dict(model: Any) -> dict[str, Any]:
    if hasattr(model, "model_dump"):
        return model.model_dump()
    return model.dict()


class ResetRequest(BaseModel):
    task_name: Optional[str] = "easy"
    seed: int = 42


class GradeRequest(BaseModel):
    task_name: str
    actions: list


def _normalize_task_name(task_name: Optional[str]) -> str:
    candidate = (task_name or "easy").strip()
    return candidate if candidate in TASK_REGISTRY else "easy"


def _resolve_reset_request(
    body_payload: Optional[dict[str, Any]] = None,
    task_name: Optional[str] = None,
    seed: Optional[str] = None,
) -> ResetRequest:
    payload: dict[str, Any] = {}
    if task_name is not None:
        payload["task_name"] = task_name
    if seed is not None:
        payload["seed"] = seed
    if body_payload:
        payload.update(body_payload)

    try:
        return ResetRequest(**payload)
    except ValidationError:
        return ResetRequest()


async def _extract_reset_payload(request: Request) -> dict[str, Any]:
    try:
        raw_body = await request.body()
    except Exception:
        return {}

    if not raw_body:
        return {}

    body_text = raw_body.decode("utf-8", errors="ignore").strip()
    if not body_text:
        return {}

    content_type = request.headers.get("content-type", "").lower()
    if "application/x-www-form-urlencoded" in content_type:
        parsed = parse_qs(body_text, keep_blank_values=False)
        return {key: values[-1] for key, values in parsed.items() if values}

    try:
        decoded = json.loads(body_text)
    except json.JSONDecodeError:
        return {}

    return decoded if isinstance(decoded, dict) else {}


@app.get("/health")
def health():
    return {"status": "healthy", "environment": "SYNAPSE-X", "version": APP_VERSION}


@app.get("/metadata")
def metadata():
    return {
        "name": "synapse-x",
        "environment": "SYNAPSE-X",
        "version": APP_VERSION,
        "description": APP_DESCRIPTION,
        "mode": "simulation",
        "entrypoints": {
            "api_app": "api.app:app",
            "server_script": "server.app:main",
            "environment": "env.environment:SynapseXEnvironment",
            "inference": "inference.py",
        },
        "task_names": list(TASK_REGISTRY.keys()),
        "task_seeds": TASK_SEEDS,
        "endpoints": API_ENDPOINTS,
    }


@app.get("/schema")
def schema():
    return {
        "environment": {
            "name": "SynapseXEnvironment",
            "module": "env.environment",
        },
        "observation": Observation.model_json_schema(),
        "action": Action.model_json_schema(),
        "state": {
            "type": "object",
            "properties": {
                "time": {"type": "integer"},
                "resources": {"type": "number"},
                "total_reward": {"type": "number"},
                "done": {"type": "boolean"},
                "tasks": {"type": "array"},
                "history": {"type": "array"},
                "seed": {"type": "integer"},
            },
            "required": ["time", "resources", "done", "tasks"],
        },
        "step_result": StepResult.model_json_schema(),
        "grade_result": GradeResult.model_json_schema(),
    }


@app.get("/mcp")
def mcp_manifest():
    return {
        "status": "healthy",
        "mode": "simulation",
        "environment": "SYNAPSE-X",
        "metadata_url": "/metadata",
        "schema_url": "/schema",
    }


@app.post("/mcp")
async def mcp_rpc(request: Request):
    try:
        payload = await request.json()
    except Exception:
        payload = {}

    return {
        "jsonrpc": "2.0",
        "id": payload.get("id"),
        "result": {
            "status": "healthy",
            "mode": "simulation",
            "environment": "SYNAPSE-X",
            "metadata_url": "/metadata",
            "schema_url": "/schema",
        },
    }


@app.get("/reset")
def reset_get(task_name: Optional[str] = None, seed: Optional[str] = None):
    global _env
    req = _resolve_reset_request(task_name=task_name, seed=seed)
    task = _normalize_task_name(req.task_name)
    _env = SynapseXEnvironment(task_config=TASK_REGISTRY[task], seed=req.seed)
    return to_dict(_env.reset())


@app.post("/reset")
async def reset(request: Request, task_name: Optional[str] = None, seed: Optional[str] = None):
    global _env
    body_payload = await _extract_reset_payload(request)
    req = _resolve_reset_request(body_payload=body_payload, task_name=task_name, seed=seed)
    task = _normalize_task_name(req.task_name)
    _env = SynapseXEnvironment(task_config=TASK_REGISTRY[task], seed=req.seed)
    return to_dict(_env.reset())


@app.post("/step")
def step(action_dict: dict):
    env = get_env()
    try:
        action = Action(**action_dict)
    except ValidationError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    try:
        result = env.step(action)
    except RuntimeError as exc:
        raise HTTPException(400, str(exc)) from exc
    return {
        "observation": to_dict(result.observation),
        "reward": float(result.reward),
        "done": bool(result.done),
        "info": result.info or {},
    }


@app.get("/state")
def state():
    return get_env().state()


@app.get("/tasks")
def list_tasks():
    return {name: cfg for name, cfg in TASK_REGISTRY.items()}


@app.get("/validate")
def validate():
    return {
        "status": "ok",
        "environment": "SYNAPSE-X",
        "version": APP_VERSION,
        "mode": "simulation",
        "task_names": list(TASK_REGISTRY.keys()),
        "task_seeds": TASK_SEEDS,
        "endpoints": API_ENDPOINTS,
        "docker": {
            "image_name": LOCAL_IMAGE_NAME,
            "app_import": "api.app:app",
        },
        "multi_mode": {
            "server_script": "server.app:main",
            "lockfile": "uv.lock",
        },
        "openenv_spec": {
            "reset": "POST /reset",
            "step": "POST /step",
            "state": "GET  /state",
        },
    }


@app.post("/grade")
def grade_endpoint(req: GradeRequest):
    if req.task_name not in TASK_REGISTRY:
        raise HTTPException(400, f"Unknown task_name '{req.task_name}'. Choose from: {list(TASK_REGISTRY.keys())}")
    result: GradeResult = run_grade(req.task_name, req.actions)
    return to_dict(result)


if __name__ == "__main__":
    import uvicorn

    port = int(os.environ.get("PORT", 7860))
    uvicorn.run("api.app:app", host="0.0.0.0", port=port, reload=False)