diff --git a/README.md b/README.md index f938b1b..8dc545a 100644 --- a/README.md +++ b/README.md @@ -117,11 +117,32 @@ from the strategy budget and returns a legal-direction fallback instead of null when its request budget expires. This is best-effort deadline handling, not a hard real-time guarantee under CPU saturation. +Version 1.2.0 fixes a search-depth regression from that larger reserve. Prism +looks one turn deeper in 3-snake rollouts when more than 250 ms remain; with +Zenith's 50 ms shorter budget that gate was never reached in production, so +Zenith always stopped at depth 2. Zenith now lowers the gate by its extra +reserve and deepens after the same elapsed time as Prism. + Run it with: ```sh SNAKE=ZenithBattleSnake_ClaudeOpus5_5 python main.py ``` +Show read-only win statistics per own snake version and per opponent from the +PostgreSQL gameplay database (`GAMEPLAY_DB_PG_DSN`): +```sh +just snake-stats --since-hours 24 +just snake-stats --snake-type ZenithBattleSnake --game-type duel --min-games 5 +just snake-stats --since-hours 48 --compare # opponents side by side per own version +just snake-stats --traces --since-hours 24 # move-trace latency per version and mode +``` + +`--traces` unpacks the newest `--trace-games` (default 300) finished replays and +reports p50/p90/p99 for strategy time, server overhead (handler minus +strategy), handler time, network/outside-handler time (engine latency minus +handler) and engine latency. `ext.timeouts` counts engine timeouts where our +handler had answered at least 100 ms before the deadline. + Benchmark Apex and Prism against sampled positions from a gameplay database: ```sh python scripts/benchmark_snakes_from_db.py \ diff --git a/justfile b/justfile index a764c31..9b482bc 100644 --- a/justfile +++ b/justfile @@ -90,6 +90,15 @@ db *args: command -v psql >/dev/null || { echo "Install the PostgreSQL client (psql) first." >&2; exit 1; } exec psql --dbname="$GAMEPLAY_DB_PG_DSN" "$@" +# Read-only win stats per own snake version and per opponent (extra args are passed through) +[positional-arguments] +snake-stats *args: + #!/usr/bin/env bash + set -euo pipefail + + : "${GAMEPLAY_DB_PG_DSN:?Set GAMEPLAY_DB_PG_DSN in .env or the environment}" + PYTHONPATH="{{justfile_directory()}}" uv run python "{{justfile_directory()}}/scripts/gameplay_snake_stats.py" "$@" + cleanup-gameplay-postgres older_than_days="30" execute="false" vacuum="false": #!/usr/bin/env bash set -euo pipefail diff --git a/scripts/gameplay_snake_stats.py b/scripts/gameplay_snake_stats.py new file mode 100644 index 0000000..64b8693 --- /dev/null +++ b/scripts/gameplay_snake_stats.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +"""Read-only win statistics per own snake version and per opponent snake. + +Reads the PostgreSQL gameplay database (``GAMEPLAY_DB_PG_DSN``). The session is +switched to read-only before any query runs, so it is safe against production. +""" +import argparse +import asyncio +import os +from dataclasses import dataclass + +@dataclass(frozen=True) +class StatsFilter: + since_hours: float | None = None + snake_type: str | None = None + snake_version: str | None = None + game_type: str | None = None + source: str | None = None + +def build_where(filters: StatsFilter, alias: str = "g") -> tuple[str, list]: + """Return a WHERE clause for finished games plus its positional arguments.""" + clauses = [f"{alias}.status = 'finished'"] + args: list = [] + + def add(sql: str, value) -> None: + args.append(value) + clauses.append(sql.format(alias=alias, n=len(args))) + + if filters.since_hours is not None: + add("{alias}.started_at > now() - make_interval(secs => ${n})", float(filters.since_hours) * 3600) + if filters.snake_type: + add("{alias}.your_snake_type = ${n}", filters.snake_type) + if filters.snake_version: + add("{alias}.your_snake_version = ${n}", filters.snake_version) + if filters.game_type: + add("{alias}.game_type = ${n}", filters.game_type) + if filters.source: + add("{alias}.source = ${n}", filters.source) + return "WHERE " + " AND ".join(clauses), args + +def own_snake_query(filters: StatsFilter) -> tuple[str, list]: + where, args = build_where(filters) + return f""" + SELECT g.your_snake_type AS snake, g.your_snake_version AS version, + g.game_type, count(*) AS games, sum(g.winner_you::int) AS wins, + round(avg(g.final_turn)) AS avg_turns, max(g.started_at) AS last_game + FROM games AS g + {where} + GROUP BY 1, 2, 3 + ORDER BY 1, 2, 3 + """, args + +def opponent_query(filters: StatsFilter, min_games: int) -> tuple[str, list]: + where, args = build_where(filters) + args.append(max(1, min_games)) + return f""" + SELECT s.snake_name AS opponent, g.your_snake_type AS snake, + g.your_snake_version AS version, + count(*) AS games, sum(g.winner_you::int) AS wins, + sum((g.winner_name = s.snake_name)::int) AS their_wins, + round(avg(g.final_turn)) AS avg_turns + FROM games AS g + JOIN game_snakes AS s ON s.game_id = g.game_id AND NOT s.is_you + {where} + GROUP BY 1, 2, 3 + HAVING count(*) >= ${len(args)} + ORDER BY count(*) DESC, 1 + """, args + +def win_rate(wins: int, games: int) -> str: + return f"{100.0 * wins / games:5.1f}%" if games else " -" + +def format_table(headers: list[str], rows: list[list]) -> str: + cells = [[str(value) for value in row] for row in rows] + widths = [max(len(h), *(len(row[i]) for row in cells)) if cells else len(h) for i, h in enumerate(headers)] + line = lambda values: " ".join(v.ljust(w) for v, w in zip(values, widths)).rstrip() + return "\n".join([line(headers), line(["-" * w for w in widths]), *(line(row) for row in cells)]) + +def own_rows(records: list[dict]) -> list[list]: + return [ + [r["snake"], r["version"], r["game_type"], r["games"], r["wins"], + win_rate(r["wins"], r["games"]), r["avg_turns"], + r["last_game"].strftime("%Y-%m-%d %H:%M") if r["last_game"] else "-"] + for r in records + ] + +def opponent_rows(records: list[dict]) -> list[list]: + return [ + [r["opponent"], f"{r['snake']} {r['version']}", r["games"], r["wins"], + win_rate(r["wins"], r["games"]), r["their_wins"], + win_rate(r["their_wins"], r["games"]), r["avg_turns"]] + for r in records + ] + +def pivot_rows(records: list[dict], min_games: int) -> tuple[list[str], list[list]]: + """One row per opponent, one ``wins/games win%`` column per own snake version.""" + versions = sorted({f"{r['snake']} {r['version']}" for r in records}) + table: dict[str, dict[str, tuple[int, int]]] = {} + for r in records: + table.setdefault(r["opponent"], {})[f"{r['snake']} {r['version']}"] = (r["wins"], r["games"]) + rows = [] + for opponent, cells in sorted(table.items(), key=lambda item: -sum(g for _, g in item[1].values())): + if len(cells) < 2 or any(games < min_games for _, games in cells.values()): + continue + rows.append([opponent, *( + f"{cells[v][0]}/{cells[v][1]} {win_rate(*cells[v]).strip()}" if v in cells else "-" + for v in versions + )]) + return ["opponent", *versions], rows + +async def _read_only_connection(dsn: str): + import asyncpg # noqa: PLC0415 + + connection = await asyncpg.connect(dsn=dsn) + await connection.execute("SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY") + return connection + +async def collect_traces(dsn: str, filters: StatsFilter, max_games: int): + from scripts.move_trace_stats import load_groups # noqa: PLC0415 + + connection = await _read_only_connection(dsn) + try: + where, args = build_where(filters) + return await load_groups(connection, where, args, max_games) + finally: + await connection.close() + +async def collect(dsn: str, filters: StatsFilter, min_games: int, limit: int) -> tuple[list[dict], list[dict]]: + connection = await _read_only_connection(dsn) + try: + own_sql, own_args = own_snake_query(filters) + opp_sql, opp_args = opponent_query(filters, min_games) + own = [dict(row) for row in await connection.fetch(own_sql, *own_args)] + opponents = [dict(row) for row in await connection.fetch(opp_sql, *opp_args)] + return own, opponents if limit <= 0 else opponents[:limit] + finally: + await connection.close() + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dsn", default=os.environ.get("GAMEPLAY_DB_PG_DSN")) + parser.add_argument("--since-hours", type=float, default=None, help="Only games started within N hours") + parser.add_argument("--snake-type", help="Own snake type, e.g. ZenithBattleSnake") + parser.add_argument("--snake-version", help="Own snake version, e.g. 1.2.0") + parser.add_argument("--game-type", help="duel, standard, constrictor, ...") + parser.add_argument("--source", help="ladder, arena, custom, ...") + parser.add_argument("--min-games", type=int, default=3, help="Hide opponents with fewer games") + parser.add_argument("--limit", type=int, default=40, help="Maximum opponent rows (0 = all)") + parser.add_argument("--compare", action="store_true", help="Side-by-side win rates of own snake versions per opponent") + parser.add_argument("--traces", action="store_true", help="Move-trace latency stats per snake version and mode") + parser.add_argument("--trace-games", type=int, default=300, help="Newest finished games to unpack for --traces") + args = parser.parse_args() + if not args.dsn: + raise SystemExit("Set GAMEPLAY_DB_PG_DSN or pass --dsn") + + filters = StatsFilter(args.since_hours, args.snake_type, args.snake_version, args.game_type, args.source) + if args.traces: + from scripts.move_trace_stats import summary_rows # noqa: PLC0415 + + groups, games = asyncio.run(collect_traces(args.dsn, filters, args.trace_games)) + print(f"Move traces from {games} newest finished games (ms; ext.timeouts = engine timeout while our handler answered >=100 ms early)") + print(format_table(*summary_rows(groups))) + return + limit = 0 if args.compare else args.limit + own, opponents = asyncio.run(collect(args.dsn, filters, args.min_games, limit)) + print("Own snakes") + print(format_table(["snake", "version", "mode", "games", "wins", "win%", "turns", "last game (UTC)"], own_rows(own))) + if args.compare: + headers, rows = pivot_rows(opponents, args.min_games) + print("\nOpponents faced by several own snakes (our wins/games win%)") + print(format_table(headers, rows[:args.limit] if args.limit > 0 else rows)) + return + print("\nOpponents (win% = our win rate against them)") + print(format_table(["opponent", "our snake", "games", "wins", "win%", "their wins", "their%", "turns"], opponent_rows(opponents))) + +if __name__ == "__main__": + main() diff --git a/scripts/move_trace_stats.py b/scripts/move_trace_stats.py new file mode 100644 index 0000000..7595f52 --- /dev/null +++ b/scripts/move_trace_stats.py @@ -0,0 +1,110 @@ +"""Aggregate stored move traces into per-snake latency statistics. + +Each move stores ``my_thinking.move_trace`` with server-side stage timings. The +engine-measured latency for turn ``T`` arrives with the board of turn ``T + 1``, +so both are paired here to split total latency into strategy, server overhead +and network/outside-handler time. +""" +from collections import defaultdict +from dataclasses import dataclass, field + +from server.database.compact.replay_blob import unpack_replay +from server.database.compact.thinking import MASK_KEY + +STAGES = ("strategy_ms", "overhead_ms", "handler_ms", "network_ms", "engine_ms") + +@dataclass +class TraceGroup: + moves: int = 0 + timeouts: int = 0 + timeouts_fast_handler: int = 0 + values: dict[str, list[float]] = field(default_factory=lambda: defaultdict(list)) + +def _thinking_dict(thinking: object) -> dict: + if not isinstance(thinking, dict): + return {} + if thinking.get(MASK_KEY) == 0 and isinstance(thinking.get("_raw"), dict): + return thinking["_raw"] + return thinking + +def _number(value: object) -> float | None: + return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None + +def _latency(value: object) -> float | None: + # Compact storage keeps latency as a small int; "" (no response) is negative. + number = _number(value) + return number if number is not None and number >= 0 else None + +def add_game(groups: dict[tuple, TraceGroup], game: dict, you_id: str, turns: list, states: list) -> None: + """Add every traced move of one game to ``groups`` keyed by snake/version/mode.""" + latency_by_turn = {s.turn: _latency(s.latency) for s in states if s.snake_id == you_id} + for record in turns: + thinking = _thinking_dict(record.thinking) + trace = thinking.get("move_trace") + if not isinstance(trace, dict): + continue + handler = _number(trace.get("handler_ms")) + strategy = _number(trace.get("strategy_ms")) + if handler is None: + continue + key = (game["your_snake_type"], game["your_snake_version"], thinking.get("mode") or "fallback") + group = groups.setdefault(key, TraceGroup()) + group.moves += 1 + group.values["handler_ms"].append(handler) + if strategy is not None: + group.values["strategy_ms"].append(strategy) + group.values["overhead_ms"].append(max(0.0, handler - strategy)) + engine = latency_by_turn.get(record.turn + 1) + if engine is None: + continue + group.values["engine_ms"].append(engine) + group.values["network_ms"].append(max(0.0, engine - handler)) + timeout_ms = (_number(trace.get("budget_ms")) or 450.0) + 50.0 + if engine >= timeout_ms: + group.timeouts += 1 + if handler < timeout_ms - 100: + group.timeouts_fast_handler += 1 + +def percentile(values: list[float], quantile: float) -> float | None: + if not values: + return None + ordered = sorted(values) + return ordered[min(len(ordered) - 1, int(len(ordered) * quantile))] + +def _fmt(value: float | None) -> str: + return "-" if value is None else f"{value:.0f}" + +def summary_rows(groups: dict[tuple, TraceGroup]) -> tuple[list[str], list[list]]: + headers = ["snake", "version", "mode", "moves", "timeouts", "ext.timeouts"] + for stage in STAGES: + headers.append(f"{stage.removesuffix('_ms')} p50/p90/p99") + rows = [] + for (snake, version, mode), group in sorted(groups.items(), key=lambda item: tuple(str(v) for v in item[0])): + row = [ + snake, version, mode, group.moves, + f"{group.timeouts} ({100.0 * group.timeouts / group.moves:.1f}%)" if group.moves else "0", + group.timeouts_fast_handler, + ] + for stage in STAGES: + values = group.values.get(stage, []) + row.append("/".join(_fmt(percentile(values, q)) for q in (0.5, 0.9, 0.99))) + rows.append(row) + return headers, rows + +async def load_groups(connection, where: str, args: list, max_games: int) -> tuple[dict[tuple, TraceGroup], int]: + """Read the newest finished replays matching ``where``; never writes.""" + args = [*args, max(1, max_games)] + rows = await connection.fetch(f""" + SELECT g.game_id, g.your_snake_id, g.your_snake_type, g.your_snake_version, + r.format, r.data + FROM games AS g + JOIN game_replays AS r ON r.gid = g.gid + {where} + ORDER BY g.started_at DESC + LIMIT ${len(args)} + """, *args) + groups: dict[tuple, TraceGroup] = {} + for row in rows: + turns, states = unpack_replay(row["data"], row["format"]) + add_game(groups, dict(row), row["your_snake_id"], turns, states) + return groups, len(rows) diff --git a/snakes/__init__.py b/snakes/__init__.py index 78d6ccd..f42abaf 100644 --- a/snakes/__init__.py +++ b/snakes/__init__.py @@ -13,7 +13,7 @@ SNAKE_REGISTRATIONS = { "snakes.strategies.prism", "1.5.0" ), "ZenithBattleSnake_ClaudeOpus5_5": SnakeRegistration( - "snakes.strategies.zenith", "1.1.0" + "snakes.strategies.zenith", "1.2.0" ), "DummSnake": SnakeRegistration("snakes.legacy.DummSnake", "1.0.0"), "LogicSnake": SnakeRegistration("snakes.legacy.LogicSnake", "1.1.0"), diff --git a/snakes/engine/survival.py b/snakes/engine/survival.py index 7904b93..4531fd0 100644 --- a/snakes/engine/survival.py +++ b/snakes/engine/survival.py @@ -4,6 +4,10 @@ from time import perf_counter from snakes.engine.survival_search import CompactSurvivalSearch class BitboardSurvivalMixin: + # Remaining budget above which rollouts against at most two enemies may look + # one turn deeper. Snakes with a larger timeout reserve lower this threshold. + _rollout_deepen_ms = 250.0 + def _future_rollout_bonus( self, move: str, safe_moves: dict, my_body: list, other_snakes: list, food_set: set, is_constrictor: bool, width: int, height: int, @@ -36,7 +40,7 @@ class BitboardSurvivalMixin: depth = min(self._planning_depth, 2 if len(other_snakes) > 1 else 3) if remaining < 90: depth = min(depth, 2) - elif remaining > 250 and len(other_snakes) <= 2: + elif remaining > self._rollout_deepen_ms and len(other_snakes) <= 2: depth = min(4, depth + 1) raw = self._survival_search_context.search_selected( my_body=my_body, diff --git a/snakes/strategies/zenith.py b/snakes/strategies/zenith.py index f2e1a12..2ab5b1d 100644 --- a/snakes/strategies/zenith.py +++ b/snakes/strategies/zenith.py @@ -1,4 +1,4 @@ -"""ZenithBattleSnake_ClaudeOpus5_5 v1.1.0 +"""ZenithBattleSnake_ClaudeOpus5_5 v1.2.0 Built on PrismBattleSnake_GPT_5_6_Sol v1.5.0. Prism's bitboard engine, duel search, and adversarial rollouts are inherited unchanged. @@ -15,6 +15,11 @@ Z2: Proven-loss veto — a candidate whose completed search proves a loss is replaced by the best non-losing candidate. Z3: Proven-win override — a candidate whose completed search proves a win is played even if the heuristic preferred another move. +Z4: Elapsed-time rollout gate — Zenith keeps a larger timeout reserve than + Prism. The inherited multiplayer "deepen when > 250 ms remain" gate is + lowered by that extra reserve, so 3-snake rollouts deepen after the same + elapsed time as Prism instead of never reaching depth 3. The duel gate is + left alone: the verdict pass already spends the duel budget. """ from time import perf_counter @@ -27,18 +32,22 @@ _PROVEN_WIN = BitboardDuelSearch.WIN / 2 _PROVEN_LOSS = BitboardDuelSearch.LOSS / 2 # Reserve kept for Apex's history bookkeeping after the verdict pass. _VERDICT_RESERVE_MS = 25.0 +_ZENITH_TIMEOUT_BUFFER_MS = 180 class ZenithBattleSnake_ClaudeOpus5_5(ZenithSurvivalMixin, PrismBattleSnake_GPT_5_6_Sol): - VERSION = "1.1.0" + VERSION = "1.2.0" def __init__(self) -> None: super().__init__() self.name = "ZenithBattleSnake" self.version = self.VERSION + # Z4: deepen after the same elapsed time as the inherited Prism gates. + extra_reserve = self._get_timeout_buffer_ms() - super()._get_timeout_buffer_ms() + self._rollout_deepen_ms = max(120.0, self._rollout_deepen_ms - extra_reserve) def _get_timeout_buffer_ms(self): # Leave room for executor scheduling, history, JSON and network delivery. - return max(180, super()._get_timeout_buffer_ms()) + return max(_ZENITH_TIMEOUT_BUFFER_MS, super()._get_timeout_buffer_ms()) def _choose_duel_move( self, safe_moves, my_body, my_len, my_health, other_snakes, food_set, diff --git a/tests/scripts/test_gameplay_snake_stats.py b/tests/scripts/test_gameplay_snake_stats.py new file mode 100644 index 0000000..3887af7 --- /dev/null +++ b/tests/scripts/test_gameplay_snake_stats.py @@ -0,0 +1,46 @@ +import unittest + +from scripts.gameplay_snake_stats import ( + StatsFilter, build_where, format_table, opponent_query, pivot_rows, win_rate, +) + +class TestGameplaySnakeStats(unittest.TestCase): + + def test_filters_use_positional_arguments(self): + where, args = build_where(StatsFilter(since_hours=2, snake_type="ZenithBattleSnake", source="ladder")) + + self.assertIn("g.status = 'finished'", where) + self.assertIn("make_interval(secs => $1)", where) + self.assertIn("g.your_snake_type = $2", where) + self.assertIn("g.source = $3", where) + self.assertEqual(args, [7200.0, "ZenithBattleSnake", "ladder"]) + + def test_opponent_query_appends_min_games_last(self): + sql, args = opponent_query(StatsFilter(game_type="duel"), min_games=5) + + self.assertIn("HAVING count(*) >= $2", sql) + self.assertEqual(args, ["duel", 5]) + + def test_win_rate_handles_zero_games(self): + self.assertEqual(win_rate(1, 4), " 25.0%") + self.assertEqual(win_rate(0, 0), " -") + + def test_pivot_keeps_opponents_faced_by_every_listed_version(self): + records = [ + {"opponent": "A", "snake": "Prism", "version": "1.5.0", "wins": 2, "games": 10}, + {"opponent": "A", "snake": "Zenith", "version": "1.2.0", "wins": 3, "games": 6}, + {"opponent": "B", "snake": "Prism", "version": "1.5.0", "wins": 1, "games": 9}, + ] + + headers, rows = pivot_rows(records, min_games=3) + + self.assertEqual(headers, ["opponent", "Prism 1.5.0", "Zenith 1.2.0"]) + self.assertEqual(rows, [["A", "2/10 20.0%", "3/6 50.0%"]]) + + def test_format_table_aligns_columns(self): + table = format_table(["name", "n"], [["long-name", 1]]) + + self.assertEqual(table.splitlines(), ["name n", "--------- -", "long-name 1"]) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/scripts/test_move_trace_stats.py b/tests/scripts/test_move_trace_stats.py new file mode 100644 index 0000000..a849c9d --- /dev/null +++ b/tests/scripts/test_move_trace_stats.py @@ -0,0 +1,61 @@ +import unittest + +from scripts.move_trace_stats import add_game, percentile, summary_rows +from server.database.compact.records import SnakeState, TurnRecord + +GAME = {"your_snake_type": "ZenithBattleSnake", "your_snake_version": "1.2.0"} + +def _turn(turn: int, thinking: object) -> TurnRecord: + return TurnRecord(turn=turn, observed_at=None, move="up", thinking=thinking, food=None, hazards=None) + +def _state(snake_id: str, turn: int, latency: int | None) -> SnakeState: + return SnakeState(snake_id=snake_id, turn=turn, health=100, latency=latency, body=[]) + +def _trace(handler: float, strategy: float) -> dict: + return {"handler_ms": handler, "strategy_ms": strategy, "budget_ms": 450.0} + +class TestMoveTraceStats(unittest.TestCase): + + def test_pairs_engine_latency_of_next_turn(self): + groups = {} + turns = [ + _turn(1, {"mode": "duel", "move_trace": _trace(200.0, 195.0)}), + _turn(2, {"mode": "duel", "move_trace": _trace(100.0, 90.0)}), + ] + states = [_state("you", 2, 260), _state("enemy", 2, 499), _state("you", 3, 500)] + + add_game(groups, GAME, "you", turns, states) + + group = groups[("ZenithBattleSnake", "1.2.0", "duel")] + self.assertEqual(group.moves, 2) + self.assertEqual(group.values["overhead_ms"], [5.0, 10.0]) + self.assertEqual(group.values["engine_ms"], [260.0, 500.0]) + self.assertEqual(group.values["network_ms"], [60.0, 400.0]) + self.assertEqual((group.timeouts, group.timeouts_fast_handler), (1, 1)) + + def test_unwraps_raw_thinking_and_labels_missing_mode(self): + groups = {} + turns = [_turn(1, {"_derived": 0, "_raw": {"move_trace": _trace(5.0, 1.0)}}), _turn(2, "text")] + + add_game(groups, GAME, "you", turns, [_state("you", 2, -32768)]) + + group = groups[("ZenithBattleSnake", "1.2.0", "fallback")] + self.assertEqual(group.moves, 1) + self.assertEqual(group.values["engine_ms"], []) + + def test_summary_formats_percentiles(self): + groups = {} + add_game(groups, GAME, "you", [_turn(1, {"mode": "multi", "move_trace": _trace(20.0, 18.0)})], []) + + headers, rows = summary_rows(groups) + + self.assertEqual(headers[:4], ["snake", "version", "mode", "moves"]) + self.assertEqual(rows[0][:6], ["ZenithBattleSnake", "1.2.0", "multi", 1, "0 (0.0%)", 0]) + self.assertEqual(rows[0][6], "18/18/18") + self.assertEqual(rows[0][-1], "-/-/-") + + def test_percentile_of_empty_list(self): + self.assertIsNone(percentile([], 0.5)) + +if __name__ == "__main__": + unittest.main() diff --git a/tests/snakes/test_ZenithBattleSnake_ClaudeOpus5_5.py b/tests/snakes/test_ZenithBattleSnake_ClaudeOpus5_5.py index c35b1d2..2307b98 100644 --- a/tests/snakes/test_ZenithBattleSnake_ClaudeOpus5_5.py +++ b/tests/snakes/test_ZenithBattleSnake_ClaudeOpus5_5.py @@ -1,5 +1,8 @@ import unittest from time import perf_counter +from unittest import mock + +from snakes.engine.survival_search import CompactSurvivalSearch from snakes import SnakeBuilder, get_snake_version from snakes.strategies.prism import PrismBattleSnake_GPT_5_6_Sol @@ -14,8 +17,8 @@ class TestZenithRegistration(unittest.TestCase): snake = ZenithBattleSnake_ClaudeOpus5_5() self.assertEqual(snake.name, "ZenithBattleSnake") - self.assertEqual(snake.version, "1.1.0") - self.assertEqual(get_snake_version("ZenithBattleSnake_ClaudeOpus5_5"), "1.1.0") + self.assertEqual(snake.version, "1.2.0") + self.assertEqual(get_snake_version("ZenithBattleSnake_ClaudeOpus5_5"), "1.2.0") self.assertIsInstance(snake, PrismBattleSnake_GPT_5_6_Sol) self.assertIsInstance( SnakeBuilder.build("ZenithBattleSnake_ClaudeOpus5_5"), @@ -80,5 +83,56 @@ class TestZenithVerdicts(unittest.TestCase): self.assertEqual(verdicts, {}) +def _rollout_depth(snake, remaining_ms: float) -> int: + """Depth requested by a 3-snake rollout with ``remaining_ms`` left.""" + snake._get_bb(11, 11) + snake._survival_search_context = None + enemies = [ + {"body": [{"x": 8, "y": 8}, {"x": 8, "y": 9}, {"x": 8, "y": 10}]}, + {"body": [{"x": 2, "y": 8}, {"x": 2, "y": 9}, {"x": 2, "y": 10}]}, + ] + with mock.patch.object(CompactSurvivalSearch, "search_selected", return_value=0.0) as search: + snake._future_rollout_bonus( + move="up", safe_moves={"up": {"x": 5, "y": 3}}, + my_body=[{"x": 5, "y": 2}, {"x": 5, "y": 1}, {"x": 5, "y": 0}], + other_snakes=enemies, food_set=set(), is_constrictor=False, + width=11, height=11, enemy_can_grow={}, + deadline=perf_counter() + remaining_ms / 1000, + ) + return search.call_args.kwargs["depth"] + +class TestZenithRolloutDepth(unittest.TestCase): + + def test_gate_is_lowered_by_the_extra_timeout_reserve(self): + zenith = ZenithBattleSnake_ClaudeOpus5_5() + prism = PrismBattleSnake_GPT_5_6_Sol() + extra = zenith._get_timeout_buffer_ms() - prism._get_timeout_buffer_ms() + + self.assertEqual(zenith._rollout_deepen_ms, prism._rollout_deepen_ms - extra) + + def test_three_snake_rollout_deepens_after_same_elapsed_time_as_prism(self): + # Zenith's budget is 50 ms shorter, so 220 ms left equals 270 ms for Prism. + self.assertEqual(_rollout_depth(ZenithBattleSnake_ClaudeOpus5_5(), 220), 3) + self.assertEqual(_rollout_depth(PrismBattleSnake_GPT_5_6_Sol(), 270), 3) + self.assertEqual(_rollout_depth(PrismBattleSnake_GPT_5_6_Sol(), 220), 2) + + def test_low_budget_still_caps_rollout_depth(self): + self.assertEqual(_rollout_depth(ZenithBattleSnake_ClaudeOpus5_5(), 150), 2) + +class TestZenithVerdictBudget(unittest.TestCase): + + def test_verdict_pass_skips_when_only_reserve_remains(self): + snake = ZenithBattleSnake_ClaudeOpus5_5() + + verdicts = snake._duel_verdicts( + safe_moves={"up": {"x": 0, "y": 2}, "right": {"x": 1, "y": 1}}, + my_body=[{"x": 0, "y": 1}], enemy={"body": [{"x": 4, "y": 4}]}, + my_health=100, food_set=set(), hazard_set=set(), hazard_damage=15, + hazard_count={}, previous_hazard_set=set(), width=5, height=5, + deadline=perf_counter() + 0.020, + ) + + self.assertEqual(verdicts, {}) + if __name__ == "__main__": unittest.main()