"""FastAPI application entry point for the HR Productivity Environment.""" import json from pathlib import Path from typing import Optional from fastapi import Query from fastapi.responses import RedirectResponse from openenv.core.env_server.http_server import create_app from hr_env.models import HRAction, HRObservation from hr_env.server.environment import HRProductivityEnvironment app = create_app( HRProductivityEnvironment, HRAction, HRObservation, env_name="hr-productivity-env", max_concurrent_envs=1, ) @app.get("/") async def root(): return RedirectResponse(url="/web") # ── Hyperparameters endpoint ──────────────────────────────────────── @app.get("/hyperparameters") async def hyperparameters(): """Return all environment hyperparameters and configuration constants.""" from hr_env.server.company import ( BENEFITS_MULTIPLIER, DEPT_CONFIGS, RECRUITING_COST_PER_HIRE, ) from hr_env.server.data_gen import ( DEPT_SIZE_FRACTIONS, DEPT_SKILLS, ROLES, SALARY_BASE, SALARY_SIGMA, ) from hr_env.server.events import ALL_EVENTS from hr_env.server.phases import PHASE_ACTIONS, PHASE_MIN_ACTIONS, PHASES from hr_env.server.scoring import compute_final_score # noqa: F401 return { "environment": { "max_quarters": 6, "max_steps": 300, "max_concurrent_envs": 1, "default_company_size": 300, "recommended_size_range": [200, 500], "departments": ["Engineering", "Sales", "Operations", "HR", "Finance"], }, "company_financials": { "base_revenue_formula": "size * 150,000", "hr_budget_formula": "size * 6,000", "quarterly_hr_budget": "hr_budget / 4", "non_employment_cost_ratio": 0.55, "benefits_multiplier": BENEFITS_MULTIPLIER, "recruiting_cost_per_hire": RECRUITING_COST_PER_HIRE, "training_cost_per_hour_per_employee": 50, }, "cobb_douglas_production": { dept: { "A": cfg["A"], "alpha": cfg["alpha"], "beta": cfg["beta"], "base_revenue_share": cfg["base_revenue_share"], } for dept, cfg in DEPT_CONFIGS.items() }, "employee_generation": { "salary_base_by_level": SALARY_BASE, "salary_lognormal_sigma": SALARY_SIGMA, "performance_distribution": {"type": "normal", "mean": 3.2, "std": 0.8, "clip": [1.0, 5.0]}, "tenure_distribution": {"type": "exponential", "lambda_months": 24}, "engagement_distribution": {"type": "beta", "alpha": 7, "beta": 3, "scale": 100}, "level_distribution": { "type": "categorical", "probabilities": {"L1": 0.35, "L2": 0.30, "L3": 0.20, "L4": 0.10, "L5": 0.05}, }, "dept_size_fractions": DEPT_SIZE_FRACTIONS, "skills_per_employee": {"min": 2, "max": 4}, }, "employee_lifecycle": { "promotion_salary_increase": 0.15, "promotion_engagement_boost": 10, "promotion_flight_risk_reduction": 0.15, "quarterly_engagement_decay": -1.5, "long_tenure_no_promotion_penalty": -3.0, "long_tenure_threshold_quarters": 4, "transfer_engagement_penalty": -5, "termination_colleague_engagement_penalty": -3, "new_hire_engagement_distribution": {"type": "beta", "alpha": 8, "beta": 2, "scale": 100}, }, "flight_risk": { "model": "logistic", "formula": "sigmoid(-1 + 2*pay_factor + 1.5*engagement_factor + 1*tenure_factor + 0.8*market_demand)", "turnover_probability": "flight_risk * 0.4 per quarter", "bounds": [0.02, 0.95], }, "phase_system": { "phases": PHASES, "actions_per_phase": PHASE_ACTIONS, "minimum_actions_per_phase": PHASE_MIN_ACTIONS, }, "stochastic_events": [ { "name": ev.name, "description": ev.description, "probability": ev.probability, } for ev in ALL_EVENTS ], "scoring": { "quarterly_reward_weights": { "hcva_improvement": 0.40, "hcroi_improvement": 0.20, "qips_improvement": 0.20, "five_indexes": 0.20, }, "final_score_weights": { "hcva_trajectory": 0.25, "hcroi_final_vs_baseline": 0.20, "employee_value": 0.20, "qips_consistency": 0.15, "five_indexes_cumulative": 0.10, "financial_health": 0.10, }, "calibration_targets": { "random_agent": "0.15-0.25", "reasonable_heuristic": "~0.50", "strong_agent": "0.70+", }, }, "heuristic_strategy": { "description": "Data-driven heuristic used in the demo endpoint", "hiring_rate": "12% of current headcount for primary dept, 8% for secondary", "training_budget_fraction": 0.30, "training_hours_per_session": 20, "compensation_adjustment_pct": 3.0, "retention_budget_fraction": 0.20, "promotion_criteria": {"min_performance": 4.0, "max_level": 4, "max_per_quarter": 3}, "termination_criteria": {"max_performance": 1.8, "max_per_quarter": 2}, }, "roles_by_department": { dept: [{"role": role, "level": level} for role, level in roles] for dept, roles in ROLES.items() }, "skills_by_department": DEPT_SKILLS, } # ── Scenarios endpoint ────────────────────────────────────────────── @app.get("/scenarios") async def scenarios(): """Return all available task scenarios with their configurations.""" tasks_dir = Path(__file__).resolve().parent.parent / "tasks" task_list = [] if tasks_dir.exists(): for f in sorted(tasks_dir.glob("*.json")): try: task_list.append(json.loads(f.read_text())) except (json.JSONDecodeError, OSError): continue # Also include scenario effect descriptions scenario_effects = { "high_eng_turnover": { "modifications": "Engineering flight_risk += 0.20, engagement -= 10", "challenge": "Stabilize retention in a bleeding engineering department", }, "budget_cuts": { "modifications": "HR budget halved (50% reduction)", "challenge": "Optimize workforce with severely constrained resources", }, "rapid_growth": { "modifications": "Revenue +30%, 15% employees deactivated (understaffing)", "challenge": "Scale the workforce while maintaining quality and culture", }, "balanced_optimization": { "modifications": "All employees: performance -0.3, engagement -8", "challenge": "Improve below-average metrics across the board", }, } return { "scenarios": [ { **task, "effects": scenario_effects.get(task.get("params", {}).get("scenario"), {}), } for task in task_list ], "default": { "seed": 42, "size": 300, "scenario": None, "description": "Standard company with no scenario modifications", }, } # ── Demo endpoint ─────────────────────────────────────────────────── @app.get("/demo") async def demo( seed: int = Query(42, description="Random seed for reproducibility"), size: int = Query(300, ge=50, le=1000, description="Company size (employees)"), scenario: Optional[str] = Query( None, description="Scenario variant", enum=["high_eng_turnover", "budget_cuts", "rapid_growth", "balanced_optimization"], ), ): """Run a full 6-quarter demo with a heuristic strategy and return results. Executes data-driven HR decisions through all HCM:21 phases for 6 quarters. No LLM API key required — uses a built-in heuristic agent. """ from demo import run_demo_json return run_demo_json(seed=seed, size=size, scenario=scenario)