CNS Bridge

A Python library that lets any agent plug into the Hermes Central Nervous System (CNS) bus. Agents communicate through a pair of filesystem inboxes and outboxes using the Universal Sensory/Command Packet (USCP) protocol.

        ┌─────────────┐          outbox          ┌─────────────┐
        │   Agent A   │ ───────────────────────► │   Hermes    │
        │  (any code) │                            │    CNS      │
        └─────────────┘          inbox           └─────────────┘
                ▲ ◄────────────────────────
                │
                └ Heartbeat poller watches for responses

Quick start

from cns_bridge import Agent, FileSystemTransport, Intent, Priority

transport = FileSystemTransport(
    inbox_path="/tmp/hermes/inbox",
    outbox_path="/tmp/hermes/outbox",
)

agent = Agent(agent_id="my_agent", transport=transport, secret="shared-secret")

agent.send(
    intent=Intent.QUERY,
    message="Hello Hermes, what is the fleet status?",
    priority=Priority.NORMAL,
)

Install

pip install /home/eileen/projects/cns-bridge

For development:

cd /home/eileen/projects/cns-bridge
pip install -e ".[dev]"
pytest

Default paths

The default inbox and outbox point to the Windows-side Hermes directories used by the SuperInstance stack:

• Inbox: /mnt/c/Users/casey/.hermes/cns_inbox/
• Outbox: /mnt/c/Users/casey/.hermes/cns_outbox/

These can be overridden per instance, through environment variables, or both:

FileSystemTransport(
    inbox_path="/custom/inbox",
    outbox_path="/custom/outbox",
)

export CNS_INBOX=/custom/inbox
export CNS_OUTBOX=/custom/outbox

The USCP protocol

Every message on the CNS bus is a Universal Sensory/Command Packet. It is a single JSON object with three top-level keys:

{
  "header": { ... },
  "body": { ... },
  "signature": { ... }
}

Header

The header introduces the message.

Field	Type	Description
origin_id	string	Identity of the sending agent.
packet_id	UUID	Unique identifier for this packet.
intent	string	Kind of message (see Intents below).
priority	string	Urgency level: low, normal, high, critical.
destination_id	string	Target agent or "hermes" for the CNS.
timestamp	ISO-8601	UTC timestamp of creation.
version	string	USCP protocol version, currently "1.0".
correlation_id	string?	Optional ID linking to a prior packet.

Body

The body carries the need: commands, queries, sensory data, or responses.

Field	Type	Description
data	object	Arbitrary structured payload.
message	string	Human-readable summary.
mime_type	string	Content type, defaults to application/json.
encoding	string	Character encoding, defaults to utf-8.
schema	string?	Optional schema identifier for the payload.

Signature

The signature records integrity metadata. The default algorithm is HMAC-SHA256 over a canonical JSON serialization of the header and body.

Field	Type	Description
value	string	Base64-encoded HMAC digest.
algorithm	string	Algorithm name, e.g. HMAC-SHA256.
key_id	string	Identifier for the signing key.
verified	bool?	Optional verification state.

Intents

• sense — Report sensory data or state.
• command — Request an action.
• query — Ask a question.
• response — Reply to a query or command.
• alert — Raise an issue.
• heartbeat — Periodic keep-alive.
• register — Announce presence on the bus.
• escalation — Priority escalation notice.

Priorities

Priorities are ordered:

low < normal < high < critical

Agents and the CNS may use priority to decide routing order, retry behavior, and alerting.

Escalation rules

An EscalationRule describes when a packet should be bumped to a higher priority because it has not received a response. For example, a high packet that is unanswered for 30 seconds may be escalated to critical.

from cns_bridge import EscalationRule, Priority, ProtocolContext

context = ProtocolContext(
    escalation_rules=[
        EscalationRule(
            min_priority=Priority.HIGH,
            no_response_seconds=30.0,
            bump_to=Priority.CRITICAL,
        )
    ]
)

API overview

PacketBuilder

Fluent construction of USCP packets:

from cns_bridge import PacketBuilder, Intent, Priority

packet = (
    PacketBuilder(origin_id="lucineer")
    .to("hermes")
    .with_intent(Intent.QUERY)
    .with_priority(Priority.HIGH)
    .with_data(question="status")
    .with_message("Fleet status request")
    .signed_with("shared-secret", key_id="lucineer")
    .build()
)

FileSystemTransport

Read/write packets to inbox/outbox directories:

transport.send(packet)
packet = transport.receive(origin_id="hermes")
packet = transport.peek()
packets = list(transport.poll(origin_id="hermes"))

Agent

Base class with send, receive, and handle methods, plus optional background heartbeat polling:

from cns_bridge import Agent

class MyAgent(Agent):
    def handle(self, packet):
        print(f"Got packet from {packet.header.origin_id}")

agent = MyAgent(agent_id="my_agent", transport=transport)
agent.start_heartbeat(interval=1.0)

HeartbeatPoller

Background thread that watches the inbox and invokes a callback for each new packet addressed to the agent:

from cns_bridge import HeartbeatPoller

poller = HeartbeatPoller(
    transport=transport,
    agent_id="my_agent",
    callback=on_packet,
    interval=1.0,
)
poller.start()

CompactionGuardian

Tracks context-window pressure and decides when to capture important state before a context compaction event:

from cns_bridge import CompactionGuardian

guardian = CompactionGuardian()
guardian.capture("decision", "Chose Redis over SQLite for speed")
state = guardian.snapshot()

LedgerGraph

A decision-consequence graph for recording how earlier choices lead to later outcomes:

from cns_bridge import LedgerGraph, DecisionNode, ConsequenceEdge

graph = LedgerGraph()
graph.add_node(DecisionNode(id="d1", summary="Shipped feature X"))

TokenEstimator

Estimate token counts for messages, assess context-window health, and decide when to trigger a creative break:

from cns_bridge import estimate_tokens, context_health, should_trigger_creative_break

tokens = estimate_tokens("Hello, world!")
health = context_health(used=4000, limit=8000)
should_break = should_trigger_creative_break(used=7500, limit=8000)

PersonalLog

File-based personal log for an agent to record thoughts, observations, and reflections:

from cns_bridge import PersonalLog

log = PersonalLog(path="/tmp/my_agent/log.jsonl")
log.append("Learned something new today")

Examples

• examples/lucineer_agent.py — Lucineer queries Hermes and receives a response.
• examples/wesley_agent.py — Wesley sends night-school training results.

Run an example:

python examples/lucineer_agent.py

Running tests

pytest

License

MIT — see LICENSE.