12.2The Message Protocol

from dataclasses import dataclass

@dataclass
class AgentMessage:
    sender: str           # "code_analyzer"
    receiver: str         # "test_generator" or "broadcast"
    message_type: str     # "analysis_complete" | "tests_ready" | "error"
    payload: dict         # Structured data
    priority: int         # 0=low, 1=normal, 2=high, 3=critical
    timestamp: float
    correlation_id: str   # Traces related messages across agents

Message Types

Type	Sender	Receiver	Purpose
analysis_complete	Code Analyzer	Test Generator	"I have analyzed the module, here are the testable functions"
tests_ready	Test Generator	Test Runner	"I have generated tests, they are ready to execute"
results_available	Test Runner	Test Fixer	"Tests ran, here are the failures"
fix_applied	Test Fixer	Test Runner	"I fixed the failing tests, please re-run"
error	Any agent	Orchestrator	"I failed, here is what happened"
budget_warning	Any agent	Orchestrator	"I am at 80% of my token budget"

Example Message Flow

messages = [
    AgentMessage(
        sender="code_analyzer",
        receiver="test_generator",
        message_type="analysis_complete",
        payload={
            "module": "auth/login.py",
            "functions_to_test": ["authenticate", "validate_token"],
            "complexity_hints": {"authenticate": "high", "validate_token": "medium"},
            "existing_test_count": 3,
            "suggested_test_count": 8
        },
        priority=1,
        timestamp=time.time(),
        correlation_id="sprint-42-auth-refactor"
    ),
    AgentMessage(
        sender="test_generator",
        receiver="test_runner",
        message_type="tests_ready",
        payload={
            "test_file": "tests/test_auth_login.py",
            "test_count": 8,
            "dependencies": ["pytest", "pytest-asyncio", "httpx"]
        },
        priority=1,
        timestamp=time.time(),
        correlation_id="sprint-42-auth-refactor"
    )
]

Correlation IDs for Tracing

The correlation_id is essential for debugging multi-agent systems. It ties together all messages related to a single task:

# All messages for the auth-refactor task share the same correlation ID
# This allows you to:
# 1. Filter logs by correlation_id to see the full pipeline
# 2. Measure total pipeline time (first message to last)
# 3. Identify where bottlenecks occur
# 4. Replay a specific task for debugging

def get_pipeline_trace(correlation_id: str) -> list[AgentMessage]:
    """Get all messages for a specific task, ordered by timestamp."""
    return sorted(
        [m for m in message_store if m.correlation_id == correlation_id],
        key=lambda m: m.timestamp
    )