Core Primitives

2.2 Core Primitives

The core primitives in simudyne.engine.sdk form the vocabulary of every model. Understanding these classes is essential before building anything.

Message

Messages are the sole mechanism for inter-agent communication. An agent never reads another agent's state directly --- it sends a message through a link and the recipient reads it from its inbox in a later phase. This no-write-sharing invariant is what makes the execution model deterministic regardless of backend parallelism.

class Message:
    def __init__(self, sender_id: str = "", body: Any = None) -> None
    def to_dict(self) -> Dict[str, Any]
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "Message"

Three convenience subtypes are provided for common payloads:

• EmptyMessage(sender_id="") --- signal-only message with no payload (e.g., "I have defaulted").
• DoubleMessage(sender_id="", body=0.0) --- carries a single float (e.g., a price quote). Provides get_body() -> float and set_body(value: float).
• IntegerMessage(sender_id="", body=0) --- carries a single integer (e.g., a quantity). Provides get_body() -> int and set_body(value: int).

Custom message types are defined by subclassing Message and adding typed attributes:

class TradeOrder(Message):
    price: float
    quantity: int
    side: str  # "buy" or "sell"

Link

A Link is a typed, weighted, directed edge between two agents. Links define the communication topology: an agent can only send messages to agents it is linked to (or broadcast to all agents sharing a link type).

class Link:
    def __init__(
        self,
        link_type: str,       # e.g., "interbank", "supply", "contact"
        source_id: str,        # sender agent ID
        target_id: str,        # receiver agent ID
        weight: float = 1.0,   # edge weight (used by topology algorithms)
        metadata: Optional[Dict[str, Any]] = None,
    ) -> None

Links are created via ABMModel.create_link() or by topology generators (Section 3). The link_type string groups links for selective messaging: agent.broadcast(DoubleMessage, link_type="interbank") sends only to agents connected via "interbank" links.

GlobalState and Accumulator

GlobalState holds model-level configuration accessible to all agents as a read-only proxy. Agents call self.get_globals() to read it; only the model can write it. This enforces the principle that agents do not share mutable state.

Accumulator is a thread-safe aggregation primitive for collecting values across agents within a step. Common uses: counting defaults, summing total demand, tracking maximum price.

class Accumulator:
    def __init__(self, name: str, initial: float = 0.0) -> None
    def add(self, value: float) -> None   # Thread-safe (uses Lock)
    def get(self) -> float
    def reset(self) -> None
    def set(self, value: float) -> None

The standard pattern is: reset_accumulators() at the start of each step, agents call self.get_accumulator("name").add(value) during their actions, and record_accumulators() writes the final values to the output recorder.

Agent

The base class for all simulation agents. Every agent has:

• Typed state via the __state_schema__ class variable (a dict mapping field names to types). State fields are initialised to their type's default (0 for int/float, "" for str, False for bool) and are readable/writable as regular attributes.
• A deterministic PRNG via self.rng (a numpy.random.Generator seeded by the SHA-256 hierarchy).
• Messaging via send(), broadcast(), and get_messages().

class Agent:
    __state_schema__: Dict[str, type] = {}
    __behavior_type__: BehaviorType = BehaviorType.RULE_BASED

    def __init__(self, agent_id: str, model: Optional["ABMModel"] = None) -> None

    # Messaging
    def send(self, msg_type: Type[Message], target: Union[Link, Agent, str],
             **payload: Any) -> None
    def broadcast(self, msg_type: Type[Message], link_type: Optional[str] = None,
                  **payload: Any) -> None
    def get_messages(self, msg_type: Optional[Type[Message]] = None) -> List[Message]
    def has_messages(self, msg_type: Optional[Type[Message]] = None) -> bool

    # Links and globals
    def get_links(self, link_type: Optional[str] = None) -> List[Link]
    def get_globals(self) -> GlobalStateProxy  # Read-only view
    def get_accumulator(self, name: str) -> Accumulator

    # Lifecycle
    def stop(self) -> None          # Mark agent as stopped (excluded from future actions)
    @property
    def is_stopped(self) -> bool

    # Properties
    @property
    def agent_id(self) -> str
    @property
    def rng(self) -> np.random.Generator

Action, Sequence, and Split

These three classes compose agent behaviour into simulation phases:

Action binds an agent type to a callable. When executed, the callable is invoked on every agent of that type.

class Action:
    def __init__(self, agent_type: Type[Agent], fn: Callable[[Agent], None],
                 name: Optional[str] = None) -> None
    @classmethod
    def create(cls, agent_type: Type[Agent], fn: Callable[[Agent], None],
               name: Optional[str] = None) -> "Action"

Sequence chains actions with message delivery between each. This is the fundamental ordering primitive: messages sent in action A are delivered to inboxes before action B begins.

class Sequence:
    def __init__(self, *actions: Action) -> None

Split executes multiple actions simultaneously. Messages from all actions in a split are delivered together after all complete. Use split when two agent types act independently in the same phase (e.g., cancer cells and normal cells both update).

class Split:
    def __init__(self, *actions: Action) -> None

ABMModel

The simulation container. Manages the agent population, link topology, accumulators, execution backend, and lifecycle. Key methods:

class ABMModel(ABC):
    # Lifecycle (override these)
    def init(self) -> None                              # Register types and accumulators
    def setup(self, config: Dict[str, Any]) -> None     # Create agents, topology, initial state
    @abstractmethod
    def step(self) -> None                              # One simulation tick
    def done(self) -> None                              # Cleanup after all steps

    # Agent management
    def register_agent_type(self, name: str, cls: Type[Agent]) -> None
    def create_agents(self, type_name: str, count: int) -> List[Agent]
    def create_agent(self, type_name: str, agent_id: Optional[str] = None) -> Agent
    def get_agent(self, agent_id: str) -> Agent
    def get_agents(self, agent_type: Optional[str] = None) -> List[Agent]
    def remove_agent(self, agent_id: str) -> bool

    # Link management
    def register_link_type(self, name: str, source_type: str, target_type: str) -> None
    def create_link(self, link_type: str, source: Union[Agent, str],
                    target: Union[Agent, str], weight: float = 1.0) -> Link
    def remove_link(self, link: Link) -> bool

    # Accumulators
    def create_accumulator(self, name: str, initial: float = 0.0) -> Accumulator
    def get_accumulator(self, name: str) -> Accumulator
    def reset_accumulators(self) -> None
    def record_accumulators(self, seed: int = 1) -> None

    # Execution
    def run(self, *sequences_or_actions: Union[Sequence, Action, Split]) -> None
    @classmethod
    def run_simulation(cls, config: Dict[str, Any]) -> "ABMModel"

    # Global state
    def get_globals(self) -> GlobalState
    def set_globals(self, gs: GlobalState) -> None

    # Properties
    @property
    def tick(self) -> int
    @property
    def hooks(self) -> HookRegistry
    @property
    def guardrails(self) -> GuardrailRegistry

The lifecycle is always: init() -> setup(config) -> step() x N -> done(). The run_simulation(config) classmethod is a convenience that executes the full lifecycle, using config["steps"] for the step count and config["seed"] for the global seed.

Grid Visualization Colors (__color_map__)

Models that use GridSpace can declare a __color_map__ class attribute on ABMModel to control how agent states are rendered in the frontend grid visualization:

class ForestFireModel(ABMModel):
    __color_map__ = {
        "TREE": "#228B22",
        "BURNING": "#FF4500",
        "ASH": "#808080",
        "EMPTY": "#F5F5DC",
    }

The __color_map__ is a Dict[str, str] mapping state values (strings) to CSS color strings. The frontend reads this attribute via the /api/v1/models/{name}/architecture endpoint to render grid cells with the correct colors. If __color_map__ is not defined, the frontend falls back to auto-assigned colors.

---