Show & Tell: Designing the AgentMesh Coordinator — Service Registry, Task Dispatch, and Role-Based Routing

[web3guru888]

TL;DR

Phase 4.2 is underway. Issue #147 introduced the AgentMesh coordinator for multi-agent task distribution. This post walks through the full architecture: how agents register themselves, how the coordinator dispatches tasks, and how the Cognitive Blackboard ties everything together.

The Problem: Scaling Beyond a Single Agent

The CognitiveCycle (Phase 4.1) gave us a real-time perception-action loop over 29 modules — but it runs single-agent. Phase 4.2 asks: what happens when you need specialized agents running in parallel?

Examples:

• A planner agent decomposes goals into sub-tasks
• Perception agents process sensor streams independently
• A memory consolidation agent runs nightly knowledge graph maintenance
• A safety monitor agent watches all Blackboard writes for policy violations

Each of these is a separate process with a dedicated AgentRole. They share state through the Cognitive Blackboard — the same shared memory that the CognitiveCycle uses for module communication.

The Architecture

┌─────────────────────────────────────────────────────────┐
│                     AgentMesh                           │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │            AgentDiscovery (registry)            │   │
│  │  PLANNER:001 ✓ | PERCEPTION:002 ✓ | MEMORY:003  │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  dispatch(task) → route → agent_id → Blackboard write   │
│                                                         │
│  Routing strategies:                                    │
│    ROUND_ROBIN    — balanced across healthy agents      │
│    LEAST_LOADED   — fewest active tasks wins            │
│    CAPABILITY     — "who can handle perception?"        │
│    ROLE_AFFINITY  — task type → preferred role          │
└─────────────────────────────────────────────────────────┘
          ↓                        ↑
    task assignment          result pickup
          ↓                        ↑
┌─────────────────┐    ┌─────────────────────────────────┐
│  Cognitive      │    │         Cognitive Blackboard     │
│  Blackboard     │←──→│  mesh.task.assigned             │
│                 │    │  mesh.task.result               │
│                 │    │  mesh.agent.registered          │
└─────────────────┘    │  mesh.agent.heartbeat           │
                       └─────────────────────────────────┘

Data Model

from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Any, Optional
import uuid, time

class AgentRole(Enum):
    COORDINATOR   = auto()   # orchestrates other agents
    PLANNER       = auto()   # goal decomposition and task planning
    PERCEPTION    = auto()   # sensory processing and feature extraction
    MEMORY        = auto()   # knowledge graph reads/writes
    REASONING     = auto()   # hybrid reasoning, PLN, logic
    SAFETY_MONITOR = auto()  # ethical verification gate
    EXECUTOR      = auto()   # action execution

class TaskStatus(Enum):
    PENDING     = auto()
    ASSIGNED    = auto()
    IN_PROGRESS = auto()
    COMPLETED   = auto()
    FAILED      = auto()
    RETRYING    = auto()

@dataclass
class AgentTask:
    task_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    task_type: str = ""            # e.g. "perception.process_frame"
    required_role: Optional[AgentRole] = None
    required_capability: Optional[str] = None
    payload: dict = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    deadline_ms: Optional[float] = None
    priority: int = 5              # 1=critical, 10=background
    status: TaskStatus = TaskStatus.PENDING
    assigned_to: Optional[str] = None  # agent_id
    result: Optional[Any] = None
    error: Optional[str] = None

Task Dispatch Flow

class AgentMesh:
    async def dispatch(self, task: AgentTask) -> str:
        """Assign task to best available agent; return task_id."""
        # 1. Find candidates
        if task.required_capability:
            candidates = self._discovery.by_capability(task.required_capability)
        elif task.required_role:
            candidates = self._discovery.healthy_agents(role=task.required_role)
        else:
            candidates = self._discovery.healthy_agents()

        if not candidates:
            raise NoAgentAvailableError(
                f"No healthy agent for task={task.task_type}"
            )

        # 2. Pick agent based on routing strategy
        agent = self._route(candidates)
        task.assigned_to = agent.agent_id
        task.status = TaskStatus.ASSIGNED

        # 3. Publish to Blackboard — agent picks it up via subscription
        await self._bb.write(
            entry_type="mesh.task.assigned",
            namespace=agent.blackboard_namespace,
            data={"task": asdict(task)},
            ttl_s=task.deadline_ms / 1000 if task.deadline_ms else 60.0,
        )

        self._active_tasks[task.task_id] = task
        return task.task_id

Result Collection

Agents write results back to the Blackboard using a well-known key:

# Inside each agent after completing a task:
await blackboard.write(
    entry_type="mesh.task.result",
    namespace="mesh.coordinator",
    data={
        "task_id": task.task_id,
        "agent_id": self.agent_id,
        "status": "COMPLETED",
        "result": my_result,
        "duration_ms": elapsed,
    },
    ttl_s=30.0,
)

The coordinator subscribes to mesh.task.result.* and reconciles completed tasks.

Service Registry (Issue #150)

AgentDiscovery handles registration, heartbeats, and health transitions:

STARTING → HEALTHY         (on first heartbeat)
HEALTHY  → DEGRADED        (heartbeat age > 15s)
DEGRADED → UNREACHABLE     (heartbeat age > 30s)
any      → DRAINING        (graceful shutdown signal)
any      → removed         (deregister() call)

The coordinator never dispatches to DEGRADED or UNREACHABLE agents. If all agents for a required role are unreachable, dispatch() raises NoAgentAvailableError.

Blackboard Topic Map

Topic	Producer	Consumer
mesh.agent.registered	AgentDiscovery	Dashboard, safety monitor
mesh.agent.heartbeat	Each agent	AgentDiscovery health sweep
mesh.agent.status_changed	AgentDiscovery sweep	Dashboard, coordinator
mesh.task.assigned.<agent_ns>	AgentMesh.dispatch()	Target agent
mesh.task.result.<task_id>	Each agent	AgentMesh result collector
mesh.task.failed.<task_id>	Each agent	AgentMesh retry logic

Open Questions

1. Persistent task queue or pure Blackboard TTL? If the coordinator crashes mid-dispatch, tasks in the Blackboard will expire. Should we add a SQLite-backed task log for durability?

2. Priority lanes? Should priority-1 tasks bypass the round-robin queue and preempt lower-priority assigned tasks?

3. Agent affinity for stateful tasks? A memory-consolidation task that operates on specific KG namespaces probably needs to go to the same agent every time. How do we express that?

Next Issues in Phase 4.2

• Add AgentDiscovery: service registry and health tracking for AgentMesh #150 — AgentDiscovery service registry (this sprint)
• Show & Tell: Designing the AgentMesh Coordinator — Service Registry, Task Dispatch, and Role-Based Routing #151 — AgentTask queue with priority lanes (planned)
• Ideas: Should AgentMesh roles be static enums or dynamic capability sets? #152 — Coordinator-to-agent protocol: request-reply over Blackboard (planned)
• Q&A: How does AgentMesh handle agent failures and crashed tasks? #153 — AgentMesh integration tests (planned)

Follow the Phase 4 roadmap in #109 for the full picture.