server/app.py

"""HTTP server for running the code-review environment in a Space/container."""

from __future__ import annotations

import os
from threading import Lock
from typing import Any, Dict

from flask import Flask, jsonify, request

from environment.env import CodeReviewEnv


app = Flask(__name__)
_env = CodeReviewEnv()
_lock = Lock()


@app.get("/")
def root() -> Any:
    return jsonify({
        "status": "ok",
        "service": "code-review-agent-env",
        "endpoints": ["/health", "/reset", "/step", "/state"],
    })


@app.get("/health")
def health() -> Any:
    return jsonify({"status": "healthy"})


@app.route("/reset", methods=["GET", "POST"])
def reset() -> Any:
    payload: Dict[str, Any] = request.get_json(silent=True) or {}
    task_id = payload.get("task_id") or request.args.get("task_id")
    with _lock:
        obs = _env.reset(task_id=task_id)
    return jsonify({"observation": obs})


@app.post("/step")
def step() -> Any:
    payload = request.get_json(silent=True) or {}
    action = payload.get("action")
    if not isinstance(action, dict):
        return jsonify({"error": "Request body must include an 'action' object"}), 400

    with _lock:
        observation, reward, done, info = _env.step(action)

    return jsonify(
        {
            "observation": observation,
            "reward": reward,
            "done": done,
            "info": info,
        }
    )


@app.get("/state")
def state() -> Any:
    with _lock:
        current_state = _env.state()
    return jsonify(current_state)


def main() -> None:
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", "7860"))
    app.run(host=host, port=port)


if __name__ == "__main__":
    main()