Does Shadowbroker Use a Message Queue or Event Bus in Its Backend Architecture?

Shadowbroker does not employ an external message broker such as RabbitMQ or Kafka; instead, it implements an internal event bus using lightweight, in-process queues including asyncio.Queue, collections.deque, and thread-protected Python lists to manage real-time agent communication and mesh networking.

Shadowbroker is an open-source operational intelligence platform designed for high-agency agent coordination and decentralized mesh operations. According to the source code in the BigBodyCobain/Shadowbroker repository, the backend architecture deliberately avoids the operational overhead of standalone message queues by utilizing native Python concurrency primitives. These in-process queues function as an event bus within the application, handling Server-Sent Events (SSE), WebSocket notifications, and mesh packet batching without requiring separate infrastructure.

SSE Queue Implementation in ai_intel.py

The primary mechanism for pushing real-time updates to HTTP clients relies on per-client asyncio.Queue instances. In backend/routers/ai_intel.py at line 2863, the system maintains a global list of active queues:

# backend/routers/ai_intel.py – line 2863

_sse_queues: list[asyncio.Queue] = []

When an agent connects to the /api/ai/channel/sse endpoint, the system instantiates a new asyncio.Queue(maxsize=512) and registers it in _sse_queues (line 2904). The broadcast_to_sse_clients function (lines 2868–2875) iterates over these queues to distribute events:

async def broadcast_to_sse_clients(event_type: str, data: dict[str, Any]):
    async with _sse_queues_lock:
        queues = list(_sse_queues)
    for q in queues:
        try:
            q.put_nowait({"event": event_type, "data": data})
        except asyncio.QueueFull:
            pass   # slow client – drop the message

If a client’s queue reaches capacity, the system applies backpressure handling by dropping the message rather than blocking the broadcaster, ensuring that slow consumers do not degrade system performance.

WebSocket Channel Management

For bidirectional real-time communication, Shadowbroker uses a similar pattern with WebSocket connections. Located in the same file at lines 3031–3033, the implementation stores active connections in a protected list:

# backend/routers/ai_intel.py – lines 3031‑3033

_ws_clients: list[WebSocket] = []
_ws_clients_lock = asyncio.Lock()

When the server generates a new task or alert, it iterates over _ws_clients and transmits the payload directly via await ws.send_json(payload). This design eliminates the need for an external pub/sub broker while maintaining low-latency push delivery to connected agents.

Mesh RNS Batch Queues

The mesh networking layer handles high-latency, covert communication channels using thread-protected lists rather than async queues. In backend/services/mesh/mesh_rns.py at lines 148–149, the MeshRNS class maintains two distinct batch buffers:

# backend/services/mesh/mesh_rns.py – line 148‑149

self._batch_queue: list[dict] = []
self._gate_batch_queue: list[tuple[str, dict]] = []

These lists aggregate outbound packets for cover-traffic timing and bandwidth optimization. A background timer (_batch_timer) periodically flushes accumulated items (line 1415–1417), ensuring that transmissions adhere to predefined timing intervals to obfuscate traffic patterns:

queued = list(self._batch_queue)
self._batch_queue.clear()

# … send `queued` to the network …

This batching strategy is thread-safe through the use of _batch_lock, allowing concurrent access from multiple mesh threads without requiring an external message queue.

Agent Actions Queue

For frontend interaction, the system uses a bounded collections.deque to buffer UI actions generated by AI agents. Defined at line 38 in backend/routers/ai_intel.py, this queue stores actions with a maximum length of 20:

from collections import deque

# backend/routers/ai_intel.py – line 38

_agent_actions: deque[dict] = deque(maxlen=20)

The frontend polls the /api/ai/agent_actions endpoint to consume these actions, creating a simple request-response event mechanism that decouples AI-generated commands from the UI rendering cycle without persistent websocket overhead for short-lived interactions.

Practical Implementation Examples

Broadcasting an SSE Event

To push a task update to all connected agents from anywhere in the backend:

from backend.routers.ai_intel import broadcast_to_sse_clients

await broadcast_to_sse_clients(
    event_type="task",
    data={
        "task_type": "fetch_satellite",
        "payload": {"lat": 40.7, "lon": -74.0}
    }
)

Queueing a Mesh Packet

To add data to the mesh batch queue for covert transmission:

from backend.services.mesh.mesh_rns import MeshRNS

mesh = MeshRNS()
packet = {"type": "dm", "dest": "gate123", "payload": {"msg": "Hello"}}
mesh._batch_queue.append(packet)  # thread-safe via _batch_lock

Pushing a UI Action

To record an action for frontend consumption:

from backend.routers.ai_intel import push_agent_action

push_agent_action({
    "type": "display_image",
    "url": "https://example.com/sat.png",
    "caption": "Recent satellite view"
})

Summary

• No external brokers: Shadowbroker avoids RabbitMQ, Kafka, or Redis by using in-process Python queues.
• Asyncio primitives: Real-time channels rely on asyncio.Queue with explicit size limits (max 512) to handle backpressure.
• Thread-safe batching: The mesh layer uses locked Python lists (_batch_queue) to aggregate packets for timing-based transmission.
• Bounded buffering: The agent actions system uses collections.deque with maxlen=20 to prevent memory growth from untended UI events.
• Single-process architecture: All queue operations occur within the application process, simplifying deployment and reducing infrastructure dependencies.

Frequently Asked Questions

Does Shadowbroker use RabbitMQ or Kafka for its message queue?

No, Shadowbroker does not integrate external message brokers. The codebase implements an internal event bus using asyncio.Queue for asynchronous streams and thread-protected lists for mesh batching, keeping all message handling within the Python application process.

How does Shadowbroker handle slow SSE clients?

The system implements drop-on-overflow backpressure. When broadcast_to_sse_clients encounters a full queue (asyncio.QueueFull), it silently drops the message for that specific client (line 2874). This prevents slow HTTP consumers from blocking the broadcaster and affecting other connected agents.

What is the maximum capacity of the agent actions queue?

The agent actions queue is initialized with maxlen=20 using collections.deque at line 38 in backend/routers/ai_intel.py. When this limit is reached, new actions automatically evict the oldest entries, ensuring the memory footprint remains bounded regardless of AI generation frequency.

Are the mesh batch queues thread-safe?

Yes, the _batch_queue and _gate_batch_queue in backend/services/mesh/mesh_rns.py are protected by _batch_lock. While they are standard Python lists rather than formal queue structures, all append and flush operations acquire this lock, making them safe for concurrent access from multiple mesh networking threads.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}