simulatecraft.core.runner

simulatecraft.core.runner

Async simulation loop: Environment + Agents + EventBus.

The Runner never inspects what kind of Brain an agent uses. All observability flows through the EventBus; with zero subscribers everything still runs.

RunnerConfig

Bases: BaseModel

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

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