"""HF Spaces server: ClaimSense adjudication gym + lightweight dashboard. Run with:: uvicorn space_app:app --host 0.0.0.0 --port 7860 Adds three things on top of the OpenEnv FastAPI scaffolding: 1. ``GET /`` — an HTML dashboard so the Space's landing page looks like a product, not raw JSON. 2. ``GET /api`` — the JSON metadata block that used to live at ``/``. 3. ``GET /info`` — verbose env description used by notebooks/training. """ from __future__ import annotations import os import sys from pathlib import Path # Make local sibling modules importable when running inside the Space's # Docker image (where the working directory is ``/app``). sys.path.insert(0, str(Path(__file__).resolve().parent)) from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from openenv.core.env_server import create_fastapi_app from models import AdjudicatorAction, AdjudicatorObservation from server.claims_environment import AdjudicationGym # --------------------------------------------------------------------------- # Compose the FastAPI app # --------------------------------------------------------------------------- app: FastAPI = create_fastapi_app( AdjudicationGym, AdjudicatorAction, AdjudicatorObservation ) # Allow notebooks/clients on any origin to call us during demos. app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # Dashboard HTML # --------------------------------------------------------------------------- DASHBOARD_HTML = """ ClaimSense AI · Adjudication Gym

🛡️ ClaimSense AI

checking… Space akhiilll/claims-env

OpenEnv Hackathon · Statement 3.1 · Scaler AI Labs

An adjudication gym for training LLM agents to triage insurance claims — partial observability, eight curated cases, fraud signals, and bank-transaction verification.

📘 OpenAPI /docs { } JSON /api

Endpoints

HTTP base

WebSocket

ResetPOST /reset

StepPOST /step

StateGET /state

HealthGET /health

Reward shaping

Correct decision+10

Fraud caught (deny)+5

Plaid discrepancy surfaced+2

Fast resolution (≤4 steps)+1

Wrong decision-5

Fraud missed (approve)-10

Query cost-0.1 … -0.5

Curated case set (8)

Routine fender-bender Burst pipe (capped) Staged accident External flood (excluded) Six-figure house fire Inflated stolen vehicle Slip-and-fall liability Lapsed policy

Action vocabulary (10)

loading…

Live API probe

click a button above to call the API

""" # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/", response_class=HTMLResponse) async def root_dashboard() -> HTMLResponse: """Single-page dashboard served at the Space root.""" return HTMLResponse(content=DASHBOARD_HTML) @app.get("/api") async def api_metadata() -> dict[str, object]: """Machine-readable metadata (was at ``/`` historically).""" return { "name": "ClaimSense Adjudication Gym", "version": "1.1.0", "hackathon": "OpenEnv Hackathon - Cerebral Valley", "problem_statement": "3.1 - Professional Tasks (World Modeling)", "partner_theme": "Scaler AI Labs - Enterprise Workflows", "status": "running", "valid_actions": list(AdjudicationGym.VALID_ACTIONS), "endpoints": { "health": "/health", "info": "/info", "reset": "POST /reset", "step": "POST /step", "state": "GET /state", "websocket": "/ws", }, } @app.get("/info") async def long_info() -> dict[str, object]: """Verbose description used by notebooks for documentation.""" return { "name": "ClaimSense Adjudication Gym", "version": "1.1.0", "description": ( "RL environment for training LLM agents to triage insurance " "claims through a sequence of evidence-gathering steps and a " "final verdict." ), "problem_statement": "3.1 - Professional Tasks (World Modeling)", "partner_theme": "Scaler AI Labs - Enterprise Workflows", "features": [ "Partial observability — agent must query for facts", "Multi-step decision making with terminal verdicts", "Fraud detection signals and Plaid-style transaction audit", "Business rule enforcement (deductibles, exclusions, lapsed)", "Enterprise-flavoured workflow with escalation paths", ], "valid_actions": list(AdjudicationGym.VALID_ACTIONS), "action_costs": AdjudicationGym.ACTION_TIME_COSTS, "reward_structure": { "correct_decision": "+10", "wrong_decision": "-5", "fraud_caught": "+5", "fraud_missed": "-10", "plaid_discrepancy_found": "+2", "query_cost": "-0.1 to -0.5", "fast_resolution_bonus": "+1 (≤ 4 steps)", "slow_resolution_penalty": "-0.2 per step beyond 8", }, "scenarios": 8, "scenario_types": [ "Routine approval", "Partial settlement (capped)", "Staged accident fraud", "Excluded coverage denial", "Six-figure escalation", "Inflated theft fraud", "Liability slip-and-fall", "Lapsed-policy denial", ], } @app.get("/scenarios") async def list_scenarios() -> dict[str, object]: """Enumerate the curated case library.""" from server.mock_systems import CASE_LIBRARY # local to avoid import cycle return { "total_scenarios": len(CASE_LIBRARY), "scenarios": [ { "index": i, "claim_id": case.claim_id, "claim_type": case.claim_type, "complexity": case.complexity, "amount": case.claim_amount, "is_fraud": case.is_fraud, } for i, case in enumerate(CASE_LIBRARY) ], } @app.get("/health") async def health_probe() -> dict[str, str]: """Liveness probe used by Spaces, monitors, and the dashboard.""" return {"status": "healthy", "environment": "claimsense"} # --------------------------------------------------------------------------- # Local dev entrypoint (``python space_app.py``) # --------------------------------------------------------------------------- if __name__ == "__main__": import uvicorn port = int(os.environ.get("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)