| """Smoke-test the live ForgeEnv Space end-to-end via the OpenEnv client.
|
|
|
| Runs one full episode against the deployed Space:
|
|
|
| reset() -> drift-gen turn
|
| step(DriftAction) -> repair turn
|
| step(RepairAction) -> reward + verifier breakdown
|
|
|
| This is the simplest possible "is the deployed env working?" check
|
| and a clean standalone artifact for the hackathon writeup/video.
|
|
|
| Usage::
|
|
|
| python scripts/test_live_env.py
|
| """
|
| from __future__ import annotations
|
|
|
| import asyncio
|
| import json
|
|
|
| from openenv.core import GenericAction, GenericEnvClient
|
|
|
| ENV_URL = "https://akhiilll-forgeenv.hf.space"
|
|
|
|
|
| def _summary(result, label: str) -> None:
|
| obs = result.observation if isinstance(result.observation, dict) else {}
|
| print(f"\n=== {label} ===")
|
| print(f"phase : {obs.get('current_phase')}")
|
| print(f"task_id : {obs.get('task_id')}")
|
| print(f"target_category : {obs.get('target_category')}")
|
| print(f"reward : {result.reward}")
|
| print(f"done : {result.done}")
|
| breakdown = obs.get("reward_breakdown")
|
| if breakdown:
|
| print("reward_breakdown:")
|
| print(json.dumps(breakdown, indent=2))
|
| script = obs.get("script_content") or obs.get("broken_script") or ""
|
| if script:
|
| preview = script.splitlines()[:8]
|
| print("script preview :")
|
| for line in preview:
|
| print(f" | {line}")
|
| if len(script.splitlines()) > 8:
|
| print(" | ...")
|
|
|
|
|
| async def main(seed: int = 42) -> None:
|
| print(f"connecting to {ENV_URL} (seed={seed}) ...")
|
| client = GenericEnvClient(base_url=ENV_URL)
|
|
|
| res = await client.reset(seed=seed, options={"difficulty": "medium"})
|
| _summary(res, "after reset()")
|
| target = res.observation.get("target_category", "RenameApiCall") if isinstance(res.observation, dict) else "RenameApiCall"
|
|
|
| res = await client.step(GenericAction(
|
| breakage={"action_type": "breakage", "primitive_type": target, "params": {}},
|
| repair=None,
|
| ))
|
| _summary(res, "after drift step (Challenger)")
|
|
|
|
|
| res = await client.step(GenericAction(
|
| breakage=None,
|
| repair={"action_type": "repair", "unified_diff": ""},
|
| ))
|
| _summary(res, "after repair step (Solver, no-op)")
|
|
|
| print("\nOK -- reset + 2 steps round-trip the deployed env.")
|
|
|
|
|
| if __name__ == "__main__":
|
| import sys
|
|
|
| seed = int(sys.argv[1]) if len(sys.argv) > 1 else 42
|
| asyncio.run(main(seed=seed))
|
|
|
