simulatecraft.core.events

simulatecraft.core.events

Typed event schemas and the central pub/sub EventBus.

Every observability concern (terminal viewer, JSONL logger, websocket broadcaster) subscribes here. Environments and agents never know they are being watched: with zero subscribers everything still works.

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).

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)