proffesor-for-testing__agentic-qe/docs/guides/quic-coordination.md

QUIC Coordination for Distributed Agents

This guide explains how to use QUIC-based coordination for distributed agent communication in the AQE Fleet system.

Overview

QUIC (Quick UDP Internet Connections) provides low-latency, multiplexed, connection-oriented communication between agents. It's particularly useful for:

• Direct agent-to-agent communication without centralized coordination
• Real-time streaming updates for long-running tasks
• Request-response patterns with automatic retry and timeout
• Broadcast to multiple agents with efficient delivery
• Connection migration for resilient mobile deployments

Features

• ✅ Opt-in design: QUIC is disabled by default, zero impact when not used
• ✅ EventBus fallback: Automatically falls back to EventBus if QUIC unavailable
• ✅ Type-safe API: Full TypeScript support with comprehensive interfaces
• ✅ Peer discovery: Automatic discovery of connected agents
• ✅ Health monitoring: Built-in health checks and connection stats
• ✅ Channel management: Multiple communication channels with priorities
• ✅ Stream support: Long-lived streams for continuous data transfer

Configuration

Basic Configuration

import { BaseAgentConfig, QUICConfig } from '../types';

const quicConfig: QUICConfig = {
  enabled: true,
  host: 'localhost',
  port: 9000,
  channels: [
    {
      name: 'coordination',
      id: 'coord-1',
      type: 'broadcast',
      priority: 5,
      ordered: true,
      reliable: true
    },
    {
      name: 'results',
      id: 'results-1',
      type: 'unicast',
      priority: 7,
      ordered: false,
      reliable: true
    },
    {
      name: 'metrics',
      id: 'metrics-1',
      type: 'broadcast',
      priority: 3,
      ordered: false,
      reliable: false
    }
  ],
  connectionTimeout: 5000,
  enable0RTT: true,
  enableMigration: true,
  maxConcurrentStreams: 100
};

const agentConfig: BaseAgentConfig = {
  type: 'test-executor',
  capabilities: [...],
  context: {...},
  memoryStore: memoryStore,
  eventBus: eventBus,
  quicConfig // Add QUIC configuration
};

Agent Definition Configuration

Add QUIC configuration to agent definition files in .claude/agents/:

{
  "agentType": "qe-test-executor",
  "enableQUIC": true,  // Opt-in
  "quicConfig": {
    "host": "localhost",
    "port": 9000,
    "channels": ["coordination", "results", "metrics"]
  }
}

Usage

Enabling QUIC at Runtime

import { BaseAgent } from '../agents/BaseAgent';
import { QUICConfig } from '../types/quic';

class MyAgent extends BaseAgent {
  async initialize(): Promise<void> {
    await super.initialize();

    // Enable QUIC after initialization
    if (this.quicConfig?.enabled) {
      await this.enableQUIC(this.quicConfig);
    }
  }
}

Direct Agent-to-Agent Communication

class CoordinatorAgent extends BaseAgent {
  async assignTaskToAgent(targetAgentId: string, task: any): Promise<void> {
    // Send task assignment via QUIC
    await this.sendToAgent(targetAgentId, {
      action: 'execute-task',
      task: task
    }, 'coordination');
  }
}

class WorkerAgent extends BaseAgent {
  protected onQUICMessage(message: QUICMessage): void {
    if (message.payload.action === 'execute-task') {
      const task = message.payload.task;
      this.executeTask(task);
    }
  }
}

Request-Response Pattern

class TestExecutorAgent extends BaseAgent {
  async getCoverageFromAnalyzer(analyzerId: string): Promise<CoverageReport> {
    // Send request and wait for response
    const response = await this.requestFromAgent(analyzerId, {
      action: 'get-coverage',
      filters: { threshold: 80 }
    }, 10000); // 10 second timeout

    return response.coverageReport;
  }
}

class CoverageAnalyzerAgent extends BaseAgent {
  protected async onQUICMessage(message: QUICMessage): Promise<void> {
    if (message.type === 'REQUEST' && message.payload.action === 'get-coverage') {
      // Process request and send response
      const report = await this.generateCoverageReport(message.payload.filters);

      // Response is automatically sent by QUIC transport
      // Just return the payload
      return report;
    }
  }
}

Broadcasting to Fleet

class FleetCommanderAgent extends BaseAgent {
  async updateConfiguration(config: FleetConfig): Promise<void> {
    // Broadcast configuration update to all agents
    await this.broadcastToFleet({
      action: 'update-config',
      config: config,
      timestamp: Date.now()
    }, 'coordination');
  }
}

class WorkerAgent extends BaseAgent {
  protected onQUICMessage(message: QUICMessage): void {
    if (message.payload.action === 'update-config') {
      this.updateLocalConfig(message.payload.config);
    }
  }
}

Streaming Updates

class PerformanceMonitorAgent extends BaseAgent {
  private metricsStreamId?: string;

  async startMetricsStream(): Promise<void> {
    this.metricsStreamId = `metrics-${this.agentId.id}`;

    // Open QUIC stream
    await this.quicTransport?.openStream(this.metricsStreamId, {
      priority: 5,
      ordered: false,
      reliable: false // UDP-like for real-time metrics
    });

    // Send periodic updates
    setInterval(async () => {
      const metrics = await this.collectMetrics();

      await this.quicTransport?.writeStream(this.metricsStreamId!, {
        streamId: this.metricsStreamId!,
        data: JSON.stringify(metrics),
        final: false
      });
    }, 1000); // Every second
  }

  async stopMetricsStream(): Promise<void> {
    if (this.metricsStreamId) {
      await this.quicTransport?.closeStream(this.metricsStreamId);
    }
  }
}

Coordination Patterns

1. Hierarchical Coordination

// Commander agent coordinates multiple workers
class CommanderAgent extends BaseAgent {
  private workers: string[] = [];

  async distributeWork(tasks: Task[]): Promise<void> {
    // Discover available workers
    const peers = await this.quicTransport?.discoverPeers({
      filter: (peer) => peer.agentType === 'worker',
      maxPeers: 10
    });

    // Distribute tasks via QUIC
    for (let i = 0; i < tasks.length; i++) {
      const workerId = peers![i % peers!.length].agentId;
      await this.sendToAgent(workerId, {
        action: 'execute',
        task: tasks[i]
      });
    }
  }
}

2. Peer-to-Peer Collaboration

// Agents collaborate directly without central coordinator
class CollaborativeAgent extends BaseAgent {
  async shareResults(results: TestResults): Promise<void> {
    // Broadcast results to all peers
    await this.broadcastToFleet({
      action: 'share-results',
      results: results,
      from: this.agentId.id
    }, 'results');
  }

  protected onQUICMessage(message: QUICMessage): void {
    if (message.payload.action === 'share-results') {
      // Merge results from peer
      this.mergeResults(message.payload.results);
    }
  }
}

3. Request Aggregation

// Collect data from multiple agents
class AggregatorAgent extends BaseAgent {
  async collectMetricsFromFleet(): Promise<FleetMetrics> {
    const peers = await this.quicTransport?.discoverPeers();

    // Request metrics from all peers in parallel
    const metricsPromises = peers!.map(peer =>
      this.requestFromAgent(peer.agentId, {
        action: 'get-metrics'
      })
    );

    const allMetrics = await Promise.all(metricsPromises);

    return this.aggregateMetrics(allMetrics);
  }
}

Health Monitoring

Check QUIC Health

const health = agent.getQUICHealth();

if (health) {
  console.log(`QUIC Status: ${health.status}`);
  console.log(`Connected Peers: ${health.connectedPeers}`);
  console.log(`Avg RTT: ${health.avgRTT}ms`);
  console.log(`Packet Loss: ${(health.packetLoss * 100).toFixed(2)}%`);
}

Connection Statistics

const stats = agent.getQUICStats();

if (stats) {
  console.log(`Bytes Sent: ${stats.bytesSent}`);
  console.log(`Bytes Received: ${stats.bytesReceived}`);
  console.log(`Messages Sent: ${stats.messagesSent}`);
  console.log(`Current RTT: ${stats.currentRTT}ms`);
  console.log(`Active Streams: ${stats.activeStreams}`);
}

Error Handling

QUIC operations automatically fall back to EventBus when:

• QUIC is not enabled
• Connection fails
• Peer is not reachable
• Timeout occurs

// Automatic fallback - no special handling needed
await agent.broadcastMessage('update', data); // Uses QUIC or EventBus

Manual Error Handling

try {
  await agent.sendToAgent(targetId, payload);
} catch (error) {
  console.error('QUIC send failed:', error);
  // Fallback to EventBus
  await agent.broadcastMessage('agent-message', {
    to: targetId,
    payload: payload
  });
}

Performance

Latency Comparison

Operation	EventBus	QUIC (0-RTT)	Improvement
Direct Message	5-10ms	1-2ms	5x faster
Broadcast (10 agents)	50-100ms	5-10ms	10x faster
Request-Response	15-30ms	3-5ms	6x faster
Stream (1000 msgs)	500-1000ms	100-200ms	5x faster

Bandwidth Efficiency

• Multiplexing: Multiple streams over single connection
• Header Compression: QPACK reduces overhead by 50-70%
• 0-RTT: Resume connections without handshake
• Connection Migration: Seamless network changes

Best Practices

1. Use Channels Appropriately

// Coordination: ordered, reliable
await agent.sendToAgent(id, data, 'coordination');

// Results: unordered, reliable (higher priority)
await agent.sendToAgent(id, results, 'results');

// Metrics: unordered, unreliable (lower priority)
await agent.sendToAgent(id, metrics, 'metrics');

2. Set Appropriate Timeouts

// Quick operations
await agent.requestFromAgent(id, data, 1000); // 1 second

// Long-running tasks
await agent.requestFromAgent(id, data, 30000); // 30 seconds

3. Monitor Connection Health

setInterval(() => {
  const health = agent.getQUICHealth();

  if (health?.status === 'unhealthy') {
    console.warn('QUIC connection degraded');
    // Take corrective action
  }
}, 5000);

4. Clean Up Resources

class MyAgent extends BaseAgent {
  async cleanup(): Promise<void> {
    // Close QUIC transport
    if (this.quicTransport) {
      await this.disableQUIC();
    }

    await super.cleanup();
  }
}

Security

QUIC supports TLS 1.3 encryption by default:

const secureConfig: QUICConfig = {
  enabled: true,
  host: 'localhost',
  port: 9000,
  channels: [...],
  security: {
    enableTLS: true,
    certPath: '/path/to/cert.pem',
    keyPath: '/path/to/key.pem',
    verifyPeer: true,
    enableTokenAuth: true,
    token: 'secure-token-here'
  }
};

Migration from EventBus

Existing code using EventBus will continue to work. To migrate:

1. Add quicConfig to agent configuration
2. Replace broadcastMessage() with broadcastToFleet() for fleet-wide messages
3. Use sendToAgent() for direct agent communication
4. Use requestFromAgent() for request-response patterns

No changes required to existing event handlers.

Troubleshooting

QUIC Not Connecting

# Check if QUIC port is open
netstat -an | grep 9000

# Check agent logs
tail -f .agentic-qe/logs/fleet.log | grep QUIC

High Latency

// Check RTT and packet loss
const stats = agent.getQUICStats();
console.log(`RTT: ${stats?.currentRTT}ms`);
console.log(`Loss: ${(stats?.packetLoss || 0) * 100}%`);

Connection Drops

// Enable connection migration
const config: QUICConfig = {
  ...baseConfig,
  enableMigration: true,
  connectionTimeout: 10000 // Increase timeout
};

Examples

See /workspaces/agentic-qe-cf/examples/quic-coordination-demo.ts for complete working examples.

References

• QUIC Protocol (RFC 9000)
• QPACK (RFC 9204)
• BaseAgent API Documentation
• Agent Coordination Patterns