Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| OpenEnv and API readiness verification for SYNAPSE-X. | |
| """ | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import httpx | |
| from fastapi.testclient import TestClient | |
| PROJECT_ROOT = Path(__file__).resolve().parents[1] | |
| if str(PROJECT_ROOT) not in sys.path: | |
| sys.path.insert(0, str(PROJECT_ROOT)) | |
| from api.app import app | |
| from agents.baseline import run_episode | |
| from env.environment import SynapseXEnvironment | |
| from env.grader import TASK_REGISTRY, grade | |
| from env.models import Action, Observation, StepResult | |
| def assert_true(condition: bool, message: str): | |
| if not condition: | |
| raise AssertionError(message) | |
| def verify_environment(): | |
| env = SynapseXEnvironment(task_config=TASK_REGISTRY["easy"], seed=42) | |
| obs = env.reset() | |
| assert_true(isinstance(obs, Observation), "reset() must return Observation") | |
| assert_true(len(obs.tasks) == 3, "reset() should expose 3 easy tasks") | |
| step_result = env.step(Action(action_type="execute", task_id=0)) | |
| assert_true(isinstance(step_result, StepResult), "step() must return StepResult") | |
| assert_true(-2.0 <= step_result.reward <= 2.0, "reward must be clamped") | |
| state = env.state() | |
| assert_true(isinstance(state, dict), "state() must return dict") | |
| assert_true({"time", "resources", "tasks", "done"} <= set(state.keys()), "state() missing keys") | |
| replay_env = SynapseXEnvironment(task_config=TASK_REGISTRY["easy"], seed=42) | |
| replay_env.reset() | |
| replay_result = replay_env.step(Action(action_type="execute", task_id=0)) | |
| assert_true( | |
| step_result.reward == replay_result.reward and step_result.info == replay_result.info, | |
| "environment should be deterministic for fixed seed", | |
| ) | |
| baseline_env = SynapseXEnvironment(task_config=TASK_REGISTRY["easy"], seed=42) | |
| actions = run_episode(baseline_env) | |
| grade_a = grade("easy", actions) | |
| grade_b = grade("easy", actions) | |
| assert_true(grade_a.score == grade_b.score, "grader should be deterministic") | |
| return { | |
| "reset_type": type(obs).__name__, | |
| "step_type": type(step_result).__name__, | |
| "state_keys": sorted(state.keys()), | |
| "deterministic_easy_score": grade_a.score, | |
| } | |
| def verify_api(): | |
| client = TestClient(app) | |
| default_reset_response = client.post("/reset") | |
| assert_true(default_reset_response.status_code == 200, "empty-body /reset should return 200") | |
| reset_response = client.post("/reset", json={"task_name": "medium", "seed": 42}) | |
| assert_true(reset_response.status_code == 200, "/reset should return 200") | |
| step_response = client.post("/step", json={"action_type": "execute", "task_id": 0}) | |
| assert_true(step_response.status_code == 200, "/step should return 200") | |
| state_response = client.get("/state") | |
| assert_true(state_response.status_code == 200, "/state should return 200") | |
| health_response = client.get("/health") | |
| assert_true(health_response.status_code == 200, "/health should return 200") | |
| validate_response = client.get("/validate") | |
| assert_true(validate_response.status_code == 200, "/validate should return 200") | |
| return { | |
| "default_reset_keys": sorted(default_reset_response.json().keys()), | |
| "reset_keys": sorted(reset_response.json().keys()), | |
| "step_keys": sorted(step_response.json().keys()), | |
| "state_keys": sorted(state_response.json().keys()), | |
| "health": health_response.json(), | |
| "validate": validate_response.json(), | |
| } | |
| def verify_live_server(base_url: str = "http://127.0.0.1:7860"): | |
| try: | |
| health_response = httpx.get(f"{base_url}/health", timeout=2.0) | |
| if health_response.status_code != 200: | |
| return {"live_server": "unavailable"} | |
| reset_response = httpx.post(f"{base_url}/reset", timeout=2.0) | |
| state_response = httpx.get(f"{base_url}/state", timeout=2.0) | |
| return { | |
| "live_server": "ok", | |
| "health_status": health_response.status_code, | |
| "reset_status": reset_response.status_code, | |
| "state_status": state_response.status_code, | |
| } | |
| except Exception: | |
| return {"live_server": "not_running"} | |
| def main(): | |
| report = { | |
| "environment": verify_environment(), | |
| "api": verify_api(), | |
| "live_api": verify_live_server(), | |
| "status": "ok", | |
| } | |
| print("[OK] API working") | |
| print("[OK] deterministic grader") | |
| print("[OK] environment valid") | |
| print(json.dumps(report, indent=2)) | |
| if __name__ == "__main__": | |
| main() | |