diff --git a/pyproject.toml b/pyproject.toml index cc69e855..1ddea86f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ repository = "http://localhost:3000/rockachopa/Timmy-time-dashboard" packages = [ { include = "config.py", from = "src" }, + { include = "bannerlord", from = "src" }, { include = "dashboard", from = "src" }, { include = "infrastructure", from = "src" }, { include = "integrations", from = "src" }, diff --git a/src/bannerlord/__init__.py b/src/bannerlord/__init__.py new file mode 100644 index 00000000..8a85a823 --- /dev/null +++ b/src/bannerlord/__init__.py @@ -0,0 +1,49 @@ +"""Bannerlord M3 — Full Campaign Strategy. + +Timmy runs a complete Bannerlord campaign: economy, diplomacy, kingdom +building, and war decisions — all via sovereign local inference. + +Key components: + gabs_client — TCP JSON-RPC client for the GABS mod (port 4825) + types — KingSubgoal, GameState, message schemas + session_memory — SQLite-backed multi-day strategic plan persistence + campaign — CampaignOrchestrator tying all agents together + adapter — WorldInterface adapter for use with the benchmark runner + agents/ — King, Vassal, and Companion agent hierarchy + +Quick start:: + + from bannerlord.campaign import CampaignOrchestrator + orch = CampaignOrchestrator() + summary = await orch.run(max_ticks=100) + +Register the world adapter:: + + from infrastructure.world import register_adapter + from bannerlord.adapter import BannerlordWorldAdapter + register_adapter("bannerlord", BannerlordWorldAdapter) + +M3 done-when condition: + Timmy establishes own kingdom with 3+ fiefs and + survives 100 in-game days as ruler. +""" + +from bannerlord.adapter import BannerlordWorldAdapter +from bannerlord.campaign import CampaignOrchestrator +from bannerlord.gabs_client import GABSClient +from bannerlord.session_memory import SessionMemory +from bannerlord.types import ( + GameState, + KingSubgoal, + SubgoalToken, +) + +__all__ = [ + "BannerlordWorldAdapter", + "CampaignOrchestrator", + "GABSClient", + "GameState", + "KingSubgoal", + "SessionMemory", + "SubgoalToken", +] diff --git a/src/bannerlord/adapter.py b/src/bannerlord/adapter.py new file mode 100644 index 00000000..0126f38b --- /dev/null +++ b/src/bannerlord/adapter.py @@ -0,0 +1,228 @@ +"""Bannerlord M3 — WorldInterface adapter wrapping the GABS TCP client. + +Plugs Bannerlord into the engine-agnostic ``WorldInterface`` contract so +the benchmark runner and heartbeat loop can drive the campaign the same way +they would drive any other game world. + +Register with:: + + from infrastructure.world import register_adapter + from bannerlord.adapter import BannerlordWorldAdapter + register_adapter("bannerlord", BannerlordWorldAdapter) +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime +from typing import Any + +from bannerlord.gabs_client import GABSClient +from bannerlord.types import GameState +from infrastructure.world.interface import WorldInterface +from infrastructure.world.types import ( + ActionResult, + ActionStatus, + CommandInput, + PerceptionOutput, +) + +logger = logging.getLogger(__name__) + + +class BannerlordWorldAdapter(WorldInterface): + """WorldInterface adapter for Bannerlord via the GABS mod. + + ``observe()`` — fetches the full GameState from GABS and maps it to a + ``PerceptionOutput`` with structured fields. + + ``act()`` — dispatches ``CommandInput.action`` as a GABS JSON-RPC call, + forwarding ``parameters`` as the call args. + + ``speak()`` — sends a chat message via GABS (e.g., for companion NPC + conversations or on-screen overlays). + + Degrades gracefully when GABS is unavailable. + """ + + def __init__( + self, + *, + host: str = "127.0.0.1", + port: int = 4825, + timeout: float = 10.0, + ) -> None: + self._client = GABSClient(host=host, port=port, timeout=timeout) + self._last_state: GameState | None = None + + # -- lifecycle --------------------------------------------------------- + + def connect(self) -> None: + """Synchronous connect wrapper (runs async in a new event loop).""" + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + # Inside async context — caller should use async connect + logger.warning( + "BannerlordWorldAdapter.connect() called from async context; " + "use 'await adapter.async_connect()' instead" + ) + return + loop.run_until_complete(self._client.connect()) + except Exception as exc: + logger.warning("BannerlordWorldAdapter.connect() failed: %s", exc) + + def disconnect(self) -> None: + """Synchronous disconnect wrapper.""" + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + return + loop.run_until_complete(self._client.disconnect()) + except Exception as exc: + logger.debug("BannerlordWorldAdapter.disconnect() error: %s", exc) + + async def async_connect(self) -> bool: + """Async connect — preferred in async contexts.""" + return await self._client.connect() + + async def async_disconnect(self) -> None: + """Async disconnect.""" + await self._client.disconnect() + + @property + def is_connected(self) -> bool: + return self._client.is_connected + + # -- WorldInterface contract ------------------------------------------ + + def observe(self) -> PerceptionOutput: + """Return a PerceptionOutput derived from the current GABS GameState. + + Falls back to an empty perception if GABS is unreachable. + """ + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + logger.warning("observe() called from async context — use async_observe()") + return self._empty_perception() + state = loop.run_until_complete(self._client.get_game_state()) + return self._state_to_perception(state) + except Exception as exc: + logger.warning("BannerlordWorldAdapter.observe() error: %s", exc) + return self._empty_perception() + + async def async_observe(self) -> PerceptionOutput: + """Async observe — preferred in async contexts.""" + try: + state = await self._client.get_game_state() + self._last_state = state + return self._state_to_perception(state) + except Exception as exc: + logger.warning("async_observe() error: %s", exc) + return self._empty_perception() + + def act(self, command: CommandInput) -> ActionResult: + """Dispatch a command to GABS. Returns success/failure.""" + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + logger.warning("act() called from async context — use async_act()") + return ActionResult(status=ActionStatus.NOOP) + result = loop.run_until_complete(self.async_act(command)) + return result + except Exception as exc: + logger.warning("BannerlordWorldAdapter.act() error: %s", exc) + return ActionResult( + status=ActionStatus.FAILURE, + message=str(exc), + ) + + async def async_act(self, command: CommandInput) -> ActionResult: + """Async command dispatch.""" + try: + result = await self._client._call( + command.action, command.parameters or {} + ) + if result is None: + return ActionResult( + status=ActionStatus.FAILURE, + message=f"GABS returned no result for {command.action}", + ) + return ActionResult( + status=ActionStatus.SUCCESS, + message=f"GABS executed: {command.action}", + data=result if isinstance(result, dict) else {"result": result}, + ) + except Exception as exc: + logger.warning("async_act(%s) error: %s", command.action, exc) + return ActionResult(status=ActionStatus.FAILURE, message=str(exc)) + + def speak(self, message: str, target: str | None = None) -> None: + """Send a message via GABS (e.g., companion dialogue or overlay).""" + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + return + loop.run_until_complete( + self._client._call("chat/send", {"message": message, "target": target}) + ) + except Exception as exc: + logger.debug("BannerlordWorldAdapter.speak() error: %s", exc) + + # -- helpers ----------------------------------------------------------- + + def _state_to_perception(self, state: GameState) -> PerceptionOutput: + """Map a GameState snapshot to a PerceptionOutput.""" + entities: list[str] = [] + events: list[str] = [] + + # Party location + if state.party.location: + entities.append(f"location:{state.party.location}") + + # Kingdom status + if state.has_kingdom(): + entities.append(f"kingdom:{state.kingdom.name}") + for fief in state.kingdom.fiefs: + entities.append(f"fief:{fief}") + + # Active wars as events + for war in state.kingdom.active_wars: + events.append(f"at_war_with:{war}") + + # Faction snapshot + for faction in state.factions: + entities.append(f"faction:{faction.name}[{faction.army_strength}]") + + # Alerts + if state.is_two_front_war(): + events.append("alert:two_front_war") + if state.party.wounded_pct > 0.30: + events.append(f"alert:wounded_{state.party.wounded_pct:.0%}") + if state.party.food_days < 3: + events.append("alert:low_food") + + return PerceptionOutput( + timestamp=datetime.now(UTC), + location=state.party.location, + entities=entities, + events=events, + raw=state.raw, + ) + + @staticmethod + def _empty_perception() -> PerceptionOutput: + return PerceptionOutput( + timestamp=datetime.now(UTC), + location="", + entities=[], + events=["gabs:unavailable"], + raw={"adapter": "bannerlord", "connected": False}, + ) + + @property + def last_game_state(self) -> GameState | None: + """Return the most recently observed GameState.""" + return self._last_state diff --git a/src/bannerlord/agents/__init__.py b/src/bannerlord/agents/__init__.py new file mode 100644 index 00000000..6067b256 --- /dev/null +++ b/src/bannerlord/agents/__init__.py @@ -0,0 +1,4 @@ +"""Bannerlord M3 — feudal agent hierarchy. + +King → Vassal → Companion, following Ahilan & Dayan (2019). +""" diff --git a/src/bannerlord/agents/companions/__init__.py b/src/bannerlord/agents/companions/__init__.py new file mode 100644 index 00000000..53ff37c9 --- /dev/null +++ b/src/bannerlord/agents/companions/__init__.py @@ -0,0 +1 @@ +"""Bannerlord M3 — Companion worker agents (lowest tier).""" diff --git a/src/bannerlord/agents/companions/caravan.py b/src/bannerlord/agents/companions/caravan.py new file mode 100644 index 00000000..ee541326 --- /dev/null +++ b/src/bannerlord/agents/companions/caravan.py @@ -0,0 +1,61 @@ +"""Bannerlord M3 — Caravan Companion (trade operations). + +Handles trade route assessment, buy/sell goods, caravan deployment. +Triggered by TRADE subgoal or when treasury is below threshold. + +Minimum margin threshold: 15% (never buy goods without ≥ 15% resale margin). +""" + +from __future__ import annotations + +import logging + +from bannerlord.types import GameState, KingSubgoal, SubgoalToken + +logger = logging.getLogger(__name__) + +_MIN_MARGIN_PCT = 0.15 # minimum profitable resale margin +_CARAVAN_DENAR_THRESHOLD = 10_000 # must have 10k denars to deploy a caravan + + +class CaravanCompanion: + """Companion worker for trade route management.""" + + AGENT_ID = "caravan_companion" + + def evaluate( + self, state: GameState, subgoal: KingSubgoal + ) -> list[dict]: + """Return trade primitives to execute. + + Returns: + List of dicts with 'primitive' and 'args' keys. + """ + if subgoal.token != SubgoalToken.TRADE: + return [] + + actions: list[dict] = [] + party = state.party + + # Always assess prices at current location first + actions.append({ + "primitive": "assess_prices", + "args": {"town": party.location}, + }) + + # Deploy a caravan if treasury is flush + if party.denars >= _CARAVAN_DENAR_THRESHOLD and party.location: + actions.append({ + "primitive": "establish_caravan", + "args": {"town": party.location}, + }) + + return actions + + @staticmethod + def is_profitable_trade(buy_price: int, sell_price: int) -> bool: + """Return True if the trade margin meets the minimum threshold.""" + if buy_price <= 0: + return False + margin = (sell_price - buy_price) / buy_price + return margin >= _MIN_MARGIN_PCT diff --git a/src/bannerlord/agents/companions/logistics.py b/src/bannerlord/agents/companions/logistics.py new file mode 100644 index 00000000..4f9a61a1 --- /dev/null +++ b/src/bannerlord/agents/companions/logistics.py @@ -0,0 +1,78 @@ +"""Bannerlord M3 — Logistics Companion (party management). + +Handles recruit, supply, rest, prisoner sale, and troop upgrade primitives. +Runs on Qwen3:8b for sub-2-second response times. + +Triggered by RECRUIT and HEAL subgoals, or by party condition thresholds. +""" + +from __future__ import annotations + +import logging + +from bannerlord.types import GameState, KingSubgoal, SubgoalToken + +logger = logging.getLogger(__name__) + +_FOOD_WARN_DAYS = 5 +_WOUND_WARN_PCT = 0.20 +_PRISONER_CAP = 20 + + +class LogisticsCompanion: + """Companion worker for party logistics. + + Evaluates the current party state and returns a list of primitive + action names + args to dispatch to the GABSClient. + """ + + AGENT_ID = "logistics_companion" + + def evaluate( + self, state: GameState, subgoal: KingSubgoal + ) -> list[dict]: + """Return primitives to execute given current state and active subgoal. + + Returns: + List of dicts with 'primitive' and 'args' keys. + """ + actions: list[dict] = [] + party = state.party + + # Subgoal-driven recruitment + if subgoal.token == SubgoalToken.RECRUIT: + qty = subgoal.quantity or 20 + actions.append({ + "primitive": "recruit_troop", + "args": {"troop_type": "infantry", "qty": qty}, + }) + + # Emergency rest on heavy wounds + if subgoal.token == SubgoalToken.HEAL or party.wounded_pct > _WOUND_WARN_PCT: + actions.append({ + "primitive": "rest_party", + "args": {"days": 3}, + }) + + # Replenish food if low + if party.food_days < _FOOD_WARN_DAYS: + actions.append({ + "primitive": "buy_supplies", + "args": {"qty": max(0, 10 - party.food_days)}, + }) + + # Sell prisoners when near cap + if party.prisoners >= _PRISONER_CAP: + actions.append({ + "primitive": "sell_prisoners", + "args": {"location": party.location}, + }) + + # Upgrade troops when stable + if subgoal.token == SubgoalToken.TRAIN: + actions.append({ + "primitive": "upgrade_troops", + "args": {}, + }) + + return actions diff --git a/src/bannerlord/agents/companions/scout.py b/src/bannerlord/agents/companions/scout.py new file mode 100644 index 00000000..d6c74ec7 --- /dev/null +++ b/src/bannerlord/agents/companions/scout.py @@ -0,0 +1,58 @@ +"""Bannerlord M3 — Scout Companion (intelligence gathering). + +Handles lord tracking, garrison assessment, and patrol route mapping. +Triggered by SPY subgoal or proactively before expansion decisions. +""" + +from __future__ import annotations + +import logging + +from bannerlord.types import GameState, KingSubgoal, SubgoalToken + +logger = logging.getLogger(__name__) + + +class ScoutCompanion: + """Companion worker for tactical intelligence.""" + + AGENT_ID = "scout_companion" + + def evaluate( + self, state: GameState, subgoal: KingSubgoal + ) -> list[dict]: + """Return scouting primitives to execute. + + Returns: + List of dicts with 'primitive' and 'args' keys. + """ + actions: list[dict] = [] + + if subgoal.token == SubgoalToken.SPY: + target = subgoal.target + if target: + actions.append({ + "primitive": "track_lord", + "args": {"lord_name": target}, + }) + + elif subgoal.token == SubgoalToken.EXPAND_TERRITORY: + target = subgoal.target + if target: + actions.append({ + "primitive": "assess_garrison", + "args": {"settlement": target}, + }) + + # Proactively map patrols in active war regions + for war_faction in state.kingdom.active_wars: + # Find a fief belonging to the enemy as the region reference + for faction in state.factions: + if faction.name == war_faction and faction.fiefs: + actions.append({ + "primitive": "map_patrol_routes", + "args": {"region": faction.fiefs[0]}, + }) + break # one region per enemy faction per tick + + return actions diff --git a/src/bannerlord/agents/diplomacy_vassal.py b/src/bannerlord/agents/diplomacy_vassal.py new file mode 100644 index 00000000..97628b8b --- /dev/null +++ b/src/bannerlord/agents/diplomacy_vassal.py @@ -0,0 +1,145 @@ +"""Bannerlord M3 — Diplomacy Vassal agent. + +Handles relations management: alliances, peace deals, tribute, marriage. +Responds to the ALLY subgoal. + +Reward function: + R_diplo = w1 * AlliesCount + + w2 * TruceDurationValue + + w3 * RelationsScore_weighted + - w4 * ActiveWarsFront + + w5 * SubgoalBonus + +Key strategic rules: + - Never start a new war if already in 2+ wars (2-front war rule) + - Prefer peace with weakest current enemy when overextended + - Time alliances before declaring war to reduce isolation risk +""" + +from __future__ import annotations + +import logging + +from bannerlord.types import ( + GameState, + KingSubgoal, + SubgoalToken, + TaskMessage, + VassalReward, +) + +logger = logging.getLogger(__name__) + +_W1_ALLIES = 0.30 +_W2_TRUCE = 0.25 +_W3_RELATIONS = 0.25 +_W4_WAR_FRONTS = 0.15 +_W5_SUBGOAL = 0.05 + +_SUBGOAL_TRIGGERS = {SubgoalToken.ALLY} +_MAX_WAR_FRONTS = 2 # flag when at 2+ simultaneous wars (two-front war) + + +class DiplomacyVassal: + """Mid-tier agent responsible for diplomatic relations.""" + + AGENT_ID = "diplomacy_vassal" + + def is_relevant(self, subgoal: KingSubgoal) -> bool: + return subgoal.token in _SUBGOAL_TRIGGERS + + def plan(self, state: GameState, subgoal: KingSubgoal) -> list[TaskMessage]: + """Return TaskMessages for the current diplomatic subgoal.""" + tasks: list[TaskMessage] = [] + + if subgoal.token == SubgoalToken.ALLY: + tasks.extend(self._plan_alliance(state, subgoal)) + + return tasks + + def _plan_alliance( + self, state: GameState, subgoal: KingSubgoal + ) -> list[TaskMessage]: + """Plan diplomatic outreach to reduce war fronts or build alliances.""" + tasks: list[TaskMessage] = [] + target = subgoal.target + + if not target: + logger.warning("DiplomacyVassal: no target for ALLY subgoal") + return tasks + + # If target is already an enemy, propose peace + if target in state.kingdom.active_wars: + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="propose_peace", + args={"faction": target, "tribute": 0}, + priority=subgoal.priority * 1.5, + ) + ) + else: + # Otherwise pursue alliance + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="send_envoy", + args={ + "faction": target, + "message": "We seek a lasting alliance for mutual defence.", + }, + priority=subgoal.priority, + ) + ) + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="request_alliance", + args={"faction": target}, + priority=subgoal.priority, + ) + ) + + return tasks + + def should_avoid_war(self, state: GameState) -> bool: + """Return True if starting a new war would be strategically unsound.""" + return state.active_war_count() >= _MAX_WAR_FRONTS # 2-front war check + + def compute_reward( + self, + prev_state: GameState, + curr_state: GameState, + active_subgoal: KingSubgoal, + ) -> VassalReward: + """Compute Diplomacy Vassal reward.""" + allies_count = len(curr_state.kingdom.active_alliances) + truce_value = 50.0 # placeholder — days of truce remaining + relations_avg = 30.0 # placeholder — weighted relations score + war_fronts = curr_state.active_war_count() + + subgoal_bonus = 1.0 if active_subgoal.token in _SUBGOAL_TRIGGERS else 0.0 + + total = ( + _W1_ALLIES * allies_count * 10 + + _W2_TRUCE * truce_value / 100 + + _W3_RELATIONS * relations_avg / 100 + - _W4_WAR_FRONTS * war_fronts * 10 + + _W5_SUBGOAL * subgoal_bonus * 10 + ) + + return VassalReward( + agent_id=self.AGENT_ID, + component_scores={ + "allies_count": allies_count, + "truce_value": truce_value, + "relations_avg": relations_avg, + "war_fronts": -war_fronts, + "subgoal_bonus": subgoal_bonus, + }, + subgoal_bonus=subgoal_bonus, + total=total, + ) diff --git a/src/bannerlord/agents/economy_vassal.py b/src/bannerlord/agents/economy_vassal.py new file mode 100644 index 00000000..3d1e18a7 --- /dev/null +++ b/src/bannerlord/agents/economy_vassal.py @@ -0,0 +1,151 @@ +"""Bannerlord M3 — Economy Vassal agent. + +Handles settlement management, tax collection, construction, and food supply. +Responds to FORTIFY and CONSOLIDATE subgoals. + +Reward function: + R_econ = w1 * DailyDenarsIncome + + w2 * FoodStockBuffer + + w3 * LoyaltyAverage + - w4 * ConstructionQueueLength + + w5 * SubgoalBonus +""" + +from __future__ import annotations + +import logging + +from bannerlord.types import ( + GameState, + KingSubgoal, + SubgoalToken, + TaskMessage, + VassalReward, +) + +logger = logging.getLogger(__name__) + +_W1_INCOME = 0.35 +_W2_FOOD = 0.25 +_W3_LOYALTY = 0.20 +_W4_CONSTRUCTION = 0.15 +_W5_SUBGOAL = 0.05 + +_SUBGOAL_TRIGGERS = {SubgoalToken.FORTIFY, SubgoalToken.CONSOLIDATE} + +_LOW_FOOD_THRESHOLD = 3 # days of food remaining +_INCOME_TARGET = 200 # daily net income target (denars) + + +class EconomyVassal: + """Mid-tier agent responsible for settlement economy.""" + + AGENT_ID = "economy_vassal" + + def is_relevant(self, subgoal: KingSubgoal) -> bool: + return subgoal.token in _SUBGOAL_TRIGGERS + + def plan(self, state: GameState, subgoal: KingSubgoal) -> list[TaskMessage]: + """Return TaskMessages for the current economic subgoal.""" + tasks: list[TaskMessage] = [] + + # Always maintain food supply + if state.party.food_days < _LOW_FOOD_THRESHOLD: + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="logistics_companion", + primitive="buy_supplies", + args={"qty": 10}, + priority=2.0, + ) + ) + + if subgoal.token == SubgoalToken.FORTIFY: + tasks.extend(self._plan_fortify(state, subgoal)) + elif subgoal.token == SubgoalToken.CONSOLIDATE: + tasks.extend(self._plan_consolidate(state)) + + return tasks + + def _plan_fortify( + self, state: GameState, subgoal: KingSubgoal + ) -> list[TaskMessage]: + """Queue construction projects in owned settlements.""" + tasks: list[TaskMessage] = [] + target = subgoal.target or (state.kingdom.fiefs[0] if state.kingdom.fiefs else None) + if not target: + return tasks + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="build_project", + args={"settlement": target, "project": "granary"}, + priority=1.2, + ) + ) + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="set_tax_policy", + args={"settlement": target, "policy": "normal"}, + priority=1.0, + ) + ) + return tasks + + def _plan_consolidate(self, state: GameState) -> list[TaskMessage]: + """Stabilise: optimise tax and food across all fiefs.""" + tasks: list[TaskMessage] = [] + net_income = state.kingdom.daily_income - state.kingdom.daily_expenses + for fief in state.kingdom.fiefs: + policy = "normal" if net_income >= _INCOME_TARGET else "low" + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="set_tax_policy", + args={"settlement": fief, "policy": policy}, + priority=0.8, + ) + ) + return tasks + + def compute_reward( + self, + prev_state: GameState, + curr_state: GameState, + active_subgoal: KingSubgoal, + ) -> VassalReward: + """Compute Economy Vassal reward.""" + income_delta = ( + curr_state.kingdom.daily_income - prev_state.kingdom.daily_income + ) + food_buffer = curr_state.party.food_days + loyalty_avg = 70.0 # placeholder — real value from GABS raw data + queue_len = 0 # placeholder + + subgoal_bonus = 1.0 if active_subgoal.token in _SUBGOAL_TRIGGERS else 0.0 + + total = ( + _W1_INCOME * income_delta + + _W2_FOOD * food_buffer + + _W3_LOYALTY * loyalty_avg / 100 + - _W4_CONSTRUCTION * queue_len + + _W5_SUBGOAL * subgoal_bonus * 10 + ) + + return VassalReward( + agent_id=self.AGENT_ID, + component_scores={ + "income_delta": income_delta, + "food_buffer": food_buffer, + "loyalty_avg": loyalty_avg, + "queue_len": -queue_len, + "subgoal_bonus": subgoal_bonus, + }, + subgoal_bonus=subgoal_bonus, + total=total, + ) diff --git a/src/bannerlord/agents/king.py b/src/bannerlord/agents/king.py new file mode 100644 index 00000000..44e2cb58 --- /dev/null +++ b/src/bannerlord/agents/king.py @@ -0,0 +1,266 @@ +"""Bannerlord M3 — King agent (Timmy, strategic tier). + +The King operates on the campaign-map timescale (1 decision per in-game day). +He reads the full GameState and emits a single KingSubgoal token that vassals +interpret. He uses Qwen3:32b via the LLM router. + +Decision rules baked in (no LLM required for simple cases): + - Never initiate a second war while already fighting one (avoid 2-front wars) + - Prioritise HEAL when party wounds > 30 % + - Prioritise RECRUIT when troops < 80 + - Prioritise TRADE when denars < 5,000 + - If kingdom has < 3 fiefs and no active war, prioritise EXPAND_TERRITORY + - Default to CONSOLIDATE when conditions are stable +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime +from typing import Any + +from bannerlord.types import ( + GameState, + KingSubgoal, + SubgoalToken, +) + +logger = logging.getLogger(__name__) + +# Hard thresholds for rule-based fallback decisions +_MIN_TROOPS = 80 +_MIN_DENARS = 5_000 +_MAX_WOUND_PCT = 0.30 +_TARGET_FIEFS = 3 +_SURVIVAL_DAYS = 100 + + +class KingAgent: + """Strategic decision-maker for the Bannerlord campaign. + + The King agent is sovereign — it cannot be terminated by vassals. + It decides the active subgoal at most once per campaign tick. + + Usage:: + + king = KingAgent(model="qwen3:32b") + subgoal = king.decide(game_state) + """ + + def __init__( + self, + *, + model: str = "qwen3:32b", + temperature: float = 0.1, + ) -> None: + self._model = model + self._temperature = temperature + self._last_subgoal: KingSubgoal | None = None + self._tick = 0 + self._session_id: str | None = None + + def set_session(self, session_id: str) -> None: + self._session_id = session_id + + # -- primary decision interface ---------------------------------------- + + def decide(self, state: GameState) -> KingSubgoal: + """Return the King's subgoal for the current campaign tick. + + Uses rule-based heuristics as the primary decision engine. + LLM override can be wired in via ``_llm_decide`` in a future PR. + + Args: + state: Full game state snapshot from GABS. + + Returns: + A KingSubgoal to be broadcast to vassals. + """ + self._tick += 1 + subgoal = self._rule_based_decide(state) + self._last_subgoal = subgoal + logger.info( + "King[tick=%d, day=%d] → %s (target=%s)", + self._tick, + state.in_game_day, + subgoal.token, + subgoal.target, + ) + return subgoal + + # -- rule-based strategy engine ---------------------------------------- + + def _rule_based_decide(self, state: GameState) -> KingSubgoal: + """Encode campaign strategy as prioritised decision rules. + + Priority order (highest to lowest): + 1. Emergency: heal if heavily wounded + 2. Survival: recruit if dangerously low on troops + 3. Economy: earn income if broke + 4. Diplomacy: seek peace if in a 2-front war + 5. Expansion: take fiefs if not at war and need more territory + 6. Alliance: seek allies when preparing for war + 7. Default: consolidate and stabilise + """ + party = state.party + kingdom = state.kingdom + + # 1. Emergency heal + if party.wounded_pct > _MAX_WOUND_PCT: + return KingSubgoal( + token=SubgoalToken.HEAL, + context=f"{party.wounded_pct:.0%} of party is wounded — rest required", + ) + + # 2. Critical recruitment + if party.troops < _MIN_TROOPS: + return KingSubgoal( + token=SubgoalToken.RECRUIT, + quantity=_MIN_TROOPS - party.troops, + context=f"Party at {party.troops} troops — must reach {_MIN_TROOPS}", + ) + + # 3. Destitute treasury + if party.denars < _MIN_DENARS: + return KingSubgoal( + token=SubgoalToken.TRADE, + context=f"Treasury at {party.denars:,} denars — run trade routes", + ) + + # 4. Avoid 2-front war: seek peace when fighting 2+ enemies + if state.is_two_front_war(): + # Pick the weakest enemy to negotiate peace with first + weakest = self._weakest_enemy(state) + return KingSubgoal( + token=SubgoalToken.ALLY, + target=weakest, + context="2-front war detected — de-escalate with weakest enemy", + ) + + # 5. Kingdom not yet established: work toward first fief + if not state.has_kingdom(): + if party.troops >= 120 and state.active_war_count() == 0: + target_fief = self._select_expansion_target(state) + return KingSubgoal( + token=SubgoalToken.EXPAND_TERRITORY, + target=target_fief, + context="No kingdom yet — capture a fief to establish one", + ) + elif state.active_war_count() == 0: + # Not ready to fight; train up first + return KingSubgoal( + token=SubgoalToken.TRAIN, + context="Building army before first expansion", + ) + + # 6. Expand if below target fief count and no active war + if state.fief_count() < _TARGET_FIEFS and state.active_war_count() == 0: + target_fief = self._select_expansion_target(state) + return KingSubgoal( + token=SubgoalToken.EXPAND_TERRITORY, + target=target_fief, + priority=1.5, + context=f"Only {state.fief_count()} fiefs — need {_TARGET_FIEFS}", + ) + + # 7. Seek allies when stable and below fief target + if not kingdom.active_alliances and state.active_war_count() == 0: + ally_candidate = self._best_alliance_candidate(state) + if ally_candidate: + return KingSubgoal( + token=SubgoalToken.ALLY, + target=ally_candidate, + context="Stable moment — pursue defensive alliance", + ) + + # 8. Fortify if kingdom exists and there are fiefs to improve + if state.has_kingdom() and state.fief_count() > 0: + if kingdom.daily_income - kingdom.daily_expenses < 100: + return KingSubgoal( + token=SubgoalToken.FORTIFY, + context="Low net income — invest in settlements", + ) + + # 9. Default: consolidate + return KingSubgoal( + token=SubgoalToken.CONSOLIDATE, + context="Stable — hold territory and recover strength", + ) + + # -- helper methods ---------------------------------------------------- + + def _weakest_enemy(self, state: GameState) -> str | None: + """Return the name of the weakest faction currently at war with us.""" + enemy_names = set(state.kingdom.active_wars) + enemies = [f for f in state.factions if f.name in enemy_names] + if not enemies: + return None + return min(enemies, key=lambda f: f.army_strength).name + + def _select_expansion_target(self, state: GameState) -> str | None: + """Select the most vulnerable enemy settlement to target.""" + # Prefer factions already at war with others (distracted) + for faction in state.factions: + if len(faction.is_at_war_with) >= 2 and faction.fiefs: + return faction.fiefs[0] + # Fallback: weakest faction with fiefs + candidates = [f for f in state.factions if f.fiefs] + if candidates: + weakest = min(candidates, key=lambda f: f.army_strength) + return weakest.fiefs[0] + return None + + def _best_alliance_candidate(self, state: GameState) -> str | None: + """Return the best faction to approach for an alliance.""" + # Prefer factions with good relations and no active war with us + enemy_names = set(state.kingdom.active_wars) + candidates = [ + f + for f in state.factions + if f.name not in enemy_names + and f.name != state.kingdom.name + ] + if not candidates: + return None + # Pick the strongest candidate (most useful ally) + return max(candidates, key=lambda f: f.army_strength).name + + # -- accessors --------------------------------------------------------- + + @property + def last_subgoal(self) -> KingSubgoal | None: + return self._last_subgoal + + @property + def tick(self) -> int: + return self._tick + + def campaign_summary(self, state: GameState) -> dict[str, Any]: + """Return a brief summary of the campaign status.""" + return { + "tick": self._tick, + "in_game_day": state.in_game_day, + "has_kingdom": state.has_kingdom(), + "fief_count": state.fief_count(), + "active_wars": state.active_war_count(), + "two_front_war": state.is_two_front_war(), + "troops": state.party.troops, + "denars": state.party.denars, + "survival_goal_met": ( + state.has_kingdom() + and state.fief_count() >= _TARGET_FIEFS + and state.in_game_day >= _SURVIVAL_DAYS + ), + } + + def is_done_condition_met(self, state: GameState) -> bool: + """Return True when the M3 done-when condition is satisfied. + + Done when: Timmy establishes own kingdom with 3+ fiefs and + survives 100 in-game days as ruler. + """ + return ( + state.has_kingdom() + and state.fief_count() >= _TARGET_FIEFS + and state.in_game_day >= _SURVIVAL_DAYS + ) diff --git a/src/bannerlord/agents/war_vassal.py b/src/bannerlord/agents/war_vassal.py new file mode 100644 index 00000000..abeb179d --- /dev/null +++ b/src/bannerlord/agents/war_vassal.py @@ -0,0 +1,236 @@ +"""Bannerlord M3 — War Vassal agent. + +Handles military operations: sieges, field battles, raids, defensive +maneuvers. Responds to EXPAND_TERRITORY, RAID_ECONOMY, and TRAIN subgoals. + +Reward function (from feudal hierarchy design): + R_war = w1 * ΔTerritoryValue + + w2 * ΔArmyStrength_ratio + - w3 * CasualtyCost + - w4 * SupplyCost + + w5 * SubgoalBonus +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from bannerlord.types import ( + GameState, + KingSubgoal, + SubgoalToken, + TaskMessage, + VassalReward, +) + +if TYPE_CHECKING: + pass + +logger = logging.getLogger(__name__) + +# Reward weights +_W1_TERRITORY = 0.40 +_W2_ARMY_RATIO = 0.25 +_W3_CASUALTY = 0.20 +_W4_SUPPLY = 0.10 +_W5_SUBGOAL = 0.05 + +_SUBGOAL_TRIGGERS = { + SubgoalToken.EXPAND_TERRITORY, + SubgoalToken.RAID_ECONOMY, + SubgoalToken.TRAIN, +} + + +@dataclass +class WarContext: + """Mutable state tracked across War Vassal decisions.""" + + active_siege: str | None = None + last_auto_resolve_result: dict = field(default_factory=dict) + territory_gained: int = 0 + casualties_taken: int = 0 + + +class WarVassal: + """Mid-tier agent responsible for military operations. + + Runs at 4× the King's decision frequency. Translates KingSubgoals + into concrete TaskMessages for the Logistics Companion (troop management) + and issues direct GABS calls for combat actions. + """ + + AGENT_ID = "war_vassal" + + def __init__(self) -> None: + self._ctx = WarContext() + self._prev_army_ratio: float = 1.0 + + def is_relevant(self, subgoal: KingSubgoal) -> bool: + """Return True if this vassal should act on *subgoal*.""" + return subgoal.token in _SUBGOAL_TRIGGERS + + def plan(self, state: GameState, subgoal: KingSubgoal) -> list[TaskMessage]: + """Return a list of TaskMessages for the current subgoal. + + Args: + state: Current game state. + subgoal: Active King subgoal. + + Returns: + Ordered list of TaskMessages to dispatch. + """ + tasks: list[TaskMessage] = [] + + if subgoal.token == SubgoalToken.EXPAND_TERRITORY: + tasks.extend(self._plan_expansion(state, subgoal)) + elif subgoal.token == SubgoalToken.RAID_ECONOMY: + tasks.extend(self._plan_raid(state, subgoal)) + elif subgoal.token == SubgoalToken.TRAIN: + tasks.extend(self._plan_training(state)) + + return tasks + + def _plan_expansion( + self, state: GameState, subgoal: KingSubgoal + ) -> list[TaskMessage]: + """Plan territory expansion toward subgoal.target.""" + tasks: list[TaskMessage] = [] + target = subgoal.target + + if not target: + logger.warning("WarVassal.EXPAND_TERRITORY: no target in subgoal") + return tasks + + # Scout garrison before sieging + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="scout_companion", + primitive="assess_garrison", + args={"settlement": target}, + priority=1.5, + ) + ) + + # Ensure troops are sufficient (delegate to logistics if thin) + if state.party.troops < 100: + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="logistics_companion", + primitive="recruit_troop", + args={"troop_type": "infantry", "qty": 100 - state.party.troops}, + priority=1.8, + ) + ) + + # Issue the siege order + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="siege_settlement", + args={"settlement": target}, + priority=subgoal.priority, + ) + ) + # Follow up with auto-resolve + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="auto_resolve_battle", + args={}, + priority=subgoal.priority, + ) + ) + return tasks + + def _plan_raid( + self, state: GameState, subgoal: KingSubgoal + ) -> list[TaskMessage]: + """Plan economy raid for denars and food.""" + tasks: list[TaskMessage] = [] + target = subgoal.target or self._nearest_enemy_village(state) + if not target: + return tasks + tasks.append( + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="gabs", + primitive="raid_village", + args={"village": target}, + priority=subgoal.priority, + ) + ) + return tasks + + def _plan_training(self, state: GameState) -> list[TaskMessage]: + """Plan troop training via auto-resolve bandit fights.""" + return [ + TaskMessage( + from_agent=self.AGENT_ID, + to_agent="logistics_companion", + primitive="upgrade_troops", + args={}, + priority=0.8, + ) + ] + + # -- reward computation ------------------------------------------------ + + def compute_reward( + self, + prev_state: GameState, + curr_state: GameState, + active_subgoal: KingSubgoal, + ) -> VassalReward: + """Compute the War Vassal reward signal for the last decision cycle.""" + territory_delta = ( + curr_state.fief_count() - prev_state.fief_count() + ) * 100.0 + + prev_strength = max(prev_state.party.troops, 1) + curr_strength = curr_state.party.troops + army_delta = (curr_strength - prev_strength) / prev_strength + + casualties = max(0, prev_state.party.troops - curr_strength) + supply_burn = max(0, prev_state.party.food_days - curr_state.party.food_days) + + subgoal_bonus = ( + 1.0 if active_subgoal.token in _SUBGOAL_TRIGGERS else 0.0 + ) + + total = ( + _W1_TERRITORY * territory_delta + + _W2_ARMY_RATIO * army_delta * 10 + - _W3_CASUALTY * casualties + - _W4_SUPPLY * supply_burn + + _W5_SUBGOAL * subgoal_bonus * 10 + ) + + return VassalReward( + agent_id=self.AGENT_ID, + component_scores={ + "territory": territory_delta, + "army_ratio": army_delta, + "casualties": -casualties, + "supply_burn": -supply_burn, + "subgoal_bonus": subgoal_bonus, + }, + subgoal_bonus=subgoal_bonus, + total=total, + ) + + # -- helpers ----------------------------------------------------------- + + @staticmethod + def _nearest_enemy_village(state: GameState) -> str | None: + enemy_names = set(state.kingdom.active_wars) + for faction in state.factions: + if faction.name in enemy_names and faction.fiefs: + return faction.fiefs[0] + return None diff --git a/src/bannerlord/campaign.py b/src/bannerlord/campaign.py new file mode 100644 index 00000000..10415499 --- /dev/null +++ b/src/bannerlord/campaign.py @@ -0,0 +1,270 @@ +"""Bannerlord M3 — Campaign orchestrator. + +Ties together the King agent, vassals, companions, GABS client, and session +memory into a single async campaign loop. + +Architecture:: + + CampaignOrchestrator.run() + ├── GABSClient.get_game_state() → GameState + ├── KingAgent.decide(state) → KingSubgoal + ├── SessionMemory.log_subgoal(...) + ├── WarVassal.plan(state, subgoal) → [TaskMessage] + ├── EconomyVassal.plan(state, subgoal) → [TaskMessage] + ├── DiplomacyVassal.plan(state, subgoal)→ [TaskMessage] + ├── [Companions].evaluate(state, subgoal) → [primitives] + └── _dispatch_tasks([...]) → GABS calls + +Usage:: + + from bannerlord.campaign import CampaignOrchestrator + orch = CampaignOrchestrator() + await orch.run(max_ticks=100) +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from typing import Any + +from bannerlord.agents.companions.caravan import CaravanCompanion +from bannerlord.agents.companions.logistics import LogisticsCompanion +from bannerlord.agents.companions.scout import ScoutCompanion +from bannerlord.agents.diplomacy_vassal import DiplomacyVassal +from bannerlord.agents.economy_vassal import EconomyVassal +from bannerlord.agents.king import KingAgent +from bannerlord.agents.war_vassal import WarVassal +from bannerlord.gabs_client import GABSClient +from bannerlord.session_memory import SessionMemory +from bannerlord.types import GameState, KingSubgoal, TaskMessage + +logger = logging.getLogger(__name__) + +_DEFAULT_TICK_INTERVAL = 1.0 # seconds between campaign ticks (real time) +_DEFAULT_DB_PATH = Path("data/bannerlord/campaign.db") +_KINGDOM_NAME = "House Timmerson" + + +class CampaignOrchestrator: + """Full-campaign strategy orchestrator for Bannerlord M3. + + Runs the King → Vassal → Companion decision loop on each campaign tick. + Persists progress to SQLite via SessionMemory. + + Args: + gabs_host: Hostname where GABS mod is listening. + gabs_port: TCP port for GABS JSON-RPC (default 4825). + tick_interval: Real-time seconds between campaign ticks. + db_path: Path to the SQLite session memory database. + session_id: Existing session to resume (None = new session). + """ + + def __init__( + self, + *, + gabs_host: str = "127.0.0.1", + gabs_port: int = 4825, + tick_interval: float = _DEFAULT_TICK_INTERVAL, + db_path: Path = _DEFAULT_DB_PATH, + session_id: str | None = None, + ) -> None: + self._gabs = GABSClient(host=gabs_host, port=gabs_port) + self._king = KingAgent() + self._war = WarVassal() + self._economy = EconomyVassal() + self._diplomacy = DiplomacyVassal() + self._logistics = LogisticsCompanion() + self._caravan = CaravanCompanion() + self._scout = ScoutCompanion() + self._memory = SessionMemory(db_path) + self._tick_interval = tick_interval + self._session_id = session_id + self._running = False + self._prev_state: GameState | None = None + + # -- lifecycle --------------------------------------------------------- + + async def start(self) -> str: + """Connect to GABS and initialise a campaign session. + + Returns the active session_id. + """ + connected = await self._gabs.connect() + if not connected: + logger.warning( + "CampaignOrchestrator: GABS unavailable — campaign will run " + "in degraded mode (no game state updates)" + ) + + if self._session_id is None: + self._session_id = self._memory.start_session() + else: + # Resume existing session + existing = self._memory.get_session(self._session_id) + if not existing: + self._session_id = self._memory.start_session(self._session_id) + + self._king.set_session(self._session_id) + logger.info("CampaignOrchestrator: session=%s", self._session_id) + return self._session_id + + async def stop(self) -> None: + """Gracefully stop the campaign and disconnect from GABS.""" + self._running = False + await self._gabs.disconnect() + logger.info("CampaignOrchestrator: stopped") + + # -- main campaign loop ------------------------------------------------ + + async def run(self, max_ticks: int = 0) -> dict[str, Any]: + """Run the campaign loop. + + Args: + max_ticks: Stop after this many ticks. 0 = run indefinitely. + + Returns: + Campaign summary dict. + """ + if not self._session_id: + await self.start() + + self._running = True + tick = 0 + + logger.info( + "CampaignOrchestrator: starting campaign loop (max_ticks=%d)", + max_ticks, + ) + + try: + while self._running: + if max_ticks > 0 and tick >= max_ticks: + break + + await self._tick(tick) + tick += 1 + + # Check done condition + if self._prev_state and self._king.is_done_condition_met( + self._prev_state + ): + logger.info( + "CampaignOrchestrator: M3 DONE condition met on tick %d!", tick + ) + self._memory.add_note( + self._session_id or "", + self._prev_state.in_game_day, + "milestone", + "M3 done condition met — kingdom with 3+ fiefs, 100 days survived", + ) + break + + await asyncio.sleep(self._tick_interval) + + except asyncio.CancelledError: + logger.info("CampaignOrchestrator: loop cancelled") + finally: + await self.stop() + + return self._summary(tick) + + async def _tick(self, tick: int) -> None: + """Execute one campaign tick.""" + # 1. Observe + state = await self._gabs.get_game_state() + state.tick = tick + + # 2. King decides + subgoal = self._king.decide(state) + + # 3. Log subgoal to session memory + if self._session_id: + row_id = self._memory.log_subgoal( + self._session_id, tick, state.in_game_day, subgoal + ) + self._memory.update_session( + self._session_id, + in_game_day=state.in_game_day, + fief_count=state.fief_count(), + kingdom_name=state.kingdom.name or None, + ) + + # 4. Vassal planning + tasks: list[TaskMessage] = [] + if self._war.is_relevant(subgoal): + tasks.extend(self._war.plan(state, subgoal)) + if self._economy.is_relevant(subgoal): + tasks.extend(self._economy.plan(state, subgoal)) + if self._diplomacy.is_relevant(subgoal): + tasks.extend(self._diplomacy.plan(state, subgoal)) + + # 5. Companion evaluation + companion_actions = ( + self._logistics.evaluate(state, subgoal) + + self._caravan.evaluate(state, subgoal) + + self._scout.evaluate(state, subgoal) + ) + + # 6. Dispatch tasks + companion primitives to GABS + await self._dispatch_tasks(tasks, state) + await self._dispatch_primitives(companion_actions) + + # 7. Kingdom establishment check + if not state.has_kingdom() and state.fief_count() > 0: + # We have a fief but no kingdom yet — establish one + ok = await self._gabs.establish_kingdom(_KINGDOM_NAME) + if ok and self._session_id: + self._memory.record_kingdom_established( + self._session_id, state.in_game_day, _KINGDOM_NAME + ) + + self._prev_state = state + logger.debug( + "Tick %d: day=%d, subgoal=%s, tasks=%d, companions=%d", + tick, + state.in_game_day, + subgoal.token, + len(tasks), + len(companion_actions), + ) + + # -- task dispatch ----------------------------------------------------- + + async def _dispatch_tasks( + self, tasks: list[TaskMessage], state: GameState + ) -> None: + """Dispatch vassal TaskMessages to GABS.""" + for task in sorted(tasks, key=lambda t: t.priority, reverse=True): + if task.to_agent != "gabs": + # Companion-directed tasks are handled via companion.evaluate() + continue + await self._gabs._call(task.primitive, task.args) + + async def _dispatch_primitives(self, actions: list[dict]) -> None: + """Dispatch companion primitive actions to GABS.""" + for action in actions: + primitive = action.get("primitive", "") + args = action.get("args", {}) + if primitive: + await self._gabs._call(primitive, args) + + # -- summary ----------------------------------------------------------- + + def _summary(self, ticks_run: int) -> dict[str, Any]: + state = self._prev_state or GameState() + summary = self._king.campaign_summary(state) + summary["ticks_run"] = ticks_run + summary["session_id"] = self._session_id + return summary + + # -- accessors --------------------------------------------------------- + + @property + def session_id(self) -> str | None: + return self._session_id + + @property + def memory(self) -> SessionMemory: + return self._memory diff --git a/src/bannerlord/gabs_client.py b/src/bannerlord/gabs_client.py new file mode 100644 index 00000000..05a16b38 --- /dev/null +++ b/src/bannerlord/gabs_client.py @@ -0,0 +1,434 @@ +"""GABS TCP JSON-RPC client — connects to the Bannerlord.GABS mod. + +The GABS (Game Agent Behavior System) mod exposes 90+ tools via a +TCP JSON-RPC 2.0 server on port 4825. This client wraps the transport +into a clean async interface used by all Bannerlord agents. + +Degrades gracefully: if GABS is unreachable, methods return sensible +fallbacks and log a warning (never crash). + +Architecture reference: + Timmy (Qwen3 on Ollama, M3 Max) + → GABSClient (this module, TCP JSON-RPC, port 4825) + → Bannerlord.GABS C# mod + → Game API + Harmony + → Bannerlord (Windows VM) +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +from bannerlord.types import ( + FactionState, + GameState, + KingdomState, + PartyState, +) + +logger = logging.getLogger(__name__) + +_DEFAULT_HOST = "127.0.0.1" +_DEFAULT_PORT = 4825 +_DEFAULT_TIMEOUT = 10.0 # seconds +_RECONNECT_DELAY = 5.0 # seconds between reconnect attempts + + +@dataclass +class GABSTool: + """Metadata for a single GABS tool.""" + + name: str + description: str + parameters: dict[str, Any] = field(default_factory=dict) + + +class GABSConnectionError(Exception): + """Raised when GABS is unreachable and a fallback is not possible.""" + + +class GABSClient: + """Async TCP JSON-RPC 2.0 client for the Bannerlord.GABS mod. + + Usage:: + + async with GABSClient() as client: + state = await client.get_game_state() + await client.move_party("Vlandia") + + All public methods degrade gracefully — they return ``None`` or an + empty structure when GABS is unavailable. + """ + + def __init__( + self, + host: str = _DEFAULT_HOST, + port: int = _DEFAULT_PORT, + timeout: float = _DEFAULT_TIMEOUT, + ) -> None: + self._host = host + self._port = port + self._timeout = timeout + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._connected = False + self._call_id = 0 + self._available_tools: list[GABSTool] = [] + + # -- lifecycle --------------------------------------------------------- + + async def connect(self) -> bool: + """Open a TCP connection to GABS. + + Returns: + True if connected successfully, False if GABS is unavailable. + """ + try: + self._reader, self._writer = await asyncio.wait_for( + asyncio.open_connection(self._host, self._port), + timeout=self._timeout, + ) + self._connected = True + logger.info("GABSClient connected to %s:%d", self._host, self._port) + await self._discover_tools() + return True + except (ConnectionRefusedError, OSError, TimeoutError, asyncio.TimeoutError) as exc: + logger.warning( + "GABSClient could not connect to %s:%d — %s", + self._host, + self._port, + exc, + ) + self._connected = False + return False + + async def disconnect(self) -> None: + """Close the TCP connection.""" + if self._writer is not None: + try: + self._writer.close() + await self._writer.wait_closed() + except Exception as exc: + logger.debug("GABSClient disconnect error: %s", exc) + self._connected = False + logger.info("GABSClient disconnected") + + async def __aenter__(self) -> GABSClient: + await self.connect() + return self + + async def __aexit__(self, *_: object) -> None: + await self.disconnect() + + @property + def is_connected(self) -> bool: + return self._connected + + # -- raw JSON-RPC transport -------------------------------------------- + + def _next_id(self) -> int: + self._call_id += 1 + return self._call_id + + async def _call(self, method: str, params: dict[str, Any] | None = None) -> Any: + """Send a JSON-RPC 2.0 request and return the result. + + Returns ``None`` and logs a warning on any error. + """ + if not self._connected: + logger.warning("GABSClient._call(%s): not connected", method) + return None + + request = { + "jsonrpc": "2.0", + "id": self._next_id(), + "method": method, + "params": params or {}, + } + + try: + payload = json.dumps(request) + "\n" + assert self._writer is not None + self._writer.write(payload.encode()) + await asyncio.wait_for(self._writer.drain(), timeout=self._timeout) + + assert self._reader is not None + raw = await asyncio.wait_for( + self._reader.readline(), timeout=self._timeout + ) + response = json.loads(raw.decode().strip()) + + if "error" in response: + logger.warning( + "GABS error for %s: %s", method, response["error"].get("message") + ) + return None + + return response.get("result") + + except (asyncio.TimeoutError, json.JSONDecodeError, AssertionError, OSError) as exc: + logger.warning("GABSClient._call(%s) failed: %s", method, exc) + self._connected = False + return None + + # -- tool discovery ---------------------------------------------------- + + async def _discover_tools(self) -> None: + """Populate self._available_tools via GABS tools/list.""" + result = await self._call("tools/list") + if not result: + return + self._available_tools = [ + GABSTool( + name=t.get("name", ""), + description=t.get("description", ""), + parameters=t.get("parameters", {}), + ) + for t in (result if isinstance(result, list) else []) + ] + logger.info("GABS: discovered %d tools", len(self._available_tools)) + + @property + def available_tools(self) -> list[GABSTool]: + """Return the list of tools discovered from GABS.""" + return list(self._available_tools) + + def tool_count(self) -> int: + return len(self._available_tools) + + # -- game state -------------------------------------------------------- + + async def get_game_state(self) -> GameState: + """Return the full campaign state snapshot. + + Falls back to an empty GameState if GABS is unavailable. + """ + raw = await self._call("game/get_state") + if raw is None: + return GameState() + + try: + return self._parse_game_state(raw) + except Exception as exc: + logger.warning("Failed to parse GABS game state: %s", exc) + return GameState() + + def _parse_game_state(self, raw: dict[str, Any]) -> GameState: + """Convert raw GABS state dict into a typed GameState.""" + party_raw = raw.get("party", {}) + kingdom_raw = raw.get("kingdom", {}) + factions_raw = raw.get("factions", []) + + party = PartyState( + location=party_raw.get("location", ""), + troops=party_raw.get("troops", 0), + food_days=party_raw.get("food_days", 0), + wounded_pct=party_raw.get("wounded_pct", 0.0), + denars=party_raw.get("denars", 0), + morale=party_raw.get("morale", 100.0), + prisoners=party_raw.get("prisoners", 0), + ) + + kingdom = KingdomState( + name=kingdom_raw.get("name", ""), + fiefs=kingdom_raw.get("fiefs", []), + daily_income=kingdom_raw.get("daily_income", 0), + daily_expenses=kingdom_raw.get("daily_expenses", 0), + vassal_lords=kingdom_raw.get("vassal_lords", []), + active_wars=kingdom_raw.get("active_wars", []), + active_alliances=kingdom_raw.get("active_alliances", []), + in_game_day=raw.get("in_game_day", 0), + ) + + factions = [ + FactionState( + name=f.get("name", ""), + leader=f.get("leader", ""), + fiefs=f.get("fiefs", []), + army_strength=f.get("army_strength", 0), + treasury=f.get("treasury", 0), + is_at_war_with=f.get("is_at_war_with", []), + relations=f.get("relations", {}), + ) + for f in (factions_raw if isinstance(factions_raw, list) else []) + ] + + return GameState( + tick=raw.get("tick", 0), + in_game_day=raw.get("in_game_day", 0), + timestamp=datetime.now(UTC), + party=party, + kingdom=kingdom, + factions=factions, + raw=raw, + ) + + # -- party actions ----------------------------------------------------- + + async def move_party(self, destination: str) -> bool: + """Command Timmy's party to move toward *destination*.""" + result = await self._call("party/move", {"destination": destination}) + return result is not None + + async def recruit_troops(self, troop_type: str, quantity: int) -> bool: + """Recruit *quantity* troops of *troop_type* at current location.""" + result = await self._call( + "party/recruit", + {"troop_type": troop_type, "quantity": quantity}, + ) + return result is not None + + async def buy_supplies(self, quantity: int) -> bool: + """Purchase food supplies for *quantity* days of march.""" + result = await self._call("party/buy_supplies", {"quantity": quantity}) + return result is not None + + async def rest_party(self, days: int) -> bool: + """Rest the party in current location for *days* in-game days.""" + result = await self._call("party/rest", {"days": days}) + return result is not None + + async def auto_resolve_battle(self) -> dict[str, Any]: + """Trigger auto-resolve for the current battle. + + Returns the battle outcome dict, or empty dict on failure. + """ + result = await self._call("battle/auto_resolve") + return result if isinstance(result, dict) else {} + + async def upgrade_troops(self) -> bool: + """Spend accumulated XP on troop tier upgrades.""" + result = await self._call("party/upgrade_troops") + return result is not None + + async def sell_prisoners(self, location: str) -> int: + """Sell prisoners at *location*. Returns denars gained.""" + result = await self._call("party/sell_prisoners", {"location": location}) + if isinstance(result, dict): + return result.get("denars_gained", 0) + return 0 + + # -- trade actions ----------------------------------------------------- + + async def assess_prices(self, town: str) -> dict[str, Any]: + """Query buy/sell prices at *town*.""" + result = await self._call("trade/assess_prices", {"town": town}) + return result if isinstance(result, dict) else {} + + async def buy_goods(self, item: str, quantity: int) -> bool: + """Purchase *quantity* of *item* at current location.""" + result = await self._call("trade/buy", {"item": item, "quantity": quantity}) + return result is not None + + async def sell_goods(self, item: str, quantity: int, location: str) -> bool: + """Sell *quantity* of *item* at *location*.""" + result = await self._call( + "trade/sell", + {"item": item, "quantity": quantity, "location": location}, + ) + return result is not None + + async def establish_caravan(self, town: str) -> bool: + """Deploy a caravan NPC at *town*.""" + result = await self._call("trade/establish_caravan", {"town": town}) + return result is not None + + # -- diplomacy actions ------------------------------------------------- + + async def send_envoy(self, faction: str, message: str) -> bool: + """Send a diplomatic message to *faction*.""" + result = await self._call( + "diplomacy/send_envoy", + {"faction": faction, "message": message}, + ) + return result is not None + + async def propose_peace(self, faction: str, tribute: int = 0) -> bool: + """Propose peace with *faction*, optionally offering *tribute* denars.""" + result = await self._call( + "diplomacy/propose_peace", + {"faction": faction, "tribute": tribute}, + ) + return result is not None + + async def request_alliance(self, faction: str) -> bool: + """Request a military alliance with *faction*.""" + result = await self._call( + "diplomacy/request_alliance", + {"faction": faction}, + ) + return result is not None + + async def request_military_access(self, faction: str) -> bool: + """Request military access through *faction*'s territory.""" + result = await self._call( + "diplomacy/military_access", + {"faction": faction}, + ) + return result is not None + + # -- settlement / kingdom actions -------------------------------------- + + async def siege_settlement(self, settlement: str) -> bool: + """Begin siege of *settlement*.""" + result = await self._call("military/siege", {"settlement": settlement}) + return result is not None + + async def raid_village(self, village: str) -> bool: + """Raid *village* for food and denars.""" + result = await self._call("military/raid_village", {"village": village}) + return result is not None + + async def build_project(self, settlement: str, project: str) -> bool: + """Queue a construction *project* in *settlement*.""" + result = await self._call( + "settlement/build", + {"settlement": settlement, "project": project}, + ) + return result is not None + + async def set_tax_policy(self, settlement: str, policy: str) -> bool: + """Set tax *policy* for *settlement* (e.g. 'low', 'normal', 'high').""" + result = await self._call( + "settlement/set_tax", + {"settlement": settlement, "policy": policy}, + ) + return result is not None + + async def appoint_governor(self, settlement: str, lord: str) -> bool: + """Appoint *lord* as governor of *settlement*.""" + result = await self._call( + "settlement/appoint_governor", + {"settlement": settlement, "lord": lord}, + ) + return result is not None + + async def establish_kingdom(self, name: str) -> bool: + """Declare a new kingdom with *name* (requires a captured fief).""" + result = await self._call("kingdom/establish", {"name": name}) + ok = result is not None + if ok: + logger.info("Kingdom '%s' established!", name) + return ok + + # -- scouting ---------------------------------------------------------- + + async def track_lord(self, lord_name: str) -> dict[str, Any]: + """Shadow *lord_name* and return their last-known position.""" + result = await self._call("scout/track_lord", {"lord": lord_name}) + return result if isinstance(result, dict) else {} + + async def assess_garrison(self, settlement: str) -> dict[str, Any]: + """Estimate defender count for *settlement*.""" + result = await self._call("scout/assess_garrison", {"settlement": settlement}) + return result if isinstance(result, dict) else {} + + async def map_patrol_routes(self, region: str) -> list[dict[str, Any]]: + """Log enemy patrol routes in *region*.""" + result = await self._call("scout/patrol_routes", {"region": region}) + return result if isinstance(result, list) else [] diff --git a/src/bannerlord/session_memory.py b/src/bannerlord/session_memory.py new file mode 100644 index 00000000..a687dcca --- /dev/null +++ b/src/bannerlord/session_memory.py @@ -0,0 +1,347 @@ +"""Bannerlord M3 — Session memory for multi-day strategic plans. + +Persists the King's strategic plans, completed subgoals, and kingdom +milestones to SQLite so the campaign can be interrupted and resumed +across multiple sessions. + +Pattern follows the existing EventBus persistence model. +""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +from contextlib import closing +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from bannerlord.types import KingSubgoal, SubgoalToken + +logger = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS campaign_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + started_at TEXT NOT NULL, + last_updated TEXT NOT NULL, + kingdom_name TEXT DEFAULT '', + in_game_day INTEGER DEFAULT 0, + fief_count INTEGER DEFAULT 0, + meta TEXT DEFAULT '{}' +); + +CREATE TABLE IF NOT EXISTS subgoal_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + tick INTEGER NOT NULL, + in_game_day INTEGER NOT NULL, + token TEXT NOT NULL, + target TEXT, + quantity INTEGER, + priority REAL DEFAULT 1.0, + deadline_days INTEGER, + context TEXT, + issued_at TEXT NOT NULL, + completed_at TEXT, + outcome TEXT DEFAULT 'pending' +); + +CREATE TABLE IF NOT EXISTS strategy_notes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + in_game_day INTEGER NOT NULL, + note_type TEXT NOT NULL, + content TEXT NOT NULL, + recorded_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_subgoal_session ON subgoal_log(session_id); +CREATE INDEX IF NOT EXISTS idx_subgoal_tick ON subgoal_log(tick); +CREATE INDEX IF NOT EXISTS idx_notes_session ON strategy_notes(session_id); +""" + + +@dataclass +class CampaignMilestone: + """A notable campaign achievement recorded in session memory.""" + + in_game_day: int + event: str + detail: str = "" + recorded_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + +class SessionMemory: + """SQLite-backed session memory for the Bannerlord campaign. + + Stores: + - Active session metadata (kingdom name, in-game day, fief count) + - Full subgoal history (every KingSubgoal issued and its outcome) + - Strategy notes / milestones for campaign reflection + + Usage:: + + mem = SessionMemory(Path("data/bannerlord/campaign.db")) + session_id = mem.start_session() + mem.log_subgoal(session_id, tick=1, day=42, subgoal) + mem.complete_subgoal(subgoal_id, outcome="success") + mem.add_note(session_id, day=42, note_type="milestone", + content="Kingdom established: House Timmerson") + history = mem.get_recent_subgoals(session_id, limit=10) + """ + + def __init__(self, db_path: Path) -> None: + self._db_path = db_path + self._init_db() + + def _init_db(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + with closing(sqlite3.connect(str(self._db_path))) as conn: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + conn.executescript(_SCHEMA) + conn.commit() + + def _conn(self) -> sqlite3.Connection: + conn = sqlite3.connect(str(self._db_path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout=5000") + return conn + + # -- session lifecycle ------------------------------------------------- + + def start_session(self, session_id: str | None = None) -> str: + """Create a new campaign session. Returns the session_id.""" + if session_id is None: + session_id = f"session_{datetime.now(UTC).strftime('%Y%m%d_%H%M%S')}" + now = datetime.now(UTC).isoformat() + with closing(self._conn()) as conn: + conn.execute( + "INSERT OR IGNORE INTO campaign_sessions " + "(session_id, started_at, last_updated) VALUES (?, ?, ?)", + (session_id, now, now), + ) + conn.commit() + logger.info("SessionMemory: started campaign session %s", session_id) + return session_id + + def update_session( + self, + session_id: str, + *, + kingdom_name: str | None = None, + in_game_day: int | None = None, + fief_count: int | None = None, + meta: dict[str, Any] | None = None, + ) -> None: + """Update the campaign session state.""" + now = datetime.now(UTC).isoformat() + with closing(self._conn()) as conn: + row = conn.execute( + "SELECT * FROM campaign_sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if not row: + return + + current_meta = json.loads(row["meta"] or "{}") + if meta: + current_meta.update(meta) + + conn.execute( + """UPDATE campaign_sessions SET + last_updated = ?, + kingdom_name = COALESCE(?, kingdom_name), + in_game_day = COALESCE(?, in_game_day), + fief_count = COALESCE(?, fief_count), + meta = ? + WHERE session_id = ?""", + ( + now, + kingdom_name, + in_game_day, + fief_count, + json.dumps(current_meta), + session_id, + ), + ) + conn.commit() + + def get_session(self, session_id: str) -> dict[str, Any] | None: + """Return session metadata dict or None if not found.""" + with closing(self._conn()) as conn: + row = conn.execute( + "SELECT * FROM campaign_sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if not row: + return None + return dict(row) + + def list_sessions(self) -> list[dict[str, Any]]: + """Return all campaign sessions, most recent first.""" + with closing(self._conn()) as conn: + rows = conn.execute( + "SELECT * FROM campaign_sessions ORDER BY last_updated DESC" + ).fetchall() + return [dict(r) for r in rows] + + # -- subgoal log ------------------------------------------------------- + + def log_subgoal( + self, + session_id: str, + tick: int, + in_game_day: int, + subgoal: KingSubgoal, + ) -> int: + """Record a subgoal emission. Returns the row id.""" + with closing(self._conn()) as conn: + cursor = conn.execute( + """INSERT INTO subgoal_log + (session_id, tick, in_game_day, token, target, quantity, + priority, deadline_days, context, issued_at, outcome) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending')""", + ( + session_id, + tick, + in_game_day, + str(subgoal.token), + subgoal.target, + subgoal.quantity, + subgoal.priority, + subgoal.deadline_days, + subgoal.context, + subgoal.issued_at.isoformat(), + ), + ) + conn.commit() + return cursor.lastrowid or 0 + + def complete_subgoal(self, row_id: int, outcome: str = "success") -> None: + """Mark a subgoal log entry as completed.""" + with closing(self._conn()) as conn: + conn.execute( + "UPDATE subgoal_log SET completed_at = ?, outcome = ? WHERE id = ?", + (datetime.now(UTC).isoformat(), outcome, row_id), + ) + conn.commit() + + def get_recent_subgoals( + self, session_id: str, limit: int = 20 + ) -> list[dict[str, Any]]: + """Return the *limit* most recent subgoal log entries.""" + with closing(self._conn()) as conn: + rows = conn.execute( + "SELECT * FROM subgoal_log WHERE session_id = ? " + "ORDER BY tick DESC LIMIT ?", + (session_id, limit), + ).fetchall() + return [dict(r) for r in rows] + + def count_token(self, session_id: str, token: SubgoalToken) -> int: + """Count how many times a subgoal token has been issued in a session.""" + with closing(self._conn()) as conn: + row = conn.execute( + "SELECT COUNT(*) as n FROM subgoal_log " + "WHERE session_id = ? AND token = ?", + (session_id, str(token)), + ).fetchone() + return row["n"] if row else 0 + + # -- strategy notes ---------------------------------------------------- + + def add_note( + self, + session_id: str, + in_game_day: int, + note_type: str, + content: str, + ) -> None: + """Record a strategy note or milestone.""" + with closing(self._conn()) as conn: + conn.execute( + "INSERT INTO strategy_notes " + "(session_id, in_game_day, note_type, content, recorded_at) " + "VALUES (?, ?, ?, ?, ?)", + ( + session_id, + in_game_day, + note_type, + content, + datetime.now(UTC).isoformat(), + ), + ) + conn.commit() + + def get_notes( + self, + session_id: str, + note_type: str | None = None, + limit: int = 50, + ) -> list[dict[str, Any]]: + """Return strategy notes, optionally filtered by type.""" + with closing(self._conn()) as conn: + if note_type: + rows = conn.execute( + "SELECT * FROM strategy_notes " + "WHERE session_id = ? AND note_type = ? " + "ORDER BY in_game_day DESC LIMIT ?", + (session_id, note_type, limit), + ).fetchall() + else: + rows = conn.execute( + "SELECT * FROM strategy_notes WHERE session_id = ? " + "ORDER BY in_game_day DESC LIMIT ?", + (session_id, limit), + ).fetchall() + return [dict(r) for r in rows] + + def get_milestones(self, session_id: str) -> list[dict[str, Any]]: + """Return all milestone notes for a session.""" + return self.get_notes(session_id, note_type="milestone", limit=200) + + # -- diplomatic memory ------------------------------------------------- + + def record_war_declared( + self, session_id: str, in_game_day: int, faction: str + ) -> None: + """Log that Timmy declared war on *faction*.""" + self.add_note( + session_id, + in_game_day, + "war_declared", + f"Declared war on {faction}", + ) + + def record_peace_agreed( + self, session_id: str, in_game_day: int, faction: str + ) -> None: + """Log that Timmy agreed to peace with *faction*.""" + self.add_note( + session_id, + in_game_day, + "peace_agreed", + f"Peace agreed with {faction}", + ) + + def record_kingdom_established( + self, session_id: str, in_game_day: int, kingdom_name: str + ) -> None: + """Record the kingdom establishment milestone.""" + self.add_note( + session_id, + in_game_day, + "milestone", + f"Kingdom established: {kingdom_name}", + ) + self.update_session(session_id, kingdom_name=kingdom_name, in_game_day=in_game_day) + logger.info( + "SessionMemory: milestone — kingdom '%s' established on day %d", + kingdom_name, + in_game_day, + ) diff --git a/src/bannerlord/types.py b/src/bannerlord/types.py new file mode 100644 index 00000000..ce7cb45a --- /dev/null +++ b/src/bannerlord/types.py @@ -0,0 +1,226 @@ +"""Bannerlord M3 — core data types for the campaign strategy system. + +KingSubgoal schema and all message types used by the feudal agent hierarchy. +Design follows the Feudal Multi-Agent Hierarchies (Ahilan & Dayan, 2019) model +specified in docs/research/bannerlord-feudal-hierarchy-design.md. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from enum import StrEnum +from typing import Any, Literal + + +# --------------------------------------------------------------------------- +# Subgoal vocabulary +# --------------------------------------------------------------------------- + + +class SubgoalToken(StrEnum): + """Fixed vocabulary of strategic intents the King can emit.""" + + EXPAND_TERRITORY = "EXPAND_TERRITORY" + RAID_ECONOMY = "RAID_ECONOMY" + FORTIFY = "FORTIFY" + RECRUIT = "RECRUIT" + TRADE = "TRADE" + ALLY = "ALLY" + SPY = "SPY" + HEAL = "HEAL" + CONSOLIDATE = "CONSOLIDATE" + TRAIN = "TRAIN" + + +@dataclass +class KingSubgoal: + """A strategic directive issued by the King agent. + + The King emits at most one subgoal per campaign tick. Vassals interpret + the token and prioritise actions accordingly. + + Attributes: + token: Intent from the SubgoalToken vocabulary. + target: Named target (settlement, lord, or faction). + quantity: Scalar for RECRUIT / TRADE operations. + priority: 0.0–2.0 weighting that scales vassal reward. + deadline_days: Campaign-map days to complete (None = open-ended). + context: Free-text hint passed verbatim to vassals (not parsed). + issued_at: Timestamp when the King emitted this subgoal. + """ + + token: SubgoalToken + target: str | None = None + quantity: int | None = None + priority: float = 1.0 + deadline_days: int | None = None + context: str | None = None + issued_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + def to_dict(self) -> dict[str, Any]: + return { + "token": str(self.token), + "target": self.target, + "quantity": self.quantity, + "priority": self.priority, + "deadline_days": self.deadline_days, + "context": self.context, + "issued_at": self.issued_at.isoformat(), + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> KingSubgoal: + return cls( + token=SubgoalToken(data["token"]), + target=data.get("target"), + quantity=data.get("quantity"), + priority=data.get("priority", 1.0), + deadline_days=data.get("deadline_days"), + context=data.get("context"), + issued_at=datetime.fromisoformat(data["issued_at"]) + if "issued_at" in data + else datetime.now(UTC), + ) + + +# --------------------------------------------------------------------------- +# Game state snapshot +# --------------------------------------------------------------------------- + + +@dataclass +class FactionState: + """Snapshot of a faction's status on the campaign map.""" + + name: str + leader: str + fiefs: list[str] = field(default_factory=list) + army_strength: int = 0 + treasury: int = 0 + is_at_war_with: list[str] = field(default_factory=list) + relations: dict[str, int] = field(default_factory=dict) + + +@dataclass +class PartyState: + """Timmy's party snapshot.""" + + location: str = "" + troops: int = 0 + food_days: int = 0 + wounded_pct: float = 0.0 + denars: int = 0 + morale: float = 100.0 + prisoners: int = 0 + + +@dataclass +class KingdomState: + """Timmy's kingdom snapshot (only populated after kingdom is established).""" + + name: str = "" + fiefs: list[str] = field(default_factory=list) + daily_income: int = 0 + daily_expenses: int = 0 + vassal_lords: list[str] = field(default_factory=list) + active_wars: list[str] = field(default_factory=list) + active_alliances: list[str] = field(default_factory=list) + in_game_day: int = 0 + + +@dataclass +class GameState: + """Full campaign state snapshot delivered by GABS on each tick. + + This is the primary input to the King agent's decision loop. + """ + + tick: int = 0 + in_game_day: int = 0 + timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) + party: PartyState = field(default_factory=PartyState) + kingdom: KingdomState = field(default_factory=KingdomState) + factions: list[FactionState] = field(default_factory=list) + raw: dict[str, Any] = field(default_factory=dict) + + def has_kingdom(self) -> bool: + return bool(self.kingdom.name) + + def fief_count(self) -> int: + return len(self.kingdom.fiefs) + + def active_war_count(self) -> int: + return len(self.kingdom.active_wars) + + def is_two_front_war(self) -> bool: + """Return True if Timmy is engaged in 2+ simultaneous wars.""" + return self.active_war_count() >= 2 + + +# --------------------------------------------------------------------------- +# Inter-agent message schemas +# --------------------------------------------------------------------------- + + +@dataclass +class SubgoalMessage: + """King → Vassal directive.""" + + msg_type: Literal["subgoal"] = "subgoal" + from_agent: Literal["king"] = "king" + to_agent: str = "" + subgoal: KingSubgoal | None = None + issued_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + +@dataclass +class TaskMessage: + """Vassal → Companion work order.""" + + msg_type: Literal["task"] = "task" + from_agent: str = "" + to_agent: str = "" + primitive: str = "" + args: dict[str, Any] = field(default_factory=dict) + priority: float = 1.0 + issued_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + +@dataclass +class ResultMessage: + """Companion/Vassal → Parent outcome report.""" + + msg_type: Literal["result"] = "result" + from_agent: str = "" + to_agent: str = "" + success: bool = True + outcome: dict[str, Any] = field(default_factory=dict) + reward_delta: float = 0.0 + completed_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + +@dataclass +class StateUpdateMessage: + """GABS → All agents broadcast.""" + + msg_type: Literal["state"] = "state" + game_state: GameState = field(default_factory=GameState) + tick: int = 0 + timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) + + +# --------------------------------------------------------------------------- +# Reward signals +# --------------------------------------------------------------------------- + + +@dataclass +class VassalReward: + """Computed reward signal for a vassal agent after one decision cycle.""" + + agent_id: str + component_scores: dict[str, float] = field(default_factory=dict) + subgoal_bonus: float = 0.0 + total: float = 0.0 + computed_at: datetime = field(default_factory=lambda: datetime.now(UTC)) diff --git a/src/config.py b/src/config.py index f0c922cc..658bec29 100644 --- a/src/config.py +++ b/src/config.py @@ -147,6 +147,15 @@ class Settings(BaseSettings): l402_macaroon_secret: str = "" lightning_backend: Literal["mock", "lnd"] = "mock" + # ── Bannerlord / GABS ──────────────────────────────────────────────── + # TCP JSON-RPC connection to the Bannerlord.GABS mod running on the + # Windows VM. Override with GABS_HOST / GABS_PORT env vars. + gabs_host: str = "127.0.0.1" + gabs_port: int = 4825 + gabs_timeout: float = 10.0 # seconds per GABS call + bannerlord_tick_interval: float = 1.0 # real-time seconds between campaign ticks + bannerlord_db_path: str = "data/bannerlord/campaign.db" + # ── Privacy / Sovereignty ──────────────────────────────────────────── # Disable Agno telemetry for air-gapped/sovereign deployments. # Default is False (telemetry disabled) to align with sovereign AI vision. diff --git a/src/infrastructure/world/__init__.py b/src/infrastructure/world/__init__.py index 4bd63406..3afa169d 100644 --- a/src/infrastructure/world/__init__.py +++ b/src/infrastructure/world/__init__.py @@ -12,6 +12,11 @@ Quick start:: register_adapter("mock", MockWorldAdapter) world = get_adapter("mock") perception = world.observe() + +Registered adapters: + "mock" — in-memory stub for testing + "tes3mp" — Morrowind multiplayer (stub, pending PR #864) + "bannerlord" — Bannerlord via GABS mod (M3 campaign strategy) """ from infrastructure.world.registry import AdapterRegistry @@ -22,6 +27,27 @@ register_adapter = _registry.register get_adapter = _registry.get list_adapters = _registry.list_adapters +# -- Built-in adapter registration ----------------------------------------- +# Adapters are registered lazily to avoid import errors when their +# optional dependencies (e.g., GABS TCP connection) are unavailable. + +def _register_builtin_adapters() -> None: + from infrastructure.world.adapters.mock import MockWorldAdapter + from infrastructure.world.adapters.tes3mp import TES3MPWorldAdapter + + _registry.register("mock", MockWorldAdapter) + _registry.register("tes3mp", TES3MPWorldAdapter) + + try: + from bannerlord.adapter import BannerlordWorldAdapter + _registry.register("bannerlord", BannerlordWorldAdapter) + except Exception: + # bannerlord package not installed or import error — skip silently + pass + + +_register_builtin_adapters() + __all__ = [ "register_adapter", "get_adapter", diff --git a/tests/bannerlord/__init__.py b/tests/bannerlord/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bannerlord/test_campaign.py b/tests/bannerlord/test_campaign.py new file mode 100644 index 00000000..ad68b295 --- /dev/null +++ b/tests/bannerlord/test_campaign.py @@ -0,0 +1,146 @@ +"""Tests for the CampaignOrchestrator — mocked GABS client.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from bannerlord.campaign import CampaignOrchestrator +from bannerlord.types import ( + FactionState, + GameState, + KingdomState, + PartyState, +) + + +def _make_state( + *, + in_game_day: int = 50, + troops: int = 150, + denars: int = 10_000, + food_days: int = 10, + kingdom_name: str = "House Timmerson", + fiefs: list | None = None, + active_wars: list | None = None, +) -> GameState: + return GameState( + in_game_day=in_game_day, + party=PartyState( + troops=troops, + denars=denars, + food_days=food_days, + location="Epicrotea", + ), + kingdom=KingdomState( + name=kingdom_name, + fiefs=fiefs or ["Epicrotea"], + active_wars=active_wars or [], + daily_income=500, + daily_expenses=300, + ), + factions=[ + FactionState( + name="Vlandia", + leader="Derthert", + fiefs=["Pravend"], + army_strength=200, + ) + ], + ) + + +@pytest.fixture +def tmp_db(tmp_path): + return tmp_path / "test_campaign.db" + + +@pytest.fixture +def orch(tmp_db): + return CampaignOrchestrator( + db_path=tmp_db, + tick_interval=0.0, + ) + + +class TestCampaignOrchestratorLifecycle: + @pytest.mark.asyncio + async def test_start_creates_session(self, orch): + with patch.object(orch._gabs, "connect", return_value=False): + sid = await orch.start() + assert sid is not None + assert orch.session_id == sid + + @pytest.mark.asyncio + async def test_start_resumes_existing_session(self, orch): + existing_sid = orch._memory.start_session("existing_run") + orch._session_id = existing_sid + with patch.object(orch._gabs, "connect", return_value=False): + sid = await orch.start() + assert sid == existing_sid + + @pytest.mark.asyncio + async def test_stop_disconnects_gabs(self, orch): + disconnect_mock = AsyncMock() + orch._gabs.disconnect = disconnect_mock + await orch.stop() + disconnect_mock.assert_awaited_once() + + +class TestCampaignTick: + @pytest.mark.asyncio + async def test_single_tick_logs_subgoal(self, orch, tmp_db): + state = _make_state() + orch._gabs.get_game_state = AsyncMock(return_value=state) + orch._gabs._call = AsyncMock(return_value=None) + orch._session_id = orch._memory.start_session() + + await orch._tick(1) + + entries = orch.memory.get_recent_subgoals(orch.session_id, limit=5) + assert len(entries) == 1 + + @pytest.mark.asyncio + async def test_run_stops_at_max_ticks(self, orch): + state = _make_state() + orch._gabs.connect = AsyncMock(return_value=False) + orch._gabs.get_game_state = AsyncMock(return_value=state) + orch._gabs._call = AsyncMock(return_value=None) + orch._gabs.disconnect = AsyncMock() + + summary = await orch.run(max_ticks=3) + assert summary["ticks_run"] == 3 + + @pytest.mark.asyncio + async def test_run_detects_done_condition(self, orch): + """Campaign stops early when M3 done condition is met.""" + state = _make_state( + in_game_day=110, + fiefs=["A", "B", "C"], + ) + orch._gabs.connect = AsyncMock(return_value=False) + orch._gabs.get_game_state = AsyncMock(return_value=state) + orch._gabs._call = AsyncMock(return_value=None) + orch._gabs.disconnect = AsyncMock() + + summary = await orch.run(max_ticks=100) + # Should stop at tick 1 because done condition is met immediately + assert summary["ticks_run"] <= 2 + assert summary["survival_goal_met"] + + @pytest.mark.asyncio + async def test_summary_shape(self, orch): + state = _make_state(in_game_day=110, fiefs=["A", "B", "C"]) + orch._gabs.connect = AsyncMock(return_value=False) + orch._gabs.get_game_state = AsyncMock(return_value=state) + orch._gabs._call = AsyncMock(return_value=None) + orch._gabs.disconnect = AsyncMock() + + summary = await orch.run(max_ticks=1) + assert "ticks_run" in summary + assert "session_id" in summary + assert "has_kingdom" in summary + assert "fief_count" in summary diff --git a/tests/bannerlord/test_companions.py b/tests/bannerlord/test_companions.py new file mode 100644 index 00000000..a51e3f9f --- /dev/null +++ b/tests/bannerlord/test_companions.py @@ -0,0 +1,161 @@ +"""Tests for Bannerlord companion worker agents.""" + +from bannerlord.agents.companions.caravan import CaravanCompanion +from bannerlord.agents.companions.logistics import LogisticsCompanion +from bannerlord.agents.companions.scout import ScoutCompanion +from bannerlord.types import ( + FactionState, + GameState, + KingSubgoal, + KingdomState, + PartyState, + SubgoalToken, +) + + +def _state( + *, + troops: int = 150, + denars: int = 10_000, + food_days: int = 10, + wounded_pct: float = 0.0, + prisoners: int = 0, + location: str = "Epicrotea", + active_wars: list | None = None, + factions: list | None = None, +) -> GameState: + return GameState( + party=PartyState( + troops=troops, + denars=denars, + food_days=food_days, + wounded_pct=wounded_pct, + prisoners=prisoners, + location=location, + ), + kingdom=KingdomState( + name="House Timmerson", + active_wars=active_wars or [], + ), + factions=factions or [], + ) + + +class TestLogisticsCompanion: + def setup_method(self): + self.companion = LogisticsCompanion() + + def test_recruits_on_recruit_subgoal(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.RECRUIT, quantity=30) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "recruit_troop" for a in actions) + + def test_rests_on_heal_subgoal(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.HEAL) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "rest_party" for a in actions) + + def test_rests_when_heavily_wounded(self): + state = _state(wounded_pct=0.25) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "rest_party" for a in actions) + + def test_buys_food_when_low(self): + state = _state(food_days=2) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "buy_supplies" for a in actions) + + def test_no_food_purchase_when_stocked(self): + state = _state(food_days=10) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert not any(a["primitive"] == "buy_supplies" for a in actions) + + def test_sells_prisoners_at_cap(self): + state = _state(prisoners=20) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "sell_prisoners" for a in actions) + + def test_upgrades_troops_on_train(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.TRAIN) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "upgrade_troops" for a in actions) + + +class TestCaravanCompanion: + def setup_method(self): + self.companion = CaravanCompanion() + + def test_no_actions_when_not_trade_subgoal(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.RAID_ECONOMY) + actions = self.companion.evaluate(state, sg) + assert actions == [] + + def test_assesses_prices_on_trade(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.TRADE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "assess_prices" for a in actions) + + def test_deploys_caravan_when_flush(self): + state = _state(denars=15_000) + sg = KingSubgoal(token=SubgoalToken.TRADE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "establish_caravan" for a in actions) + + def test_no_caravan_when_broke(self): + state = _state(denars=5_000) + sg = KingSubgoal(token=SubgoalToken.TRADE) + actions = self.companion.evaluate(state, sg) + assert not any(a["primitive"] == "establish_caravan" for a in actions) + + def test_profitable_trade_threshold(self): + assert CaravanCompanion.is_profitable_trade(100, 116) # 16% margin = ok + assert not CaravanCompanion.is_profitable_trade(100, 114) # 14% = below threshold + assert not CaravanCompanion.is_profitable_trade(0, 100) # zero buy price = no + + +class TestScoutCompanion: + def setup_method(self): + self.companion = ScoutCompanion() + + def test_tracks_lord_on_spy_subgoal(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.SPY, target="Derthert") + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "track_lord" for a in actions) + + def test_assesses_garrison_on_expand(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target="Pravend") + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "assess_garrison" for a in actions) + + def test_maps_patrols_in_war_regions(self): + state = _state( + active_wars=["Vlandia"], + factions=[ + FactionState( + name="Vlandia", + leader="Derthert", + fiefs=["Pravend"], + army_strength=300, + ) + ], + ) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert any(a["primitive"] == "map_patrol_routes" for a in actions) + + def test_no_patrol_map_when_no_wars(self): + state = _state(active_wars=[]) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + actions = self.companion.evaluate(state, sg) + assert not any(a["primitive"] == "map_patrol_routes" for a in actions) diff --git a/tests/bannerlord/test_gabs_client.py b/tests/bannerlord/test_gabs_client.py new file mode 100644 index 00000000..f5239457 --- /dev/null +++ b/tests/bannerlord/test_gabs_client.py @@ -0,0 +1,162 @@ +"""Tests for the GABSClient — uses mocked asyncio streams, no real TCP.""" + +from __future__ import annotations + +import json +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + +from bannerlord.gabs_client import GABSClient +from bannerlord.types import GameState + + +@pytest.fixture +def client(): + return GABSClient(host="127.0.0.1", port=4825, timeout=2.0) + + +class TestGABSClientConnection: + @pytest.mark.asyncio + async def test_connect_returns_false_when_refused(self, client): + with patch( + "asyncio.open_connection", + side_effect=ConnectionRefusedError("refused"), + ): + result = await client.connect() + assert result is False + assert not client.is_connected + + @pytest.mark.asyncio + async def test_connect_success(self, client): + mock_reader = AsyncMock() + mock_writer = MagicMock() + mock_writer.drain = AsyncMock() + mock_writer.close = MagicMock() + mock_writer.wait_closed = AsyncMock() + + # Simulate tools/list response on connect + tools_response = json.dumps({"jsonrpc": "2.0", "id": 1, "result": []}) + "\n" + mock_reader.readline = AsyncMock(return_value=tools_response.encode()) + + with patch("asyncio.open_connection", return_value=(mock_reader, mock_writer)): + with patch("asyncio.wait_for", side_effect=_passthrough_wait_for): + result = await client.connect() + + assert result is True + assert client.is_connected + + @pytest.mark.asyncio + async def test_disconnect_when_not_connected(self, client): + # Should not raise + await client.disconnect() + assert not client.is_connected + + +class TestGABSClientCall: + @pytest.mark.asyncio + async def test_call_returns_none_when_disconnected(self, client): + result = await client._call("game/get_state") + assert result is None + + @pytest.mark.asyncio + async def test_call_id_increments(self, client): + assert client._next_id() == 1 + assert client._next_id() == 2 + assert client._next_id() == 3 + + @pytest.mark.asyncio + async def test_get_game_state_returns_empty_when_disconnected(self, client): + state = await client.get_game_state() + assert isinstance(state, GameState) + assert state.tick == 0 + assert not state.has_kingdom() + + @pytest.mark.asyncio + async def test_move_party_returns_false_when_disconnected(self, client): + result = await client.move_party("Vlandia") + assert result is False + + @pytest.mark.asyncio + async def test_propose_peace_returns_false_when_disconnected(self, client): + result = await client.propose_peace("Vlandia") + assert result is False + + @pytest.mark.asyncio + async def test_assess_prices_returns_empty_dict_when_disconnected(self, client): + result = await client.assess_prices("Pravend") + assert result == {} + + @pytest.mark.asyncio + async def test_map_patrol_routes_returns_empty_list_when_disconnected(self, client): + result = await client.map_patrol_routes("Vlandia") + assert result == [] + + +class TestGameStateParsing: + def test_parse_game_state_full(self): + client = GABSClient() + raw = { + "tick": 5, + "in_game_day": 42, + "party": { + "location": "Pravend", + "troops": 200, + "food_days": 8, + "wounded_pct": 0.1, + "denars": 15000, + "morale": 85.0, + "prisoners": 3, + }, + "kingdom": { + "name": "House Timmerson", + "fiefs": ["Pravend", "Epicrotea"], + "daily_income": 500, + "daily_expenses": 300, + "vassal_lords": ["Lord A"], + "active_wars": ["Sturgia"], + "active_alliances": ["Battania"], + }, + "factions": [ + { + "name": "Sturgia", + "leader": "Raganvad", + "fiefs": ["Varcheg"], + "army_strength": 250, + "treasury": 5000, + "is_at_war_with": ["House Timmerson"], + "relations": {"House Timmerson": -50}, + } + ], + } + state = client._parse_game_state(raw) + assert state.tick == 5 + assert state.in_game_day == 42 + assert state.party.location == "Pravend" + assert state.party.troops == 200 + assert state.kingdom.name == "House Timmerson" + assert state.fief_count() == 2 + assert len(state.factions) == 1 + assert state.factions[0].name == "Sturgia" + assert state.has_kingdom() + assert not state.is_two_front_war() + + def test_parse_game_state_minimal(self): + client = GABSClient() + state = client._parse_game_state({}) + assert isinstance(state, GameState) + assert not state.has_kingdom() + + def test_tool_count_zero_before_connect(self): + client = GABSClient() + assert client.tool_count() == 0 + assert client.available_tools == [] + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + +async def _passthrough_wait_for(coro, timeout=None): + """Stand-in for asyncio.wait_for that just awaits the coroutine.""" + import asyncio + return await coro diff --git a/tests/bannerlord/test_king_agent.py b/tests/bannerlord/test_king_agent.py new file mode 100644 index 00000000..79edec17 --- /dev/null +++ b/tests/bannerlord/test_king_agent.py @@ -0,0 +1,176 @@ +"""Tests for the King agent decision rules.""" + +import pytest + +from bannerlord.agents.king import KingAgent, _MIN_DENARS, _MIN_TROOPS, _TARGET_FIEFS +from bannerlord.types import ( + FactionState, + GameState, + KingdomState, + PartyState, + SubgoalToken, +) + + +def _make_state( + *, + troops: int = 150, + denars: int = 10_000, + wounded_pct: float = 0.0, + food_days: int = 10, + kingdom_name: str = "", + fiefs: list | None = None, + active_wars: list | None = None, + active_alliances: list | None = None, + factions: list | None = None, + in_game_day: int = 50, +) -> GameState: + return GameState( + in_game_day=in_game_day, + party=PartyState( + troops=troops, + denars=denars, + wounded_pct=wounded_pct, + food_days=food_days, + location="Epicrotea", + ), + kingdom=KingdomState( + name=kingdom_name, + fiefs=fiefs or [], + active_wars=active_wars or [], + active_alliances=active_alliances or [], + ), + factions=factions or [], + ) + + +class TestKingAgentRules: + def setup_method(self): + self.king = KingAgent() + + def test_heal_when_heavily_wounded(self): + state = _make_state(wounded_pct=0.35) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.HEAL + + def test_recruit_when_low_troops(self): + state = _make_state(troops=30, wounded_pct=0.0) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.RECRUIT + assert sg.quantity == _MIN_TROOPS - 30 + + def test_trade_when_broke(self): + state = _make_state(troops=150, denars=2_000) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.TRADE + + def test_no_two_front_war_rule(self): + """King must avoid 2-front wars by seeking peace.""" + state = _make_state( + active_wars=["Vlandia", "Sturgia"], + kingdom_name="House Timmerson", + factions=[ + FactionState(name="Vlandia", leader="Derthert", army_strength=500), + FactionState(name="Sturgia", leader="Raganvad", army_strength=200), + ], + ) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.ALLY + # Should target weakest enemy (Sturgia at 200 strength) + assert sg.target == "Sturgia" + + def test_expand_territory_when_no_kingdom(self): + state = _make_state( + troops=150, + denars=10_000, + kingdom_name="", + factions=[ + FactionState( + name="Vlandia", + leader="Derthert", + fiefs=["Pravend"], + army_strength=100, + is_at_war_with=["Battania", "Aserai"], + ) + ], + ) + sg = self.king.decide(state) + # Distracted faction should be the expansion target + assert sg.token == SubgoalToken.EXPAND_TERRITORY + + def test_train_when_troops_insufficient_for_expansion(self): + state = _make_state( + troops=90, + denars=10_000, + kingdom_name="", + factions=[ + FactionState( + name="Vlandia", + leader="Derthert", + fiefs=["Pravend"], + army_strength=100, + ) + ], + ) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.TRAIN + + def test_expand_when_below_target_fiefs(self): + state = _make_state( + kingdom_name="House Timmerson", + fiefs=["Epicrotea"], + factions=[ + FactionState( + name="Vlandia", + leader="Derthert", + fiefs=["Pravend", "Sargot"], + army_strength=100, + ) + ], + ) + sg = self.king.decide(state) + assert sg.token == SubgoalToken.EXPAND_TERRITORY + + def test_consolidate_when_stable(self): + state = _make_state( + kingdom_name="House Timmerson", + fiefs=["Epicrotea", "Pravend", "Sargot"], + active_alliances=["Battania"], + ) + sg = self.king.decide(state) + assert sg.token in {SubgoalToken.CONSOLIDATE, SubgoalToken.FORTIFY} + + def test_tick_increments(self): + king = KingAgent() + state = _make_state() + king.decide(state) + king.decide(state) + assert king.tick == 2 + + def test_done_condition_not_met_without_kingdom(self): + state = _make_state(in_game_day=200) + assert not self.king.is_done_condition_met(state) + + def test_done_condition_met(self): + state = _make_state( + kingdom_name="House Timmerson", + fiefs=["A", "B", "C"], + in_game_day=110, + ) + assert self.king.is_done_condition_met(state) + + def test_done_condition_not_met_insufficient_days(self): + state = _make_state( + kingdom_name="House Timmerson", + fiefs=["A", "B", "C"], + in_game_day=50, + ) + assert not self.king.is_done_condition_met(state) + + def test_campaign_summary_shape(self): + state = _make_state(kingdom_name="House Timmerson", fiefs=["A"]) + summary = self.king.campaign_summary(state) + assert "tick" in summary + assert "has_kingdom" in summary + assert "fief_count" in summary + assert "survival_goal_met" in summary diff --git a/tests/bannerlord/test_session_memory.py b/tests/bannerlord/test_session_memory.py new file mode 100644 index 00000000..f377759d --- /dev/null +++ b/tests/bannerlord/test_session_memory.py @@ -0,0 +1,140 @@ +"""Tests for SessionMemory — SQLite-backed campaign persistence.""" + +import tempfile +from pathlib import Path + +import pytest + +from bannerlord.session_memory import SessionMemory +from bannerlord.types import KingSubgoal, SubgoalToken + + +@pytest.fixture +def memory(tmp_path): + return SessionMemory(tmp_path / "test_campaign.db") + + +class TestSessionLifecycle: + def test_start_session_returns_id(self, memory): + sid = memory.start_session() + assert sid.startswith("session_") + + def test_start_session_with_explicit_id(self, memory): + sid = memory.start_session("my_run_001") + assert sid == "my_run_001" + + def test_start_idempotent(self, memory): + sid1 = memory.start_session("run") + sid2 = memory.start_session("run") + assert sid1 == sid2 + + def test_get_session_returns_dict(self, memory): + sid = memory.start_session() + row = memory.get_session(sid) + assert row is not None + assert row["session_id"] == sid + + def test_get_unknown_session_returns_none(self, memory): + assert memory.get_session("does_not_exist") is None + + def test_list_sessions(self, memory): + memory.start_session("s1") + memory.start_session("s2") + sessions = memory.list_sessions() + ids = [s["session_id"] for s in sessions] + assert "s1" in ids + assert "s2" in ids + + def test_update_session(self, memory): + sid = memory.start_session() + memory.update_session(sid, kingdom_name="House Timmerson", fief_count=2, in_game_day=45) + row = memory.get_session(sid) + assert row["kingdom_name"] == "House Timmerson" + assert row["fief_count"] == 2 + assert row["in_game_day"] == 45 + + +class TestSubgoalLog: + def test_log_and_retrieve_subgoal(self, memory): + sid = memory.start_session() + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target="Epicrotea") + row_id = memory.log_subgoal(sid, tick=1, in_game_day=10, subgoal=sg) + assert row_id > 0 + entries = memory.get_recent_subgoals(sid, limit=5) + assert len(entries) == 1 + assert entries[0]["token"] == "EXPAND_TERRITORY" + assert entries[0]["target"] == "Epicrotea" + assert entries[0]["outcome"] == "pending" + + def test_complete_subgoal(self, memory): + sid = memory.start_session() + sg = KingSubgoal(token=SubgoalToken.TRADE) + row_id = memory.log_subgoal(sid, tick=2, in_game_day=11, subgoal=sg) + memory.complete_subgoal(row_id, outcome="success") + entries = memory.get_recent_subgoals(sid, limit=5) + assert entries[0]["outcome"] == "success" + assert entries[0]["completed_at"] is not None + + def test_count_token(self, memory): + sid = memory.start_session() + for i in range(3): + memory.log_subgoal( + sid, tick=i, in_game_day=i, + subgoal=KingSubgoal(token=SubgoalToken.RECRUIT) + ) + memory.log_subgoal( + sid, tick=10, in_game_day=10, + subgoal=KingSubgoal(token=SubgoalToken.TRADE) + ) + assert memory.count_token(sid, SubgoalToken.RECRUIT) == 3 + assert memory.count_token(sid, SubgoalToken.TRADE) == 1 + assert memory.count_token(sid, SubgoalToken.ALLY) == 0 + + def test_recent_subgoals_respects_limit(self, memory): + sid = memory.start_session() + for i in range(10): + memory.log_subgoal( + sid, tick=i, in_game_day=i, + subgoal=KingSubgoal(token=SubgoalToken.CONSOLIDATE) + ) + entries = memory.get_recent_subgoals(sid, limit=3) + assert len(entries) == 3 + + +class TestStrategyNotes: + def test_add_and_get_notes(self, memory): + sid = memory.start_session() + memory.add_note(sid, in_game_day=5, note_type="intel", content="Vlandia weakened") + notes = memory.get_notes(sid) + assert len(notes) == 1 + assert notes[0]["content"] == "Vlandia weakened" + + def test_filter_notes_by_type(self, memory): + sid = memory.start_session() + memory.add_note(sid, 1, "milestone", "Kingdom established") + memory.add_note(sid, 2, "intel", "Enemy sighted") + milestones = memory.get_notes(sid, note_type="milestone") + assert len(milestones) == 1 + assert milestones[0]["content"] == "Kingdom established" + + def test_record_kingdom_established(self, memory): + sid = memory.start_session() + memory.record_kingdom_established(sid, in_game_day=42, kingdom_name="House Timmerson") + milestones = memory.get_milestones(sid) + assert any("House Timmerson" in m["content"] for m in milestones) + row = memory.get_session(sid) + assert row["kingdom_name"] == "House Timmerson" + + def test_record_war_declared(self, memory): + sid = memory.start_session() + memory.record_war_declared(sid, in_game_day=20, faction="Vlandia") + notes = memory.get_notes(sid, note_type="war_declared") + assert len(notes) == 1 + assert "Vlandia" in notes[0]["content"] + + def test_record_peace_agreed(self, memory): + sid = memory.start_session() + memory.record_peace_agreed(sid, in_game_day=30, faction="Sturgia") + notes = memory.get_notes(sid, note_type="peace_agreed") + assert len(notes) == 1 + assert "Sturgia" in notes[0]["content"] diff --git a/tests/bannerlord/test_types.py b/tests/bannerlord/test_types.py new file mode 100644 index 00000000..c0c70cf6 --- /dev/null +++ b/tests/bannerlord/test_types.py @@ -0,0 +1,103 @@ +"""Tests for Bannerlord M3 core data types.""" + +from datetime import UTC, datetime + +from bannerlord.types import ( + FactionState, + GameState, + KingSubgoal, + KingdomState, + PartyState, + SubgoalToken, + VassalReward, +) + + +class TestSubgoalToken: + def test_all_tokens_are_strings(self): + for token in SubgoalToken: + assert isinstance(str(token), str) + + def test_round_trip_from_string(self): + assert SubgoalToken("EXPAND_TERRITORY") == SubgoalToken.EXPAND_TERRITORY + assert SubgoalToken("ALLY") == SubgoalToken.ALLY + + +class TestKingSubgoal: + def test_defaults(self): + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + assert sg.priority == 1.0 + assert sg.target is None + assert sg.quantity is None + assert isinstance(sg.issued_at, datetime) + + def test_to_dict_round_trip(self): + sg = KingSubgoal( + token=SubgoalToken.EXPAND_TERRITORY, + target="Epicrotea", + quantity=None, + priority=1.5, + deadline_days=10, + context="capture castle", + ) + d = sg.to_dict() + assert d["token"] == "EXPAND_TERRITORY" + assert d["target"] == "Epicrotea" + assert d["priority"] == 1.5 + + restored = KingSubgoal.from_dict(d) + assert restored.token == SubgoalToken.EXPAND_TERRITORY + assert restored.target == "Epicrotea" + assert restored.priority == 1.5 + assert restored.deadline_days == 10 + + def test_from_dict_without_issued_at(self): + d = {"token": "TRADE"} + sg = KingSubgoal.from_dict(d) + assert sg.token == SubgoalToken.TRADE + assert isinstance(sg.issued_at, datetime) + + +class TestGameState: + def test_empty_state(self): + state = GameState() + assert not state.has_kingdom() + assert state.fief_count() == 0 + assert state.active_war_count() == 0 + assert not state.is_two_front_war() + + def test_has_kingdom(self): + state = GameState(kingdom=KingdomState(name="House Timmerson")) + assert state.has_kingdom() + + def test_fief_count(self): + state = GameState( + kingdom=KingdomState(name="House Timmerson", fiefs=["Pravend", "Epicrotea"]) + ) + assert state.fief_count() == 2 + + def test_two_front_war(self): + state = GameState( + kingdom=KingdomState( + name="House Timmerson", + active_wars=["Vlandia", "Sturgia"], + ) + ) + assert state.is_two_front_war() + + def test_single_war_not_two_front(self): + state = GameState( + kingdom=KingdomState( + name="House Timmerson", + active_wars=["Vlandia"], + ) + ) + assert not state.is_two_front_war() + + +class TestVassalReward: + def test_defaults(self): + reward = VassalReward(agent_id="war_vassal") + assert reward.total == 0.0 + assert reward.subgoal_bonus == 0.0 + assert isinstance(reward.computed_at, datetime) diff --git a/tests/bannerlord/test_vassals.py b/tests/bannerlord/test_vassals.py new file mode 100644 index 00000000..d6aa9372 --- /dev/null +++ b/tests/bannerlord/test_vassals.py @@ -0,0 +1,179 @@ +"""Tests for Bannerlord vassal agents (War, Economy, Diplomacy).""" + +from bannerlord.agents.diplomacy_vassal import DiplomacyVassal +from bannerlord.agents.economy_vassal import EconomyVassal +from bannerlord.agents.war_vassal import WarVassal +from bannerlord.types import ( + FactionState, + GameState, + KingSubgoal, + KingdomState, + PartyState, + SubgoalToken, +) + + +def _state( + *, + troops: int = 150, + denars: int = 10_000, + food_days: int = 10, + kingdom_name: str = "House Timmerson", + fiefs: list | None = None, + active_wars: list | None = None, + active_alliances: list | None = None, + factions: list | None = None, +) -> GameState: + return GameState( + party=PartyState( + troops=troops, + denars=denars, + food_days=food_days, + location="Epicrotea", + ), + kingdom=KingdomState( + name=kingdom_name, + fiefs=fiefs or ["Epicrotea"], + active_wars=active_wars or [], + active_alliances=active_alliances or [], + ), + factions=factions or [], + ) + + +class TestWarVassal: + def setup_method(self): + self.vassal = WarVassal() + + def test_is_relevant_expand(self): + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY) + assert self.vassal.is_relevant(sg) + + def test_is_relevant_raid(self): + sg = KingSubgoal(token=SubgoalToken.RAID_ECONOMY) + assert self.vassal.is_relevant(sg) + + def test_not_relevant_for_trade(self): + sg = KingSubgoal(token=SubgoalToken.TRADE) + assert not self.vassal.is_relevant(sg) + + def test_plan_expansion_with_target(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target="Pravend") + tasks = self.vassal.plan(state, sg) + primitives = [t.primitive for t in tasks] + assert "siege_settlement" in primitives + assert "auto_resolve_battle" in primitives + + def test_plan_expansion_scouts_garrison_first(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target="Pravend") + tasks = self.vassal.plan(state, sg) + # Scout garrison should be the first task (highest priority) + assert tasks[0].primitive == "assess_garrison" + + def test_plan_expansion_recruits_when_low(self): + state = _state(troops=80) + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target="Pravend") + tasks = self.vassal.plan(state, sg) + primitives = [t.primitive for t in tasks] + assert "recruit_troop" in primitives + + def test_plan_expansion_no_target(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY, target=None) + tasks = self.vassal.plan(state, sg) + assert tasks == [] + + def test_plan_raid_with_target(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.RAID_ECONOMY, target="Enemy Village") + tasks = self.vassal.plan(state, sg) + assert any(t.primitive == "raid_village" for t in tasks) + + def test_compute_reward_territory_gain(self): + prev = _state(fiefs=["A"]) + curr = _state(fiefs=["A", "B"]) + sg = KingSubgoal(token=SubgoalToken.EXPAND_TERRITORY) + reward = self.vassal.compute_reward(prev, curr, sg) + assert reward.total > 0 + assert reward.component_scores["territory"] > 0 + + +class TestEconomyVassal: + def setup_method(self): + self.vassal = EconomyVassal() + + def test_is_relevant_fortify(self): + sg = KingSubgoal(token=SubgoalToken.FORTIFY) + assert self.vassal.is_relevant(sg) + + def test_is_relevant_consolidate(self): + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + assert self.vassal.is_relevant(sg) + + def test_not_relevant_for_raid(self): + sg = KingSubgoal(token=SubgoalToken.RAID_ECONOMY) + assert not self.vassal.is_relevant(sg) + + def test_plan_fortify_queues_build(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.FORTIFY, target="Epicrotea") + tasks = self.vassal.plan(state, sg) + primitives = [t.primitive for t in tasks] + assert "build_project" in primitives + assert "set_tax_policy" in primitives + + def test_plan_buys_food_when_low(self): + state = _state(food_days=2) + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + tasks = self.vassal.plan(state, sg) + primitives = [t.primitive for t in tasks] + assert "buy_supplies" in primitives + + def test_plan_consolidate_sets_tax(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.CONSOLIDATE) + tasks = self.vassal.plan(state, sg) + assert any(t.primitive == "set_tax_policy" for t in tasks) + + +class TestDiplomacyVassal: + def setup_method(self): + self.vassal = DiplomacyVassal() + + def test_is_relevant_ally(self): + sg = KingSubgoal(token=SubgoalToken.ALLY) + assert self.vassal.is_relevant(sg) + + def test_not_relevant_for_train(self): + sg = KingSubgoal(token=SubgoalToken.TRAIN) + assert not self.vassal.is_relevant(sg) + + def test_plan_proposes_peace_with_enemy(self): + state = _state(active_wars=["Vlandia"]) + sg = KingSubgoal(token=SubgoalToken.ALLY, target="Vlandia") + tasks = self.vassal.plan(state, sg) + assert any(t.primitive == "propose_peace" for t in tasks) + + def test_plan_requests_alliance_with_non_enemy(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.ALLY, target="Battania") + tasks = self.vassal.plan(state, sg) + primitives = [t.primitive for t in tasks] + assert "send_envoy" in primitives + assert "request_alliance" in primitives + + def test_plan_no_target_returns_empty(self): + state = _state() + sg = KingSubgoal(token=SubgoalToken.ALLY, target=None) + tasks = self.vassal.plan(state, sg) + assert tasks == [] + + def test_should_avoid_war_when_two_fronts(self): + state = _state(active_wars=["A", "B"]) + assert self.vassal.should_avoid_war(state) + + def test_should_not_avoid_war_when_one_front(self): + state = _state(active_wars=["A"]) + assert not self.vassal.should_avoid_war(state)