fix: dispatch strategy pattern, context view protocols, lock-timeout, tick hardening

Author: tmcarmichael

src/brain/completion.py

@@ -34,7 +34,22 @@ def tick_active_routine(brain: Brain, state: GameState, now: float) -> None:
 
     rt0 = brain.perf_clock()
     brain._active._tick_deadline = rt0 + 0.200  # 200ms cooperative budget
-    status = brain._active.tick(state)
+    try:
+        status = brain._active.tick(state)
+    except Exception as exc:
+        brain.routine_tick_ms = (brain.perf_clock() - rt0) * 1000
+        brain._ticked_routine_name = brain._active_name
+        log.warning(
+            "[DECISION] Routine %s tick() raised %s -- treating as FAILURE",
+            brain._active_name,
+            exc,
+        )
+        brain._active.failure_reason = f"tick_exception: {exc}"
+        from core.types import FailureCategory as _FC
+
+        brain._active.failure_category = _FC.EXECUTION
+        handle_routine_completion(brain, state, RoutineStatus.FAILURE, now)
+        return
     brain.routine_tick_ms = (brain.perf_clock() - rt0) * 1000
 
     # Capture name before completion/hard-kill clears it (for profiling)

src/brain/context.py

@@ -3,7 +3,7 @@
 Contains 12 focused sub-state objects for clean grouping:
   ctx.combat     -  CombatState (engaged, pull_target_id, dot/lifetap timers)
   ctx.pet        -  PetState (alive, spawn_id, name, has_add)
-  ctx.camp       -  CampConfig (camp_x/y, guard_x/y, hunt zone, danger points)
+  ctx.camp       -  CampConfig (camp_pos, guard_pos, hunt zone, danger points)
   ctx.inventory  -  InventoryState (weight tracking, loot count)
   ctx.plan       -  PlanState (typed travel plans)
   ctx.player     -  PlayerState (death, position, engagement)

src/brain/context_views.py

@@ -0,0 +1,93 @@
+"""Narrowed view protocols for AgentContext consumers.
+
+Each rule module depends on a subset of AgentContext's surface area.
+These Protocol classes make that dependency explicit at the type level:
+mypy enforces that closures only access attributes defined on their view.
+
+AgentContext satisfies all views via structural subtyping -- no changes
+to AgentContext are needed.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    from brain.learning.encounters import FightHistory
+    from brain.state.camp import CampConfig
+    from brain.state.combat import CombatState
+    from brain.state.kill_tracker import DefeatTracker
+    from brain.state.loot_config import LootConfig
+    from brain.state.pet import PetState
+    from brain.state.plan import PlanState
+    from brain.state.player import PlayerState
+    from brain.state.threat import ThreatState
+    from brain.state.zone import ZoneState
+    from brain.world.model import WorldModel
+    from nav.travel_planner import TunnelRoute
+    from nav.waypoint_graph import WaypointGraph
+    from perception.state import GameState
+
+
+@runtime_checkable
+class SurvivalView(Protocol):
+    """Context surface used by survival rules (FLEE, REST, FEIGN_DEATH, etc.)."""
+
+    combat: CombatState
+    pet: PetState
+    player: PlayerState
+    threat: ThreatState
+    defeat_tracker: DefeatTracker
+    fight_history: FightHistory | None
+    world: WorldModel | None
+
+    # Rest thresholds
+    rest_hp_entry: float
+    rest_mana_entry: float
+    rest_hp_threshold: float
+    rest_mana_threshold: float
+
+    @property
+    def in_active_combat(self) -> bool: ...
+
+
+@runtime_checkable
+class CombatView(Protocol):
+    """Context surface used by combat rules (IN_COMBAT, ACQUIRE, PULL, etc.)."""
+
+    combat: CombatState
+    pet: PetState
+    zone: ZoneState
+    plan: PlanState
+    defeat_tracker: DefeatTracker
+    fight_history: FightHistory | None
+    loot: LootConfig
+    world: WorldModel | None
+
+    @property
+    def in_active_combat(self) -> bool: ...
+
+    def nearby_player_count(self, state: GameState, radius: float = ...) -> int: ...
+
+
+@runtime_checkable
+class MaintenanceView(Protocol):
+    """Context surface used by maintenance rules (BUFF, SUMMON_PET, etc.)."""
+
+    combat: CombatState
+    pet: PetState
+    plan: PlanState
+    player: PlayerState
+
+
+@runtime_checkable
+class NavigationView(Protocol):
+    """Context surface used by navigation rules (TRAVEL, WANDER)."""
+
+    combat: CombatState
+    pet: PetState
+    plan: PlanState
+    camp: CampConfig
+    world: WorldModel | None
+    tunnel_routes: list[TunnelRoute]
+    waypoint_graph: WaypointGraph | None

src/brain/decision.py

@@ -36,12 +36,10 @@
 from brain.circuit_breaker import CircuitBreaker
 from brain.completion import tick_active_routine
 from brain.profiling import tick_profiling
-from brain.rule_def import RuleDef
+from brain.rule_def import Consideration, RuleDef
 from brain.scoring_phases import (
+    build_phase_selector,
     compute_divergence,
-    select_by_tier,
-    select_weighted,
-    select_with_considerations,
 )
 from brain.transitions import handle_transition
 from perception.state import GameState
@@ -90,6 +88,7 @@ def __init__(
         self._active_start_time: float = 0.0  # when current routine activated
         self._last_lock_blocked: str = ""  # suppress repeated lock messages
         self.utility_phase: int = utility_phase
+        self._phase_selector = build_phase_selector(utility_phase)
 
         # Per-tick profiling (rule eval + routine tick times in ms)
         self.rule_times: dict[str, float] = {}  # rule_name -> eval ms
@@ -120,7 +119,7 @@ def add_rule(
         score_fn: Callable[[GameState], float] | None = None,
         tier: int = 0,
         weight: float = 1.0,
-        considerations: list | None = None,
+        considerations: list[Consideration] | None = None,
         breaker_max_failures: int = 5,
         breaker_window: float = 300.0,
         breaker_recovery: float = 120.0,
@@ -170,6 +169,23 @@ def add_rule(
                 recovery_seconds=breaker_recovery,
             )
 
+    # -- Public accessors (used by BrainRunner, simulator) --
+
+    @property
+    def active_routine(self) -> RoutineBase | None:
+        """The currently running routine, or None."""
+        return self._active
+
+    @property
+    def active_routine_name(self) -> str:
+        """Name of the currently running routine."""
+        return self._active_name
+
+    @property
+    def last_matched_rule(self) -> str:
+        """Name of the rule that matched on the most recent tick."""
+        return self._last_matched_rule
+
     def tick(self, state: GameState) -> None:
         """Evaluate rules and run the appropriate routine."""
         tick_start = self.perf_clock()
@@ -199,51 +215,9 @@ def _evaluate_rules(
         rule_eval: dict[str, str] = {}
         rule_times: dict[str, float] = {}
 
-        if self.utility_phase >= 4:
-            selected, selected_name, selected_emergency = select_with_considerations(
-                self, state, now, rule_eval, diag_results, rule_times
-            )
-        elif self.utility_phase >= 3:
-            selected, selected_name, selected_emergency = select_weighted(
-                self, state, now, rule_eval, diag_results, rule_times
-            )
-        elif self.utility_phase >= 2:
-            selected, selected_name, selected_emergency = select_by_tier(
-                self, state, now, rule_eval, diag_results, rule_times
-            )
-        else:
-            # Phase 0/1: binary conditions, insertion-order priority.
-            # Short-circuit after winner found: skip condition evaluation
-            # for lower-priority rules. This is safe because detection
-            # logic (threat scan, add detection) runs pre-rule in the
-            # tick pipeline, not inside condition functions.
-            for r in self._rules:
-                if r.name in self._cooldowns and now < self._cooldowns[r.name]:
-                    remaining = self._cooldowns[r.name] - now
-                    rule_eval[r.name] = f"cooldown({remaining:.0f}s)"
-                    diag_results.append(f"{r.name}=CD")
-                    rule_times[r.name] = 0.0
-                    continue
-                breaker = self._breakers.get(r.name)
-                if breaker and not breaker.allow():
-                    rule_eval[r.name] = "OPEN"
-                    diag_results.append(f"{r.name}=OPEN")
-                    rule_times[r.name] = 0.0
-                    continue
-                if selected is not None:
-                    rule_eval[r.name] = "skip"
-                    diag_results.append(f"{r.name}=skip")
-                    rule_times[r.name] = 0.0
-                    continue
-                t0 = self.perf_clock()
-                matched = r.condition(state)
-                rule_times[r.name] = (self.perf_clock() - t0) * 1000
-                rule_eval[r.name] = "YES" if matched else "no"
-                diag_results.append(f"{r.name}={'YES' if matched else 'no'}")
-                if matched:
-                    selected = r.routine
-                    selected_name = r.name
-                    selected_emergency = r.emergency
+        selected, selected_name, selected_emergency = self._phase_selector.select(
+            self, state, now, rule_eval, diag_results, rule_times
+        )
 
         # Phase 1+: compute scores in parallel for divergence logging
         if self.utility_phase >= 1:

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/goap/world_state.py

@@ -72,9 +72,7 @@ def build_world_state(state: GameState, ctx: AgentContext) -> PlanWorldState:
     # At camp
     at_camp = True
     if ctx.camp.roam_radius > 0:
-        from core.types import Point
-
-        d = state.pos.dist_to(Point(ctx.camp.camp_x, ctx.camp.camp_y, 0.0))
+        d = state.pos.dist_to(ctx.camp.camp_pos)
         at_camp = d <= ctx.camp.roam_radius * 1.2  # slight buffer
 
     return PlanWorldState(

src/brain/rule_def.py

@@ -39,7 +39,7 @@ class RuleDef:
     # Phase 3+: weight for cross-tier scoring
     weight: float = 1.0
     # Phase 4: considerations (list of Consideration objects)
-    considerations: list = field(default_factory=list)
+    considerations: list[Consideration] = field(default_factory=list)
     # Circuit breaker: max failures in window before tripping (0 = disabled)
     breaker_max_failures: int = 5
     breaker_window: float = 300.0
@@ -74,7 +74,7 @@ def score_from_considerations(
     weight_sum = 0.0
     for c in considerations:
         raw = c.input_fn(state, ctx)
-        mapped = c.curve(raw)
+        mapped = max(0.0, min(c.curve(raw), 1.0))
         if mapped <= 0.0:
             return 0.0  # hard gate
         total_log += c.weight * math.log(mapped)