Update Best Battle Snake Logic

server/metrics/ServerMetricsCollector.py

@@ -20,19 +20,19 @@ class ServerMetricsCollector:
       'active_games_peak': 0,
       'games_autocreated': 0,
       'http_requests_total': 0,
-      'http_requests_by_endpoint': {
+      # 'http_requests_by_endpoint': {
-        'info': 0,
+      #   'info': 0,
-        'start': 0,
+      #   'start': 0,
-        'move': 0,
+      #   'move': 0,
-        'end': 0,
+      #   'end': 0,
-      },
+      # },
-      'move_direction_counts': {
+      # 'move_direction_counts': {
-        'up': 0,
+      #   'up': 0,
-        'down': 0,
+      #   'down': 0,
-        'left': 0,
+      #   'left': 0,
-        'right': 0,
+      #   'right': 0,
-        'unknown': 0,
+      #   'unknown': 0,
-      },
+      # },
       'move_response_time_ms_total': 0.0,
       'move_response_time_ms_max': 0.0,
       'last_game_start_unix': 0,
@@ -53,8 +53,8 @@ class ServerMetricsCollector:
 
   def record_http_request(self, endpoint:str) -> None:
     self._metrics['http_requests_total'] += 1
-    endpoint_counts = self._metrics['http_requests_by_endpoint']
+    # endpoint_counts = self._metrics['http_requests_by_endpoint']
-    endpoint_counts[endpoint] = endpoint_counts.get(endpoint, 0) + 1
+    # endpoint_counts[endpoint] = endpoint_counts.get(endpoint, 0) + 1
 
   async def record_game_started(self, active_count:int) -> None:
     self._metrics['games_started'] += 1
@@ -76,12 +76,12 @@ class ServerMetricsCollector:
       self._metrics['move_response_time_ms_max'],
       elapsed_ms,
     )
-    move_counts = self._metrics['move_direction_counts']
+    # move_counts = self._metrics['move_direction_counts']
-    if direction in move_counts:
+    # if direction in move_counts:
-      move_counts[direction] += 1
+    #   move_counts[direction] += 1
-    else:
+    # else:
-      move_counts['unknown'] += 1
+    #   move_counts['unknown'] += 1
-    self._metrics['last_move_unix'] = int(time.time())
+    # self._metrics['last_move_unix'] = int(time.time())
     await self._auto_publish()
 
   async def record_game_end(self, game_state:dict) -> None:
@@ -156,8 +156,8 @@ class ServerMetricsCollector:
       'avg_turns_per_game': round(avg_turns, 2),
       'win_rate': round(win_rate, 4),
       'avg_move_response_ms': round(avg_move_ms, 2),
-      'http_requests_by_endpoint': dict(self._metrics['http_requests_by_endpoint']),
+      # 'http_requests_by_endpoint': dict(self._metrics['http_requests_by_endpoint']),
-      'move_direction_counts': dict(self._metrics['move_direction_counts']),
+      # 'move_direction_counts': dict(self._metrics['move_direction_counts']),
       'oldest_active_game_age_sec': oldest_active_age,
       'stale_game_timeout_sec': self._stale_game_timeout_sec,
       'active_games_stale': report_stale_candidates,
@@ -243,19 +243,19 @@ class ServerMetricsCollector:
       f'snake_active_games_stale {snapshot["active_games_stale"]}',
     ]
 
-    lines.extend([
+    # lines.extend([
-      '# HELP snake_http_requests_by_endpoint_total Requests served grouped by endpoint.',
+    #   '# HELP snake_http_requests_by_endpoint_total Requests served grouped by endpoint.',
-      '# TYPE snake_http_requests_by_endpoint_total counter',
+    #   '# TYPE snake_http_requests_by_endpoint_total counter',
-    ])
+    # ])
-    for endpoint, count in snapshot['http_requests_by_endpoint'].items():
+    # for endpoint, count in snapshot['http_requests_by_endpoint'].items():
-      lines.append(f'snake_http_requests_by_endpoint_total{{endpoint="{endpoint}"}} {count}')
+    #   lines.append(f'snake_http_requests_by_endpoint_total{{endpoint="{endpoint}"}} {count}')
 
-    lines.extend([
+    # lines.extend([
-      '# HELP snake_moves_by_direction_total Move responses grouped by direction.',
+    #   '# HELP snake_moves_by_direction_total Move responses grouped by direction.',
-      '# TYPE snake_moves_by_direction_total counter',
+    #   '# TYPE snake_moves_by_direction_total counter',
-    ])
+    # ])
-    for direction, count in snapshot['move_direction_counts'].items():
+    # for direction, count in snapshot['move_direction_counts'].items():
-      lines.append(f'snake_moves_by_direction_total{{direction="{direction}"}} {count}')
+    #   lines.append(f'snake_moves_by_direction_total{{direction="{direction}"}} {count}')
 
     return '\n'.join(lines) + '\n'
 

snakes/BestBattleSnake.py

@@ -864,6 +864,15 @@ class BestBattleSnake(TemplateSnake):
       if self._is_tail_stacked(snake["body"]):
         continue
 
+      snake_id = snake.get("id")
+      enemy_can_grow = None
+      if enemy_can_grow_cache is not None and snake_id is not None:
+        enemy_can_grow = enemy_can_grow_cache.get(snake_id)
+      if enemy_can_grow is None:
+        enemy_can_grow = self._enemy_can_grow_this_turn(snake, food_set)
+      if enemy_can_grow:
+        continue
+
       enemy_tail = snake["body"][-1]
       blocked.discard((enemy_tail["x"], enemy_tail["y"]))
 
@@ -881,6 +890,12 @@ class BestBattleSnake(TemplateSnake):
       enemy_len = enemy.get("length", len(enemy["body"]))
       enemy_tail = (enemy["body"][-1]["x"], enemy["body"][-1]["y"])
       enemy_tail_stacked = self._is_tail_stacked(enemy["body"])
+      enemy_id = enemy.get("id")
+      enemy_can_grow = None
+      if enemy_can_grow_cache is not None and enemy_id is not None:
+        enemy_can_grow = enemy_can_grow_cache.get(enemy_id)
+      if enemy_can_grow is None:
+        enemy_can_grow = self._enemy_can_grow_this_turn(enemy, food_set)
       enemy_head = enemy["head"]
       for dx, dy in self.DIRECTIONS.values():
         point = (enemy_head["x"] + dx, enemy_head["y"] + dy)
@@ -891,6 +906,7 @@ class BestBattleSnake(TemplateSnake):
           not is_constrictor
           and point == enemy_tail
           and not enemy_tail_stacked
+          and not enemy_can_grow
         )
         can_contest_my_tail = (not is_constrictor and point == my_tail and not my_tail_stacked)
 
@@ -988,7 +1004,8 @@ class BestBattleSnake(TemplateSnake):
     if self._time_exceeded(deadline) or self._remaining_ms(deadline) < self.future_planning_min_time_ms:
       return random.choice(top_moves)
 
-    candidate_moves = sorted(top_moves, key=lambda move: scores[move], reverse=True)[:3]
+    candidate_limit = max(1, self.future_planning_branch)
+    candidate_moves = sorted(top_moves, key=lambda move: scores[move], reverse=True)[:candidate_limit]
     lookahead_bonus:dict[str, float] = {}
     for move in candidate_moves:
       if self._time_exceeded(deadline):

tests/test_BestBattleSnake.py

@@ -905,7 +905,7 @@ class TestBestBattleSnake(unittest.TestCase):
       enemy_can_grow_cache={"enemy": True},
     )
 
-    self.assertNotIn((0, 1), blocked)
+    self.assertIn((0, 1), blocked)
 
   def test_enemy_attack_map_allows_enemy_tail_move_when_enemy_can_grow(self):
     snake = BestBattleSnake()
@@ -948,7 +948,7 @@ class TestBestBattleSnake(unittest.TestCase):
       enemy_can_grow_cache={"enemy": True},
     )
 
-    self.assertEqual(attack_map.get((2, 3)), 4)
+    self.assertIsNone(attack_map.get((2, 3)))
 
   def test_future_planning_prefers_non_trap_path(self):
     snake = BestBattleSnake()