Timmy-time-dashboard/src/bannerlord/models.py

"""Bannerlord feudal hierarchy data models.

All inter-agent communication uses typed Pydantic models.  No raw dicts
cross agent boundaries — every message is validated at construction time.

Design: Ahilan & Dayan (2019) Feudal Multi-Agent Hierarchies.
Refs: #1097, #1099.
"""

from __future__ import annotations

from datetime import datetime
from typing import Any, Literal

from pydantic import BaseModel, Field

# ── Subgoal vocabulary ────────────────────────────────────────────────────────

SUBGOAL_TOKENS = frozenset(
    {
        "EXPAND_TERRITORY",  # Take or secure a fief — War Vassal
        "RAID_ECONOMY",  # Raid enemy villages for denars — War Vassal
        "FORTIFY",  # Upgrade or repair a settlement — Economy Vassal
        "RECRUIT",  # Fill party to capacity — Logistics Companion
        "TRADE",  # Execute profitable trade route — Caravan Companion
        "ALLY",  # Pursue non-aggression / alliance — Diplomacy Vassal
        "SPY",  # Gain information on target faction — Scout Companion
        "HEAL",  # Rest party until wounds recovered — Logistics Companion
        "CONSOLIDATE",  # Hold territory, no expansion — Economy Vassal
        "TRAIN",  # Level troops via auto-resolve bandits — War Vassal
    }
)


# ── King subgoal ──────────────────────────────────────────────────────────────


class KingSubgoal(BaseModel):
    """Strategic directive issued by the King agent to vassals.

    The King operates on campaign-map timescale (days to weeks of in-game
    time).  His sole output is one subgoal token plus optional parameters.
    He never micro-manages primitives.
    """

    token: str = Field(..., description="One of SUBGOAL_TOKENS")
    target: str | None = Field(None, description="Named target (settlement, lord, faction)")
    quantity: int | None = Field(None, description="For RECRUIT, TRADE tokens", ge=1)
    priority: float = Field(1.0, ge=0.0, le=2.0, description="Scales vassal reward weighting")
    deadline_days: int | None = Field(None, ge=1, description="Campaign-map days to complete")
    context: str | None = Field(None, description="Free-text hint; not parsed by workers")

    def model_post_init(self, __context: Any) -> None:  # noqa: ANN401
        if self.token not in SUBGOAL_TOKENS:
            raise ValueError(
                f"Unknown subgoal token {self.token!r}. Must be one of: {sorted(SUBGOAL_TOKENS)}"
            )


# ── Inter-agent messages ──────────────────────────────────────────────────────


class SubgoalMessage(BaseModel):
    """King → Vassal direction."""

    msg_type: Literal["subgoal"] = "subgoal"
    from_agent: Literal["king"] = "king"
    to_agent: str = Field(..., description="e.g. 'war_vassal', 'economy_vassal'")
    subgoal: KingSubgoal
    issued_at: datetime = Field(default_factory=datetime.utcnow)


class TaskMessage(BaseModel):
    """Vassal → Companion direction."""

    msg_type: Literal["task"] = "task"
    from_agent: str = Field(..., description="e.g. 'war_vassal'")
    to_agent: str = Field(..., description="e.g. 'logistics_companion'")
    primitive: str = Field(..., description="One of the companion primitives")
    args: dict[str, Any] = Field(default_factory=dict)
    priority: float = Field(1.0, ge=0.0, le=2.0)
    issued_at: datetime = Field(default_factory=datetime.utcnow)


class ResultMessage(BaseModel):
    """Companion / Vassal → Parent direction."""

    msg_type: Literal["result"] = "result"
    from_agent: str
    to_agent: str
    success: bool
    outcome: dict[str, Any] = Field(default_factory=dict, description="Primitive-specific result")
    reward_delta: float = Field(0.0, description="Computed reward contribution")
    completed_at: datetime = Field(default_factory=datetime.utcnow)


class StateUpdateMessage(BaseModel):
    """GABS → All agents (broadcast).

    Sent every campaign tick.  Agents consume at their own cadence.
    """

    msg_type: Literal["state"] = "state"
    game_state: dict[str, Any] = Field(..., description="Full GABS state snapshot")
    tick: int = Field(..., ge=0)
    timestamp: datetime = Field(default_factory=datetime.utcnow)


# ── Reward snapshots ──────────────────────────────────────────────────────────


class WarReward(BaseModel):
    """Computed reward for the War Vassal at a given tick."""

    territory_delta: float = 0.0
    army_strength_ratio: float = 1.0
    casualty_cost: float = 0.0
    supply_cost: float = 0.0
    subgoal_bonus: float = 0.0

    @property
    def total(self) -> float:
        w1, w2, w3, w4, w5 = 0.40, 0.25, 0.20, 0.10, 0.05
        return (
            w1 * self.territory_delta
            + w2 * self.army_strength_ratio
            - w3 * self.casualty_cost
            - w4 * self.supply_cost
            + w5 * self.subgoal_bonus
        )


class EconomyReward(BaseModel):
    """Computed reward for the Economy Vassal at a given tick."""

    daily_denars_income: float = 0.0
    food_stock_buffer: float = 0.0
    loyalty_average: float = 50.0
    construction_queue_length: int = 0
    subgoal_bonus: float = 0.0

    @property
    def total(self) -> float:
        w1, w2, w3, w4, w5 = 0.35, 0.25, 0.20, 0.15, 0.05
        return (
            w1 * self.daily_denars_income
            + w2 * self.food_stock_buffer
            + w3 * self.loyalty_average
            - w4 * self.construction_queue_length
            + w5 * self.subgoal_bonus
        )


class DiplomacyReward(BaseModel):
    """Computed reward for the Diplomacy Vassal at a given tick."""

    allies_count: int = 0
    truce_duration_value: float = 0.0
    relations_score_weighted: float = 0.0
    active_wars_front: int = 0
    subgoal_bonus: float = 0.0

    @property
    def total(self) -> float:
        w1, w2, w3, w4, w5 = 0.30, 0.25, 0.25, 0.15, 0.05
        return (
            w1 * self.allies_count
            + w2 * self.truce_duration_value
            + w3 * self.relations_score_weighted
            - w4 * self.active_wars_front
            + w5 * self.subgoal_bonus
        )


# ── Victory condition ─────────────────────────────────────────────────────────


class VictoryCondition(BaseModel):
    """Sovereign Victory (M5) — evaluated each campaign tick."""

    holds_king_title: bool = False
    territory_control_pct: float = Field(
        0.0, ge=0.0, le=100.0, description="% of Calradia fiefs held"
    )
    majority_threshold: float = Field(
        51.0, ge=0.0, le=100.0, description="Required % for majority control"
    )

    @property
    def achieved(self) -> bool:
        return self.holds_king_title and self.territory_control_pct >= self.majority_threshold