Spaces:
Sleeping
Sleeping
File size: 4,516 Bytes
cb330aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | #!/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()
|