WIP: Claude Code progress on #1095

Automated salvage commit — agent session ended (exit 124).
Work in progress, may need continuation.

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" },

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",
+]

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

src/bannerlord/agents/__init__.py

@@ -0,0 +1,4 @@
+"""Bannerlord M3 — feudal agent hierarchy.
+
+King → Vassal → Companion, following Ahilan & Dayan (2019).
+"""

src/bannerlord/agents/companions/__init__.py

@@ -0,0 +1 @@
+"""Bannerlord M3 — Companion worker agents (lowest tier)."""

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

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

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

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,
+        )

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,
+        )

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
+        )

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

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

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 []

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,
+        )

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))

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.

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",

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

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)

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

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

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"]

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)

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)