core.event_bus module

Redis Streams event bus for the Stargazer distributed architecture.

Defines RedisEventBus, the thin publish/provision layer that ties the five microservices (gateway / inference / agents / consolidation / web) together over Redis Streams. Gateways publish inbound message envelopes onto a single shared worker stream, inference/agents workers publish responses onto per-platform outbound streams, and each side reads with its own consumer group; this module owns the stream and group names and the XGROUP / XADD calls that create and feed them.

Stream and group names are resolved once at import time from environment variables (SG_INBOUND_STREAM, SG_OUTBOUND_STREAM_PREFIX, SG_WORKER_GROUP, SG_GATEWAY_GROUP_PREFIX) so every service agrees on the topology without hard-coding it. Envelopes are msgpack-framed by core.serialization before they hit the wire. The bus is constructed once per service by the gateway_main / inference_main / agents_main entrypoints.

core.event_bus.tools_reply_receipt_key(reply_stream, correlation_id)

Validate a tools reply route and return its colocated receipt key.

The tools service calls this before claiming or executing a source request, while RedisEventBus.publish_tools_reply() calls it again immediately before publication. Keeping one validator on both sides prevents malformed reply_to or correlation_id values from producing an external effect whose terminal reply can never be published.

Return type:

str

Parameters:

  • reply_stream (Any)

  • correlation_id (Any)

class core.event_bus.RedisEventBus(redis, node_role, node_id, *, stream_maxlen=100000, stream_retention_seconds=172800, ingress_ledger_ttl_seconds=691200, turn_checkpoint_proof_ttl_seconds=604800, gateway_rpc_proof_ttl_seconds=691200, durable_tools_proof_ttl_seconds=691200)

Bases: object

Manages Redis Streams for inbound/outbound message routing.

Responsibilities: - Publishing inbound message envelopes (gateway → worker) - Publishing outbound response envelopes (worker → gateway) - Creating and managing consumer groups - Providing stream metadata (lag, pending counts)

Parameters:

  • redis (redis.asyncio.Redis)

  • node_role (str)

  • node_id (str)

  • stream_maxlen (int)

  • stream_retention_seconds (float | None)

  • ingress_ledger_ttl_seconds (int)

  • turn_checkpoint_proof_ttl_seconds (int)

  • gateway_rpc_proof_ttl_seconds (int)

  • durable_tools_proof_ttl_seconds (int)

__init__(redis, node_role, node_id, *, stream_maxlen=100000, stream_retention_seconds=172800, ingress_ledger_ttl_seconds=691200, turn_checkpoint_proof_ttl_seconds=604800, gateway_rpc_proof_ttl_seconds=691200, durable_tools_proof_ttl_seconds=691200)

Initialize the event bus with its Redis client and node identity.

Stores the shared redis.asyncio.Redis client used for every subsequent XGROUP/XADD/XINFO/XPENDING/XLEN call, plus the node’s role and id, which decide (in ensure_streams()) which inbound/outbound streams and consumer groups this node provisions. The legacy stream length cap is floored at 1000. Inbound and durable effect lanes deliberately do not use MAXLEN because it can evict pending or never-delivered entries without consulting a consumer group.

This is a pure in-memory initializer: it opens no connections and issues no Redis commands. It is constructed once per service by the gateway, inference, and agents entrypoints (gateway_main.py, inference_main.py, agents_main.py) and by tests/core/test_event_bus.py.

Parameters:

  • redis (Redis) – Shared async Redis client backing every stream operation performed by this bus.

  • node_role (str) – One of "gateway", "worker", or "standalone"; determines which consumer groups ensure_streams() creates.

  • node_id (str) – Stable identifier for this process, used only in diagnostic log extra fields.

  • stream_maxlen (int) – Approximate maximum number of entries to retain per stream. Values below 1000 are clamped up to 1000. Defaults to 100_000.

  • stream_retention_seconds (float | None) – Age horizon for the time-based stream reaper. Entries older than this are trimmed via consumer-group-safe XTRIM MINID on the throttled backlog-check cadence. The entry-count cap never fired in practice (entries average 78-256KB, so streams ballooned to gigabytes at a fraction of the count cap); the age horizon bounds BYTES via time instead. Consumed/ACKed entries are transport history only — every payload is durably stored elsewhere (message cache, conversation lists, toolcall records). None disables reaping. Defaults to 48 hours.

  • ingress_ledger_ttl_seconds (int)

  • turn_checkpoint_proof_ttl_seconds (int)

  • gateway_rpc_proof_ttl_seconds (int)

  • durable_tools_proof_ttl_seconds (int)

Return type:

None

async ensure_streams(platforms=None)

Create consumer groups for all known streams.

Called once at startup. Uses XGROUP CREATE with mkstream=True so the stream is auto-created if it doesn’t exist yet.

Return type:

None

Parameters:

platforms (list[str] | None)

async publish_inbound(envelope)

Publish an inbound message envelope to the worker stream.

Called by gateway nodes when they receive a platform message. Returns the Redis Stream message ID.

Return type:

str

Parameters:

envelope (dict[str, Any])

async publish_outbound(platform, envelope)

Publish an outbound response envelope to the platform-specific stream.

Called by worker nodes after LLM inference completes. Returns the Redis Stream message ID.

Return type:

str

Parameters:
async publish_tools_request(envelope)

Publish a tool-execution request to the shared tools stream.

Called by the inference tier’s RemoteToolRegistry to delegate a non-pinned tool call. The tools service instances load-balance the stream via the sg:tools consumer group. The envelope must carry the reply_to stream name + correlation_id so the result can be routed back. Returns the Redis Stream message ID.

Return type:

str

Parameters:

envelope (dict[str, Any])

async publish_tools_reply(reply_stream, envelope)

Publish a tool-execution result onto a worker’s reply stream.

Called by the tools service after running a delegated tool. The reply_stream is the reply_to value from the originating request (sg:tools:reply:{worker_id}); a demux reader on that worker matches correlation_id and resolves the waiting call. The reply may contain raw media bytes (writeback.sent_files) — msgpack framing preserves them. Reply rows are never MAXLEN-trimmed because that can evict unread or pending durable evidence. Publication and its colocated per-correlation receipt are one Lua operation, so source redelivery after a lost reply returns the original Stream ID instead of appending a duplicate. The reader explicitly ACKs and deletes terminal rows; a refreshed proof-horizon TTL bounds abandoned per-process lanes and their independently expiring receipts. Returns the canonical Redis Stream message ID.

Return type:

str

Parameters:
async get_stream_info(stream)

Return XINFO STREAM metadata for a stream, or empty on error.

Surfaces the raw Redis XINFO STREAM mapping (length, first/last entry ids, group count, and so on) so monitoring code can report on a stream’s depth and shape. Any error – most commonly the stream not existing yet – is swallowed and reported as an empty dict so a probe never crashes the caller.

Issues a single read-only XINFO STREAM against Redis. No internal callers were found by grep; it is consumed by monitoring/diagnostic code and the bus tests in tests/core/test_event_bus.py.

Parameters:

stream (str) – The Redis Stream name to introspect.

Return type:

dict[str, Any]

Returns:

The XINFO STREAM field mapping, or an empty dict if the stream is missing or the call fails.

async get_pending_count(stream, group)

Return the consumer-group backlog (unacknowledged message count).

Reports how many messages a consumer group has read but not yet XACK-ed – the pending entries list (PEL) size – which is the primary signal that workers are falling behind. The result feeds the eviction-risk check in _check_stream_backlog() and any external monitoring. Errors (including a missing stream or group) are swallowed and reported as 0 so a diagnostic call never disrupts publishing.

Issues a single read-only XPENDING summary against Redis. Called by _check_stream_backlog() in this class and by the bus tests in tests/core/test_event_bus.py.

Parameters:

  • stream (str) – The Redis Stream to inspect.

  • group (str) – The consumer group whose pending count is requested.

Return type:

int

Returns:

The number of pending (read but unacknowledged) messages, or 0 when the information is unavailable.