fix: resolve duel roots and recover legacy snake data

- Resolve selected moves and enemy replies on the same simulated turn.
- Add an Apex candidate hook and bump the Prism snake to version 1.1.0.
- Rebuild benchmark states from normalized turn data when snapshots are empty.
- Synthesize missing game snake identities during legacy database migration.
- Add regression coverage for duel timing and partial legacy schemas.
This commit is contained in:
2026-08-01 19:11:32 +02:00
parent c646392b84
commit 6643eb35af
8 changed files with 329 additions and 39 deletions
+47 -11
View File
@@ -31,7 +31,8 @@ def load_states(db_path: str, samples: int, stride: int) -> list[tuple[dict, dic
states: list[tuple[dict, dict]] = []
next_id = max(1, max_id - (samples - 1) * stride)
query = """
SELECT t.board_state_json, t.you_json, g.your_snake_id,
SELECT t.id, t.board_state_json, t.you_json, t.food_json, t.hazards_json,
g.your_snake_id, g.your_snake_name, g.width, g.height,
g.game_id, g.source, g.map_name,
g.ruleset_name, g.ruleset_version, t.turn
FROM turns AS t
@@ -40,30 +41,65 @@ def load_states(db_path: str, samples: int, stride: int) -> list[tuple[dict, dic
ORDER BY t.id
LIMIT 1
"""
snake_query = """
SELECT st.snake_id, COALESCE(gs.snake_name, st.snake_name),
st.health, st.length, st.head_x, st.head_y, st.body_json,
COALESCE(gs.customizations_json, '{}')
FROM snake_turns AS st
LEFT JOIN game_snakes AS gs
ON gs.game_id = st.game_id AND gs.snake_id = st.snake_id
WHERE st.game_id = ? AND st.turn = ?
ORDER BY st.id
"""
while len(states) < samples and next_id <= max_id:
row = connection.execute(query, (next_id,)).fetchone()
if row is None:
break
board = json.loads(row[0])
you = json.loads(row[1])
board = json.loads(row[1])
you = json.loads(row[2])
if not board.get("snakes"):
snakes = []
for snake_row in connection.execute(snake_query, (row[9], row[14])):
snake_id = snake_row[0]
snake_name = snake_row[1] or (row[6] if snake_id == row[5] else snake_id)
body = json.loads(snake_row[6])
snakes.append({
"id": snake_id,
"name": snake_name,
"health": snake_row[2],
"length": snake_row[3],
"head": {"x": snake_row[4], "y": snake_row[5]},
"body": body,
"customizations": json.loads(snake_row[7]),
})
board = {
"width": row[7],
"height": row[8],
"food": json.loads(row[3]),
"hazards": json.loads(row[4]),
"snakes": snakes,
}
if not you:
you = next(
(snake for snake in board.get("snakes", []) if snake.get("id") == row[2]),
(snake for snake in board.get("snakes", []) if snake.get("id") == row[5]),
{},
)
if not you or not board.get("snakes"):
next_id = int(row[0]) + stride
continue
metadata = {
"game_id": row[3],
"source": row[4] or "custom",
"map": row[5] or "standard",
"game_id": row[9],
"source": row[10] or "custom",
"map": row[11] or "standard",
"ruleset": {
"name": row[6] or "standard",
"version": row[7] or "v1.0.0",
"name": row[12] or "standard",
"version": row[13] or "v1.0.0",
"settings": {},
},
"turn": int(row[8]),
"turn": int(row[14]),
}
states.append((board, {"you": you, **metadata}))
next_id += stride
next_id = int(row[0]) + stride
connection.close()
return states
+23 -15
View File
@@ -143,6 +143,12 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection,
has_game_snakes = source.execute("""
SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'game_snakes'
""").fetchone() is not None
sql = """
INSERT OR IGNORE INTO game_snakes (
game_id, snake_id, snake_name, is_you, customizations_json
) VALUES (?, ?, ?, ?, ?)
"""
count = 0
if has_game_snakes:
columns = object_columns(source, "game_snakes")
customizations = (
@@ -153,24 +159,26 @@ def copy_game_snakes(source:sqlite3.Connection, destination:sqlite3.Connection,
SELECT game_id, snake_id, snake_name, is_you, {customizations}
FROM game_snakes ORDER BY game_id, snake_id
""")
else:
cursor = source.execute("""
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you),
'{}' AS customizations_json
FROM snake_turns
GROUP BY game_id, snake_id
ORDER BY game_id, snake_id
""")
sql = """
INSERT INTO game_snakes (
game_id, snake_id, snake_name, is_you, customizations_json
) VALUES (?, ?, ?, ?, ?)
"""
count = 0
while rows := cursor.fetchmany(batch_size):
retained_rows = [tuple(row) for row in rows if row[0] in retained_ids]
before = destination.total_changes
destination.executemany(sql, retained_rows)
count += destination.total_changes - before
# Older databases can contain an empty or only partially populated
# game_snakes table. Always synthesize missing identities from snake_turns.
cursor = source.execute("""
SELECT game_id, snake_id, MAX(snake_name), MAX(is_you),
'{}' AS customizations_json
FROM snake_turns
GROUP BY game_id, snake_id
ORDER BY game_id, snake_id
""")
while rows := cursor.fetchmany(batch_size):
retained_rows = [tuple(row) for row in rows if row[0] in retained_ids]
before = destination.total_changes
destination.executemany(sql, retained_rows)
count += len(retained_rows)
count += destination.total_changes - before
return count
def decode_json(value:str|None, fallback):
+53 -8
View File
@@ -468,15 +468,11 @@ class ApexBattleSnake(TemplateSnake):
if self._time_exceeded(deadline):
break
pos = safe_moves[m]
ate = (pos["x"], pos["y"]) in food_set
fb = self._future_body(my_body, pos, ate, False)
nmy_h = 100 if ate else my_health - 1
if (pos["x"], pos["y"]) in hazard_set and not ate:
nmy_h -= hazard_damage * hazard_count.get((pos["x"], pos["y"]), 1)
mm_val, depth_done = self._minimax_sim_id(
my_body=fb, enemy_body=enemy["body"],
mm_val, depth_done = self._minimax_candidate_id(
my_body=my_body, enemy_body=enemy["body"],
my_target=(pos["x"], pos["y"]),
food_set=food_set, hazard_set=hazard_set,
my_health=nmy_h, enemy_health=enemy_health,
my_health=my_health, enemy_health=enemy_health,
hazard_damage=hazard_damage, hazard_count=hazard_count,
width=width, height=height,
max_depth=self._planning_depth,
@@ -893,6 +889,55 @@ class ApexBattleSnake(TemplateSnake):
# ── A1: Iterative deepening minimax ──────────────────────────────────────────
def _minimax_candidate_id(
self,
my_body: list,
enemy_body: list,
my_target: tuple[int, int],
food_set: set,
hazard_set: set,
my_health: int,
enemy_health: int,
hazard_damage: int,
hazard_count: dict,
width: int,
height: int,
max_depth: int,
alpha: float,
beta: float,
deadline: float | None,
previous_hazard_set: set | None = None,
) -> tuple[float, int]:
"""Evaluate a selected move before continuing the legacy duel search.
Optimized subclasses can override this hook to resolve our selected move
and the opponent's reply simultaneously at the search root.
"""
pos = {"x": my_target[0], "y": my_target[1]}
ate = my_target in food_set
future_body = self._future_body(my_body, pos, ate, False)
future_health = 100 if ate else my_health - 1
effective_previous = previous_hazard_set if previous_hazard_set is not None else hazard_set
if my_target in hazard_set and my_target in effective_previous and not ate:
future_health -= hazard_damage * hazard_count.get(my_target, 1)
return self._minimax_sim_id(
my_body=future_body,
enemy_body=enemy_body,
food_set=food_set,
hazard_set=hazard_set,
my_health=future_health,
enemy_health=enemy_health,
hazard_damage=hazard_damage,
hazard_count=hazard_count,
width=width,
height=height,
max_depth=max_depth,
alpha=alpha,
beta=beta,
deadline=deadline,
previous_hazard_set=previous_hazard_set,
)
def _minimax_sim_id(
self,
my_body: list,
+25 -2
View File
@@ -1,4 +1,4 @@
"""PrismBattleSnake_GPT_5_6_Sol v1.0.1
"""PrismBattleSnake_GPT_5_6_Sol v1.1.0
Built on ApexBattleSnake v1.0.0. All strategic logic is inherited.
Performance improvement: all spatial primitives (flood fill, territory,
@@ -25,6 +25,7 @@ Key speedups:
S11: _future_survival_tree inlines legal-move check with bitboard ops.
S12: Duel minimax uses tuple bodies and bitboard move generation.
S13: Iterative deepening reuses a transposition table and move-order hints.
S14: Candidate duel moves and enemy replies resolve on the same root turn.
"""
from __future__ import annotations
@@ -41,7 +42,7 @@ _DIR_DELTAS = ((0, 1), (0, -1), (-1, 0), (1, 0))
_DIR_NAMES = ("up", "down", "left", "right")
class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake):
VERSION = "1.0.1"
VERSION = "1.1.0"
def __init__(self) -> None:
super().__init__()
@@ -268,6 +269,28 @@ class PrismBattleSnake_GPT_5_6_Sol(ApexBattleSnake):
deadline=deadline,
)
def _minimax_candidate_id(
self, my_body: list, enemy_body: list, my_target: tuple[int, int],
food_set: set, hazard_set: set,
my_health: int, enemy_health: int, hazard_damage: int, hazard_count: dict,
width: int, height: int, max_depth: int, alpha: float, beta: float,
deadline: float | None, previous_hazard_set: set | None = None,
) -> tuple[float, int]:
"""Resolve our selected move and every enemy reply simultaneously."""
search = self._new_duel_search(
food_set, hazard_set, hazard_count, hazard_damage,
width, height, deadline,
)
return search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=my_target,
my_health=my_health,
enemy_health=enemy_health,
max_depth=max_depth,
previous_hazards=previous_hazard_set if previous_hazard_set is not None else hazard_set,
)
def _minimax_sim_id(
self, my_body: list, enemy_body: list, food_set: set, hazard_set: set,
my_health: int, enemy_health: int, hazard_damage: int, hazard_count: dict,
+1 -1
View File
@@ -11,7 +11,7 @@ SNAKE_REGISTRY = {
"UltimateBattleSnake": "4.5.0",
"ApexBattleSnake": "1.0.0",
"SupremeBattleSnake_ClaudeOpus4_6": "1.0.0",
"PrismBattleSnake_GPT_5_6_Sol": "1.0.1",
"PrismBattleSnake_GPT_5_6_Sol": "1.1.0",
}
DEFAULT_SNAKE_CONFIG = {
+81
View File
@@ -87,6 +87,47 @@ class BitboardDuelSearch:
return result, completed_depth
def search_candidate(
self,
my_body: list[dict],
enemy_body: list[dict],
my_target: tuple[int, int],
my_health: int,
enemy_health: int,
max_depth: int,
previous_hazards: Iterable[tuple[int, int]],
) -> tuple[float, int]:
"""Evaluate one selected move against every simultaneous enemy reply.
``max_depth`` counts the selected root turn, so a completed depth of one
means all opponent replies to that move were resolved.
"""
state = DuelState(
my_body=self.body_from_dicts(my_body),
enemy_body=self.body_from_dicts(enemy_body),
food_bits=self.food_bits,
my_health=my_health,
enemy_health=enemy_health,
previous_hazard_bits=self.board.set_to_bits(set(previous_hazards)),
)
target_idx = self.board.idx(my_target[0], my_target[1])
if not self.board.neighbors_of(state.my_body[0]) & (1 << target_idx):
return self.LOSS, 0
result = self._evaluate(state)
completed_depth = 0
for depth in range(1, max_depth + 1):
if self._out_of_time(5.0):
break
value, completed = self._search_selected_move(
state, target_idx, depth, -float("inf"), float("inf")
)
if not completed:
break
result = value
completed_depth = depth
return result, completed_depth
def search_depth(
self,
my_body: list[dict],
@@ -107,6 +148,46 @@ class BitboardDuelSearch:
value, _ = self._search(state, depth, -float("inf"), float("inf"))
return value
def _search_selected_move(
self,
state: DuelState,
my_target: int,
depth: int,
alpha: float,
beta: float,
) -> tuple[float, bool]:
"""Resolve the selected root move with the opponent on the same turn."""
self.nodes += 1
if self._out_of_time():
return self._evaluate(state), False
enemy_moves = self._candidate_targets(state.enemy_body)
if not enemy_moves:
return self.WIN + depth, True
enemy_moves = self._ordered_moves(enemy_moves, state, depth, False)
worst = float("inf")
for enemy_target in enemy_moves:
if self._out_of_time():
return (worst if worst != float("inf") else self._evaluate(state)), False
child, terminal = self._advance(state, my_target, enemy_target)
if terminal is not None:
value = terminal
completed = True
elif depth <= 1:
value = self._evaluate(child)
completed = True
else:
value, completed = self._search(child, depth - 1, alpha, beta)
if not completed:
return (worst if worst != float("inf") else value), False
worst = min(worst, value)
beta = min(beta, worst)
if beta <= alpha:
break
return worst, True
def _search(self, state: DuelState, depth: int, alpha: float, beta: float) -> tuple[float, bool]:
self.nodes += 1
if self._out_of_time():
@@ -51,8 +51,8 @@ class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase):
snake = PrismBattleSnake_GPT_5_6_Sol()
self.assertEqual(snake.name, "PrismBattleSnake")
self.assertEqual(snake.version, "1.0.1")
self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.0.1")
self.assertEqual(snake.version, "1.1.0")
self.assertEqual(get_snake_version("PrismBattleSnake_GPT_5_6_Sol"), "1.1.0")
self.assertIsInstance(SnakeBuilder.build("PrismBattleSnake_GPT_5_6_Sol"), PrismBattleSnake_GPT_5_6_Sol)
def test_bitboard_primitives_match_apex(self):
@@ -100,6 +100,50 @@ class TestPrismBattleSnake_GPT_5_6_Sol(unittest.TestCase):
self.assertGreater(value, 0)
def test_candidate_search_resolves_enemy_reply_on_the_same_turn(self):
board = BitBoard(3, 3)
search = BitboardDuelSearch(
board=board, food=set(), hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
my_body = [{"x": 0, "y": 1}, {"x": 0, "y": 0}]
enemy_body = [{"x": 2, "y": 1}, {"x": 2, "y": 0}]
value, depth = search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=(1, 1),
my_health=100,
enemy_health=100,
max_depth=1,
previous_hazards=set(),
)
self.assertEqual(value, -500.0)
self.assertEqual(depth, 1)
def test_candidate_search_does_not_advance_our_snake_twice_at_root(self):
board = BitBoard(4, 1)
search = BitboardDuelSearch(
board=board, food=set(), hazards=set(), hazard_count={},
hazard_damage=15, deadline=None,
)
my_body = [{"x": 0, "y": 0}]
enemy_body = [{"x": 3, "y": 0}]
value, depth = search.search_candidate(
my_body=my_body,
enemy_body=enemy_body,
my_target=(1, 0),
my_health=100,
enemy_health=100,
max_depth=1,
previous_hazards=set(),
)
self.assertEqual(value, 0.0)
self.assertEqual(depth, 1)
def test_duel_search_keeps_tail_blocked_when_its_snake_eats(self):
board = BitBoard(3, 3)
search = BitboardDuelSearch(
+53
View File
@@ -47,6 +47,54 @@ class TestMigrateGameplayDatabase(unittest.TestCase):
'{"color":"#663399","head":"ferret","tail":"swirl"}',
))
def test_copy_game_snakes_synthesizes_rows_when_table_is_empty(self):
source = sqlite3.connect(":memory:")
destination = sqlite3.connect(":memory:")
try:
source.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,
customizations_json TEXT NOT NULL DEFAULT '{}'
)
""")
source.execute("""
CREATE TABLE snake_turns (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER
)
""")
source.executemany(
"INSERT INTO snake_turns VALUES (?, ?, ?, ?)",
[
("game-1", "snake-1", "PrismBattleSnake", 1),
("game-1", "snake-1", "PrismBattleSnake", 1),
("game-1", "snake-2", "Enemy", 0),
],
)
destination.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,
customizations_json TEXT NOT NULL DEFAULT '{}',
PRIMARY KEY (game_id, snake_id)
)
""")
copied = copy_game_snakes(
source, destination, batch_size=10, retained_ids={"game-1"},
)
rows = destination.execute("""
SELECT snake_id, snake_name, is_you, customizations_json
FROM game_snakes ORDER BY snake_id
""").fetchall()
finally:
source.close()
destination.close()
self.assertEqual(copied, 2)
self.assertEqual(rows, [
("snake-1", "PrismBattleSnake", 1, "{}"),
("snake-2", "Enemy", 0, "{}"),
])
def test_copy_game_snakes_defaults_legacy_schema_to_empty_customizations(self):
source = sqlite3.connect(":memory:")
destination = sqlite3.connect(":memory:")
@@ -60,6 +108,11 @@ class TestMigrateGameplayDatabase(unittest.TestCase):
"INSERT INTO game_snakes VALUES (?, ?, ?, ?)",
("game-1", "snake-1", "LegacySnake", 0),
)
source.execute("""
CREATE TABLE snake_turns (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER
)
""")
destination.execute("""
CREATE TABLE game_snakes (
game_id TEXT, snake_id TEXT, snake_name TEXT, is_you INTEGER,