Skills › Scientific Agent Skills › pufferlib › env_contract_validator.py
env_contract_validator.py
198 lines · skills/pufferlib/scripts/env_contract_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate a built-in synthetic environment without importing plug-ins."""
4from __future__ import annotations
6import argparse
7import math
8import random
9from typing import Any
11try:
12 from ._common import UserInputError, bounded_int, emit_json
13 from .env_template import SyntheticGymEnv
14except ImportError: # Direct script execution.
15 from _common import UserInputError, bounded_int, emit_json
16 from env_template import SyntheticGymEnv
19def _require(condition: bool, message: str, errors: list[str]) -> None:
20 if not condition:
21 errors.append(message)
24def _validate_reset(result: Any, env: SyntheticGymEnv, errors: list[str]) -> Any:
25 _require(
26 isinstance(result, tuple) and len(result) == 2,
27 "reset() must return (observation, info)",
28 errors,
29 )
30 if not isinstance(result, tuple) or len(result) != 2:
31 return None
32 observation, info = result
33 _require(
34 env.observation_space.contains(observation),
35 "reset observation is outside observation_space",
36 errors,
37 )
38 _require(isinstance(info, dict), "reset info must be a dict", errors)
39 return observation
42def _validate_step(result: Any, env: SyntheticGymEnv, errors: list[str]) -> tuple[bool, bool]:
43 _require(
44 isinstance(result, tuple) and len(result) == 5,
45 "step() must return (observation, reward, terminated, truncated, info)",
46 errors,
47 )
48 if not isinstance(result, tuple) or len(result) != 5:
49 return False, False
50 observation, reward, terminated, truncated, info = result
51 _require(
52 env.observation_space.contains(observation),
53 "step observation is outside observation_space",
54 errors,
55 )
56 _require(
57 isinstance(reward, (int, float))
58 and not isinstance(reward, bool)
59 and math.isfinite(float(reward)),
60 "reward must be a finite scalar",
61 errors,
62 )
63 _require(type(terminated) is bool, "terminated must be bool", errors)
64 _require(type(truncated) is bool, "truncated must be bool", errors)
65 _require(isinstance(info, dict), "step info must be a dict", errors)
66 _require(
67 not (terminated and truncated),
68 "synthetic environment must not terminate and truncate simultaneously",
69 errors,
70 )
71 return bool(terminated), bool(truncated)
74def _check_determinism(seed: int, max_steps: int, errors: list[str]) -> None:
75 env_a = SyntheticGymEnv(max_steps=max_steps)
76 env_b = SyntheticGymEnv(max_steps=max_steps)
77 first_a = env_a.reset(seed=seed)
78 first_b = env_b.reset(seed=seed)
79 _require(first_a == first_b, "same reset seed produced different results", errors)
80 actions = [0, 2, 1, 2, 0, 1]
81 for action in actions:
82 result_a = env_a.step(action)
83 result_b = env_b.step(action)
84 _require(result_a == result_b, "same action trace produced different results", errors)
85 if result_a[2] or result_a[3]:
86 break
87 env_a.close()
88 env_b.close()
91def validate_synthetic(
92 *,
93 seed: int,
94 steps: int,
95 episodes: int,
96 max_steps: int,
97) -> dict[str, Any]:
98 """Run bounded API, space, reset, termination, and determinism checks."""
99 errors: list[str] = []
100 env = SyntheticGymEnv(max_steps=max_steps)
101 _require(hasattr(env, "observation_space"), "missing observation_space", errors)
102 _require(hasattr(env, "action_space"), "missing action_space", errors)
104 action_rng = random.Random(seed + 1)
105 observation = _validate_reset(env.reset(seed=seed), env, errors)
106 total_steps = 0
107 completed_episodes = 0
108 terminations = 0
109 truncations = 0
111 while total_steps < steps and completed_episodes < episodes:
112 action = env.action_space.sample(action_rng)
113 _require(env.action_space.contains(action), "sampled action is invalid", errors)
114 terminated, truncated = _validate_step(env.step(action), env, errors)
115 total_steps += 1
116 if terminated or truncated:
117 completed_episodes += 1
118 terminations += int(terminated)
119 truncations += int(truncated)
120 observation = _validate_reset(
121 env.reset(seed=seed + completed_episodes), env, errors
122 )
124 _require(
125 observation is None or env.observation_space.contains(observation),
126 "final observation is invalid",
127 errors,
128 )
129 _check_determinism(seed, max_steps, errors)
130 env.close()
131 return {
132 "checks": {
133 "deterministic_seed": "passed" if not errors else "see errors",
134 "reset_two_tuple": True,
135 "spaces": True,
136 "step_five_tuple": True,
137 },
138 "contract": "gymnasium",
139 "environment": "synthetic",
140 "errors": errors,
141 "network_used": False,
142 "observed": {
143 "episodes": completed_episodes,
144 "steps": total_steps,
145 "terminations": terminations,
146 "truncations": truncations,
147 },
148 "seed": seed,
149 "status": "passed" if not errors else "failed",
150 "vector_shape_expectation": {
151 "actions": ["num_envs"],
152 "observations": ["num_envs", 4],
153 "rewards": ["num_envs"],
154 "terminations": ["num_envs"],
155 "truncations": ["num_envs"],
156 },
157 }
160def build_parser() -> argparse.ArgumentParser:
161 parser = argparse.ArgumentParser(
162 description=(
163 "Validate only the allowlisted built-in synthetic environment; "
164 "external modules and dotted import paths are not supported."
165 )
166 )
167 parser.add_argument("--environment", choices=["synthetic"], default="synthetic")
168 parser.add_argument("--seed", type=int, default=42)
169 parser.add_argument("--steps", type=int, default=64, help="1..10000")
170 parser.add_argument("--episodes", type=int, default=8, help="1..100")
171 parser.add_argument("--max-steps", type=int, default=16, help="1..10000")
172 parser.add_argument("--compact", action="store_true")
173 return parser
176def main(argv: list[str] | None = None) -> int:
177 parser = build_parser()
178 args = parser.parse_args(argv)
179 try:
180 steps = bounded_int(args.steps, name="steps", minimum=1, maximum=10_000)
181 episodes = bounded_int(args.episodes, name="episodes", minimum=1, maximum=100)
182 max_steps = bounded_int(
183 args.max_steps, name="max_steps", minimum=1, maximum=10_000
184 )
185 report = validate_synthetic(
186 seed=args.seed,
187 steps=steps,
188 episodes=episodes,
189 max_steps=max_steps,
190 )
191 except (UserInputError, ValueError, RuntimeError) as exc:
192 parser.error(str(exc))
193 emit_json(report, pretty=not args.compact)
194 return 0 if report["status"] == "passed" else 1
197if __name__ == "__main__":
198 raise SystemExit(main())