| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import sys |
| import time |
| from dataclasses import replace |
| from pathlib import Path |
| from typing import Any |
|
|
| import numpy as np |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
| from life_game.arcade import ( |
| ArcadeConfig, |
| ChallengeSpecError, |
| LLMChallengeError, |
| apply_challenge_spec, |
| parse_challenge_spec, |
| request_llm_challenge, |
| ) |
| from life_game.board import BoardModel |
| from life_game.game import INTERVENTION_MODE, TOWER_MODE, new_game |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description="Probe a challenge-spec LLM endpoint.") |
| parser.add_argument("--base-url", default="http://127.0.0.1:8080/v1") |
| parser.add_argument("--model", default="signal-garden-qwen35-code-mutation") |
| parser.add_argument("--timeout-seconds", type=float, default=120.0) |
| parser.add_argument("--fail-on-error", action="store_true") |
| args = parser.parse_args() |
|
|
| config = ArcadeConfig( |
| llm_base_url=args.base_url, |
| llm_model=args.model, |
| llm_timeout_seconds=max(1.0, args.timeout_seconds), |
| ) |
| rng = np.random.default_rng(3407) |
| cases = [ |
| { |
| "name": "intervention_start", |
| "mode": INTERVENTION_MODE, |
| "preference": "readable survival pressure", |
| "phase": "start", |
| }, |
| { |
| "name": "intervention_mutation", |
| "mode": INTERVENTION_MODE, |
| "preference": "readable survival pressure", |
| "phase": "mutation", |
| "score": 42, |
| "data": {"discharges": 1}, |
| }, |
| { |
| "name": "tower_start", |
| "mode": TOWER_MODE, |
| "preference": "tower defense with light pressure", |
| "phase": "start", |
| }, |
| ] |
|
|
| results = [] |
| for case in cases: |
| result = _run_case(config, case, rng) |
| results.append(result) |
| print(json.dumps(result, sort_keys=True)) |
|
|
| passed = sum(1 for result in results if result["ok"]) |
| summary = {"ok": passed == len(results), "passed": passed, "total": len(results)} |
| print(json.dumps({"summary": summary}, sort_keys=True)) |
| return 1 if args.fail_on_error and not summary["ok"] else 0 |
|
|
|
|
| def _run_case(config: ArcadeConfig, case: dict[str, Any], rng: np.random.Generator) -> dict[str, Any]: |
| model = BoardModel(grid=np.zeros((16, 16), dtype=np.uint8)) |
| model = replace(model, running=True) |
| game = new_game(model.size, str(case["mode"]), rng) |
| if "score" in case: |
| game = replace(game, score=int(case["score"])) |
| if "data" in case: |
| game = replace(game, data={**game.data, **case["data"]}) |
|
|
| started = time.monotonic() |
| try: |
| raw = request_llm_challenge( |
| config, |
| str(case["mode"]), |
| str(case["preference"]), |
| case["phase"], |
| model, |
| game, |
| ) |
| spec = parse_challenge_spec(raw, case["phase"]) |
| next_model, next_game = apply_challenge_spec(model, game, spec, case["phase"], rng) |
| except (ChallengeSpecError, LLMChallengeError) as exc: |
| return { |
| "name": case["name"], |
| "ok": False, |
| "phase": case["phase"], |
| "error": str(exc), |
| "elapsed_seconds": round(time.monotonic() - started, 3), |
| } |
|
|
| return { |
| "name": case["name"], |
| "ok": True, |
| "phase": case["phase"], |
| "title": spec.title, |
| "difficulty": spec.difficulty, |
| "actions": [action.type for action in spec.actions], |
| "action_count": len(spec.actions), |
| "goal_updates": dict(spec.goal_updates), |
| "enemy_delta": next_game.enemy_count - game.enemy_count, |
| "live_cell_delta": int(next_model.grid.sum()) - int(model.grid.sum()), |
| "elapsed_seconds": round(time.monotonic() - started, 3), |
| } |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|