Core Engine

The engine core is in src/engine/. It contains four major components that work together to run a simulation tick: Simulation, Scheduler, AgentRuntime, and EventBus. The engine never hard-codes world semantics, action meanings, or brain logic — those belong in contracts and plugins.

Simulation

Simulation is the top-level orchestrator. It owns the tick loop, snapshot/restore, and wires all components together. Source: src/engine/simulation.py.

Construction

python

from src.engine.simulation import Simulation, SimulationConfig

# From YAML (recommended)
sim = Simulation.from_config("examples/engine_owned/sim.yaml")

# Programmatic
from src.engine.agent import Agent
from src.engine.registry import ActionRegistry
from src.plugins.external.world import HostedWorld

sim = Simulation(
    agents       = [agent1, agent2],
    world        = HostedWorld(),
    registry     = registry,
    bus          = None,   # defaults to a new EventBus
    scheduler    = None,   # defaults to SimultaneousScheduler
    config       = SimulationConfig(ticks=20, seed=42, log_level="normal"),
    social       = None,   # optional SocialSystem
    obs_registry = None,   # optional ObservationRegistry
)

SimulationConfig

@dataclass class SimulationConfig: ticks: int = 20 seed: int = 42 log_level: str = "normal" # "normal" | "verbose" | "quiet"

Running

python

import asyncio

# Run all ticks (autonomous / CLI mode)
asyncio.run(sim.run())

# Run one tick (external-world integration mode)
summary = asyncio.run(sim.run_tick())
print(f"Tick {summary.tick}: {len(summary.agent_results)} agents stepped")
for ar in summary.agent_results:
    print(f"  {ar.agent_id}: {ar.intent.action} → {ar.result.outcome_text}")

# Relay engine_commands back to host
commands = summary.engine_commands()   # list[dict]

TickSummary

run_tick() returns a TickSummary — an aggregated view of every agent's decision and outcome for that tick.

@dataclass class AgentTickResult: agent_id: str agent_name: str intent: Intent # never None (agents always produce an intent) result: ActionResult # never None @dataclass class TickSummary: tick: int agent_results: list[AgentTickResult] errors: list[tuple[str, str]] # (agent_id, error_message) def engine_commands(self) -> list[dict[str, Any]]: ...

Snapshot and restore

Simulations can be checkpointed and restored. Components that implement the Snapshotable protocol are serialized; others are skipped gracefully.

python

# In-memory snapshot
snap = sim.snapshot()
print(f"Snapshot at tick {snap.tick}, complete={snap.is_complete()}")

# Restore to snapshotted state (agents mutated in-place)
sim.restore(snap)

# File persistence
sim.save_snapshot("checkpoints/tick_10.pkl")
sim.load_snapshot("checkpoints/tick_10.pkl")

Snapshot guarantees

Agents are mutated in-place during restore, so all existing object references (Brain instances, EventBus subscribers) remain valid. No new objects are created. Check snap.is_complete() to verify all components were captured.

Agent

Agents are identity + runtime state containers. They do not own personality, prompts, or action semantics.

@dataclass class Agent: id: str name: str brain: Brain memory: Memory inventory: dict[str, Any] = # item counts state_ext: AgentStateExtension | None = None # optional domain vitals capabilities: frozenset[str] = frozenset() # action capability tags metadata: dict[str, Any] = # arbitrary client-supplied data

capabilities is matched against ActionSchema.tags each tick to determine which actions the agent is offered. See Per-agent capability filtering.

metadata is opaque to the tick loop. It is stored on the agent, included in the AGENT_REGISTERED event, and preserved through snapshot/restore. Use it to tag agents with scene context, ownership info, or any client-defined data.

Runtime Agent Registration

Agents can be added to and removed from a running simulation without restarting the backend. This is the foundation for Unity's host-owned ownership model where prefabs register themselves on connect.

AgentDefinition

The AgentDefinition dataclass is the domain model for runtime registration. It carries the full specification and is validated before any Python import is attempted. Source: src/engine/agent_definition.py.

@dataclass class AgentDefinition: id: str name: str brain_class: str brain_config: dict[str, Any] = memory_class: str | None = None memory_config: dict[str, Any] = capabilities: list[str] = [] inventory: dict[str, Any] = metadata: dict[str, Any] =

Simulation.register_agent() / unregister_agent()

python

from src.engine.agent_definition import AgentDefinition

defn = AgentDefinition(
    id           = "scout_001",
    name         = "Scout",
    brain_class  = "src.plugins.builtin.idle_brain.brain.IdleBrain",
    capabilities = ["patrol"],
    metadata     = {"scene": "level_01", "owner": "client_A"},
)

# Register — validates, imports, instantiates, appends to scheduler
agent = sim.register_agent(defn)

# Unregister — closes brain, removes from scheduler, emits event
sim.unregister_agent("scout_001")

Registration lifecycle

Validate — validate_definition() checks structural constraints (id format, non-empty strings) with no side effects. Failures raise AgentDefinitionError with a field-annotated message.
Duplicate check — if the id is already registered, the call raises AgentDefinitionError. The WebSocket transport handles reconnect logic at the protocol layer via the reconnect param in agent.register — see Transport docs.
Construct — build_agent_from_definition() imports brain_class (and optionally memory_class) by dotted path, instantiates them with their config kwargs.
Register — appends to Simulation.agents, calls social.add_agent() if a social system is configured, calls world.register_agents() if the world supports it, emits AGENT_REGISTERED on the event bus.
Tick — the agent enters the scheduler immediately; the next tick includes it.

Unregistration reverses steps 4 and 3: removes from the agents list, calls world.register_agents(), calls brain.close() if the brain implements the Closeable protocol, emits AGENT_UNREGISTERED.

YAML agents are unaffected

Runtime registration is a purely additive pathway. Agents declared in sim.yaml load via Simulation.from_config() as before and are never touched by register_agent() / unregister_agent(). Both YAML-static and runtime-dynamic agents participate in the same tick loop and are returned together by agent.list.

Scheduler

The Scheduler controls the order and concurrency of agent steps within a tick. Source: src/engine/scheduler.py.

SimultaneousScheduler (default)

All agents run concurrently using asyncio.gather(). Each agent observes the world state as it was at the start of the tick. This is the default and is fast at scale.

python

from src.engine.scheduler import SimultaneousScheduler

sim = Simulation(
    agents   = agents,
    world    = world,
    registry = registry,
    scheduler = SimultaneousScheduler(),   # this is the default
)

SequentialScheduler

Agents step one at a time in a fixed order. Each agent sees the mutations from all prior agents this tick. Fully deterministic when combined with a seeded RNG and ReplayBrain.

python

from src.engine.scheduler import SequentialScheduler

# Use agent list order
scheduler = SequentialScheduler()

# Explicit ordering by agent_id
scheduler = SequentialScheduler(order=["agent_001", "agent_003", "agent_002"])

sim = Simulation(agents=agents, world=world, registry=registry, scheduler=scheduler)

In YAML:

yaml

engine:
  scheduler: sequential   # or "simultaneous" (default)

Custom scheduler

Implement the Scheduler protocol to add priority-based ordering, weighted random ordering, or turn-based mechanics:

The Scheduler protocol requires one method: async run_tick(agents, step_fn) -> list[tuple[Agent, Any]]. Exceptions are returned as the result value, not raised.

python

class PriorityScheduler:
    async def run_tick(self, agents, step_fn) -> list[tuple]:
        sorted_agents = sorted(agents, key=lambda a: a.state_ext.priority, reverse=True)
        results = []
        for agent in sorted_agents:
            try:
                result = await step_fn(agent)
            except Exception as e:
                result = e
            results.append((agent, result))
        return results

Agent Runtime

AgentRuntime drives a single agent through one complete tick step. One AgentRuntime instance is shared across all agents per simulation run. Source: src/engine/agent_runtime.py.

Step pipeline

Each call to step(agent) executes this pipeline in order:

Tick state extension — calls state_ext.tick() to advance domain-specific vitals (health decay, hunger, etc.)
Build observation — calls world.observe(agent_id), merges in nearby agents, identity fields, and state snapshot
Brain decides — calls registry.schemas_for(agent.capabilities) to build the filtered action list, then await agent.brain.decide(view, observation, schemas, context) returning an Intent
Dispatch — registry.dispatch(intent, view, world) executes the matching ActionHandler
Apply mutations — applies state_mutations to state_ext and inventory; calls world.apply() for cross-agent effects
Store memory — calls agent.memory.store(tick, obs_str, intent, outcome_text)
Emit events — emits ACTION_COMPLETED or ACTION_FAILED on the event bus

Observation construction

The observation dict sent to the brain is assembled from multiple sources. The engine merges these fields on top of whatever world.observe() returns:

Field	Source
`nearby_agents`	From pushed observation (HostedWorld), or `VisibilityWorld.get_nearby_agents()`, or `[]`
`agent_id`	Injected by engine
`agent_name`	Injected by engine
`inventory`	Defensive copy of `agent.inventory`
`state_ext`	Snapshot dict from `state_ext.snapshot()`
`state_advice`	Urgent one-line warning from `state_ext.urgent_advice()`
`state_str`	Human-readable vitals from `state_ext.to_prompt_str()`
everything else	Verbatim from `world.observe(agent_id)`

ActionRegistry

The registry maps action names to (schema, handler) pairs. The engine never hard-codes action semantics. Source: src/engine/registry.py.

python

from src.engine.registry import ActionRegistry
from src.contracts.action import ActionHint, ActionSchema, ActionResult, StateMutations

class GatherFoodHandler:
    def execute(self, agent, intent, context) -> ActionResult:
        amount = context.rng.randint(1, 5)
        return ActionResult(
            success=True,
            outcome_text=f"gathered {amount} food",
            mutations=StateMutations(inventory={"food": amount}),
        )

registry = ActionRegistry()
registry.register(
    ActionSchema(
        "gather_food",
        "Collect food from your current location.",
        execution_hint=ActionHint.ENGINE,
        required_capabilities=frozenset({"forager"}),   # only agents with "forager" capability
        example={"action": "gather_food", "parameters": {}},
    ),
    GatherFoodHandler(),
)

# Dispatch an intent
from src.contracts.action import Intent
result = registry.dispatch(Intent(action="gather_food"), agent_view, world)

schemas_for()

registry.schemas_for(capabilities) returns only the schemas visible to an agent with the given capability set. This is what AgentRuntime calls on every step — the brain receives the pre-filtered list.

python

# All schemas (universal + all tagged)
all_schemas = registry.schemas()

# Only schemas visible to an agent with {"forager", "guard"} capabilities
filtered = registry.schemas_for(frozenset({"forager", "guard"}))

# frozenset() → only universal schemas (no tags)
universal_only = registry.schemas_for(frozenset())

Building the action prompt

LLM brains receive the pre-filtered actions list in Brain.decide(). Call ActionSchema.prompt_block() on each schema to render the system prompt section:

python

# `actions` is already filtered to this agent's capabilities by AgentRuntime
action_lines = ["AVAILABLE ACTIONS (use exactly these names):"]
for schema in actions:
    action_lines.append(schema.prompt_block())
system_prompt = "\n".join(action_lines)

Event Bus

EventBus is a synchronous publish/subscribe bus. It is the primary observability mechanism — the WebSocket event stream, replay recorders, analytics, and logging all hook in here. Source: src/engine/event_bus.py.

Standard event types

Constant	String value	Emitted by	agent_id
`TICK_START`	`"tick_start"`	Simulation._tick()	`"engine"`
`TICK_END`	`"tick_end"`	Simulation._tick()	`"engine"`
`ACTION_COMPLETED`	`"action_completed"`	AgentRuntime.step()	agent id
`ACTION_FAILED`	`"action_failed"`	AgentRuntime.step()	agent id
`BRAIN_DECIDED`	`"brain_decided"`	Brain impls (optional)	agent id
`AGENT_STEP_ERROR`	`"agent_step_error"`	Simulation._tick()	agent id
`AGENT_REGISTERED`	`"agent_registered"`	Simulation.register_agent()	agent id
`AGENT_UNREGISTERED`	`"agent_unregistered"`	Simulation.unregister_agent()	agent id

Basic usage

python

from src.engine.event_bus import EventBus, Event, TICK_END, ACTION_COMPLETED

bus = EventBus()

# Subscribe to a specific event type
def on_tick_end(event: Event):
    print(f"Tick {event.tick} complete")

bus.subscribe(TICK_END, on_tick_end)

# Subscribe to all events (wildcard)
bus.subscribe("*", lambda e: print(f"[{e.type}] tick={e.tick} agent={e.agent_id}"))

# Emit an event (the engine does this automatically; shown for custom use)
bus.emit(Event(type=TICK_END, tick=5, agent_id="engine", data={"agent_count": 3}))

# Unsubscribe
bus.unsubscribe(TICK_END, on_tick_end)

Built-in subscribers

Several built-in subscriber classes handle common cross-cutting concerns:

python

from src.engine.event_bus import (
    SocialEffectSubscriber,   # reads social side_effects, updates SocialSystem
    EventLogSubscriber,       # appends plain-text log for ACTION_COMPLETED events
    ObservabilitySubscriber,  # user-provided hooks for tick/action/failure events
)

# Log subscriber
log_sub = EventLogSubscriber()
bus.subscribe(ACTION_COMPLETED, log_sub)
# later: log_sub.tail(20)  # last 20 log lines

# Observability hooks
obs = ObservabilitySubscriber(
    on_tick   = lambda tick: print(f"=== Tick {tick} ==="),
    on_action = lambda ev:   print(f"  {ev.data['agent_name']}: {ev.data['action']}"),
    on_failure= lambda ev:   print(f"  FAIL: {ev.data.get('outcome')}"),
)
bus.subscribe("*", obs)

Async note

The event bus is synchronous. Subscribers must not be async coroutines. Async event subscribers are not supported — this is a known limitation.

ObservationRegistry

ObservationRegistry manages named observation providers and merges their output into agent observations before each brain call. It is optional — simulations without one work identically to pre-observation behavior. Source: src/engine/obs_registry.py.

Architecture mirrors ActionRegistry: register a schema + provider pair, and the registry handles capability filtering and fault isolation.

python

from src.contracts.observation import ObservationSchema
from src.engine.obs_registry import ObservationRegistry
from src.plugins.builtin.observations.providers import (
    SimulationTimeProvider,
    SocialContextProvider,
)

obs_registry = ObservationRegistry()

# Universal — every agent sees this
obs_registry.register(
    ObservationSchema("simulation_tick", "Current simulation tick number.",
                      {"simulation_tick": "int"}),
    SimulationTimeProvider(),
)

# Capability-gated — only agents with "social" capability see this
obs_registry.register(
    ObservationSchema("social_relationships", "Summary of agent relationships.",
                      {"social_relationships": "str"},
                      tags=frozenset({"social"})),
    SocialContextProvider(social_system),
)

# Pass to Simulation
sim = Simulation(agents=[...], world=world, registry=registry,
                 obs_registry=obs_registry)

Fault isolation

Provider exceptions are caught, logged at WARNING level (naming the provider and agent), and skipped. The remaining providers still run. A buggy provider cannot crash a tick or silence other providers.

Capability filtering

Works identically to action filtering: schemas with empty tags are universal; schemas with non-empty tags are only sent to agents whose capabilities intersect the tags. The merge result is what the brain sees in its observation dict.

Observation construction order

Priority	Source	Notes
3 (lowest)	`ObservationRegistry.collect()`	All matching providers merged in registration order
2	`world.observe(agent_id)`	Overwrites registry output on key conflicts
1 (highest)	Engine identity fields	`agent_id`, `agent_name`, `inventory`, `state_ext`, `state_advice`, `state_str`, `nearby_agents` — always overwritten by engine

Event data shapes

Event type	data keys
`tick_start`	World metadata dict
`tick_end`	`{"agent_count": N}`
`action_completed`	`{"agent_name", "action", "target", "reasoning", "outcome", "side_effects", "location"}`
`action_failed`	Same as action_completed
`brain_decided`	`{"agent_name", "system", "prompt", "raw_output", "intent": {...}}`
`agent_step_error`	`{"error": "...", "agent_name": "..."}`
`agent_registered`	`{"agent_name", "capabilities", "metadata"}`
`agent_unregistered`	`{"agent_name"}`