server/app.py

"""
LogSentinel v2 FastAPI application — canonical definition.

Both server.py (root) and server/__init__.py import from here.
"""

import sys
import os

# Ensure parent directory (repo root) is on sys.path for module imports
_repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _repo_root not in sys.path:
    sys.path.insert(0, _repo_root)

from typing import Any, Dict

from fastapi import Body, FastAPI
from environment import LogSentinelEnv
from tasks import TASKS

app = FastAPI(
    title="LogSentinel v2 — Multi-Agent SOC War-Room",
    description="OpenEnv-compliant multi-agent SOC environment for RLVR training.",
    version="2.0.0",
)
env = LogSentinelEnv()


@app.get("/")
async def root() -> Dict[str, Any]:
    """Root endpoint — environment info."""
    return {
        "name": "LogSentinel",
        "version": "2.0.0",
        "description": (
            "Adaptive Multi-Agent SOC War-Room environment for training AI agents "
            "on real-world incident response with verifiable rewards."
        ),
        "endpoints": {
            "POST /reset": "Reset environment, returns initial observation",
            "POST /step": "Execute action, returns {observation, reward, done, info}",
            "GET /state": "Current environment state with reward breakdown",
            "GET /health": "Health check",
            "GET /tasks": "List available tasks",
        },
        "status": "healthy",
    }


@app.post("/reset")
async def reset(request: Dict[str, Any] = Body(default={})) -> Dict[str, Any]:
    """Reset the environment and return initial observation."""
    task_name = request.get("task_name")
    agent_role = request.get("agent_role")
    seed = request.get("seed")
    return env.reset(task_name=task_name, agent_role=agent_role, seed=seed)


@app.post("/step")
async def step(request: Dict[str, Any] = Body(...)) -> Dict[str, Any]:
    """Take a step with the given action."""
    action_data = request.get("action", {})
    return env.step(action_data)


@app.get("/state")
async def get_state() -> Dict[str, Any]:
    """Return current environment state."""
    return env.state


@app.get("/health")
async def health() -> Dict[str, str]:
    """Health check endpoint."""
    return {"status": "healthy"}


@app.get("/tasks")
async def list_tasks() -> Dict[str, Any]:
    """List available tasks."""
    return {
        "tasks": [
            {
                "name": t.name,
                "description": t.description,
                "difficulty": t.difficulty,
                "max_steps": t.max_steps,
                "num_logs": t.num_logs,
                "multi_agent": t.multi_agent,
            }
            for t in TASKS.values()
        ]
    }


def main() -> None:
    """Entry point for [project.scripts] server command."""
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)


if __name__ == "__main__":
    main()