fix: enforce rule conditions in scoring phases, add GOAP closed set

Author: tmcarmichael

src/brain/goap/planner.py

@@ -484,6 +484,11 @@ def _search(self, start: PlanWorldState, goal: Goal, ctx: AgentContext | None) -
         counter = 1
         visited = 0
 
+        # Best g-cost seen per state.  PlanWorldState is frozen/hashable,
+        # so equivalent states reached via different action orderings are
+        # detected and the worse path is pruned.
+        best_g: dict[PlanWorldState, float] = {start: 0.0}
+
         while open_list:
             if visited >= MAX_SEARCH_NODES:
                 log.log(VERBOSE, "[GOAP] Search exhausted: %d nodes", visited)
@@ -493,6 +498,11 @@ def _search(self, start: PlanWorldState, goal: Goal, ctx: AgentContext | None) -
                 break
 
             _, _, node = heapq.heappop(open_list)
+
+            # Skip if we already expanded this state at equal or lower cost.
+            prev = best_g.get(node.state)
+            if prev is not None and node.g_cost > prev:
+                continue
             visited += 1
 
             # Goal test: deterministic satisfaction check
@@ -542,8 +552,14 @@ def _search(self, start: PlanWorldState, goal: Goal, ctx: AgentContext | None) -
 
                 new_state = action.apply_effects(node.state)
                 new_cost = node.g_cost + self.get_corrected_cost(action, ctx)
-                new_actions = node.actions + [action]
 
+                # Prune if we already reached this state at equal or lower cost.
+                prev_best = best_g.get(new_state)
+                if prev_best is not None and new_cost >= prev_best:
+                    continue
+                best_g[new_state] = new_cost
+
+                new_actions = node.actions + [action]
                 child = _Node(
                     state=new_state,
                     g_cost=new_cost,

src/brain/scoring_phases.py

@@ -48,14 +48,19 @@ def _apply_modifiers(score: float, phase: str, rule_name: str, goap_hint: str) -
 def _is_rule_blocked(
     brain: Brain,
     r: RuleDef,
+    state: GameState,
     now: float,
     rule_eval: dict,
     diag_results: list,
     rule_times: dict,
 ) -> bool:
-    """Check cooldown and circuit breaker. Returns True if rule should be skipped.
+    """Check condition, cooldown, and circuit breaker.  Returns True if
+    rule should be skipped.
 
     Shared by Phases 2/3/4 to avoid duplicating eligibility logic.
+    The condition check is critical: without it, a rule whose predicate
+    returns False (e.g. ACQUIRE when suppressed by hard gates) could
+    still win selection based on its utility score alone.
     """
     if r.name in brain._cooldowns and now < brain._cooldowns[r.name]:
         remaining = brain._cooldowns[r.name] - now
@@ -69,6 +74,11 @@ def _is_rule_blocked(
         diag_results.append(f"{r.name}=OPEN")
         rule_times[r.name] = 0.0
         return True
+    if not r.condition(state):
+        rule_eval[r.name] = "cond=F"
+        diag_results.append(f"{r.name}=cond_F")
+        rule_times[r.name] = 0.0
+        return True
     return False
 
 
@@ -134,7 +144,7 @@ def select_by_tier(
     tier_groups: dict[int, list[RuleDef]] = defaultdict(list)
 
     for r in brain._rules:
-        if _is_rule_blocked(brain, r, now, rule_eval, diag_results, rule_times):
+        if _is_rule_blocked(brain, r, state, now, rule_eval, diag_results, rule_times):
             continue
         tier_groups[r.tier].append(r)
 
@@ -169,7 +179,7 @@ def select_weighted(
     phase, goap_hint = _resolve_phase_context(brain)
 
     for r in brain._rules:
-        if _is_rule_blocked(brain, r, now, rule_eval, diag_results, rule_times):
+        if _is_rule_blocked(brain, r, state, now, rule_eval, diag_results, rule_times):
             continue
         t0 = time.perf_counter()
         s = _apply_modifiers(_safe_score(r.score_fn, state), phase, r.name, goap_hint)
@@ -210,7 +220,7 @@ def select_with_considerations(
     phase, goap_hint = _resolve_phase_context(brain)
 
     for r in brain._rules:
-        if _is_rule_blocked(brain, r, now, rule_eval, diag_results, rule_times):
+        if _is_rule_blocked(brain, r, state, now, rule_eval, diag_results, rule_times):
             continue
         t0 = time.perf_counter()
         # Phase 4: prefer considerations over score_fn when defined

src/nav/pathfinding.py

@@ -61,15 +61,15 @@ def _validate_endpoints(
 
     snap_radius = max(10, int(100.0 / terrain.cell_size))
     if not _cell_walkable(terrain, sc, sr, near_z):
-        snapped = _snap_to_walkable(terrain, sc, sr, radius=snap_radius)
+        snapped = _snap_to_walkable(terrain, sc, sr, radius=snap_radius, near_z=near_z)
         if snapped:
             sc, sr = snapped
         else:
             log.warning("[NAV] Pathfind: start is blocked and no walkable cell nearby")
             return None
 
     if not _cell_walkable(terrain, gc, gr, near_z):
-        snapped = _snap_to_walkable(terrain, gc, gr, radius=snap_radius)
+        snapped = _snap_to_walkable(terrain, gc, gr, radius=snap_radius, near_z=near_z)
         if snapped:
             gc, gr = snapped
         else:
@@ -199,11 +199,15 @@ def find_path(
     rows = terrain._rows
     explored = 0
 
-    # Extract bitfield for hot-path (local refs = faster than attr access)
-    if not terrain._walk_bits:
-        terrain._build_walk_bits()
-    wb = terrain._walk_bits
-    wbc = terrain._walk_byte_cols
+    # Build a Z-filtered walkability bitfield so the search never routes
+    # through cells on a different vertical level (bridges, multi-floor).
+    if not math.isnan(near_z) and terrain._z_ceiling:
+        wb, wbc = terrain.build_walk_bits_z(near_z)
+    else:
+        if not terrain._walk_bits:
+            terrain._build_walk_bits()
+        wb = terrain._walk_bits
+        wbc = terrain._walk_byte_cols
 
     # All 8 directions for the start node (no parent to prune from)
     _ALL_DIRS = [
@@ -297,8 +301,9 @@ def _cell_walkable(terrain: ZoneTerrain, col: int, row: int, near_z: float = flo
     """Check if a grid cell is walkable with Z-level filtering.
 
     When near_z is provided (not NaN), cells on a different vertical
-    level (bridges, multi-floor buildings) are rejected. This prevents
-    pathfinding from routing through geometry on the wrong floor.
+    level (bridges, multi-floor buildings) are rejected. Bridge cells
+    with water/lava underneath are only walkable for agents at bridge
+    height, not ground level.
     """
     idx = terrain._grid_idx(col, row)
     if idx < 0:
@@ -310,6 +315,18 @@ def _cell_walkable(terrain: ZoneTerrain, col: int, row: int, near_z: float = flo
     if not terrain.is_walkable(game_x, game_y):
         return False
     if not math.isnan(near_z):
+        # Multi-level check: reject bridge cells when agent is at ground
+        # level and ground surface is hazardous (water/lava).
+        from nav.terrain.heightmap import SURFACE_BRIDGE, SURFACE_LAVA, SURFACE_WATER
+
+        f = terrain._flags[idx]
+        z_ceil = terrain.get_z_ceiling(game_x, game_y)
+        if not math.isnan(z_ceil):
+            z_ground = terrain.get_z(game_x, game_y)
+            mid = (z_ground + z_ceil) / 2.0
+            on_upper = near_z > mid
+            if not on_upper and (f & SURFACE_BRIDGE) and (f & (SURFACE_WATER | SURFACE_LAVA)):
+                return False
         level_z = terrain.get_level_z(game_x, game_y, near_z)
         if abs(level_z - near_z) > _WALKABLE_CLIMB:
             return False
@@ -352,14 +369,16 @@ def _cell_cost(terrain: ZoneTerrain, col: int, row: int, from_col: int = -1, fro
     return cost
 
 
-def _snap_to_walkable(terrain: ZoneTerrain, col: int, row: int, radius: int = 5) -> tuple[int, int] | None:
-    """Find nearest walkable cell within radius."""
+def _snap_to_walkable(
+    terrain: ZoneTerrain, col: int, row: int, radius: int = 5, near_z: float = float("nan")
+) -> tuple[int, int] | None:
+    """Find nearest walkable cell within radius, respecting Z-level."""
     best = None
     best_dist = float("inf")
     for dr in range(-radius, radius + 1):
         for dc in range(-radius, radius + 1):
             nc, nr = col + dc, row + dr
-            if _cell_walkable(terrain, nc, nr):
+            if _cell_walkable(terrain, nc, nr, near_z):
                 d = dc * dc + dr * dr
                 if d < best_dist:
                     best_dist = d
@@ -399,13 +418,13 @@ def _find_path_astar(
 
     snap_radius = max(10, int(100.0 / terrain.cell_size))
     if not _bit_walkable(wb, wbc, sc, sr, cols, rows):
-        snapped = _snap_to_walkable(terrain, sc, sr, radius=snap_radius)
+        snapped = _snap_to_walkable(terrain, sc, sr, radius=snap_radius, near_z=near_z)
         if snapped:
             sc, sr = snapped
         else:
             return None
     if not _bit_walkable(wb, wbc, gc, gr, cols, rows):
-        snapped = _snap_to_walkable(terrain, gc, gr, radius=snap_radius)
+        snapped = _snap_to_walkable(terrain, gc, gr, radius=snap_radius, near_z=near_z)
         if snapped:
             gc, gr = snapped
         else:

src/nav/terrain/heightmap.py

@@ -38,6 +38,11 @@
 SURFACE_OBSTACLE = 0x0040  # static object (tree, rock, building, pillar)
 SURFACE_BRIDGE = 0x0080  # bridge deck (walkable over water, auto-detected)
 
+# Max Z difference between agent and cell surface to consider walkable.
+# Shared with pathfinding (which defines its own copy for module independence).
+_WALKABLE_CLIMB = 15.0
+_NAN = float("nan")
+
 # Material IDs (uint8, stored per cell in v3 cache)
 MAT_UNKNOWN = 0
 MAT_STONE = 1
@@ -268,6 +273,78 @@ def _build_walk_bits(self) -> None:
         walkable = sum(bin(b).count("1") for b in wb)
         log.debug("[NAV] Walk bitfield built: %d walkable bits, %d bytes", walkable, len(wb))
 
+    def build_walk_bits_z(self, near_z: float) -> tuple[bytearray, int]:
+        """Build a Z-filtered walkability bitfield for pathfinding.
+
+        Like ``_build_walk_bits`` but rejects cells whose surface Z is
+        too far from *near_z* (the agent's current level).  This prevents
+        paths from routing across bridges or floors the agent is not on.
+
+        Returns ``(bitfield, byte_cols)`` without mutating the cached
+        ``_walk_bits`` (which is Z-agnostic and used elsewhere).
+        """
+        if not self._flags:
+            return bytearray(), 0
+
+        cols = self._cols
+        rows = self._rows
+        byte_cols = (cols + 7) >> 3
+        wb = bytearray(rows * byte_cols)
+        blocked = SURFACE_WATER | SURFACE_LAVA | SURFACE_CLIFF | SURFACE_OBSTACLE
+        has_ceil = bool(self._z_ceiling)
+        _z = self._z
+        _zc = self._z_ceiling
+        climb = _WALKABLE_CLIMB
+
+        for row in range(rows):
+            rb = row * byte_cols
+            rf = row * cols
+            for col in range(cols):
+                idx = rf + col
+                f = self._flags[idx]
+
+                if has_ceil:
+                    z_ceil = _zc[idx]
+                else:
+                    z_ceil = _NAN
+
+                is_multi = z_ceil == z_ceil  # not NaN → multi-level cell
+
+                if is_multi:
+                    # Multi-level cell (bridge, multi-floor): decide
+                    # walkability based on which surface the agent is on.
+                    z_ground = _z[idx]
+                    mid = (z_ground + z_ceil) / 2.0
+                    on_upper = near_z > mid
+                    if on_upper:
+                        # Agent on the upper surface (bridge deck) — walkable
+                        # only if BRIDGE flag is set.
+                        walkable = bool(f & SURFACE_BRIDGE)
+                    else:
+                        # Agent at ground level — walkable only if the
+                        # ground surface itself is walkable (not water/lava
+                        # under a bridge).
+                        walkable = bool((f & SURFACE_WALKABLE) and not (f & blocked))
+                        # Bridge-only cells are not ground-walkable
+                        if (f & SURFACE_BRIDGE) and (f & (SURFACE_WATER | SURFACE_LAVA)):
+                            walkable = False
+                    level_z = z_ceil if on_upper else z_ground
+                    if abs(level_z - near_z) > climb:
+                        walkable = False
+                else:
+                    # Single-level cell: standard walkability check
+                    if f & SURFACE_BRIDGE:
+                        walkable = True
+                    elif (f & SURFACE_WALKABLE) and not (f & blocked):
+                        walkable = True
+                    else:
+                        walkable = False
+
+                if walkable:
+                    wb[rb + (col >> 3)] |= 1 << (col & 7)
+
+        return wb, byte_cols
+
     def invalidate_walk_bits(self) -> None:
         """Force rebuild of the walkability bitfield.
 

src/perception/log_parser.py

@@ -99,13 +99,16 @@ def __init__(self, log_path: str, pet_names: set[str] | None = None) -> None:
 
     def _open(self) -> None:
         """Open the log file and seek to the end (only read new lines)."""
+        f = None
         try:
             f = open(self._path, encoding="utf-8", errors="replace")
-            self._file = f
             f.seek(0, os.SEEK_END)
             log.info("[PERCEPTION] LogParser tailing %s (pos=%d)", self._path, f.tell())
+            self._file = f
         except OSError as e:
             log.warning("[PERCEPTION] Could not open log file %s: %s", self._path, e)
+            if f is not None:
+                f.close()
             self._file = None
 
     def _parse_line(self, body: str, events: LogEvents) -> None: