Event Bus

WebSocket pub/sub server. 6 channels, ring buffers, thread-safe publishing.

The event bus is a WebSocket server that connects every component of the swarm. The 13 Hive subsystems publish state changes, agents publish lifecycle events, the circuit breaker publishes alerts, the kill chain publishes incident phases. Any client can subscribe to any channel and receive events in real time.

Channels

Channel	What Gets Published
agents	Phase changes, spawned, stopped, errors, heartbeats, directives
hive	Proposal verdicts, capital allocated, agents spawned/killed
risk	Circuit breaker triggered, drawdown alerts, risk limit hits
knowledge	Insights shared between agents, discoveries
system	Status updates, metrics, startup/shutdown
ui	User commands forwarded from external clients

Usage

from horizon.swarm import EventBus, SwarmEvent, SwarmEventType
import time

bus = EventBus(host="127.0.0.1", port=8765)
await bus.start()

# Publish from async context
await bus.publish("agents", SwarmEvent(
    type=SwarmEventType.AGENT_PHASE_CHANGED,
    source="agent_001",
    data={"phase": "analyzing", "reason": "found 3 candidate markets"},
    timestamp=time.time(),
))

# Publish from threads (circuit breaker, engine callbacks)
bus.publish_sync("risk", SwarmEvent(
    type=SwarmEventType.CIRCUIT_BREAKER_TRIGGERED,
    source="circuit_breaker",
    data={"reason": "portfolio drawdown 12% > 10% limit"},
    timestamp=time.time(),
))

Ring Buffers

Each channel keeps the last 1,000 events in a ring buffer. Late-connecting clients can request history:

{"get_history": "agents", "limit": 50}

The server responds with the most recent events from that channel’s buffer.

Client Protocol

Clients connect via WebSocket to ws://host:port. On connect, automatically subscribed to all channels. They can manage subscriptions and send commands:

// Subscribe to specific channels only
{"subscribe": ["agents", "hive"]}

// Unsubscribe
{"unsubscribe": ["system"]}

// Send a command to the Hive
{"command": "spawn", "args": "Research weather markets on Kalshi"}

// Request history
{"get_history": "risk", "limit": 100}

Commands sent via command are forwarded to the hive channel as HIVE_DIRECTIVE events. The Hive picks them up on its next oversight cycle.

Event Format

Every event follows the SwarmEvent structure:

@dataclass
class SwarmEvent:
    type: SwarmEventType     # enum: AGENT_SPAWNED, FILL_RECEIVED, etc.
    source: str              # "agent_001", "hive", "circuit_breaker"
    data: dict[str, Any]     # event-specific payload
    timestamp: float         # time.time()

Over the wire, events are JSON:

{
  "channel": "agents",
  "event": {
    "type": "agent_phase_changed",
    "source": "agent_003",
    "data": {"phase": "backtesting", "reason": "edge estimate 3.2% on KSH-FED-RATE"},
    "timestamp": 1710768000.0
  }
}

Event Types

Type	Channel	When
agent_spawned	agents	Agent created by AgentFactory
agent_promoted	agents	Agent moved paper to shadow to live
agent_culled	agents	Agent removed by SwarmCoordinator
agent_phase_changed	agents	Agent transitions to new phase
agent_stopped	agents	Agent finished or errored
order_submitted	agents	Agent places an order
fill_received	agents	Order matched
position_opened	agents	New position created
position_closed	agents	Position fully closed
pheromone_deposit	hive	Agent deposits pheromone signal
consensus_update	hive	ConsensusEngine view updated
capital_rebalanced	hive	SwarmCoordinator reallocated capital
evolution_complete	hive	EvolutionEngine finished a generation
autonomy_graduated	hive	AutonomyController promoted to new mode
circuit_breaker_triggered	risk	Hard limit breached, emergency stop
criticality_alert	risk	CriticalityMonitor detected systemic risk
kill_chain_triggered	risk	KillChain incident response started
kill_chain_phase	risk	KillChain advanced to next phase
drawdown_alert	risk	Drawdown approaching limit
knowledge_shared	knowledge	Agent shared insight to knowledge graph
causal_edge_detected	knowledge	CausalEngine found new relationship
heartbeat	agents	Periodic agent state snapshot
metrics_update	system	Hive oversight cycle metrics
system_status	system	Startup, shutdown, status change

Thread Safety

The event bus runs as an asyncio WebSocket server. Agents and the Hive publish from the async event loop using await bus.publish(). The circuit breaker and engine callbacks run in threads, so they use bus.publish_sync() which calls asyncio.run_coroutine_threadsafe() to cross the thread/async boundary safely.

Building a Custom Client

Any WebSocket client can connect:

import asyncio, json, websockets

async def monitor():
    async with websockets.connect("ws://127.0.0.1:8765") as ws:
        # Subscribe to risk events only
        await ws.send(json.dumps({"subscribe": ["risk"]}))

        async for msg in ws:
            event = json.loads(msg)
            if event.get("event", {}).get("type") == "circuit_breaker_triggered":
                print("CIRCUIT BREAKER:", event["event"]["data"])

asyncio.run(monitor())

JavaScript/TypeScript, Go, Rust, or any language with WebSocket support works the same way. Plain JSON over WebSocket.