Core API

Stateless, SOLID building blocks for the ECS framework.

Overview

The core API provides the fundamental abstractions for entities, components, systems, and queries. These are pure, stateless functionalities that can be composed to build complex agent behaviors.

Key Principles: - Protocols over inheritance: Use runtime-checkable protocols for flexibility - Functional core: All operations are deterministic and stateless - Composability: Mix and match protocols to create rich component behaviors

---

Component Module

Define components with optional operation protocols.

Component Decorator

component(cls=None)

component(cls: type) -> type

component(cls: None = None) -> Callable[[type], type]

Register a dataclass or Pydantic model as a component type.

Supports two forms

@component # bare decorator @component() # parenthesized, no args

Parameters:

Name	Type	Description	Default
cls	type | None	The class to register, or None if called with arguments.	None

Returns:

Type	Description
type | Callable[[type], type]	Decorated class or decorator function.

Raises:

Type	Description
TypeError	If class is neither a dataclass nor Pydantic model.

Note

Apply @component AFTER @dataclass:

@component ... @dataclass(slots=True) ... class MyComponent: ... value: int

Source code in src/agentecs/core/component/core.py

def component(cls: type | None = None) -> type | Callable[[type], type]:
    """Register a dataclass or Pydantic model as a component type.

    Supports two forms:
        @component                    # bare decorator
        @component()                  # parenthesized, no args

    Args:
        cls: The class to register, or None if called with arguments.

    Returns:
        Decorated class or decorator function.

    Raises:
        TypeError: If class is neither a dataclass nor Pydantic model.

    Note:
        Apply @component AFTER @dataclass:

        >>> @component
        ... @dataclass(slots=True)
        ... class MyComponent:
        ...     value: int
    """

    def decorator(c: type) -> type:
        if not (is_dataclass(c) or _is_pydantic(c)):
            raise TypeError(
                f"Component {c.__name__} must be a dataclass or Pydantic model. "
                f"Did you forget @dataclass decorator?"
            )
        meta = _registry.register(c)
        c.__component_meta__ = meta  # type: ignore
        return c

    if cls is None:
        # Called with args: @component()
        return decorator
    else:
        # Called bare: @component
        return decorator(cls)

Component Registry

ComponentRegistry

Process-local registry mapping component types to deterministic type IDs.

Maintains bidirectional mapping between component types and their type IDs. Deterministic IDs ensure same code produces same IDs across nodes.

TODO: Figure out distributed syncing of local registries if needed.

Source code in src/agentecs/core/component/core.py

class ComponentRegistry:
    """Process-local registry mapping component types to deterministic type IDs.

    Maintains bidirectional mapping between component types and their type IDs.
    Deterministic IDs ensure same code produces same IDs across nodes.

    # TODO: Figure out distributed syncing of local registries if needed.
    """

    def __init__(self) -> None:
        """Initialize empty component registry."""
        self._by_type: dict[type, ComponentTypeMeta] = {}
        self._by_type_id: dict[int, type] = {}

    def register(self, cls: type) -> ComponentTypeMeta:
        """Register a component type and return its metadata.

        Args:
            cls: Component class to register.

        Returns:
            Component metadata including ID and type name.

        Raises:
            RuntimeError: If component ID collides with another registered type.
        """
        if cls in self._by_type:
            return self._by_type[cls]

        component_type_id = _stable_component_type_id(cls)

        if component_type_id in self._by_type_id:
            existing = self._by_type_id[component_type_id]
            raise RuntimeError(
                f"Component ID collision: {cls} and {existing} hash to {component_type_id}"
            )

        meta = ComponentTypeMeta(
            component_type_id=component_type_id,
            type_name=f"{cls.__module__}.{cls.__qualname__}",
        )
        self._by_type[cls] = meta
        self._by_type_id[component_type_id] = cls
        return meta

    def get_meta(self, cls: type) -> ComponentTypeMeta | None:
        """Get metadata for a registered component type.

        Args:
            cls: Component class to look up.

        Returns:
            Component metadata if registered, None otherwise.
        """
        return self._by_type.get(cls)

    def get_type(self, component_type_id: int) -> type | None:
        """Get component type by its type ID.

        Args:
            component_type_id: Component type ID to look up.

        Returns:
            Component class if found, None otherwise.
        """
        return self._by_type_id.get(component_type_id)

    def is_registered(self, cls: type) -> bool:
        """Check if a type is registered as a component.

        Args:
            cls: Class to check.

        Returns:
            True if class is registered as component, False otherwise.
        """
        return cls in self._by_type

register(cls)

Register a component type and return its metadata.

Parameters:

Name	Type	Description	Default
cls	type	Component class to register.	required

Returns:

Type	Description
ComponentTypeMeta	Component metadata including ID and type name.

Raises:

Type	Description
RuntimeError	If component ID collides with another registered type.

Source code in src/agentecs/core/component/core.py

def register(self, cls: type) -> ComponentTypeMeta:
    """Register a component type and return its metadata.

    Args:
        cls: Component class to register.

    Returns:
        Component metadata including ID and type name.

    Raises:
        RuntimeError: If component ID collides with another registered type.
    """
    if cls in self._by_type:
        return self._by_type[cls]

    component_type_id = _stable_component_type_id(cls)

    if component_type_id in self._by_type_id:
        existing = self._by_type_id[component_type_id]
        raise RuntimeError(
            f"Component ID collision: {cls} and {existing} hash to {component_type_id}"
        )

    meta = ComponentTypeMeta(
        component_type_id=component_type_id,
        type_name=f"{cls.__module__}.{cls.__qualname__}",
    )
    self._by_type[cls] = meta
    self._by_type_id[component_type_id] = cls
    return meta

get_meta(cls)

Get metadata for a registered component type.

Parameters:

Name	Type	Description	Default
cls	type	Component class to look up.	required

Returns:

Type	Description
ComponentTypeMeta | None	Component metadata if registered, None otherwise.

Source code in src/agentecs/core/component/core.py

def get_meta(self, cls: type) -> ComponentTypeMeta | None:
    """Get metadata for a registered component type.

    Args:
        cls: Component class to look up.

    Returns:
        Component metadata if registered, None otherwise.
    """
    return self._by_type.get(cls)

get_type(component_type_id)

Get component type by its type ID.

Parameters:

Name	Type	Description	Default
component_type_id	int	Component type ID to look up.	required

Returns:

Type	Description
type | None	Component class if found, None otherwise.

Source code in src/agentecs/core/component/core.py

def get_type(self, component_type_id: int) -> type | None:
    """Get component type by its type ID.

    Args:
        component_type_id: Component type ID to look up.

    Returns:
        Component class if found, None otherwise.
    """
    return self._by_type_id.get(component_type_id)

is_registered(cls)

Check if a type is registered as a component.

Parameters:

Name	Type	Description	Default
cls	type	Class to check.	required

Returns:

Type	Description
bool	True if class is registered as component, False otherwise.

Source code in src/agentecs/core/component/core.py

def is_registered(self, cls: type) -> bool:
    """Check if a type is registered as a component.

    Args:
        cls: Class to check.

    Returns:
        True if class is registered as component, False otherwise.
    """
    return cls in self._by_type

get_registry()

Access the global component registry.

Returns:

Type	Description
ComponentRegistry	The process-local ComponentRegistry instance.

Source code in src/agentecs/core/component/core.py

def get_registry() -> ComponentRegistry:
    """Access the global component registry.

    Returns:
        The process-local ComponentRegistry instance.
    """
    return _registry

Operation Protocols

Components can optionally implement these protocols to enable advanced operations:

Combinable

Bases: Protocol

Component knows how to accumulate multiple writes.

When multiple ops target the same (entity, type), the framework folds them with combine instead of overwriting.

Source code in src/agentecs/core/component/models.py

@runtime_checkable
class Combinable(Protocol):
    """Component knows how to accumulate multiple writes.

    When multiple ops target the same (entity, type), the framework
    folds them with __combine__ instead of overwriting.
    """

    def __combine__(self, other: Self) -> Self: ...

Splittable

Bases: Protocol

One instance → two instances (for agent splitting).

Source code in src/agentecs/core/component/models.py

@runtime_checkable
class Splittable(Protocol):
    """One instance → two instances (for agent splitting)."""

    def __split__(self) -> tuple[Self, Self]: ...

Utility Functions

combine_protocol_or_fallback(comp1, comp2)

Combine two components, using Combinable protocol with LWW fallback.

If comp1 implements Combinable and comp2 is the same type, delegates to comp1.combine(comp2). Otherwise returns comp2 (last-writer-wins).

Parameters:

Name	Type	Description	Default
comp1	T	Existing component value.	required
comp2	T	Incoming component value.	required

Returns:

Type	Description
T	Combined result or comp2 as fallback.

Source code in src/agentecs/core/component/operations.py

def combine_protocol_or_fallback[T](comp1: T, comp2: T) -> T:
    """Combine two components, using Combinable protocol with LWW fallback.

    If comp1 implements Combinable and comp2 is the same type, delegates
    to comp1.__combine__(comp2). Otherwise returns comp2 (last-writer-wins).

    Args:
        comp1: Existing component value.
        comp2: Incoming component value.

    Returns:
        Combined result or comp2 as fallback.
    """
    if not isinstance(comp1, Combinable) or not isinstance(comp2, type(comp1)):
        return comp2
    else:
        return comp1.__combine__(comp2)

split_protocol_or_fallback(comp)

Split a component, using Splittable protocol with deepcopy fallback.

If comp implements Splittable, delegates to comp.split(). Otherwise returns two independent deep copies.

Parameters:

Name	Type	Description	Default
comp	T	Component to split.	required

Returns:

Type	Description
tuple[T, T]	Tuple of two components.

Source code in src/agentecs/core/component/operations.py

def split_protocol_or_fallback[T](comp: T) -> tuple[T, T]:
    """Split a component, using Splittable protocol with deepcopy fallback.

    If comp implements Splittable, delegates to comp.__split__().
    Otherwise returns two independent deep copies.

    Args:
        comp: Component to split.

    Returns:
        Tuple of two components.
    """
    if not isinstance(comp, Splittable):
        return (copy.deepcopy(comp), copy.deepcopy(comp))
    return cast(tuple[T, T], comp.__split__())

reduce_components(items)

Reduce a list of components into one.

Uses sequential combines

if combine is defined, merges items using that pairwise; otherwise, takes the last item as the result.

Parameters:

Name	Type	Description	Default
items	list[T]	List of components to reduce (must be same type).	required

Returns:

Type	Description
T	Single component resulting from reduction.

Raises:

Type	Description
ValueError	If items list is empty.

Source code in src/agentecs/core/component/operations.py

def reduce_components[T](items: list[T]) -> T:
    """Reduce a list of components into one.

    Uses sequential combines:
        if __combine__ is defined, merges items using that pairwise;
        otherwise, takes the last item as the result.

    Args:
        items: List of components to reduce (must be same type).

    Returns:
        Single component resulting from reduction.

    Raises:
        ValueError: If items list is empty.
    """
    if not items:
        raise ValueError("Cannot reduce empty list")
    if len(items) == 1:
        return items[0]
    result = items[0]
    for item in items[1:]:
        result = combine_protocol_or_fallback(result, item)
    return result

---

Identity Module

Entity identification with generational indices.

EntityId dataclass

Lightweight entity identifier with generation for safe handle reuse.

Shard field enables future distributed scaling - each shard allocates indices independently within its range.

Source code in src/agentecs/core/identity/models.py

@dataclass(frozen=True, slots=True)
class EntityId:
    """Lightweight entity identifier with generation for safe handle reuse.

    Shard field enables future distributed scaling - each shard allocates
    indices independently within its range.
    """

    shard: int = 0  # 0 = local, >0 = remote shard
    index: int = 0
    generation: int = 0

    def __hash__(self) -> int:
        return hash((self.shard, self.index, self.generation))

    def is_local(self) -> bool:
        """Check if this entity belongs to the local shard.

        Returns:
            True if entity is on shard 0 (local), False otherwise.
        """
        return self.shard == 0

is_local()

Check if this entity belongs to the local shard.

Returns:

Type	Description
bool	True if entity is on shard 0 (local), False otherwise.

Source code in src/agentecs/core/identity/models.py

def is_local(self) -> bool:
    """Check if this entity belongs to the local shard.

    Returns:
        True if entity is on shard 0 (local), False otherwise.
    """
    return self.shard == 0

SystemEntity

Reserved entity IDs for singletons. Always on shard 0.

Source code in src/agentecs/core/identity/models.py

class SystemEntity:
    """Reserved entity IDs for singletons. Always on shard 0."""

    WORLD = EntityId(shard=0, index=0, generation=0)
    CLOCK = EntityId(shard=0, index=1, generation=0)
    SCHEDULER = EntityId(shard=0, index=2, generation=0)

    _RESERVED_COUNT = 1000  # First 1000 indices reserved

---

Query Module

Query builder for filtering entities by component types.

Query dataclass

Declarative query for access pattern declarations.

Immutable - each method returns a new Query instance.

Source code in src/agentecs/core/query/models.py

@dataclass(frozen=True)
class Query:
    """Declarative query for access pattern declarations.

    Immutable - each method returns a new Query instance.
    """

    required: tuple[type, ...] = ()
    excluded: tuple[type, ...] = ()

    def __init__(self, *required: type):
        object.__setattr__(self, "required", required)
        object.__setattr__(self, "excluded", ())

    def having(self, *types: type) -> Query:
        """Entities must also have these component types."""
        new = Query(*self.required, *types)
        object.__setattr__(new, "excluded", self.excluded)
        return new

    def excluding(self, *types: type) -> Query:
        """Entities must NOT have these component types."""
        new = Query(*self.required)
        object.__setattr__(new, "excluded", self.excluded + types)
        return new

    def __iter__(self) -> Iterator[type]:
        """Allow Query to be used where tuple of types expected."""
        return iter(self.required)

    def __contains__(self, item: type) -> bool:
        return item in self.required

    def types(self) -> frozenset[type]:
        """All types this query accesses (required only)."""
        return frozenset(self.required)

    def matches_archetype(self, has: frozenset[type]) -> bool:
        """Check if an archetype (set of component types) matches this query."""
        return all(t in has for t in self.required) and all(t not in has for t in self.excluded)

having(*types)

Entities must also have these component types.

Source code in src/agentecs/core/query/models.py

def having(self, *types: type) -> Query:
    """Entities must also have these component types."""
    new = Query(*self.required, *types)
    object.__setattr__(new, "excluded", self.excluded)
    return new

excluding(*types)

Entities must NOT have these component types.

Source code in src/agentecs/core/query/models.py

def excluding(self, *types: type) -> Query:
    """Entities must NOT have these component types."""
    new = Query(*self.required)
    object.__setattr__(new, "excluded", self.excluded + types)
    return new

types()

All types this query accesses (required only).

Source code in src/agentecs/core/query/models.py

def types(self) -> frozenset[type]:
    """All types this query accesses (required only)."""
    return frozenset(self.required)

matches_archetype(has)

Check if an archetype (set of component types) matches this query.

Source code in src/agentecs/core/query/models.py

def matches_archetype(self, has: frozenset[type]) -> bool:
    """Check if an archetype (set of component types) matches this query."""
    return all(t in has for t in self.required) and all(t not in has for t in self.excluded)

AccessPattern = AllAccess | TypeAccess | QueryAccess | NoAccess module-attribute

---

System Module

Define systems with declared access patterns.

System Decorator

system = _SystemDecorator() module-attribute

System Metadata

SystemDescriptor dataclass

Metadata about a registered system.

Source code in src/agentecs/core/system/models.py

@dataclass(frozen=True)
class SystemDescriptor:
    """Metadata about a registered system."""

    name: str
    run: Callable[..., Any]
    reads: AccessPattern
    writes: AccessPattern
    mode: SystemMode
    is_async: bool = False
    frequency: float = 1.0
    phase: str = "update"
    runs_alone: bool = False  # If True, runs in its own execution group (dev mode)

    def can_read_type(self, component_type: type) -> bool:
        """Check whether this system can read the given component type.

        Write access implies read access.
        """
        return self._pattern_allows(self.reads, component_type) or self._pattern_allows(
            self.writes, component_type
        )

    def can_write_type(self, component_type: type) -> bool:
        """Check whether this system can write the given component type."""
        return self._pattern_allows(self.writes, component_type)

    def _pattern_allows(self, pattern: AccessPattern, component_type: type) -> bool:
        from agentecs.core.query.models import AllAccess, NoAccess, QueryAccess, TypeAccess

        if isinstance(pattern, AllAccess):
            return True
        if isinstance(pattern, NoAccess):
            return False
        if isinstance(pattern, TypeAccess):
            return component_type in pattern.types
        if isinstance(pattern, QueryAccess):
            return component_type in pattern.types()
        return False

    def is_dev_mode(self) -> bool:
        """Check if system should run in isolation (dev mode).

        Returns:
            True if system should run alone in its own execution group.
        """
        return self.runs_alone

can_read_type(component_type)

Check whether this system can read the given component type.

Write access implies read access.

Source code in src/agentecs/core/system/models.py

def can_read_type(self, component_type: type) -> bool:
    """Check whether this system can read the given component type.

    Write access implies read access.
    """
    return self._pattern_allows(self.reads, component_type) or self._pattern_allows(
        self.writes, component_type
    )

can_write_type(component_type)

Check whether this system can write the given component type.

Source code in src/agentecs/core/system/models.py

def can_write_type(self, component_type: type) -> bool:
    """Check whether this system can write the given component type."""
    return self._pattern_allows(self.writes, component_type)

is_dev_mode()

Check if system should run in isolation (dev mode).

Returns:

Type	Description
bool	True if system should run alone in its own execution group.

Source code in src/agentecs/core/system/models.py

def is_dev_mode(self) -> bool:
    """Check if system should run in isolation (dev mode).

    Returns:
        True if system should run alone in its own execution group.
    """
    return self.runs_alone

SystemMode

Bases: Enum

Execution mode controlling access capabilities.

Source code in src/agentecs/core/system/models.py

class SystemMode(Enum):
    """Execution mode controlling access capabilities."""

    INTERACTIVE = auto()  # Full ScopedAccess, writes during execution
    PURE = auto()  # ReadOnlyAccess, must return all changes
    READONLY = auto()  # ReadOnlyAccess, no writes allowed