Hierarchical Tree-Like Messaging Contract

Philosophy of Ignorance

VOX implements a hierarchical communication model where each agent communicates exclusively with its immediate superior (master) and its direct subordinates (children). No peer-level discovery exists.

Example Hierarchy

root (id: <root_uuid>)
  master_id: ""
  └── coordinator (id: <coordinator_uuid>)
        master_id: <root_uuid>
        ├── specialist_a (id: <specialist_a_uuid>)
        │     master_id: <coordinator_uuid>
        └── specialist_b (id: <specialist_b_uuid>)
              master_id: <coordinator_uuid>

Design consequences:

• The root does not know the specialists exist. If the root needs information from a specialist domain, it sends a message to the coordinator.
• The specialists do not know the root exists. They only respond to the coordinator.
• The specialists do not know each other. No lateral communication is possible.
• This architecture acts as a conceptual and security firewall: a compromise in one specialist does not expose another specialist's attack surface, and vice versa.

Implementation in VOXOrchestrator

The orchestrator constructs the parent→children graph (get_hierarchy_snapshot) and exposes direct children via get_children(agent_id). Command delegation is tagged with the delegating parent, so a command from a specialist appears in the root as handled through the coordinator.

---

VOXMessage — The Universal Envelope

The messaging contract is defined in vox/messaging/models.py. VOXMessage is a frozen pydantic.BaseModel with UUID-typed identity fields:

class VOXMessage(BaseModel):
    message_id: UUID                 # globally unique UUID
    message_source: str              # origin system
    emitted_at: str                  # ISO-8601 with timezone
    source: UUID                     # sender agent UUID
    target: UUID                     # recipient agent UUID
    type: str                        # semantic message type
    details: dict[str, Any]          # event-specific payload
    reply_to: UUID | None = None     # correlation to previous message

Envelope Fields

Field	Type	Required	Description
message_id	UUID	Yes	Globally unique identifier
message_source	str	Yes	Origin system (e.g., interface name)
emitted_at	str (ISO-8601)	Yes	Emission timestamp with timezone
source	UUID	Yes	Sender agent UUID
target	UUID	Yes	Destination agent UUID
type	str	Yes	Semantic message intent
details	dict	Yes	Variable event payload
reply_to	UUID | None	No	message_id of the message being replied to

Defined Message Types

The framework defines the envelope-level constants in vox.messaging (MSG_COMMAND_REQUEST, MSG_ACKNOWLEDGED, MSG_COMMAND_RESULT, MSG_COMMAND_ERROR). Domain messages (job_detected, data_request, system_failure, …) are application-specific type strings carried in the details payload:

1. command_request — Parent → Child: request execution of a named command
2. message_acknowledged — Any → Any: receipt confirmation
3. command_result — Child → Parent: execution result
4. command_error — Any → Any: delivery or execution failure
5. system_failure / alerts — Any agent → War Room (the only message that crosses hierarchy levels)

---

Example Flow: Vertical Escalation and Asynchronous Delegation

Scenario: the root receives an operator command requiring specialist information

1. root receives a command from the operator that it does not implement
   └── root consults its command map
       └── command → handled by a subordinate (through the coordinator)

2. root constructs a VOXMessage:
   {
     "message_id": "a1b2c3d4-...",
     "source": "<root_uuid>",        # root
     "target": "<coordinator_uuid>", # coordinator
     "type": "command_request",
     "details": {"command": "analyze", "params": {"target": "..."}}
   }

3. coordinator receives the message, records it in its VOXAgentMemory
   └── coordinator classifies the intent
       └── coordinator delegates to the matching specialist

4. coordinator constructs a VOXMessage:
   {
     "message_id": "e5f6g7h8-...",
     "source": "<coordinator_uuid>",   # coordinator
     "target": "<specialist_a_uuid>",  # specialist
     "type": "command_request",
     "details": {"command": "analyze", "params": {"target": "..."}},
     "reply_to": "a1b2c3d4-..."
   }

5. specialist executes the task and responds to coordinator:
   {
     "message_id": "i9j0k1l2-...",
     "source": "<specialist_a_uuid>",  # specialist
     "target": "<coordinator_uuid>",   # coordinator
     "type": "command_result",
     "details": {"result": "..."},
     "reply_to": "e5f6g7h8-..."
   }

6. coordinator adds context, records in its ledger, responds to root:
   {
     "message_id": "m3n4o5p6-...",
     "source": "<coordinator_uuid>", # coordinator
     "target": "<root_uuid>",        # root
     "type": "command_result",
     "details": {"command": "analyze", "result": "...", "summary": "..."},
     "reply_to": "a1b2c3d4-..."
   }

7. root presents the result to the operator.

Key flow points:

• The root never interacts directly with the specialist
• The coordinator is the sole coupling point between levels
• Each message is recorded with its UUID in every involved agent's ledger
• reply_to enables reconstruction of the full causality chain
• The architecture supports asynchronous delegation: the specialist can process while the root and coordinator continue with other tasks

---

Internal Event Router

Each VOXAgent maintains an event_router mapping event names to role lists:

self.event_router: dict[str, list[Any]] = {}

_register_role_routes() in VOXAgent._bootstrap automatically registers handlers:

def _register_role_routes(self, role: Any) -> None:
    handlers = getattr(role, "_handlers", {})
    command_names = set(role.get_commands().keys())
    for route_name in handlers.keys():
        is_command = route_name in command_names
        if is_command:
            self.commands.add(route_name)
        else:
            self.events.add(route_name)
        if route_name not in self.event_router:
            self.event_router[route_name] = []
        if role not in self.event_router[route_name]:
            self.event_router[route_name].append(role)

Commands (decorated with @command) and events (registered via role.on("event_name")) share the same router but are distinguished by membership in the command_names set.