12.4Communication Patterns

Pattern 1: Pipeline (Sequential)

Analyzer -> Generator -> Runner -> Fixer -> Runner

Each agent passes output to the next. Simple and predictable.

Pattern 2: Publish-Subscribe

class MessageBus:
    def __init__(self):
        self.subscribers = defaultdict(list)

    def subscribe(self, message_type: str, handler: Callable):
        self.subscribers[message_type].append(handler)

    def publish(self, message: AgentMessage):
        for handler in self.subscribers[message.message_type]:
            handler(message)

# Usage
bus = MessageBus()
bus.subscribe("analysis_complete", test_generator.on_analysis_complete)
bus.subscribe("analysis_complete", coverage_tracker.on_analysis_complete)
bus.subscribe("tests_ready", test_runner.on_tests_ready)
bus.subscribe("error", alerting_service.on_error)

Multiple agents can react to the same event. The Coverage Tracker and Test Generator both receive the analysis result.

Pattern 3: Request-Response

class AgentProxy:
    """Synchronous request-response to another agent."""

    async def request(self, target_agent: str, action: str, data: dict) -> dict:
        request_id = str(uuid4())
        self.send(AgentMessage(
            sender=self.agent_id,
            receiver=target_agent,
            message_type=f"request:{action}",
            payload={"request_id": request_id, **data},
            priority=2,
        ))

        # Wait for response with timeout
        response = await self.wait_for_response(request_id, timeout=30)
        return response.payload