""" server/app.py — Step 4: Create Server FastAPI application for the Cross-Session Continuity environment. Uses openenv.core's create_app() factory, which wires up: POST /reset — start a new episode POST /step — execute an action GET /state — current episode state GET /health — health check WS /ws — WebSocket session (stateful) GET /docs — Swagger UI Run with: uvicorn server.app:app --host 0.0.0.0 --port 7860 """ import sys import os _HERE = os.path.dirname(os.path.abspath(__file__)) _ROOT = os.path.dirname(_HERE) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) try: from openenv.core.env_server.http_server import create_app except Exception as e: raise ImportError( "openenv-core is required. Install with: pip install openenv-core" ) from e try: from models import ContinuityAction, ContinuityObservation from server.env import CrossSessionContinuityEnv except ImportError: from models import ContinuityAction, ContinuityObservation # type: ignore from server.env import CrossSessionContinuityEnv # type: ignore from fastapi.responses import HTMLResponse app = create_app( CrossSessionContinuityEnv, ContinuityAction, ContinuityObservation, env_name="cross-session-continuity-env", max_concurrent_envs=1, ) _LANDING_HTML = """ Cross-Session Continuity Env

🧠 Cross-Session Continuity Env

• Running

An OpenEnv-compliant RL environment where an LLM agent must complete a coding task across two sessions with zero shared memory. Session 1 writes a structured handoff note; Session 2 starts cold with only that note.

GET /healthServer health check
GET /docsInteractive API (Swagger)
POST /resetStart a new episode
POST /stepExecute an action
WS /wsWebSocket (stateful session)
GET /stateCurrent episode state

Quick start:

pip install openenv-core

from client import ContinuityEnvClient, ContinuityAction

with ContinuityEnvClient(
    base_url="https://aswini-kumar-cross-session-continuity-env.hf.space"
) as env:
    obs = env.reset(difficulty="easy", seed=42)
    result = env.step(ContinuityAction(tool="run_tests"))
    print(result.observation.output)

Tools: read_file   write_file   run_tests   write_handoff   parse_handoff   submit

API Docs  |  Health  |  GitHub  |  HF Space
""" @app.get("/", response_class=HTMLResponse, include_in_schema=False) @app.get("/web", response_class=HTMLResponse, include_in_schema=False) async def landing(): """Human-readable landing page.""" return HTMLResponse(content=_LANDING_HTML) def main(host: str = "0.0.0.0", port: int = 7860): import uvicorn uvicorn.run(app, host=host, port=port) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=7860) args = parser.parse_args() main(host=args.host, port=args.port)