How OpenMAIC Handles Multi-Agent Discussions and SSE Streaming: A Deep Dive

OpenMAIC orchestrates multi-agent discussions through a turn-based runtime that emits agent_start and agent_end events, while streaming partial outputs via Server-Sent Events (SSE) using the PBLSSEEvent contract defined in lib/pbl/v2/api/sse.ts.

OpenMAIC treats every participant—users, AI assistants, and tools—as agents operating within a unified runtime environment. When multiple AI agents engage in collaborative reasoning, the system leverages Server-Sent Events (SSE) to deliver real-time updates while maintaining strict turn-based ordering through session state management. This architecture ensures deterministic multi-agent discussions without fragmented UI experiences.

Multi-Agent Architecture and Turn-Based Coordination

The Agent Lifecycle

In lib/agent/runtime/lifecycle.ts, OpenMAIC implements a strict turn-based protocol that governs when agents may generate output. The runtime treats every interaction as a logical conversation where each agent receives dedicated window boundaries marked by agent_start and agent_end events.

When an agent begins processing, the runtime emits an agent_start event to the session state queue. This signals the frontend to display a "thinking" indicator for that specific agent. The agent's turn continues until it completes reasoning, at which point agent_end is emitted and the next agent in the queue may begin. This sequential execution model prevents race conditions and guarantees deterministic ordering of outputs.

Session State and Buffering

While an agent is "thinking," its partial results are buffered in memory rather than immediately committed to the conversation history. The runtime aggregates these chunks throughout the agent's turn, ensuring that users see a continuous stream of text within a single cohesive UI card. Only when the agent_end event fires does the system finalize the aggregated content and render the complete response, preventing interleaved or fragmented output from multiple simultaneous agents.

SSE Streaming Infrastructure

The PBLSSEEvent Contract

The messaging protocol between the runtime and frontend is defined in lib/pbl/v2/api/sse.ts. This module exports the PBLSSEEvent interface, which standardizes all SSE communications:

// lib/pbl/v2/api/sse.ts
export interface PBLSSEEvent {
  type: 'partial' | 'error' | 'done';
  payload?: string;
  code?: string;   // error code, if any
}

/** Helper that builds a JSON‑encoded SSE line */
export function makeSSEMessage(event: PBLSSEEvent): string {
  return `data: ${JSON.stringify(event)}\n\n`;
}

The makeSSEMessage function encodes event objects into proper SSE format, ensuring compliance with the text/event-stream content type requirements.

Converting Generators to Event Streams

The transformation from async generators to SSE streams occurs in lib/agent/runtime/stream-fn.ts. The createPartMapper function accepts an AsyncGenerator<string> and returns a ReadableStream that yields properly formatted SSE chunks:

// lib/agent/runtime/stream-fn.ts
import { makeSSEMessage } from '@/lib/pbl/v2/api/sse';

export function createPartMapper(gen: AsyncGenerator<string>) {
  return new ReadableStream({
    async start(controller) {
      for await (const chunk of gen) {
        const ev = { type: 'partial', payload: chunk };
        controller.enqueue(new TextEncoder().encode(makeSSEMessage(ev)));
      }
      // Signal completion
      controller.enqueue(new TextEncoder().encode(makeSSEMessage({ type: 'done' })));
      controller.close();
    },
  });
}

This utility enables language models to stream tokens as they are generated, with each chunk wrapped in a PBLSSEEvent envelope and encoded for transmission over HTTP.

Implementation Examples

Backend API Route

To expose a multi-agent discussion endpoint, you implement a route that instantiates the generator and applies the part mapper:

// server/api/agent/generate.ts
import { createPartMapper } from '@/lib/agent/runtime/stream-fn';

export async function GET(req: Request) {
  const prompt = new URL(req.url).searchParams.get('prompt') ?? '';
  const gen = someLLMGenerator(prompt);           // AsyncGenerator<string>
  const stream = createPartMapper(gen);
  return new Response(stream, {
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

The createPartMapper handles the backpressure and encoding, allowing the API route to remain agnostic to the streaming implementation details.

Frontend Consumption

The frontend consumes these streams using the standard EventSource API, parsing the structured events to update React state incrementally:

useEffect(() => {
  const es = new EventSource('/api/agent/generate?prompt=hello');
  es.onmessage = e => {
    const ev = JSON.parse(e.data) as PBLSSEEvent;
    if (ev.type === 'partial') setAnswer(a => a + ev.payload);
    if (ev.type === 'done')   es.close();
    if (ev.type === 'error')  setError(ev.message);
  };
  return () => es.close();
}, []);

Summary

• Turn-based coordination in lib/agent/runtime/lifecycle.ts ensures agents execute sequentially via agent_start and agent_end events, preventing output interleaving.
• SSE streaming utilizes the PBLSSEEvent interface from lib/pbl/v2/api/sse.ts to standardize partial outputs, errors, and completion signals.
• Stream transformation is handled by createPartMapper in lib/agent/runtime/stream-fn.ts, which converts async generators into RFC-compliant text/event-stream responses.
• Validation is enforced through test suites including tests/workbench/thinking-stream-seam.test.ts and tests/workbench/session-events-subscription.test.ts.

Frequently Asked Questions

How does OpenMAIC prevent race conditions between multiple agents?

OpenMAIC implements a session state queue in lib/agent/runtime/lifecycle.ts that processes agents sequentially. Each agent must wait for the previous agent's agent_end event before its agent_start event fires, ensuring deterministic ordering and preventing concurrent modification of shared conversation state.

What is the structure of a PBLSSEEvent message?

According to lib/pbl/v2/api/sse.ts, a PBLSSEEvent is a TypeScript interface with a type field that can be 'partial', 'error', or 'done', plus optional payload (string content) and code (error identifier) properties. The makeSSEMessage utility serializes this to data: ${JSON.stringify(event)}\n\n format.

How does the frontend distinguish between different agents in the same stream?

The frontend uses the agent_start and agent_end events emitted by the lifecycle manager to delimit agent boundaries. When agent_start arrives, the UI initializes a new card for that agent's output; subsequent partial events append to that card until agent_end signals completion, at which point the UI finalizes the card and awaits the next agent's turn.

Where is the agent turn order managed in the codebase?

Agent turn ordering is managed in lib/agent/runtime/lifecycle.ts, which maintains the session state and orchestrates the transition between agents. This module ensures that the runtime respects the logical conversation flow and that SSE streams are properly flushed before the next agent begins processing.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}