simulatecraft.core

simulatecraft.core

Core simulation loop: agents, environments, events, and the runner.

Agent

Bases: BaseModel

Wraps a Brain; the runner only ever talks to this interface.

on_human_message

on_human_message(sender: str, text: str) -> None

Hook for human-in-the-loop chat. Brains may override via attribute.

Source code in src/simulatecraft/core/agent.py

def on_human_message(self, sender: str, text: str) -> None:
    """Hook for human-in-the-loop chat. Brains may override via attribute."""
    handler = getattr(self.brain, "on_human_message", None)
    if handler is not None:
        handler(sender, text)

Environment

Environment()

Bases: ABC

Owns all mutable simulation state.

Contract

• observe may return partial state (partial observability is supported).
• step mutates state for one agent and returns a StepResult.
• agent_ids must reflect dynamic membership (spawn/death/exit).
• tick advances environment-owned state (weather, NPC timers, physics).

Source code in src/simulatecraft/core/environment.py

def __init__(self) -> None:
    self._tick_count: int = 0
    self._registered: set[str] = set()

observe abstractmethod

observe(agent_id: str) -> Observation | Awaitable[Observation]

Return the state visible to agent_id (may be partial).

Source code in src/simulatecraft/core/environment.py

@abstractmethod
def observe(self, agent_id: str) -> Observation | Awaitable[Observation]:
    """Return the state visible to ``agent_id`` (may be partial)."""

step abstractmethod

step(agent_id: str, action: Action) -> StepResult | Awaitable[StepResult]

Apply action for agent_id, mutating environment state.

Source code in src/simulatecraft/core/environment.py

@abstractmethod
def step(self, agent_id: str, action: Action) -> StepResult | Awaitable[StepResult]:
    """Apply ``action`` for ``agent_id``, mutating environment state."""

tick

tick() -> None | Awaitable[None]

Advance environment-owned dynamics once per simulation tick.

Source code in src/simulatecraft/core/environment.py

def tick(self) -> None | Awaitable[None]:
    """Advance environment-owned dynamics once per simulation tick."""
    self._tick_count += 1
    return None

reset

reset(seed: int | None = None) -> None

Reset to the initial episode state. Subclasses should override.

Source code in src/simulatecraft/core/environment.py

def reset(self, seed: int | None = None) -> None:
    """Reset to the initial episode state. Subclasses should override."""
    self._tick_count = 0

snapshot

snapshot() -> Snapshot

Full-state view for REST/viewers. Override to include world details.

Source code in src/simulatecraft/core/environment.py

def snapshot(self) -> Snapshot:
    """Full-state view for REST/viewers. Override to include world details."""
    return Snapshot(tick=self._tick_count)

Snapshot

Bases: BaseModel

Domain-agnostic full-state snapshot served over REST / rendered by viewers.

EventBus

EventBus()

Ordered pub/sub with isolated subscriber errors and an inbound queue.

Source code in src/simulatecraft/core/events.py

def __init__(self) -> None:
    self._handlers: list[tuple[EventHandler, bool]] = []
    self._inbound: asyncio.Queue[InboundEvent] = asyncio.Queue()
    self._loop: asyncio.AbstractEventLoop | None = None

publish_inbound

publish_inbound(event: InboundEvent) -> None

Queue an inbound event AND mirror it onto the outbound bus for viewers.

Source code in src/simulatecraft/core/events.py

def publish_inbound(self, event: InboundEvent) -> None:
    """Queue an inbound event AND mirror it onto the outbound bus for viewers."""
    try:
        current = asyncio.get_running_loop()
    except RuntimeError:
        current = None
    if current is not None:
        if self._loop is None:
            self.bind_loop(current)
        if current is self._loop:
            self._inbound.put_nowait(event)
            current.create_task(self.publish(event))
            return
    if self._loop is not None and self._loop.is_running():

        def _queue_and_broadcast() -> None:
            self._inbound.put_nowait(event)
            asyncio.ensure_future(self.publish(event), loop=self._loop)

        self._loop.call_soon_threadsafe(_queue_and_broadcast)
    else:
        self._inbound.put_nowait(event)

HumanChat

Bases: Event

Inbound: a human sent a message, optionally targeting one agent.

HumanControl

Bases: Event

Inbound: viewer control commands (pause/resume/step/stop/reset).

Runner dataclass

Runner(environment: Environment, agents: list[Agent] = list(), bus: EventBus = EventBus(), config: RunnerConfig = RunnerConfig(), _running: bool = False, _paused: bool = False, _step_requests: int = 0, _stop_reason: str = '', _control_lock: Lock = Lock())

remove_agent

remove_agent(agent_id: str) -> bool

Remove an agent from the runner (does not disconnect Minecraft).

Source code in src/simulatecraft/core/runner.py

def remove_agent(self, agent_id: str) -> bool:
    """Remove an agent from the runner (does not disconnect Minecraft)."""
    before = len(self.agents)
    self.agents = [a for a in self.agents if a.id != agent_id]
    self._known_ids.discard(agent_id)
    return len(self.agents) < before

start async

start() -> None

Run until max_ticks / empty env / stop(). Returns when finished.

Source code in src/simulatecraft/core/runner.py

async def start(self) -> None:
    """Run until max_ticks / empty env / stop(). Returns when finished."""
    if self._running:
        raise RuntimeError("Runner already running")
    await self._emit(SimulationStarted(agent_ids=[a.id for a in self.agents]))
    self._running = True
    self._stop_reason = "max_ticks"
    await self._sync_membership()
    try:
        while self._running and self.environment.tick_count < self.config.max_ticks:
            await self._process_inbound()
            if not self._running:
                break
            if self._paused:
                if self._step_requests > 0:
                    self._step_requests -= 1
                else:
                    await asyncio.sleep(0.05)
                    continue
            await self.run_tick()
            if not self._running:
                break
            if self.config.stop_when_env_empty and not self.environment.agent_ids:
                self._stop_reason = "no_agents_left"
                break
            await self._pace()
    finally:
        self._running = False
        await self._emit(SimulationEnded(reason=self._stop_reason))

set_tick_rate

set_tick_rate(rate: float | None) -> float | None

Ticks per second. None = run as fast as possible.

Source code in src/simulatecraft/core/runner.py

def set_tick_rate(self, rate: float | None) -> float | None:
    """Ticks per second. ``None`` = run as fast as possible."""
    if rate is None:
        self.config.tick_rate = None
    else:
        value = float(rate)
        if value <= 0:
            self.config.tick_rate = None
        else:
            self.config.tick_rate = max(0.05, min(50.0, value))
    return self.config.tick_rate

adjust_tick_rate

adjust_tick_rate(factor: float) -> float | None

Multiply current rate (e.g. 2.0 faster, 0.5 slower). Overspeed → unlimited.

Source code in src/simulatecraft/core/runner.py

def adjust_tick_rate(self, factor: float) -> float | None:
    """Multiply current rate (e.g. 2.0 faster, 0.5 slower). Overspeed → unlimited."""
    current = self.config.tick_rate
    if factor <= 0:
        raise ValueError("factor must be positive")
    if current is None or current <= 0:
        if factor < 1.0:
            return self.set_tick_rate(50.0)
        return None
    new_rate = current * float(factor)
    if new_rate > 50.0:
        return self.set_tick_rate(None)
    return self.set_tick_rate(new_rate)

step_once async

step_once() -> None

Execute exactly one tick regardless of pause state.

Source code in src/simulatecraft/core/runner.py

async def step_once(self) -> None:
    """Execute exactly one tick regardless of pause state."""
    was_paused = self._paused
    self._paused = False
    try:
        await self.run_tick()
    finally:
        self._paused = was_paused

RunnerConfig

Bases: BaseModel

tick_rate=None runs as fast as possible (batch mode); a number paces realtime.

Action

Bases: StrictModel

Base class for domain-specific actions.

Subclass this in your environment package (e.g. MoveAction(kind="move")) and pass the discriminated union to brains so LLM/RL outputs arrive validated.

AgentState

Bases: StrictModel

Flexible per-agent state container (position, inventory, mood, ...).

Observation

Bases: StrictModel

Structured state visible to one agent. Supports partial observability.

The data payload is domain-specific; define typed subclasses for richer schemas (e.g. GridObservation) when you want validation at the edges.

StepResult

Bases: StrictModel

Outcome of applying one action for one agent.