The Sixth Sense: Design & Implementation

1 Motivation

Before the Sixth Sense, Bedsheet agents were limited to single-process execution. A Supervisor could coordinate multiple agents, but they all lived in the same Python process. This created three problems:

1. Scalability: A single process can only run so many agents before resource contention kills performance. CPU-intensive agents block the event loop for others.
2. Isolation: One crashing agent can take down the whole system. No way to restart a single agent without restarting everything.
3. Distribution: No way to run agents across machines or cloud providers. A monitoring agent on AWS can't talk to an analyzer on GCP.

The Sixth Sense solves this by giving agents distributed communication over PubNub. Each agent runs in its own process (or container, or cloud function) and communicates through publish/subscribe messaging. They find each other by subscribing to shared channels.

2 Architecture Overview

The Sixth Sense is organized as a layered module inside bedsheet/sense/:

Each layer has a clear responsibility:

3 Signal Design

The Signal Dataclass

@dataclass
class Signal:
    kind: SignalKind          # What type of signal
    sender: str               # Who sent it (agent name)
    payload: dict[str, Any]   # Arbitrary data
    correlation_id: str       # Links requests to responses
    target: str | None        # Intended recipient (None = broadcast)
    timestamp: float          # Unix timestamp (auto-set)
    source_channel: str | None  # Which channel it arrived on

The Seven Signal Kinds

We use a Literal type (not an Enum) for signal kinds. This gives us type-checking without the overhead of enum instances:

SignalKind = Literal[
    "request",    # Ask another agent to do something
    "response",   # Reply to a request
    "alert",      # Broadcast an alert to all listeners
    "heartbeat",  # Periodic "I'm alive" signal
    "claim",      # Attempt to claim ownership of an incident
    "release",    # Release a claimed incident
    "event",      # General-purpose notification
]

Correlation IDs

Every signal gets a correlation_id — a 12-character hex string from uuid4(). This is how request/response pairs are matched:

# Commander sends a request with correlation_id="a1b2c3d4e5f6"
signal = Signal(kind="request", sender="commander",
                payload={"task": "check CPU"}, target="cpu-watcher")

# cpu-watcher responds with the SAME correlation_id
response = Signal(kind="response", sender="cpu-watcher",
                  payload={"result": "CPU at 45%"},
                  correlation_id="a1b2c3d4e5f6",  # same ID
                  target="commander")

The mixin's signal loop checks self._pending_requests[signal.correlation_id] to resolve the correct asyncio.Future. This means an agent can have multiple concurrent requests in flight without confusion.

4 Transport Protocol

@runtime_checkable
class SenseTransport(Protocol):
    async def connect(self, agent_id: str, namespace: str) -> None: ...
    async def disconnect(self) -> None: ...
    async def broadcast(self, channel: str, signal: Signal) -> None: ...
    async def subscribe(self, channel: str) -> None: ...
    async def unsubscribe(self, channel: str) -> None: ...
    def signals(self) -> AsyncIterator[Signal]: ...
    async def get_online_agents(self, channel: str) -> list[AgentPresence]: ...

The Seven Methods

The protocol is intentionally minimal. Every method maps to a fundamental pub/sub operation:

AgentPresence

AgentPresence is a simple dataclass representing a remote agent's identity on the network:

@dataclass
class AgentPresence:
    agent_id: str
    agent_name: str
    namespace: str
    capabilities: list[str] = field(default_factory=list)
    status: str = "online"
    metadata: dict[str, Any] = field(default_factory=dict)

This is returned by get_online_agents() and populated from PubNub's here_now() API (or from heartbeat signals in the mock transport).

5 Why PubNub

We evaluated several messaging backends before choosing PubNub:

PubNub's Key Features We Use

• Publish/Subscribe: Core messaging. Each channel is a topic agents subscribe to.
• Presence: PubNub tracks who's on each channel via here_now(). We get agent discovery for free.
• UUID identity: Each PubNub connection has a UUID. We set this to the agent's name for presence identification.
• Auto-reconnection: PNReconnectionPolicy.EXPONENTIAL handles network flaps automatically.
• Global CDN: Messages route through PubNub's edge network, so agents in different regions get low latency.

6 The Thread-to-Asyncio Bridge

This is the trickiest piece of the PubNub integration. PubNub's Python SDK uses threaded callbacks for incoming messages, but Bedsheet is fully async. We need to safely cross the thread boundary.

The Problem

# PubNub calls this from a BACKGROUND THREAD:
class _SignalListener(SubscribeCallback):
    def message(self, pubnub, message):
        # We're on PubNub's thread, NOT the asyncio event loop!
        # Cannot await anything, cannot use asyncio directly
        signal = deserialize(message.message)
        # How do we get this signal to the async signal_loop?

The Solution: call_soon_threadsafe

class _SignalListener(SubscribeCallback):
    def __init__(self, queue: asyncio.Queue[Signal], loop: asyncio.AbstractEventLoop):
        self._queue = queue
        self._loop = loop

    def message(self, pubnub, message):
        signal = deserialize(message.message, source_channel=message.channel)
        # Thread-safe way to put into the asyncio queue:
        self._loop.call_soon_threadsafe(self._queue.put_nowait, signal)

call_soon_threadsafe is the standard Python mechanism for thread-to-asyncio communication. It's safe to call from any thread and guarantees the callback runs on the event loop's thread during the next iteration.

7 Channel Naming Convention

PubNub channels are just strings. We use a namespaced convention to prevent collisions:

def _full_channel(self, channel: str) -> str:
    """Expand short channel name to full namespaced channel."""
    if channel.startswith("bedsheet."):
        return channel
    return f"bedsheet.{self._namespace}.{channel}"

# Examples:
# "alerts"  → "bedsheet.cloud-ops.alerts"
# "tasks"   → "bedsheet.cloud-ops.tasks"
# "agent-1" → "bedsheet.cloud-ops.agent-1"  (direct channel)

Every agent also subscribes to its own direct channel (bedsheet.{namespace}.{agent_name}). This enables targeted messaging — when you call send_to("cpu-watcher", signal), it publishes to bedsheet.cloud-ops.cpu-watcher.

8 The Mixin Pattern

SenseMixin is the heart of the Sixth Sense. It adds distributed capabilities to any Agent without modifying the Agent class itself:

class SenseMixin:
    """Mixin that adds distributed sensing to any Agent."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)  # Calls Agent.__init__()
        self._transport: SenseTransport | None = None
        self._namespace: str = ""
        self._signal_handlers: dict[SignalKind, list[SignalHandler]] = {}
        self._signal_task: asyncio.Task | None = None
        self._pending_requests: dict[str, asyncio.Future[Signal]] = {}
        self._claimed_incidents: set[str] = set()
        self._heartbeat_task: asyncio.Task | None = None

How Python MRO Makes This Work

class MyAgent(SenseMixin, Agent):
    pass

# Method Resolution Order (MRO):
# MyAgent → SenseMixin → Agent → object

# When you call MyAgent(name="x", instruction="y", model_client=client):
# 1. MyAgent.__init__ → not defined, goes to next
# 2. SenseMixin.__init__(self, name="x", ...)
#    → super().__init__(name="x", ...) → Agent.__init__
# 3. Agent.__init__ sets up name, instruction, model_client
# 4. Back in SenseMixin.__init__, adds _transport, _signal_handlers, etc.

The type: ignore[attr-defined] Pattern

You'll notice # type: ignore[attr-defined] comments throughout the mixin. This is because SenseMixin accesses self.name and self.invoke(), which are defined on Agent, not on the mixin itself:

await transport.connect(self.name, namespace)  # type: ignore[attr-defined]

Mypy can't know that SenseMixin will always be combined with Agent. The type ignores acknowledge this limitation. An alternative would be a generic Protocol for the host class, but that adds complexity for no runtime benefit.

9 Request/Response Over PubNub

The most complex pattern is request/response — asking a remote agent to do something and waiting for the answer. This implements an RPC-like pattern over pub/sub messaging.

The Flow

Key Implementation Detail: _handle_request

async def _handle_request(self, signal: Signal) -> None:
    task = signal.payload.get("task", "")
    session_id = f"sense-{signal.correlation_id}"

    # Run the full agent loop (LLM + tools)
    result = ""
    async for event in self.invoke(session_id, task):
        if isinstance(event, CompletionEvent):
            result = event.response

    # Send response back to requester
    response_signal = Signal(
        kind="response",
        sender=self.name,
        payload={"result": result},
        correlation_id=signal.correlation_id,
        target=signal.sender,
    )
    await self.send_to(signal.sender, response_signal)

This is where the Sixth Sense connects to Bedsheet's core: the request handler calls self.invoke(), which runs the agent's full ReAct loop (LLM reasoning → tool calls → final response). The remote agent isn't just a message router — it's a full AI agent that thinks about the request and uses its tools to answer it.

10 Claim Protocol

When multiple agents see the same alert, we need exactly one to handle it. The claim protocol provides leaderless conflict resolution — no central coordinator required.

How It Works

async def claim_incident(self, incident_id: str, channel: str) -> bool:
    # 1. Broadcast our claim
    signal = Signal(kind="claim", sender=self.name,
                    payload={"incident_id": incident_id})
    await self.broadcast(channel, signal)

    # 2. Wait 500ms for competing claims
    await asyncio.sleep(0.5)

    # 3. If still in our claimed set, we won
    return incident_id in self._claimed_incidents

The conflict resolution happens in _handle_claim():

def _handle_claim(self, signal: Signal) -> None:
    incident_id = signal.payload.get("incident_id")
    if incident_id in self._claimed_incidents:
        # Tie-breaking: lower sender name wins
        if signal.sender < self.name:
            self._claimed_incidents.discard(incident_id)

11 Testing Strategy

The MockSenseTransport

The key testing challenge: how to test distributed agent communication without PubNub? The answer is MockSenseTransport, which follows the same pattern as MockLLMClient.

The Hub Pattern

class _MockSenseHub:
    """Shared state for mock transports. Routes signals between agents."""

    def __init__(self):
        self.queues: dict[str, asyncio.Queue[Signal]] = {}
        self.subscriptions: dict[str, set[str]] = {}
        self.presences: dict[str, AgentPresence] = {}

class MockSenseTransport:
    def __init__(self, hub: _MockSenseHub | None = None):
        self.hub = hub or _MockSenseHub()

    def create_peer(self) -> "MockSenseTransport":
        """Create a sibling transport sharing the same hub."""
        return MockSenseTransport(self.hub)

Test Design Lessons

Test coverage (31 tests in tests/test_sense.py):

12 Trade-offs & Future Work

What We Chose

Known Limitations

• No message persistence: PubNub messages are ephemeral (unless you enable PubNub Storage, a paid feature). If an agent is offline when a signal is sent, it misses it.
• No message ordering guarantee: PubNub guarantees per-channel ordering, but cross-channel ordering is not guaranteed.
• Claim protocol is probabilistic: Under very high load or network partitions, two agents might both win a claim. Acceptable for monitoring; not suitable for financial transactions.
• 32KB message limit: The auto-truncation handles this, but very large payloads lose data. If you need to send large data, send a reference (URL, S3 key) instead.

Future Directions

• Redis transport: For teams with existing Redis infrastructure
• NATS transport: For high-throughput scenarios
• Message persistence: Store signals for replay and audit trails
• Encryption: End-to-end encryption for signal payloads
• Agent discovery: Automatic capability-based routing (find the agent that can handle this request)