proffesor-for-testing__agentic-qe/docs/streaming-mcp-tools.md

Streaming MCP Tools - v1.0.5

Overview

Agentic QE Fleet v1.0.5 introduces streaming support for long-running MCP operations, providing real-time progress updates for improved user experience and better visibility into test execution and coverage analysis.

Features

✅ Real-Time Progress Updates

• Progress events emitted at configurable intervals (default: every 2-5 seconds)
• Per-test completion tracking with status, duration, and metrics
• Incremental file-by-file coverage analysis reporting

✅ Memory Efficient

• AsyncGenerator-based streaming (no buffering of entire results)
• Progressive result disclosure
• Automatic resource cleanup

✅ Error Resilient

• Proper error handling maintains stream integrity
• Recoverable vs non-recoverable error classification
• Session state persistence for recovery

✅ Backward Compatible

• Non-streaming tools continue to work unchanged
• Streaming tools can be used interchangeably with non-streaming versions
• Same input parameters as non-streaming equivalents

Architecture

StreamingMCPTool Base Class

abstract class StreamingMCPTool {
  // Returns AsyncGenerator yielding StreamEvent objects
  async *execute(params: any): AsyncGenerator<StreamEvent, void, undefined>

  // Subclasses implement this with progress reporting
  protected abstract executeWithProgress(
    params: any,
    reporter: ProgressReporter
  ): Promise<any>
}

Event Types

Progress Event:

{
  type: 'progress',
  message: string,
  percent: number,
  currentItem?: string,
  itemsProcessed?: number,
  itemsTotal?: number,
  timestamp: string,
  metadata?: Record<string, any>
}

Result Event:

{
  type: 'result',
  data: any,
  timestamp: string,
  executionTime?: number,
  metadata?: Record<string, any>
}

Error Event:

{
  type: 'error',
  error: string,
  details?: any,
  timestamp: string,
  recoverable?: boolean
}

Available Streaming Tools

1. Test Execution Stream (test_execute_stream)

Use Case: Long-running test suites (>30 seconds)

Features:

• Real-time test completion updates
• Per-suite and per-test progress tracking
• Status updates (passed/failed/skipped)
• Duration metrics
• Parallel execution support with coordinated progress

Example:

{
  spec: {
    testSuites: ['tests/**/*.test.js'],
    parallelExecution: true,
    retryCount: 3,
    timeoutSeconds: 300,
    reportFormat: 'json'
  },
  enableRealtimeUpdates: true
}

Progress Events:

{
  "type": "progress",
  "message": "Completed suite: integration-tests (45 tests)",
  "percent": 33,
  "currentItem": "integration-tests",
  "itemsProcessed": 1,
  "itemsTotal": 3,
  "metadata": {
    "suite": "integration-tests",
    "environment": "test",
    "type": "suite-complete",
    "result": {
      "status": "passed",
      "testsCount": 45,
      "duration": 8234
    }
  }
}

2. Coverage Analysis Stream (coverage_analyze_stream)

Use Case: Large codebases with many files to analyze

Features:

• File-by-file analysis progress
• Incremental gap detection reporting
• O(log n) Johnson-Lindenstrauss optimization for faster analysis
• Real-time recommendations as gaps are discovered

Example:

{
  sourceFiles: ['src/**/*.ts'],
  coverageThreshold: 0.8,
  useJohnsonLindenstrauss: true,
  includeUncoveredLines: true,
  analysisDepth: 'detailed'
}

Progress Events:

{
  "type": "progress",
  "message": "Completed: src/agents/TestExecutor.ts (87.3% coverage)",
  "percent": 45,
  "currentItem": "src/agents/TestExecutor.ts",
  "itemsProcessed": 23,
  "itemsTotal": 51,
  "metadata": {
    "type": "file-analysis-complete",
    "file": "src/agents/TestExecutor.ts",
    "coverage": 87.3,
    "gaps": 2
  }
}

Usage Examples

Basic Streaming Test Execution

// Using MCP tool directly
const result = await mcpClient.callTool('mcp__agentic_qe__test_execute_stream', {
  spec: {
    testSuites: ['tests/integration/**/*.test.js'],
    parallelExecution: true,
    retryCount: 3
  }
});

// Result includes streaming metadata
console.log(result.streaming); // true
console.log(result.events); // Array of all progress events
console.log(result.summary); // { totalEvents: 45, progressUpdates: 42, errors: 0 }

Monitoring Progress with Event Bus

// Subscribe to streaming progress events
eventBus.on('streaming:progress', (event) => {
  console.log(`[${event.percent}%] ${event.message}`);

  if (event.metadata?.type === 'test-complete') {
    console.log(`  Test: ${event.currentItem} - ${event.metadata.status}`);
  }
});

// Execute streaming operation
await testExecuteStreamHandler.execute(params);

Session State Tracking

// Retrieve streaming session state from memory
const session = await memoryStore.get(`streaming/session-${sessionId}`);

console.log(session.status); // 'active' | 'completed' | 'failed' | 'cancelled'
console.log(session.progress.percent); // Current progress percentage
console.log(session.progress.message); // Latest progress message

// Cancel ongoing operation
await progressReporter.cancel();

Configuration

StreamingConfig Options

{
  progressInterval: 5000,      // Min time between updates (ms)
  bufferEvents: false,         // Buffer events for batching
  maxBufferSize: 10,           // Max buffer size before flush
  timeout: 600000,             // Operation timeout (10 min)
  persistSession: true         // Save session to memory
}

Adjusting Progress Frequency

// Faster updates for interactive scenarios
new TestExecuteStreamHandler(memoryStore, eventBus, {
  progressInterval: 2000  // Update every 2 seconds
});

// Slower updates to reduce overhead
new CoverageAnalyzeStreamHandler(memoryStore, eventBus, {
  progressInterval: 10000  // Update every 10 seconds
});

Performance Characteristics

Memory Usage

• Non-Streaming: O(n) - stores entire result before returning
• Streaming: O(1) - constant memory, progressive disclosure

Responsiveness

• Non-Streaming: No feedback until completion
• Streaming: Updates every 2-5 seconds

Network Overhead

• Minimal: Progress events are small (~200-500 bytes)
• Configurable interval prevents excessive updates
• Optional event buffering for batching

Best Practices

When to Use Streaming

✅ Use Streaming For:

• Test suites that run > 30 seconds
• Coverage analysis of > 50 files
• Parallel test execution with multiple suites
• Interactive CLI/UI applications
• Long-running integration tests

❌ Don't Use Streaming For:

• Quick unit tests (< 10 seconds)
• Single test execution
• Small codebases (< 10 files)
• Automated CI/CD pipelines (unless monitoring is needed)

Error Handling

try {
  for await (const event of handler.execute(params)) {
    if (event.type === 'error') {
      if (event.recoverable) {
        // Log and continue
        console.warn(`Recoverable error: ${event.error}`);
      } else {
        // Stop execution
        throw new Error(event.error);
      }
    }
  }
} catch (error) {
  // Handle fatal errors
  console.error('Streaming execution failed:', error);
}

Resource Cleanup

// Streaming handlers automatically cleanup on completion/error
// But you can also cancel explicitly:
const session = await handler.getSession();
if (session && session.status === 'active') {
  await reporter.cancel();
}

Integration with AQE Agents

Streaming tools integrate seamlessly with TestExecutorAgent via AQE hooks:

class TestExecutorAgent extends BaseAgent {
  protected async onPreTask(data: PreTaskData): Promise<void> {
    // Subscribe to streaming progress
    this.eventBus.on('streaming:progress', (event) => {
      this.logger.info(`Test progress: ${event.percent}%`);
    });
  }

  protected async onPostTask(data: PostTaskData): Promise<void> {
    // Store streaming session results
    await this.memoryStore.store('aqe/last-execution', {
      streaming: true,
      result: data.result
    });
  }
}

Troubleshooting

Progress Updates Not Appearing

Check:

1. enableRealtimeUpdates is set to true
2. Progress interval hasn't been set too high
3. Event bus listeners are properly registered
4. MCP server notifications are enabled

Memory Leaks with Long Sessions

Solution:

• Set persistSession: false for ephemeral operations
• Implement periodic cleanup of old sessions
• Use TTL on memory store keys

Slow Performance

Optimization:

• Increase progressInterval to reduce event overhead
• Enable bufferEvents for batching
• Use Johnson-Lindenstrauss optimization for coverage analysis

API Reference

StreamingMCPTool

class StreamingMCPTool {
  constructor(
    memoryStore: Map<string, any>,
    eventBus: EventEmitter,
    config?: Partial<StreamingConfig>
  )

  async *execute(params: any): AsyncGenerator<StreamEvent, void>
  getSession(): StreamingSession | null
  isCancelled(): boolean
}

ProgressReporter

interface ProgressReporter {
  report(progress: Omit<ToolProgress, 'type' | 'timestamp'>): Promise<void>
  complete(result: Omit<ToolResult, 'type' | 'timestamp'>): Promise<void>
  error(error: Omit<ToolError, 'type' | 'timestamp'>): Promise<void>
  getProgress(): ToolProgress | null
  cancel(): Promise<void>
}

Migration Guide

From Non-Streaming to Streaming

Before (non-streaming):

const result = await mcpClient.callTool('mcp__agentic_qe__test_execute', {
  spec: { testSuites: ['tests/**/*.test.js'] }
});
console.log(result); // All results at once

After (streaming):

const result = await mcpClient.callTool('mcp__agentic_qe__test_execute_stream', {
  spec: { testSuites: ['tests/**/*.test.js'] },
  enableRealtimeUpdates: true
});

// Access streaming events
result.events.forEach(event => {
  if (event.type === 'progress') {
    console.log(`Progress: ${event.percent}% - ${event.message}`);
  }
});

// Final result is the same structure
console.log(result.result);

Future Enhancements

• WebSocket support for push-based streaming
• Graphical progress visualization
• Advanced cancellation and pause/resume
• Distributed streaming across fleet agents
• Real-time metrics dashboard integration

---

Version: 1.0.5 Last Updated: October 2025 Author: Agentic QE Team