NPC Autonomy & Living World — Design Document¶
Version: 3.1
Status: Draft
Author: Systems Architecture Team
Date: 2025-02-09
Priority: P1 — Core Experience
Dependencies: Doc 03 (NPC Memory & Relationships), Doc 01 (Durable Persistence)
1. Executive Summary¶
MAID's NPC Memory & Relationships system (Doc 03) gives NPCs the ability to remember — but memory without agency produces characters that passively accumulate data without acting on it. A shopkeeper who remembers a player stole from them but continues trading normally is worse than one with no memory at all: the awareness of inaction breaks immersion more deeply than amnesia.
This design introduces a Goal-Driven NPC Autonomy system that transforms NPCs from reactive dialogue endpoints into autonomous agents pursuing their own goals, maintaining daily routines, forming opinions, gossiping with each other, and taking meaningful actions that advance their personal arcs. NPCs become the primary driver of emergent narrative: a blacksmith who needs rare ore sends adventurers to find it; a jealous merchant spreads rumors about a rival; a guard captain worried about bandit attacks organizes patrols.
The system consists of five integrated subsystems:
- Need & Desire Model — A utility-based motivation framework giving NPCs persistent wants (safety, profit, social standing, personal goals) that drive autonomous decision-making
- Daily Life Simulation — Schedules, routines, and activities that make NPCs feel like inhabitants of a living world rather than quest dispensers rooted to one spot
- Social Fabric — NPC-to-NPC interactions including gossip, trade, friendship, rivalry, and political maneuvering that create an evolving social ecosystem
- Autonomous Action System — A tick-budgeted decision engine that selects and executes NPC actions based on current needs, available opportunities, and world state
- Story Signal Emitter — A detection layer that identifies when NPC actions and world conditions create narratively interesting situations, emitting events that downstream systems (especially Doc 08's quest generation) can consume
The architecture is designed for hundreds of autonomous NPCs running within the existing tick budget through aggressive tiering: only nearby NPCs get full AI-powered autonomy; distant NPCs run simplified state machines; offscreen NPCs advance via statistical simulation.
No backwards compatibility concerns — MAID has not been deployed. This design can make breaking changes to any existing system.
2. Problem Statement & Current State¶
What Exists Today¶
| Component | File | Capability | Limitation |
|---|---|---|---|
| BehaviorSystem | classic_rpg/systems/npc/behavior.py |
Patrol, wander, aggro, flee, combat | Purely reactive; no goals, no decision-making beyond combat |
| NPCSpawner | classic_rpg/systems/npc/spawner.py |
Spawn/respawn NPCs with population limits | NPCs appear but have no purpose or daily routine |
| NPCDialogueSystem | classic_rpg/systems/npc/dialogue.py |
AI-powered conversation | Player-initiated only; NPCs never initiate or talk to each other |
| NPCState | classic_rpg/models/npc/behavior.py |
Patrol state, combat state, territory | No goal tracking, no need state, no schedule |
| GossipSystem (Doc 03) | Designed, not implemented | Knowledge propagation between co-located NPCs | Message-passing only; no motivation for why an NPC gossips |
| GameTimeSystem | classic_rpg/systems/world/time.py |
Day/night cycle, seasons | Time advances but NPCs ignore it completely |
| Memory System (Doc 03) | Designed, not implemented | Episodic/semantic/procedural memory | Passive storage; memories don't drive behavior |
The Problem¶
- NPCs are furniture — They stand in fixed locations waiting for players to interact with them. A blacksmith who never smiths, a guard who never patrols, a farmer who never farms.
- No emergent narrative — Story only happens when a designer writes it. There's no mechanism for interesting situations to arise organically from NPC interactions.
- Gossip without purpose — Doc 03's gossip system propagates knowledge, but NPCs have no reason to care about what they learn. Gossip should change behavior.
- Time is cosmetic — The day/night cycle exists but NPCs don't respond to it. The world feels static despite a ticking clock.
- No social dynamics — NPCs exist in isolation. There are no friendships, rivalries, trade relationships, or political tensions between NPCs that players can observe and influence.
- Memory without agency — Doc 03 gives NPCs memories, but no mechanism to act on those memories. An NPC remembers being threatened but takes no precautions.
Success Criteria¶
- NPCs follow daily routines that vary by time of day, weather, and personal state
- NPCs pursue personal goals that create observable world changes
- NPC-to-NPC interactions generate emergent social dynamics
- Gossip meaningfully changes NPC behavior and creates narrative hooks
- Players observe NPCs "living their lives" even without direct interaction
- System scales to 200+ NPCs within tick budget (target: <5ms total per tick)
- NPC actions are deterministic/reproducible given the same world state (for debugging)
- Story-relevant situations are automatically detected and signaled
3. Architecture Overview¶
┌─────────────────────────────────────────────────────────────┐
│ Story Signal Emitter │
│ Detects narratively interesting patterns from NPC actions │
│ Emits: StoryOpportunityEvent, TensionEscalationEvent, etc. │
└──────────────────────────┬──────────────────────────────────┘
│ consumes
┌──────────────────────────┴──────────────────────────────────┐
│ Autonomous Action System │
│ ┌────────────┐ ┌──────────────┐ ┌─────────────────────┐ │
│ │ ActionPlan │ │UtilityScorer │ │ ActionExecutor │ │
│ │ Generator │◄─┤ (per-need │ │ (move, trade, talk, │ │
│ │ │ │ evaluation) │ │ craft, rest, ...) │ │
│ └────────────┘ └──────────────┘ └─────────────────────┘ │
└──────────────────────────┬──────────────────────────────────┘
│ reads
┌──────────────────────────┴──────────────────────────────────┐
│ NPC State Layer │
│ ┌──────────┐ ┌────────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Needs │ │ Schedule │ │ Goals │ │ Social │ │
│ │Component │ │ Component │ │Component │ │Component │ │
│ └──────────┘ └────────────┘ └──────────┘ └──────────┘ │
└──────────────────────────┬──────────────────────────────────┘
│ built on
┌──────────────────────────┴──────────────────────────────────┐
│ Existing Infrastructure │
│ World │ ECS │ EventBus │ Memory (Doc03) │ GameTime │ Grid │
└─────────────────────────────────────────────────────────────┘
Processing Tiers¶
Not all NPCs need the same level of simulation fidelity. The system uses three processing tiers to maintain performance:
| Tier | Criteria | Update Rate | Simulation Depth | Budget |
|---|---|---|---|---|
| Active | Same room as a player, or adjacent rooms | Every tick (1s) | Full utility evaluation, AI-eligible actions, social interactions | 2ms/tick |
| Nearby | Within 5 rooms of any player | Every 10s | Simplified needs decay, schedule adherence, pre-scripted routines | 1ms/tick |
| Background | No players nearby | Every 60s | Statistical advancement only — needs drift toward schedule baseline | 0.5ms/tick |
Tier transitions happen automatically as players move through the world. When a player enters an area, background NPCs are "caught up" to their expected state based on elapsed time and their schedule.
Global Tick-Budget Governor: A TickBudgetGovernor enforces a hard ceiling on total autonomy processing per tick. The governor wraps the entire AutonomySystem.update() loop, tracking elapsed wall-clock time:
class TickBudgetGovernor:
"""Enforces a hard ceiling on per-tick autonomy processing time.
If measured tick time exceeds the soft ceiling, the governor dynamically
demotes the lowest-priority Active NPCs to Nearby tier (LOD reduction).
If it exceeds the hard ceiling, remaining unprocessed NPCs are skipped
for this tick and a TickBudgetExceededEvent is emitted for monitoring.
"""
SOFT_CEILING_MS: float = 4.0 # Begin LOD demotion
HARD_CEILING_MS: float = 5.0 # Stop processing, skip remaining NPCs
RECOVERY_TICKS: int = 10 # Ticks below soft ceiling before restoring demoted NPCs
def __init__(self) -> None:
self._dynamic_demotions: set[uuid.UUID] = set()
self._ticks_under_budget: int = 0
def should_process(self, npc_id: uuid.UUID, elapsed_ms: float) -> bool:
"""Return False if the budget is exhausted for this tick."""
if elapsed_ms >= self.HARD_CEILING_MS:
return False
if elapsed_ms >= self.SOFT_CEILING_MS and npc_id not in self._dynamic_demotions:
self._dynamic_demotions.add(npc_id)
return False # Demote to Nearby processing this tick
return True
def end_tick(self, elapsed_ms: float) -> None:
"""Track recovery — restore demoted NPCs after sustained low load."""
if elapsed_ms < self.SOFT_CEILING_MS:
self._ticks_under_budget += 1
if self._ticks_under_budget >= self.RECOVERY_TICKS:
self._dynamic_demotions.clear()
self._ticks_under_budget = 0
else:
self._ticks_under_budget = 0
Individual _process_npc() calls are also wrapped with a per-NPC timeout of 2ms (matching the Active tier budget for ~20 NPCs). If a single NPC exceeds this, its processing is cancelled for the tick and its failure counter is incremented.
Tier Transition Rules:
- Incremental tier assignment via AOI index: Tier assignment does not scan all NPCs each tick. Instead, it uses the engine's spatial AOI (area-of-interest) index, which is maintained incrementally as entities move. When a player moves, only the NPCs entering/leaving that player's AOI range are re-evaluated. The AOI index maps
room_id → set[entity_id]and supports range queries via theGridManager's adjacency graph. This keeps tier assignment O(changed entities) rather than O(total NPCs). - Highest tier wins: If an NPC is within range of multiple players, it uses the highest applicable tier (e.g., same room as Player A = Active, even if 5 rooms from Player B = Nearby).
- Demotion hysteresis: An NPC must remain outside Active range for 3 consecutive ticks before demoting from Active → Nearby, preventing flicker at room boundaries.
- Catch-up budget: At most 5 NPCs per tick are caught up from Background → Active/Nearby, spreading the cost over multiple ticks when a player enters a populated area. Catch-up is two-phase: (1) immediately teleport NPCs to their schedule-predicted location so players never see un-caught-up NPCs, then (2) run full state catch-up (needs, goals) in the background at 5 NPCs/tick.
- Error demotion: NPCs demoted due to repeated errors (see §4.5) remain in Background until manually reviewed via admin tools. Demotion state is persisted to DocumentStore so demoted NPCs stay demoted across server restarts.
4. Detailed Design¶
4.1 Need & Desire Model¶
NPCs are driven by a hierarchy of needs, loosely inspired by Maslow but tuned for gameplay relevance:
@dataclass
class Need:
"""A single NPC need with current satisfaction level."""
category: NeedCategory
value: float # 0.0 (desperate) to 1.0 (fully satisfied)
decay_rate: float # How fast this need depletes per hour (game time)
last_satisfied: GameTime # When this need was last addressed (see §6.3)
class NeedCategory(str, Enum):
"""Categories of NPC needs, ordered by priority."""
SURVIVAL = "survival" # Health, safety from threats
ECONOMIC = "economic" # Money, resources, trade goods
SOCIAL = "social" # Companionship, reputation, belonging
PURPOSE = "purpose" # Fulfilling their role (smithing, guarding, etc.)
AMBITION = "ambition" # Personal goals, advancement, power
COMFORT = "comfort" # Rest, shelter, food, drink
class NeedsComponent(Component):
"""Attached to NPCs to track their motivational state.
Need prioritization is controlled solely by `personality_weights`.
Individual `Need` objects do NOT carry a weight field — this prevents
the ambiguity of having two weight sources. The per-need weight used
in utility scoring is always `personality_weights[need.category]`.
"""
needs: dict[NeedCategory, Need]
personality_weights: dict[NeedCategory, float] # Single source of truth for need priority
mood: float = 0.5 # 0.0 (miserable) to 1.0 (elated)
stress: float = 0.0 # Accumulated from unmet needs
GameTime convention: All timestamps in the autonomy system use
GameTime(an opaque wrapper around the game-world clock fromGameTimeSystem) rather thandatetime. This ensures NPC state is consistent with the in-game clock, survives time-scaling changes, and avoids confusion between wall-clock and game-clock. See §6.3 for theGameTimeSystemintegration.
Need Decay: Each need decays over game time at its configured rate. A guard's SURVIVAL need decays slowly (they're trained), while a merchant's ECONOMIC need decays faster (they need to make sales). Need decay rates are configurable per NPC archetype.
Mood Calculation: Mood is a weighted average of all need satisfaction levels, modified by recent events (memory-driven). An NPC who just had a good conversation gets a temporary social mood boost. Mood affects dialogue tone, willingness to help, and action selection.
def calculate_mood(needs: NeedsComponent, recent_memories: list[BaseMemory]) -> float:
total_weight = sum(
needs.personality_weights.get(cat, 0.0) for cat in needs.needs
)
if total_weight == 0.0:
base_mood = 0.5 # Default when no needs are configured
else:
base_mood = sum(
need.value * needs.personality_weights.get(cat, 0.0)
for cat, need in needs.needs.items()
) / total_weight
# Apply memory-driven modifiers (last 24h game time)
memory_modifier = sum(
m.emotional_valence * m.importance * recency_weight(m.created_at)
for m in recent_memories
) * 0.2 # Memories can shift mood by ±20%
return clamp(base_mood + memory_modifier, 0.0, 1.0)
4.2 Goal System¶
Goals represent medium-to-long-term objectives that NPCs pursue over time. Unlike needs (which are perpetual), goals can be completed, failed, or abandoned.
@dataclass
class Goal:
"""A specific objective an NPC is working toward."""
id: uuid.UUID
category: GoalCategory
description: str # Human-readable, used in dialogue context
priority: float # 0.0 to 1.0, affects action selection
progress: float # 0.0 to 1.0, how close to completion
conditions: list[GoalCondition] # Typed predicates that must be satisfied (see below)
deadline: GameTime | None # Optional deadline (game time)
created_at: GameTime
source: GoalSource # What generated this goal
target: uuid.UUID | str | None = None # Primary target entity/item (used for dedup)
# Behavioral modifiers
required_resources: dict[str, int] # Items/currency needed
preferred_helpers: list[uuid.UUID] # NPCs or players who could help
blockers: list[str] # What's preventing progress
class GoalCondition:
"""A typed predicate that must be satisfied for a goal to complete.
Replaces untyped string hooks. Each condition is a concrete predicate
that the goal system can evaluate deterministically.
"""
predicate: GoalPredicate
description: str # Human-readable explanation for debug/dialogue
class GoalPredicate(Protocol):
"""Protocol for goal completion predicates. Content packs implement concrete types."""
def evaluate(self, world: World, npc_id: uuid.UUID) -> bool: ...
def progress_estimate(self, world: World, npc_id: uuid.UUID) -> float:
"""Return 0.0–1.0 estimate of how close the condition is to being met."""
...
# Built-in predicate types
@dataclass(frozen=True)
class HasItemPredicate:
"""NPC possesses a specific item or quantity."""
item_id: str
quantity: int = 1
def evaluate(self, world: World, npc_id: uuid.UUID) -> bool: ...
def progress_estimate(self, world: World, npc_id: uuid.UUID) -> float: ...
@dataclass(frozen=True)
class RelationshipThresholdPredicate:
"""Relationship with target exceeds a threshold on a given dimension."""
target_id: uuid.UUID
dimension: str # "trust", "respect", "loyalty"
threshold: float
def evaluate(self, world: World, npc_id: uuid.UUID) -> bool: ...
def progress_estimate(self, world: World, npc_id: uuid.UUID) -> float: ...
@dataclass(frozen=True)
class LocationPredicate:
"""NPC is at a specific location."""
room_id: uuid.UUID
def evaluate(self, world: World, npc_id: uuid.UUID) -> bool: ...
def progress_estimate(self, world: World, npc_id: uuid.UUID) -> float: ...
@dataclass(frozen=True)
class CurrencyThresholdPredicate:
"""NPC possesses at least a given amount of currency."""
amount: int
def evaluate(self, world: World, npc_id: uuid.UUID) -> bool: ...
def progress_estimate(self, world: World, npc_id: uuid.UUID) -> float: ...
class GoalCategory(str, Enum):
ACQUIRE = "acquire" # Get an item, resource, or amount of gold
CRAFT = "craft" # Create something
SOCIAL = "social" # Build a relationship, gain reputation
PROTECT = "protect" # Keep something/someone safe
EXPLORE = "explore" # Visit a location, discover something
REVENGE = "revenge" # Retaliate against a perceived wrong
AMBITION = "ambition" # Advance in rank, gain power
DUTY = "duty" # Fulfill role obligations
class GoalSource(str, Enum):
INNATE = "innate" # From NPC archetype definition
DERIVED = "derived" # Generated from unmet needs
REACTIVE = "reactive" # Response to a world event or memory
SOCIAL = "social" # Learned from gossip or conversation
QUEST = "quest" # Assigned by quest system
class GoalsComponent(Component):
"""Tracks an NPC's current goals."""
active_goals: list[Goal] # Max 5 active goals
completed_goals: list[uuid.UUID] # History (last 20)
failed_goals: list[uuid.UUID]
goal_generation_cooldown: float = 0.0 # Prevents goal churn
Goal Generation: Goals are derived from three sources:
- Innate goals — Defined in the NPC's archetype (e.g., a blacksmith has an innate CRAFT goal to produce weapons)
- Need-derived goals — When a need drops below a threshold, the system generates a goal to address it (e.g., low ECONOMIC → goal to sell more goods)
- Reactive goals — Triggered by events or memories (e.g., NPC witnesses a theft → PROTECT goal to secure their shop; NPC hears gossip about bandits → SURVIVAL goal to fortify defenses)
class GoalGenerator:
"""Derives new goals from NPC state, memories, and world events."""
EVALUATE_TIMEOUT = 0.002 # 2ms hard ceiling per NPC — matches Active tier per-NPC budget
async def evaluate(
self,
npc_id: uuid.UUID,
needs: NeedsComponent,
memories: list[BaseMemory],
relationships: dict[uuid.UUID, RelationshipState],
world_context: WorldContext,
) -> list[Goal]:
"""Generate candidate goals, with timeout and deduplication.
Goals are deduplicated by (category, source, target, description_hash)
to prevent the same trigger from producing overlapping goals while
allowing distinct goals of the same category+source (e.g., two
different ACQUIRE goals from the same DERIVED source).
"""
try:
goals = await asyncio.wait_for(
self._evaluate_inner(npc_id, needs, memories, relationships, world_context),
timeout=self.EVALUATE_TIMEOUT,
)
except asyncio.TimeoutError:
logger.warning(f"GoalGenerator.evaluate timed out for NPC {npc_id}")
return []
# Deduplicate by (category, source, target, description hash)
seen: set[tuple[GoalCategory, GoalSource, uuid.UUID | str | None, int]] = set()
deduped: list[Goal] = []
for goal in goals:
key = (goal.category, goal.source, goal.target, hash(goal.description))
if key not in seen:
seen.add(key)
deduped.append(goal)
return deduped
async def _evaluate_inner(
self,
npc_id: uuid.UUID,
needs: NeedsComponent,
memories: list[BaseMemory],
relationships: dict[uuid.UUID, RelationshipState],
world_context: WorldContext,
) -> list[Goal]:
goals: list[Goal] = []
# Need-derived goals
for category, need in needs.needs.items():
if need.value < 0.3: # Below critical threshold
goal = self._derive_goal_from_need(category, need, world_context)
if goal:
goals.append(goal)
# Memory-reactive goals
recent_memories = [m for m in memories if m.importance > 0.7]
for memory in recent_memories:
goal = self._derive_goal_from_memory(memory, relationships)
if goal:
goals.append(goal)
return goals
4.3 Daily Life & Schedules¶
NPCs follow configurable daily schedules that give them predictable-but-not-rigid routines. Schedules are defined per archetype and can be overridden per individual NPC.
@dataclass
class ScheduleCondition:
"""Typed condition for schedule block activation."""
category: str # e.g., "weather", "season", "event"
operator: str # e.g., "not", "is", "gt"
value: str # e.g., "storm", "winter", "festival"
@dataclass
class ScheduleBlock:
"""A single time block in an NPC's daily schedule.
Overnight-aware: when end_hour < start_hour (e.g., start=20, end=6),
the block wraps across midnight. Matching logic must treat this as
covering hours 20-23 AND 0-5 inclusive.
"""
start_hour: int # 0-23 game time
end_hour: int # 0-23 game time; end < start means overnight wrap
activity: ActivityType
location: str | uuid.UUID # Room name or ID
priority: float # How important this block is (can be overridden by urgent needs)
conditions: list[ScheduleCondition] # Typed conditions for activation
class ActivityType(str, Enum):
WORK = "work" # Fulfills PURPOSE need
SLEEP = "sleep" # Fulfills COMFORT need
EAT = "eat" # Fulfills COMFORT need
SOCIALIZE = "socialize" # Fulfills SOCIAL need
PATROL = "patrol" # Fulfills DUTY (for guards)
TRADE = "trade" # Fulfills ECONOMIC need
CRAFT = "craft" # Fulfills PURPOSE need
WORSHIP = "worship" # Fulfills PURPOSE/SOCIAL
TRAIN = "train" # Fulfills AMBITION need
WANDER = "wander" # Default idle behavior
CUSTOM = "custom" # Content-pack defined
class ScheduleComponent(Component):
"""An NPC's daily routine."""
blocks: list[ScheduleBlock]
current_activity: ActivityType | None = None
schedule_adherence: float = 0.8 # How strictly they follow schedule (0-1)
override_until: GameTime | None = None # Temporary schedule override
override_reason: str | None = None
Schedule Execution Flow:
GameTimeSystem ticks → emits TimeAdvancedEvent
│
▼
ScheduleSystem checks current game hour
│
├─ For each NPC with ScheduleComponent:
│ ├─ Find applicable ScheduleBlock for current hour
│ ├─ Check conditions (weather, season, etc.)
│ ├─ Compare schedule priority vs urgent need priority
│ │ ├─ If urgent need wins → override schedule
│ │ └─ If schedule wins → set current_activity
│ └─ If activity changed → emit NPCActivityChangedEvent
│ └─ Movement system handles relocation to activity location
│
└─ Background NPCs: advance state statistically (skip pathfinding)
Example Schedule — Village Blacksmith:
archetype: blacksmith
schedule:
- { start: 6, end: 7, activity: eat, location: "tavern" }
- { start: 7, end: 12, activity: work, location: "smithy" }
- { start: 12, end: 13, activity: eat, location: "tavern" }
- { start: 13, end: 17, activity: work, location: "smithy" }
- { start: 17, end: 19, activity: socialize, location: "tavern" }
- { start: 19, end: 20, activity: eat, location: "home" }
- { start: 20, end: 6, activity: sleep, location: "home" }
# Weather override (uses typed ScheduleCondition — see §4.3)
- { start: 7, end: 17, activity: work, location: "smithy",
conditions: [{ category: "weather", operator: "not", value: "storm" }] }
- { start: 7, end: 17, activity: socialize, location: "tavern",
conditions: [{ category: "weather", operator: "is", value: "storm" }] }
4.4 Social Fabric¶
The social fabric system extends Doc 03's gossip mechanism into a full NPC-to-NPC interaction system. NPCs don't just pass data — they have conversations, form opinions, make deals, and scheme.
4.4.1 NPC-to-NPC Interaction Types¶
class SocialInteractionType(str, Enum):
GOSSIP = "gossip" # Share knowledge/rumors
TRADE = "trade" # Buy/sell goods
CONVERSATION = "conversation" # General social interaction
ARGUMENT = "argument" # Disagreement (relationship tension)
FAVOR = "favor" # Ask for or offer help
INTIMIDATION = "intimidation" # Threaten or bully
SCHEMING = "scheming" # Plot together against a third party
MENTORING = "mentoring" # Teach or advise
COURTSHIP = "courtship" # Romantic interest
4.4.2 Gossip as Behavioral Driver¶
Doc 03 defines gossip as knowledge propagation. This design extends gossip to be a behavioral driver: NPCs gossip with purpose, and the information they receive changes their behavior.
@dataclass
class GossipIntent:
"""Why an NPC is sharing this particular piece of gossip."""
intent_type: GossipIntentType
target_npc_id: uuid.UUID | None # Who the gossip is about
desired_outcome: str # What the speaker hopes to achieve
source_event_id: uuid.UUID | None = None # Originating event for depth tracking
class GossipIntentType(str, Enum):
WARN = "warn" # Alert ally about danger
DEFAME = "defame" # Damage someone's reputation
BOAST = "boast" # Elevate own status
BOND = "bond" # Build social connection by sharing
MANIPULATE = "manipulate" # Influence listener's behavior
INFORM = "inform" # Neutral information sharing
SEEK_HELP = "seek_help" # Describe a problem to find assistance
Gossip-Driven Behavior Changes:
When an NPC receives gossip, they don't just store it — they react:
class GossipReactionProcessor:
"""Determines how an NPC reacts to received gossip.
Gossip impact is dampened by propagation depth to prevent
amplification cascades. Each re-telling reduces the behavioral
effect via a depth_factor. Gossip originating from the same
source_event_id inherits depth+1 (not reset to 0) when an NPC
re-originates a rumor as behavioral gossip. Maximum propagation
depth is capped at MAX_PROPAGATION_DEPTH to prevent runaway chains.
"""
DEPTH_DAMPENING = 0.7 # Each propagation hop multiplies effect by this
MAX_PROPAGATION_DEPTH = 5 # Hard cap on gossip re-telling depth
async def process_gossip_reaction(
self,
listener: NPCContext,
gossip: GossipMessage,
intent: GossipIntent,
) -> list[BehaviorModification]:
modifications: list[BehaviorModification] = []
# Evaluate trust in the source
source_trust = listener.relationships.get(gossip.current_holder_id)
credibility = gossip.confidence * (source_trust.trust if source_trust else 0.5)
# Dampen by propagation depth (first-hand = depth 0 = full effect)
depth_factor = self.DEPTH_DAMPENING ** gossip.propagation_depth
credibility *= depth_factor
# Hard cap: discard gossip that has propagated too far
if gossip.propagation_depth >= self.MAX_PROPAGATION_DEPTH:
return []
if credibility < 0.3:
return [] # Dismiss unreliable gossip
# React based on gossip content
if gossip.concerns_threat and credibility > 0.6:
modifications.append(BehaviorModification(
type=ModType.INCREASE_NEED,
target_need=NeedCategory.SURVIVAL,
amount=credibility * 0.3,
))
# May generate a PROTECT goal
if gossip.concerns_opportunity and listener.needs.economic.value < 0.5:
modifications.append(BehaviorModification(
type=ModType.ADD_GOAL,
goal=Goal(category=GoalCategory.ACQUIRE), # ...
))
if gossip.concerns_person:
# Adjust relationship with the person being discussed
modifications.append(BehaviorModification(
type=ModType.ADJUST_RELATIONSHIP,
target_id=gossip.subject_entity_id,
dimension="trust",
delta=-0.1 * credibility if intent.intent_type == GossipIntentType.DEFAME else 0.05,
))
return modifications
4.4.3 Social Interaction Resolution¶
NPC-to-NPC interactions are resolved through a lightweight system that doesn't require LLM calls for most interactions:
class SocialInteractionResolver:
"""Resolves NPC-to-NPC interactions without LLM calls.
Uses deterministic rules for common interactions.
Reserves LLM for player-witnessed interactions that need
natural language output.
"""
async def resolve(
self,
initiator: NPCContext,
target: NPCContext,
interaction_type: SocialInteractionType,
player_witnessed: bool,
) -> InteractionOutcome:
match interaction_type:
case SocialInteractionType.TRADE:
return self._resolve_trade(initiator, target)
case SocialInteractionType.GOSSIP:
return self._resolve_gossip(initiator, target)
case SocialInteractionType.ARGUMENT:
return self._resolve_argument(initiator, target)
# ...
# If a player is watching, enqueue an LLM narrative request.
# Use fire-and-forget: return a placeholder immediately, fill in
# the real text on the next tick when the LLM responds.
if player_witnessed:
outcome.narrative = self._enqueue_narrative_request(
initiator, target, interaction_type, outcome
)
return outcome
def _enqueue_narrative_request(
self,
initiator: NPCContext,
target: NPCContext,
interaction_type: SocialInteractionType,
outcome: InteractionOutcome,
) -> str:
"""Queue an LLM call with a 500ms timeout; return placeholder text.
The LLM response is delivered via NarrativeReadyEvent on the next tick.
If the call times out, the placeholder remains.
The queue is bounded (MAX_PENDING_NARRATIVES). When full, new requests
are dropped with a log warning — backpressure prevents unbounded growth
during LLM outages. Before emitting NarrativeReadyEvent, the system
verifies that at least one player is still in the room; stale narratives
are silently discarded.
LLM calls go through a provider-level circuit breaker (see §7.3)
that trips after consecutive failures, preventing cascading timeouts.
"""
MAX_PENDING_NARRATIVES = 50
if len(self._pending_narratives) >= MAX_PENDING_NARRATIVES:
logger.warning("Narrative queue full — dropping request")
return f"{initiator.name} and {target.name} {interaction_type.value}."
placeholder = f"{initiator.name} and {target.name} {interaction_type.value}."
self._pending_narratives.append(NarrativeRequest(
initiator=initiator,
target=target,
interaction_type=interaction_type,
outcome=outcome,
timeout=0.5, # 500ms hard ceiling per call
room_id=initiator.current_room, # For presence check on delivery
))
return placeholder
Key design decision: NPC-to-NPC interactions that happen off-screen use deterministic rules (no LLM). Only interactions witnessed by a player generate narrative text via the LLM, keeping AI costs bounded.
4.4.4 Social Network Component¶
class SocialComponent(Component):
"""Tracks an NPC's social standing and interaction state.
Note: Friends, rivals, and allies are derived from Doc 03 relationship data
and queried via SocialFabricSystem methods — not stored here to avoid duplication.
"""
faction_standing: dict[str, float] # Standing with factions
social_influence: float = 0.5 # 0-1, affects gossip credibility
last_social_interaction: GameTime | None = None
social_cooldown: float = 0.0 # Prevents spam interactions
interaction_reservation: uuid.UUID | None = None # NPC reserved for interaction
reservation_expires: GameTime | None = None
Querying Social Relationships: Friends, rivals, and allies are derived from Doc 03's relationship data at query time. SocialFabricSystem provides helper methods:
class SocialFabricSystem(System):
"""Manages NPC-to-NPC social interactions and relationship queries.
Relationship lookups are backed by a per-tick cache to avoid repeated
scans of Doc 03's relationship data. The cache is invalidated at the
start of each tick and lazily populated on first access per NPC.
"""
priority: ClassVar[int] = 120
def __init__(self, world: World) -> None:
super().__init__(world)
self._friends_cache: dict[uuid.UUID, list[uuid.UUID]] = {}
self._rivals_cache: dict[uuid.UUID, list[uuid.UUID]] = {}
self._allies_cache: dict[uuid.UUID, list[uuid.UUID]] = {}
async def update(self, delta: float) -> None:
"""Process social interactions, sweep reservations, and clear caches."""
# Invalidate caches at the start of each tick
self._friends_cache.clear()
self._rivals_cache.clear()
self._allies_cache.clear()
now = current_game_time()
for entity_id in self.entities.with_components(SocialComponent):
entity = self.entities.get(entity_id)
social = entity.get(SocialComponent)
if (
social.reservation_expires is not None
and social.reservation_expires < now
):
social.interaction_reservation = None
social.reservation_expires = None
def get_friends(self, npc_id: uuid.UUID) -> list[uuid.UUID]:
"""Return NPCs with trust > 0.7 from Doc 03 relationship data (cached)."""
if npc_id not in self._friends_cache:
self._friends_cache[npc_id] = self._query_relationships(
npc_id, lambda r: r.trust > 0.7
)
return self._friends_cache[npc_id]
def get_rivals(self, npc_id: uuid.UUID) -> list[uuid.UUID]:
"""Return NPCs with trust < 0.3 and respect < 0.3 (cached)."""
if npc_id not in self._rivals_cache:
self._rivals_cache[npc_id] = self._query_relationships(
npc_id, lambda r: r.trust < 0.3 and r.respect < 0.3
)
return self._rivals_cache[npc_id]
def get_allies(self, npc_id: uuid.UUID) -> list[uuid.UUID]:
"""Return NPCs with loyalty > 0.6 (cached)."""
if npc_id not in self._allies_cache:
self._allies_cache[npc_id] = self._query_relationships(
npc_id, lambda r: r.loyalty > 0.6
)
return self._allies_cache[npc_id]
def reserve_interaction(
self, initiator_id: uuid.UUID, target_id: uuid.UUID, duration: float
) -> bool:
"""Reserve a target NPC for social interaction, preventing chase behavior.
Returns False if the target is already reserved by another NPC.
"""
...
4.5 Autonomous Action System¶
The action system is the decision engine that evaluates NPC state and selects actions each tick.
4.5.1 Action Definition & Lifecycle¶
Actions have a clear lifecycle represented by ActionState. An NPC owns at most one active action at a time; starting a new action implicitly cancels the current one (emitting IntentCancelledEvent).
class ActionState(str, Enum):
"""Lifecycle state of an NPC's current action."""
PENDING = "pending" # Selected but not yet started
RUNNING = "running" # In progress (may span multiple ticks)
COMPLETE = "complete" # Finished successfully
FAILED = "failed" # Could not complete
INTERRUPTED = "interrupted" # Cancelled by a higher-priority action or combat
@dataclass(frozen=True)
class ActionCost:
"""Resource cost of performing an action."""
time_cost: float # Estimated game-hours to complete
risk: float # 0.0 (safe) to 1.0 (life-threatening)
currency: int = 0 # Gold cost, if any
stamina: float = 0.0 # Fraction of COMFORT need consumed (0.0–1.0)
@dataclass
class Action:
"""A possible action an NPC can take (before scoring)."""
action_type: ActionType
target: uuid.UUID | str | None # Entity, location, or item
cost: ActionCost # Time, resources, risk
prerequisites: list[GoalPredicate] # Typed predicates that must hold (reuses GoalPredicate)
need_effects: dict[NeedCategory, float] # Expected satisfaction delta per need
# Units: absolute change to need.value (0.0–1.0 scale).
# Positive = satisfies need, negative = depletes.
# Capped to keep need.value within [0.0, 1.0] after application.
# Example: {NeedCategory.COMFORT: 0.3} means "restores 0.3 comfort points".
context: dict[str, Any] = field(default_factory=dict)
# Action-type-specific payload (e.g., {"recipe": "sword_iron"} for CRAFT_ITEM,
# {"gossip_message_id": uuid} for GOSSIP). Typed per ActionType — see ActionExecutor.
def advances_goal(self, goal: Goal) -> bool:
"""Return True if this action progresses the given goal.
Checks whether any of the goal's conditions would be moved closer
to completion by this action's effects (e.g., an ACQUIRE action
for the same item a goal requires).
"""
...
def aligns_with(self, activity: ActivityType) -> bool:
"""Return True if this action is compatible with the given schedule activity."""
...
@dataclass
class ScoredAction:
"""An Action paired with its computed utility score.
Produced by UtilityScorer; consumed by AutonomySystem for selection.
Separating the score from the action keeps Action immutable and
allows re-scoring without mutation.
"""
action: Action
utility: float # Computed score from UtilityScorer
score_breakdown: dict[str, float] | None = None # Debug: per-factor scores
class ActiveAction:
"""Tracks the NPC's currently executing action with lifecycle state."""
scored_action: ScoredAction
state: ActionState = ActionState.PENDING
started_at: GameTime | None = None
ticks_elapsed: int = 0
def is_terminal(self) -> bool:
return self.state in (ActionState.COMPLETE, ActionState.FAILED, ActionState.INTERRUPTED)
class ActionType(str, Enum):
MOVE_TO = "move_to"
CRAFT_ITEM = "craft_item"
SELL_ITEM = "sell_item"
BUY_ITEM = "buy_item"
REST = "rest"
EAT = "eat"
SOCIALIZE = "socialize"
GOSSIP = "gossip"
PATROL = "patrol"
GUARD = "guard"
SEEK_HELP = "seek_help"
OFFER_HELP = "offer_help"
FLEE = "flee"
CONFRONT = "confront"
INVESTIGATE = "investigate"
COMPLAIN = "complain" # To authority figure about a problem
PRAY = "pray"
TRAIN = "train"
SCHEME = "scheme" # Plot against rival
CELEBRATE = "celebrate" # After goal completion
Utility Scoring:
Each possible action is scored based on how well it addresses the NPC's most pressing needs and goals:
class UtilityScorer:
"""Evaluates the utility of potential actions for an NPC.
Produces ScoredAction instances from raw Actions. The score is NOT
stored on the Action itself — this separation keeps Actions immutable
and allows re-scoring with different contexts.
"""
def score(
self,
action: Action,
npc: NPCDecisionContext,
world: WorldDecisionContext,
) -> ScoredAction:
score = 0.0
# Need satisfaction score — uses personality_weights as the single
# source of truth for per-need importance
for need_category, satisfaction_delta in action.need_effects.items():
need = npc.needs.get(need_category)
if need is None:
continue
weight = npc.personality_weights.get(need_category, 0.0)
urgency = 1.0 - need.value # Lower need = higher urgency
weighted = satisfaction_delta * urgency * weight
score += weighted
# Goal advancement score — uses typed predicate check
for goal in npc.active_goals:
if action.advances_goal(goal):
score += goal.priority * 0.5
# Schedule alignment bonus
if npc.schedule.current_block:
if action.aligns_with(npc.schedule.current_block.activity):
score += 0.2 * npc.schedule.schedule_adherence
# Social relationship modifiers
if action.target:
rivals = self.social_fabric.get_rivals(npc.npc_id)
if action.target in rivals:
if action.action_type in (ActionType.CONFRONT, ActionType.SCHEME):
score += 0.1 # Slight preference for acting against rivals
# Personality modifiers
score *= npc.personality_modifier(action.action_type)
# Cost penalty
score -= action.cost.time_cost * 0.1
score -= action.cost.risk * 0.3
return ScoredAction(
action=action,
utility=score,
) # Negative scores are filtered at selection time
4.5.2 Decision Context (Snapshot DTOs)¶
The AutonomySystem builds deep-copy snapshot DTOs of NPC and world state each tick, preventing systems from accidentally mutating shared ECS state during scoring. These are intentionally separate types from the live ECS components — they contain only the fields needed for decision-making, and use plain Python types (not mutable component references):
@dataclass(frozen=True)
class NPCDecisionContext:
"""Deep-copy snapshot of an NPC's state for decision-making.
All fields are immutable value types or frozen copies of component data.
This is NOT the live NeedsComponent/GoalsComponent — mutations here have
no effect on the ECS. Built via NPCDecisionContext.snapshot(entity).
"""
npc_id: uuid.UUID
# Snapshot copies (plain dicts/lists, not live components)
needs: dict[NeedCategory, float] # category → current value
personality_weights: dict[NeedCategory, float] # category → weight
active_goals: tuple[Goal, ...] # Frozen copy of active goals
current_activity: ActivityType | None
schedule_adherence: float
current_room: uuid.UUID
personality: NPCPersonality
current_action: ActiveAction | None
in_combat: bool
mood: float
stress: float
@staticmethod
def snapshot(entity: Entity) -> "NPCDecisionContext":
"""Build an immutable snapshot from a live entity's components."""
needs_comp = entity.get(NeedsComponent)
goals_comp = entity.get(GoalsComponent)
schedule_comp = entity.get(ScheduleComponent)
return NPCDecisionContext(
npc_id=entity.id,
needs={cat: need.value for cat, need in needs_comp.needs.items()},
personality_weights=dict(needs_comp.personality_weights),
active_goals=tuple(goals_comp.active_goals),
current_activity=schedule_comp.current_activity,
schedule_adherence=schedule_comp.schedule_adherence,
current_room=entity.room_id,
personality=entity.get(NPCPersonality), # Personality is already immutable
current_action=entity.get(ActiveAction, None),
in_combat=entity.has(CombatComponent),
mood=needs_comp.mood,
stress=needs_comp.stress,
)
def personality_modifier(self, action_type: ActionType) -> float:
"""Return a personality-based multiplier for an action type."""
...
@dataclass(frozen=True)
class WorldDecisionContext:
"""Immutable snapshot of relevant world state for NPC decisions."""
current_hour: int
current_weather: str
current_season: str
nearby_entities: tuple[uuid.UUID, ...] # Frozen tuple, not mutable list
nearby_threats: tuple[uuid.UUID, ...]
available_resources: dict[str, int]
4.5.3 Action Execution Pipeline¶
NeedDecaySystem.update(delta) [priority=110]
│
├─ For each NPC with NeedsComponent:
│ ├─ Decay needs based on elapsed time and current activity
│ ├─ Recalculate mood via calculate_mood()
│ └─ If mood changed significantly → emit NPCMoodChangedEvent
│
└─ Background NPCs: advance needs toward schedule-predicted baseline
AutonomySystem.update(delta) [priority=115]
│
├─ Start TickBudgetGovernor for this tick
│
├─ Tier assignment (incremental via AOI index — see §3)
│
├─ For each Active NPC (with per-NPC error isolation + 2ms timeout):
│ ├─ Check TickBudgetGovernor.should_process() — skip if budget exhausted
│ ├─ Verify entity still exists (skip if deleted — not an autonomy error)
│ ├─ Build NPCDecisionContext.snapshot(entity) and WorldDecisionContext
│ ├─ Check ActiveAction lifecycle state:
│ │ ├─ RUNNING → continue (no new decision needed)
│ │ ├─ COMPLETE/FAILED/INTERRUPTED → clear, proceed to selection
│ │ └─ None → proceed to selection
│ ├─ If needs new action:
│ │ ├─ Generate candidate Actions from:
│ │ │ ├─ Schedule (what should I be doing?)
│ │ │ ├─ Needs (what do I need most?)
│ │ │ ├─ Goals (what am I working toward?)
│ │ │ ├─ Opportunities (what's available nearby?)
│ │ │ └─ Threats (is anything dangerous?)
│ │ ├─ Score each candidate via UtilityScorer → ScoredAction
│ │ ├─ Filter out negative utility scores
│ │ ├─ Select highest-utility ScoredAction (with noise for variety)
│ │ ├─ Set ActiveAction(state=RUNNING) on entity
│ │ └─ Begin executing via ActionExecutor
│ └─ Emit NPCActionEvent for story signal detection
│
├─ For each Nearby NPC (every 10th tick):
│ ├─ Follow schedule (simplified pathfinding)
│ └─ Execute pre-computed routine actions
│
├─ For each Background NPC (every 60th tick):
│ └─ Track "virtual" goal progress (analytic — see §7.4)
│
└─ TickBudgetGovernor.end_tick(elapsed_ms)
Per-NPC Error Isolation: Each NPC's update is wrapped in a try/except with a per-NPC 2ms timeout via asyncio.wait_for. If an NPC throws an exception or times out, the error is logged and an internal failure counter is incremented. After 3 consecutive failures, the NPC is demoted to Background tier and flagged for admin review. This prevents one buggy NPC from crashing the entire autonomy loop.
Entity deletion guard: Before processing, the system checks whether the entity still exists in the ECS. Deleted entities are silently skipped — they are not counted as autonomy failures, since the deletion originated outside this system.
Failure count persistence: Failure counts are persisted alongside demotion state in DocumentStore. This prevents a known-bad NPC from getting a fresh failure budget on every server restart and causing a predictable error storm during startup.
class AutonomySystem(System):
priority: ClassVar[int] = 115
MAX_FAILURES = 3
PER_NPC_TIMEOUT = 0.002 # 2ms hard ceiling per NPC
ARCHETYPE_FAILURE_THRESHOLD = 3 # If N NPCs of same archetype fail, demote all
def __init__(self, world: World) -> None:
super().__init__(world)
self._failure_counts: dict[uuid.UUID, int] = {}
self._demoted_npcs: set[uuid.UUID] = set()
self._archetype_failure_counts: dict[str, int] = {} # archetype_id → count
self._demoted_archetypes: set[str] = set()
self._tick_budget = TickBudgetGovernor()
async def startup(self) -> None:
# Restore persisted demotion state AND failure counts to survive restarts
stored = await self.world.document_store.get("npc_demotions", "state")
if stored:
self._demoted_npcs = set(uuid.UUID(nid) for nid in stored.get("npc_ids", []))
self._demoted_archetypes = set(stored.get("archetypes", []))
self._failure_counts = {
uuid.UUID(k): v
for k, v in stored.get("failure_counts", {}).items()
}
async def _persist_demotions(self) -> None:
"""Persist demotion set and failure counts to DocumentStore.
Called from inside the error handler — wrapped in its own try/except
to prevent a persistence failure from crashing the autonomy loop.
"""
try:
await self.world.document_store.put("npc_demotions", "state", {
"npc_ids": [str(nid) for nid in self._demoted_npcs],
"archetypes": list(self._demoted_archetypes),
"failure_counts": {str(k): v for k, v in self._failure_counts.items()},
})
except Exception:
logger.error("Failed to persist NPC demotion state", exc_info=True)
async def update(self, delta: float) -> None:
tick_start = time.monotonic()
for npc_id in self._get_active_npcs():
elapsed_ms = (time.monotonic() - tick_start) * 1000
if not self._tick_budget.should_process(npc_id, elapsed_ms):
continue # Budget exhausted — skip remaining NPCs this tick
if npc_id in self._demoted_npcs:
continue
archetype_id = self._get_archetype_id(npc_id)
if archetype_id in self._demoted_archetypes:
continue
# Guard: skip deleted entities (not an autonomy failure)
if not self.world.entities.exists(npc_id):
self._failure_counts.pop(npc_id, None)
continue
try:
await asyncio.wait_for(
self._process_npc(npc_id, delta),
timeout=self.PER_NPC_TIMEOUT,
)
self._failure_counts.pop(npc_id, None) # Reset on success
except (asyncio.TimeoutError, Exception):
count = self._failure_counts.get(npc_id, 0) + 1
self._failure_counts[npc_id] = count
logger.error(f"NPC {npc_id} autonomy error ({count}/{self.MAX_FAILURES})")
if count >= self.MAX_FAILURES:
self._demoted_npcs.add(npc_id)
logger.warning(f"NPC {npc_id} demoted to Background tier")
# Track archetype-level failures
if archetype_id:
arch_count = self._archetype_failure_counts.get(archetype_id, 0) + 1
self._archetype_failure_counts[archetype_id] = arch_count
if arch_count >= self.ARCHETYPE_FAILURE_THRESHOLD:
self._demoted_archetypes.add(archetype_id)
logger.warning(
f"Archetype '{archetype_id}' demoted — "
f"{arch_count} NPCs failed"
)
await self.events.emit(ArchetypeCircuitBreakerEvent(
archetype_id=archetype_id,
failed_npc_count=arch_count,
))
await self._persist_demotions()
total_ms = (time.monotonic() - tick_start) * 1000
self._tick_budget.end_tick(total_ms)
4.5.4 Action Execution¶
Rather than calling subsystems directly (which creates tight coupling), ActionExecutor translates decisions into intent components and events. Existing systems (movement, crafting, social) handle execution by reacting to these intents:
class NavigationIntent(Component):
"""Set by ActionExecutor; consumed by NPCMovementHandler."""
destination: uuid.UUID
reason: str # e.g., "schedule:work", "goal:acquire_ore"
class SocialIntent(Component):
"""Set by ActionExecutor; consumed by SocialFabricSystem."""
target_id: uuid.UUID
interaction_type: SocialInteractionType
gossip_to_share: uuid.UUID | None = None # Optional gossip message ID
class ActionExecutor:
"""Executes NPC actions by setting intent components and emitting events.
Does NOT call subsystems directly — sets NavigationIntent, SocialIntent,
etc. and lets the owning systems handle execution on their next tick.
All code paths return an ActionResult; failed actions fall back to idle.
Before overwriting an existing intent component, emits an IntentCancelledEvent
so other systems can clean up resources associated with the prior intent.
On completion or failure, updates the NPC's ActiveAction.state to the
appropriate terminal state (COMPLETE, FAILED) and emits an
ActionCompletedEvent or ActionFailedEvent.
"""
async def execute(
self,
npc_id: uuid.UUID,
scored_action: ScoredAction,
world: World,
) -> ActionResult:
entity = world.entities.get(npc_id)
action = scored_action.action
try:
result = await self._execute_action(entity, action, world)
except Exception:
logger.error(f"Action execution failed for NPC {npc_id}: {action.action_type}")
result = None
if result is None:
# Fallback to idle on any failure
entity.add(NavigationIntent(
destination=world.get_entity_room(npc_id), reason="idle_fallback"
))
return ActionResult(success=False, fallback="idle")
return result
async def _execute_action(
self, entity: Entity, action: Action, world: World
) -> ActionResult | None:
match action.action_type:
case ActionType.MOVE_TO:
entity.add(NavigationIntent(
destination=action.target,
reason=f"action:{action.action_type.value}",
))
return ActionResult(success=True, partial=True)
case ActionType.CRAFT_ITEM:
await self.events.emit(NPCCraftRequestEvent(
npc_id=entity.id, recipe=action.context["recipe"]
))
return ActionResult(success=True)
case ActionType.SOCIALIZE:
entity.add(SocialIntent(
target_id=action.target,
interaction_type=SocialInteractionType.CONVERSATION,
))
return ActionResult(success=True)
case ActionType.GOSSIP:
entity.add(SocialIntent(
target_id=action.target,
interaction_type=SocialInteractionType.GOSSIP,
gossip_to_share=action.context.get("gossip_message_id"),
))
return ActionResult(success=True)
# ... other action types set appropriate intents
return None # Unhandled action type
4.6 Story Signal Emitter¶
The story signal system observes NPC actions and world state to detect narratively interesting patterns. These signals are consumed by the quest generation system (Doc 08) and can also trigger world events.
@dataclass
class StorySignal:
"""A detected narrative opportunity in the world."""
id: uuid.UUID
signal_type: StorySignalType
importance: float # 0.0 to 1.0
involved_npcs: list[uuid.UUID]
involved_players: list[uuid.UUID]
location: uuid.UUID # Room/area where this is happening
context: dict[str, Any] # Signal-specific data
created_at: GameTime
expires_at: GameTime # Signals decay if not acted upon
class StorySignalType(str, Enum):
# Conflict signals
RIVALRY_ESCALATION = "rivalry_escalation" # Two NPCs' tension is rising
FACTION_TENSION = "faction_tension" # Inter-faction conflict brewing
THREAT_DETECTED = "threat_detected" # NPC(s) aware of a danger
RESOURCE_SCARCITY = "resource_scarcity" # NPCs competing for limited resource
# Opportunity signals
TRADE_OPPORTUNITY = "trade_opportunity" # Supply/demand mismatch
ALLIANCE_FORMING = "alliance_forming" # NPCs bonding against common threat
DISCOVERY = "discovery" # NPC learned something significant
GOAL_BLOCKED = "goal_blocked" # NPC can't achieve goal alone
# Drama signals
BETRAYAL = "betrayal" # NPC acted against ally
SECRET_REVEALED = "secret_revealed" # Hidden information became gossip
POWER_SHIFT = "power_shift" # Change in social influence hierarchy
UNREQUITED = "unrequited" # One-sided relationship tension
# World state signals
CRISIS = "crisis" # Multiple NPCs distressed
CELEBRATION = "celebration" # Collective positive event
MIGRATION = "migration" # NPCs changing locations en masse
class StorySignalDetector:
"""Detects story-worthy patterns by accumulating events, not polling World state.
Subscribes to relevant events (NPCGoalCreatedEvent, NPCSocialInteractionEvent,
NPCMoodChangedEvent, etc.) and accumulates them into a per-detection-interval
buffer. Pattern detectors analyze the buffer periodically.
"""
DETECTION_INTERVAL_TICKS = 30 # Every 30 seconds at 1 tick/s
MAX_BUFFER_SIZE = 1000 # Drop oldest events when exceeded
async def startup(self) -> None:
"""Subscribe to events that may produce story signals."""
self._event_buffer: list[Event] = []
self.events.subscribe(NPCGoalCreatedEvent, self._accumulate)
self.events.subscribe(NPCGoalFailedEvent, self._accumulate)
self.events.subscribe(NPCSocialInteractionEvent, self._accumulate)
self.events.subscribe(NPCMoodChangedEvent, self._accumulate)
self.events.subscribe(NPCActivityChangedEvent, self._accumulate)
async def _accumulate(self, event: Event) -> None:
self._event_buffer.append(event)
if len(self._event_buffer) > self.MAX_BUFFER_SIZE:
self._event_buffer = self._event_buffer[-self.MAX_BUFFER_SIZE:]
async def detect(self) -> list[StorySignal]:
"""Analyze accumulated events for narrative patterns."""
signals: list[StorySignal] = []
# Atomic swap prevents losing events appended between the two statements
events, self._event_buffer = self._event_buffer, []
# Pass accumulated events to each detector so they analyze the
# buffer rather than polling world state
signals.extend(await self._detect_rivalry_escalation(events))
signals.extend(await self._detect_alliance_formation(events))
# Check goal states
signals.extend(await self._detect_blocked_goals(events))
signals.extend(await self._detect_resource_conflicts(events))
# Check gossip patterns
signals.extend(await self._detect_spreading_rumors(events))
signals.extend(await self._detect_secret_exposure(events))
# Check faction dynamics
signals.extend(await self._detect_faction_tensions(events))
signals.extend(await self._detect_power_shifts(events))
return signals
async def _detect_rivalry_escalation(self, events: list[Event]) -> list[StorySignal]:
"""Detect when NPC rivalries are intensifying."""
signals = []
# Find NPC pairs where mutual trust has dropped significantly
# in the last N game hours
for npc_a, npc_b, trend in self._get_declining_relationships(events):
if trend.trust_delta < -0.3 and trend.period_hours < 48:
signals.append(StorySignal(
signal_type=StorySignalType.RIVALRY_ESCALATION,
importance=abs(trend.trust_delta),
involved_npcs=[npc_a, npc_b],
context={
"cause": trend.primary_cause,
"trust_level": trend.current_trust,
"recent_conflicts": trend.conflict_count,
},
))
return signals
Story Signals → Quest Generation Pipeline (see Doc 08):
StorySignalDetector emits StorySignal events
↓
StorySignalEvent on EventBus
↓
QuestGenerationSystem (Doc 08) evaluates signals
↓
High-importance signals become quest seeds
↓
Quest generated and offered to relevant players
5. NPC Archetype System¶
Rather than configuring every NPC individually, the system uses archetypes as templates that define default needs, schedules, goals, and personality.
Memory optimization: Archetype-level data (schedules, need configs, personality defaults) is shared across all NPCs of the same archetype. Individual NPCs store only their deltas (e.g., overridden schedule blocks, current need values). This reduces per-NPC memory from ~2KB to ~200B for NPCs that don't deviate from their archetype defaults.
@dataclass
class NPCArchetype:
"""Template for NPC behavior configuration."""
archetype_id: str # e.g., "blacksmith", "guard", "merchant"
display_name: str
parent: str | None = None # Inherit from another archetype (e.g., "tradesperson")
# Need configuration (single source of truth — maps to NeedsComponent.personality_weights)
need_weights: dict[NeedCategory, float] # Priority of each need
need_decay_rates: dict[NeedCategory, float] # How fast needs deplete
# Schedule
default_schedule: list[ScheduleBlock]
# Goals
innate_goals: list[GoalTemplate] # Goals this archetype always has
forbidden_goals: list[GoalCategory] = field(default_factory=list) # Goals this archetype rejects
required_location_tags: list[str] = field(default_factory=list) # Tags rooms must have for this NPC
# Need change dampening — caps external need modifications per tick
max_need_delta_per_tick: float = 0.15 # Prevents grief exploitation
# Personality (affects utility scoring)
personality: NPCPersonality
# Social
preferred_interaction_types: list[SocialInteractionType]
gossip_tendency: float # 0.0 (secretive) to 1.0 (chatterbox)
social_initiative: float # How likely to initiate interaction
@dataclass
class NPCPersonality:
"""Personality traits that modify action selection."""
bravery: float = 0.5 # Affects willingness to face threats
greed: float = 0.5 # Affects economic motivation
sociability: float = 0.5 # Affects social interaction frequency
diligence: float = 0.5 # Affects schedule adherence
ambition: float = 0.5 # Affects goal-seeking behavior
curiosity: float = 0.5 # Affects exploration and investigation
loyalty: float = 0.5 # Affects faction/relationship behaviors
temperament: float = 0.5 # Low = volatile, high = calm
Archetype YAML Definition (loaded via Doc 04's data loaders):
Archetypes support inheritance via the parent field. Child archetypes inherit all fields from the parent and can selectively override them:
_meta:
schema: "stdlib:npc_archetypes:v1"
archetypes:
# Base archetype — not instantiated directly
tradesperson:
display_name: "Tradesperson"
need_weights:
purpose: 0.8
economic: 0.7
comfort: 0.6
social: 0.5
survival: 0.3
ambition: 0.4
personality:
diligence: 0.7
greed: 0.4
loyalty: 0.6
gossip_tendency: 0.4
social_initiative: 0.4
blacksmith:
parent: tradesperson # Inherits from tradesperson, overrides below
display_name: "Blacksmith"
need_weights:
purpose: 0.9 # Override: smithing is life
social: 0.4 # Override: somewhat solitary
need_decay_rates:
purpose: 0.04 # Need to smith regularly
economic: 0.03 # Moderate income pressure
comfort: 0.02 # Standard rest needs
social: 0.02 # Don't need much socializing
personality:
bravery: 0.6
greed: 0.3
sociability: 0.4
diligence: 0.8 # Hard workers
ambition: 0.6
curiosity: 0.3
loyalty: 0.7
temperament: 0.7 # Steady disposition
gossip_tendency: 0.3
social_initiative: 0.3
innate_goals:
- category: craft
description: "Maintain stock of quality weapons and tools"
priority: 0.7
- category: acquire
description: "Source quality iron and steel"
priority: 0.5
schedule:
- { start: 6, end: 7, activity: eat, location: "{tavern}" }
- { start: 7, end: 12, activity: work, location: "{workplace}" }
- { start: 12, end: 13, activity: eat, location: "{tavern}" }
- { start: 13, end: 17, activity: work, location: "{workplace}" }
- { start: 17, end: 19, activity: socialize, location: "{tavern}" }
- { start: 19, end: 20, activity: eat, location: "{home}" }
- { start: 20, end: 6, activity: sleep, location: "{home}" }
6. Integration with Existing Systems¶
6.1 Integration with Doc 03 (Memory & Relationships)¶
The autonomy system is the consumer of Doc 03's data:
| Doc 03 Provides | Autonomy System Uses For |
|---|---|
| Episodic memories | Goal generation (reactive goals from important events) |
| Semantic memories | Knowledge-informed decisions (NPC "knows" things) |
| Relationship state | Social target selection, trust evaluation for gossip |
| Gossip messages | Behavioral modifications via GossipReactionProcessor |
| Memory salience | Prioritizing which memories influence current decisions |
6.2 Integration with Existing BehaviorSystem¶
The existing BehaviorSystem handles combat behaviors (aggro, flee, call-for-help). The new AutonomySystem handles non-combat behaviors (daily life, goals, social). They coexist:
Priority ordering:
1. Combat (BehaviorSystem, priority=100) — Always takes precedence
2. Schedule (ScheduleSystem, priority=105) — Time-driven activity transitions
3. Need Decay (NeedDecaySystem, priority=110) — Need decay and mood calculation
4. Autonomy (AutonomySystem, priority=115) — Only runs if NPC is not in combat
5. Social (SocialFabricSystem, priority=120) — NPC-to-NPC interactions
6. Story (StorySignalSystem, priority=130) — Detects narrative patterns
When an NPC enters combat, the AutonomySystem suspends their current action and defers to BehaviorSystem. When combat ends, autonomy resumes.
6.3 Integration with GameTimeSystem¶
The ScheduleSystem subscribes to TimeAdvancedEvent to trigger schedule transitions:
class ScheduleSystem(System):
priority: ClassVar[int] = 105 # After time, before need decay
async def startup(self) -> None:
self.events.subscribe(TimeAdvancedEvent, self._on_time_advanced)
self.events.subscribe(WeatherChangedEvent, self._on_weather_changed)
async def _on_time_advanced(self, event: TimeAdvancedEvent) -> None:
# Check if any NPCs need to transition activities
for entity_id in self.entities.with_components(ScheduleComponent):
entity = self.entities.get(entity_id)
schedule = entity.get(ScheduleComponent)
new_block = self._get_block_for_hour(schedule, event.current_hour)
if new_block and new_block != schedule.current_block:
await self._transition_activity(entity_id, schedule, new_block)
6.4 Integration with Grid/Pathfinding¶
NPC movement uses the existing GridManager for A* pathfinding:
class NPCMovementHandler:
"""Handles NPC pathfinding and movement for autonomous actions."""
async def navigate_to(
self,
npc_id: uuid.UUID,
destination: uuid.UUID,
world: World,
) -> MovementResult:
current_room = world.get_entity_room(npc_id)
path = world.grid_manager.find_path(current_room, destination)
if not path:
return MovementResult(success=False, reason="no_path")
# Move one step per tick (NPCs don't teleport)
next_room = path[0]
await world.move_entity(npc_id, next_room)
return MovementResult(
success=True,
partial=len(path) > 1,
remaining_path=path[1:],
)
6.5 New Events¶
# NPC Autonomy Events
class NPCActivityChangedEvent(Event):
"""Fired when an NPC transitions between schedule activities."""
npc_id: uuid.UUID
previous_activity: ActivityType | None
new_activity: ActivityType
location: uuid.UUID
class NPCGoalCreatedEvent(Event):
"""Fired when an NPC forms a new goal."""
npc_id: uuid.UUID
goal_id: uuid.UUID
goal_category: GoalCategory
class NPCGoalCompletedEvent(Event):
"""Fired when an NPC achieves a goal."""
npc_id: uuid.UUID
goal_id: uuid.UUID
goal_category: GoalCategory
class NPCGoalFailedEvent(Event):
"""Fired when an NPC abandons or fails a goal."""
npc_id: uuid.UUID
goal_id: uuid.UUID
goal_category: GoalCategory
reason: str
class NPCSocialInteractionEvent(Event):
"""Fired when two NPCs interact socially."""
initiator_id: uuid.UUID
target_id: uuid.UUID
interaction_type: SocialInteractionType
outcome_type: str # Enum discriminator: "success", "rejected", "interrupted"
player_witnessed: bool
class StorySignalEvent(Event):
"""Fired when a narratively interesting pattern is detected."""
signal_id: uuid.UUID
signal_type: StorySignalType
importance: float
involved_npc_ids: list[uuid.UUID]
class NPCMoodChangedEvent(Event):
"""Fired when an NPC's mood shifts significantly."""
npc_id: uuid.UUID
previous_mood: float
new_mood: float
cause: str
class ArchetypeCircuitBreakerEvent(Event):
"""Fired when an entire archetype is demoted due to repeated NPC failures."""
archetype_id: str
failed_npc_count: int
class IntentCancelledEvent(Event):
"""Fired before an NPC's intent component is overwritten with a new intent."""
npc_id: uuid.UUID
cancelled_intent_type: str # e.g., "NavigationIntent", "SocialIntent"
reason: str
class NarrativeReadyEvent(Event):
"""Fired when an async LLM narrative request completes."""
interaction_id: uuid.UUID
narrative_text: str
class TickBudgetExceededEvent(Event):
"""Fired when the autonomy tick budget hard ceiling is reached."""
elapsed_ms: float
npcs_processed: int
npcs_skipped: int
class ActionCompletedEvent(Event):
"""Fired when an NPC's active action reaches COMPLETE state."""
npc_id: uuid.UUID
action_type: ActionType
ticks_elapsed: int
class ActionFailedEvent(Event):
"""Fired when an NPC's active action reaches FAILED state."""
npc_id: uuid.UUID
action_type: ActionType
reason: str
7. Performance Considerations¶
7.1 Tick Budget¶
Target: <5ms total for all NPC autonomy processing per tick, enforced by the TickBudgetGovernor (see §3).
| Component | Budget | Strategy |
|---|---|---|
| Tier assignment (AOI delta) | 0.1ms | Incremental — only re-evaluate NPCs in changed rooms |
| Need decay | 0.5ms | Simple arithmetic, all NPCs |
| Schedule checks | 0.4ms | Only on hour transitions |
| Action selection (Active tier) | 2.0ms | Utility scoring for ~20 Active NPCs, 2ms/NPC timeout |
| Action execution | 1.0ms | One action per NPC per tick max |
| Story signal detection | 0.5ms | Runs every 30 ticks, amortized |
| Social interactions | 0.5ms | Max 3 per tick (budget like gossip) |
Budget enforcement: The TickBudgetGovernor monitors wall-clock time during AutonomySystem.update(). At the soft ceiling (4ms), it begins dynamically demoting the lowest-priority Active NPCs to Nearby-tier processing for the remainder of the tick. At the hard ceiling (5ms), it stops processing entirely and emits a TickBudgetExceededEvent. Per-NPC processing is individually capped at 2ms via asyncio.wait_for. See §3 for the full governor design.
7.2 Scaling Strategy¶
200 NPCs total:
~20 Active (full simulation) → 2ms/tick
~50 Nearby (simplified) → 1ms/tick
~130 Background (statistical) → 0.5ms/tick
Total: ~3.5ms/tick ✓
7.3 LLM Cost Control¶
LLM calls are never made on the tick thread. All LLM work is dispatched to a bounded async queue processed off-tick:
class OffTickLLMQueue:
"""Bounded async queue for LLM work, processed outside the tick loop.
The queue has a hard capacity (MAX_QUEUED). When full, new requests are
dropped (backpressure). A dedicated asyncio task drains the queue,
respecting QPS limits and per-call timeouts. Results are delivered via
events on the next tick.
"""
MAX_QUEUED: int = 50
MAX_QPS: float = 10.0 # Max LLM calls per second
PER_CALL_TIMEOUT: float = 2.0 # 2s hard ceiling per LLM call
LLM calls are only used for: 1. Player-witnessed NPC-to-NPC interactions (narrative text generation) 2. Goal generation from complex memories (optional, can use rule-based fallback) 3. Player dialogue (existing system, Doc 03)
Budget: Max 5 LLM calls per tick for autonomy-related tasks, sharing the global rate limiter with the dialogue system. Calls are dispatched via a priority queue ordered by: 1. Player-witnessed interactions (highest priority — players are waiting for text) 2. Story-signal-relevant (medium — may produce quest hooks) 3. Routine flavor (lowest — nice-to-have ambient text, dropped first under budget pressure)
Rule-based fallback: All autonomy decisions can run without LLM calls. The LLM adds flavor text to witnessed events but is not required for NPC behavior.
Provider-Level Circuit Breaker: Each LLM provider is wrapped in a circuit breaker that tracks consecutive failures. After FAILURE_THRESHOLD (default: 5) consecutive errors or timeouts, the circuit opens and all calls to that provider are immediately rejected for RECOVERY_WINDOW (default: 60s). After the recovery window, a single probe call is allowed; success closes the circuit, failure re-opens it. This prevents cascading timeouts when a provider is down:
class LLMCircuitBreaker:
"""Circuit breaker for LLM provider calls."""
FAILURE_THRESHOLD: int = 5
RECOVERY_WINDOW: float = 60.0 # seconds
state: Literal["closed", "open", "half_open"] = "closed"
consecutive_failures: int = 0
opened_at: float | None = None
def allow_call(self) -> bool:
if self.state == "closed":
return True
if self.state == "open":
if time.monotonic() - self.opened_at >= self.RECOVERY_WINDOW:
self.state = "half_open"
return True # Allow one probe
return False
return False # half_open: only one probe at a time
def record_success(self) -> None:
self.consecutive_failures = 0
self.state = "closed"
def record_failure(self) -> None:
self.consecutive_failures += 1
if self.consecutive_failures >= self.FAILURE_THRESHOLD:
self.state = "open"
self.opened_at = time.monotonic()
LLM Budget Exhaustion Degradation Path: When the per-tick LLM budget is exhausted: 1. Social interactions resolve deterministically (no narrative text generation) 2. Narrative text is skipped, not queued — a generic description is shown instead (e.g., "The blacksmith and merchant have a conversation.") 3. Goal generation falls back to rule-based derivation 4. System continues functioning normally; only flavor text is affected
Narrative delivery guard: Before emitting a NarrativeReadyEvent, the system checks whether at least one player is still present in the interaction's room. If all players have left, the narrative is silently discarded — no one would see it.
7.4 NPC "Bark" System¶
To make the world feel alive without LLM calls, NPCs emit short pre-written phrases ("barks") based on their current need state, mood, and activity. Barks are displayed to players in the same room as ambient flavor:
@dataclass
class BarkTemplate:
"""A short phrase an NPC can say based on state conditions."""
text: str # e.g., "I'm famished...", "*yawns*"
trigger_need: NeedCategory | None # Which need triggers this bark
need_threshold: float = 0.3 # Bark when need drops below this
mood_range: tuple[float, float] = (0.0, 1.0) # Valid mood range
cooldown_minutes: float = 15.0 # Game-time minutes between barks
class BarkLibrary:
"""Loads bark templates per archetype from YAML data files."""
def select_bark(
self, needs: NeedsComponent, mood: float, archetype_id: str
) -> str | None:
"""Return an appropriate bark or None if on cooldown / no match."""
...
Example barks (loaded from YAML per archetype):
barks:
blacksmith:
- { text: "*wipes brow* This heat never gets easier.", trigger_need: comfort, need_threshold: 0.4 }
- { text: "Running low on good steel...", trigger_need: economic, need_threshold: 0.3 }
- { text: "*hums a working tune*", trigger_need: null, mood_range: [0.6, 1.0] }
- { text: "*grumbles under breath*", trigger_need: null, mood_range: [0.0, 0.3] }
7.5 Background NPC Catch-Up¶
When a player enters an area with background NPCs, those NPCs need to be "caught up" to their expected state. Catch-up is two-phase: Phase 1 (immediate) teleports NPCs to their schedule-predicted location so players never see stale positions; Phase 2 (budgeted at 5 NPCs/tick) runs the full state simulation below.
Analytic catch-up: Rather than simulating each elapsed hour in a loop (which is O(hours) and can be expensive for long absences), catch-up uses closed-form analytic formulas wherever possible:
MAX_CATCH_UP_HOURS = 168 # 1 game week — prevents unbounded simulation
async def catch_up_npc(
npc_id: uuid.UUID,
elapsed_game_hours: float,
schedule: ScheduleComponent,
needs: NeedsComponent,
goals: GoalsComponent,
) -> None:
"""Fast-forward an NPC to their expected current state.
Uses analytic (closed-form) progression where possible to avoid
O(hours) per-hour simulation loops. Schedule-based need effects are
pre-aggregated per 24-hour cycle and scaled by elapsed days.
"""
capped_hours = min(elapsed_game_hours, MAX_CATCH_UP_HOURS)
# --- Analytic need advancement ---
# Pre-compute the net need effect of one full 24-hour schedule cycle
# (sum of activity effects + decay for each schedule block's duration).
# Then multiply by (capped_hours / 24) for full days, and simulate
# only the fractional remainder.
full_days = int(capped_hours) // 24
remainder_hours = capped_hours - (full_days * 24)
if full_days > 0:
daily_net = compute_daily_need_delta(schedule, needs) # O(schedule_blocks)
for category, delta in daily_net.items():
need = needs.needs.get(category)
if need:
need.value = clamp(need.value + delta * full_days, 0.0, 1.0)
# Simulate only the fractional remainder (at most 23 hours)
if remainder_hours > 0:
start_hour = (current_game_hour() - int(remainder_hours)) % 24
for hour_offset in range(int(remainder_hours)):
block = get_block_for_hour(schedule, (start_hour + hour_offset) % 24)
if block:
apply_activity_effects(needs, block.activity, duration=1.0)
decay_needs(needs, hours=1.0)
# --- Batched goal advancement ---
# Partition into kept/expired to avoid mutating the list during iteration
now = current_game_time()
expired = [g for g in goals.active_goals if g.deadline and g.deadline < now]
goals.active_goals = [g for g in goals.active_goals if g not in expired]
goals.failed_goals.extend(g.id for g in expired)
for goal in goals.active_goals:
# Analytic progress: innate/duty goals advance ~2% per hour (scaled)
if goal.source in (GoalSource.INNATE, GoalSource.QUEST):
goal.progress = min(1.0, goal.progress + 0.02 * capped_hours)
# Move NPC to where they should be according to schedule
expected_block = get_block_for_hour(schedule, current_game_hour())
if expected_block:
teleport_entity(npc_id, expected_block.location)
compute_daily_need_delta() runs once per archetype (not per NPC) and is cached, since all NPCs of the same archetype share the same schedule. This makes catch-up for N NPCs of the same archetype O(1) per need category rather than O(N × hours).
8. Persistence¶
8.1 What Gets Persisted¶
| Data | Storage | Save Frequency |
|---|---|---|
| NPC needs state | DocumentStore: npc_needs |
On significant change or every 5 min |
| Active goals | DocumentStore: npc_goals |
On creation/update/completion |
| Schedule overrides | DocumentStore: npc_schedule_overrides |
On change |
| Social relationships | Extends Doc 03 relationship data | On interaction |
| Demotion state + failure counts | DocumentStore: npc_demotions |
On demotion change |
| Story signals | In-memory only (ephemeral) | Not persisted |
| Current action (ActiveAction) | In-memory only | Not persisted (reconstructed on load) |
8.2 Serialization Format¶
All enum types (NeedCategory, GoalCategory, GoalSource, ActionType, ActivityType, SocialInteractionType, GossipIntentType) are str enums and serialize to their string values in JSON. GameTime serializes as an integer tick count (not ISO 8601) to avoid wall-clock/game-clock confusion.
NPC Needs (npc_needs/{npc_id}):
{
"needs": {
"survival": {"value": 0.91, "decay_rate": 0.01, "last_satisfied": 48000},
"economic": {"value": 0.41, "decay_rate": 0.03, "last_satisfied": 47200}
},
"personality_weights": {"survival": 0.3, "economic": 0.7, "purpose": 0.9},
"mood": 0.72,
"stress": 0.1
}
NPC Goals (npc_goals/{npc_id}):
{
"active_goals": [
{
"id": "uuid-...",
"category": "craft",
"description": "Maintain stock of quality weapons",
"priority": 0.7,
"progress": 0.6,
"conditions": [{"predicate_type": "has_item", "item_id": "sword_iron", "quantity": 5}],
"deadline": null,
"created_at": 45000,
"source": "innate",
"target": null
}
],
"completed_goals": ["uuid-..."],
"failed_goals": ["uuid-..."]
}
Goal predicates are serialized using a predicate_type discriminator field. Content packs that define custom predicates must register a serializer/deserializer pair with the GoalPredicateRegistry.
8.3 Save Strategy¶
Uses Doc 01's dirty-tracking and persistence pipeline. The NeedsComponent and GoalsComponent register with the DirtyTracker and are saved through the standard EntityPersistenceManager.
9. Content Pack Integration¶
The autonomy systems and events are implemented in maid-stdlib so they are available to all content packs. Game-specific content (archetype definitions, bark text) is provided by each content pack:
# maid-stdlib registers the reusable systems and events
class StdlibContentPack(ContentPack):
def get_systems(self, world: World) -> list[System]:
return [
# ... existing systems ...
ScheduleSystem(world), # priority=105
NeedDecaySystem(world), # priority=110
AutonomySystem(world), # priority=115
SocialFabricSystem(world), # priority=120
StorySignalSystem(world), # priority=130
]
def get_events(self) -> list[type[Event]]:
return [
# ... existing events ...
NPCActivityChangedEvent,
NPCGoalCreatedEvent,
NPCGoalCompletedEvent,
NPCGoalFailedEvent,
NPCSocialInteractionEvent,
StorySignalEvent,
NPCMoodChangedEvent,
ArchetypeCircuitBreakerEvent,
IntentCancelledEvent,
NarrativeReadyEvent,
TickBudgetExceededEvent,
ActionCompletedEvent,
ActionFailedEvent,
]
# Content packs provide game-specific data (archetypes, barks, etc.)
class ClassicRPGContentPack(ContentPack):
async def on_load(self, engine: GameEngine) -> None:
# Load RPG-specific archetype definitions
registry = engine.world.get_system(AutonomySystem).archetype_registry
registry.load_from_yaml(self.data_path / "npcs" / "archetypes.yaml")
# Load RPG-specific bark text
bark_system = engine.world.get_system(BarkSystem)
bark_system.load_barks_from_yaml(self.data_path / "npcs" / "barks.yaml")
10. Testing Strategy¶
10.1 Unit Tests¶
| Component | Test Focus |
|---|---|
UtilityScorer |
Scoring determinism, personality modifiers, edge cases |
GoalGenerator |
Need→goal mapping, memory→goal mapping, goal limits |
ScheduleSystem |
Time transitions, weather overrides, catch-up logic |
GossipReactionProcessor |
Trust thresholds, behavior modifications |
StorySignalDetector |
Pattern detection accuracy, false positive rate |
10.2 Integration Tests¶
- Full tick cycle with autonomous NPCs (needs decay → action selection → execution)
- NPC schedule transitions across day/night boundary
- Gossip propagation → behavioral change → story signal chain
- Multi-NPC social interaction resolution
- Background → Active tier transition with catch-up
10.3 Performance Tests¶
- 200 NPC simulation tick budget validation
- Background catch-up time for 24h elapsed
- Memory usage per NPC with full state
10.4 Admin Debug Tool¶
The @debug_brain command allows admins/builders to inspect an NPC's autonomy state in real time:
@debug_brain <npc_name>
Output:
[Blacksmith Gundren] Tier: Active | Mood: 0.72 (content)
Needs:
PURPOSE: ████████░░ 0.82 (decay: 0.04/hr, weight: 0.9)
ECONOMIC: ████░░░░░░ 0.41 (decay: 0.03/hr, weight: 0.7) ← URGENT
COMFORT: ██████░░░░ 0.63 (decay: 0.02/hr, weight: 0.6)
SOCIAL: █████░░░░░ 0.55 (decay: 0.02/hr, weight: 0.4)
SURVIVAL: █████████░ 0.91 (decay: 0.01/hr, weight: 0.3)
AMBITION: ██████░░░░ 0.60 (decay: 0.02/hr, weight: 0.5)
Current Action: CRAFT_ITEM (sword_iron) — 0.73 utility
Schedule: WORK at smithy (7:00-12:00) — adherence: 0.8
Top Utility Scores:
1. CRAFT_ITEM (sword_iron) → 0.73
2. SELL_ITEM (shield_bronze) → 0.58
3. EAT (tavern) → 0.31
4. SOCIALIZE → 0.22
Active Goals:
[CRAFT] Maintain weapon stock — progress: 0.6, priority: 0.7
[ACQUIRE] Source quality steel — progress: 0.2, priority: 0.5
Error Count: 0 | Demoted: No
11. Open Questions¶
-
NPC Death and Goals: When an NPC dies and respawns, should they retain their goals and social state, or reset? Recommendation: retain goals and social state (memories already persist per Doc 03).
-
Player Influence on NPC Goals: Should players be able to directly suggest goals to NPCs (e.g., "You should try to become the head blacksmith")? Recommendation: yes, via dialogue system — NPC evaluates the suggestion based on personality and relationship.
-
NPC-to-NPC LLM Conversations: Should NPCs have full LLM-powered conversations with each other when a player is watching? Recommendation: yes, but capped at 3 exchanges and sharing the dialogue rate limiter.
-
Goal Conflict Resolution: When two NPCs have conflicting goals (both want the same rare resource), how is this resolved? Recommendation: emerges naturally — both pursue independently, creating a rivalry story signal.
-
Schedule vs. Emergency: How aggressively should urgent needs override schedules? Recommendation: configurable per archetype via
schedule_adherence(0.0 = always follows needs, 1.0 = always follows schedule).
12. Design Decisions Log¶
12.1 Utility-Based AI over Behavior Trees¶
Decision: Use utility scoring for action selection rather than behavior trees.
Rationale: Behavior trees produce predictable, designer-authored sequences. Utility scoring produces emergent behavior where NPCs respond dynamically to changing world state. Since the goal is a "living world," emergent behavior is the explicit design target. Behavior trees are still used for combat (existing BehaviorSystem) where predictability is desirable.
12.2 Three-Tier Processing¶
Decision: Tier NPCs into Active/Nearby/Background rather than simulating all equally.
Rationale: Full utility evaluation for 200 NPCs per tick would exceed the performance budget. Players can only observe NPCs near them, so full simulation is only needed for visible NPCs. Background NPCs advance statistically, and catch-up ensures consistency when players arrive.
12.3 Deterministic Social Resolution¶
Decision: NPC-to-NPC interactions use deterministic rules, not LLM calls.
Rationale: With potentially dozens of NPC interactions per tick cycle, LLM calls would be prohibitively expensive and slow. Deterministic rules ensure consistent behavior and bounded performance. LLM is reserved for generating narrative text when a player witnesses the interaction.
12.4 Goals Capped at 5¶
Decision: Limit active goals to 5 per NPC.
Rationale: More goals create decision paralysis in the utility scorer and increase cognitive load for content designers debugging NPC behavior. Five goals is enough for: 1 innate role goal, 1-2 need-derived goals, 1-2 reactive goals.
12.5 Story Signals as Ephemeral Events¶
Decision: Story signals are not persisted — they are emitted as events and consumed immediately.
Rationale: Signals represent opportunities for narrative, not narrative itself. If no system acts on a signal (e.g., no quest is generated), the moment passes naturally. This prevents accumulation of stale signals and keeps the system lightweight.
13. Implementation Plan¶
Phase 1: Core Autonomy (Foundation)¶
- NeedsComponent, GoalsComponent, ScheduleComponent
- ScheduleSystem (time-driven activity transitions)
- NeedDecaySystem (need decay and mood calculation)
- AutonomySystem (utility-based action selection)
- ActionExecutor with intent components (NavigationIntent, SocialIntent)
- NPC archetype YAML loader with inheritance
- Basic bark system for ambient NPC flavor
Phase 2: Social Fabric¶
- SocialComponent and social network tracking
- NPC-to-NPC interaction resolution
- Gossip intent and reaction processing
- Social interaction types (trade, argument, favor)
- Player-witnessed narrative generation (LLM)
Phase 3: Story Signals & Goals¶
- GoalGenerator (need-derived and reactive goals)
- StorySignalDetector with pattern library
- StorySignalEvent emission on EventBus
- Goal lifecycle (creation → progress → completion/failure)
- Integration with Doc 08 quest generation
Phase 4: Polish & Optimization¶
- Three-tier processing optimization with hysteresis
- Background NPC catch-up system (with goal advancement)
- Performance profiling and budget enforcement
@debug_brainadmin tool for NPC behavior inspection- LLM degradation path and budget exhaustion handling
- Content creator documentation
This design document establishes the behavioral foundation for MAID's living world. NPCs driven by needs, goals, and social dynamics create the emergent narrative fabric that the quest generation system (Doc 08) weaves into player-facing stories. The architecture prioritizes performance through tiered processing and reserves LLM usage for player-facing interactions, ensuring the system scales to hundreds of NPCs within strict tick budgets.