proffesor-for-testing__agentic-qe/docs/guides/STREAMING-API.md

Streaming API Tutorial

Version: 1.0.5 Last Updated: October 16, 2025

---

Introduction

The Streaming API provides real-time progress updates for long-running QE operations. Instead of waiting minutes for test generation to complete without feedback, you get live updates every second showing exactly what's happening.

Key Benefits

• Real-Time Visibility: See progress as tests execute
• Better UX: No more black-box waiting
• Early Error Detection: Catch issues before completion
• Resource Monitoring: Track memory, CPU, test count in real-time
• Cancellation Support: Stop operations mid-stream

When to Use Streaming

✅ Use Streaming For:

• Test generation (>50 tests)
• Test execution (>100 tests)
• Coverage analysis (>10k LOC)
• Security scanning (>5 min)
• Performance testing (load tests)

❌ Don't Use Streaming For:

• Quick operations (<5 seconds)
• Single unit test generation
• Simple validations
• Configuration checks

---

Quick Start

1. Enable Streaming

# .agentic-qe/config.yaml
features:
  streamingTools: true

streaming:
  enabled: true
  updateInterval: 1000  # 1 second
  bufferSize: 100

2. Basic Streaming Example

import { FleetManager } from 'agentic-qe';

const fleet = new FleetManager();

// Subscribe to test generation stream
const stream = fleet.streamTestGeneration({
  sourceFile: 'src/services/user-service.ts',
  framework: 'jest',
  targetCoverage: 95
});

// Listen for progress updates
stream.on('progress', (update) => {
  console.log(`[${update.timestamp}] ${update.status}`);
  console.log(`Progress: ${update.progress}% (${update.completed}/${update.total})`);
  console.log(`Current: ${update.currentOperation}`);
});

// Listen for completion
stream.on('complete', (result) => {
  console.log('✓ Test generation complete!');
  console.log(`Generated ${result.testCount} tests`);
  console.log(`Coverage: ${result.coverage}%`);
});

// Handle errors
stream.on('error', (error) => {
  console.error('✗ Error:', error.message);
});

// Start the operation
await stream.start();

Output:

[2025-10-16T10:30:01] Analyzing source file...
Progress: 10% (1/10)
Current: Extracting functions and classes

[2025-10-16T10:30:02] Generating unit tests...
Progress: 30% (3/10)
Current: Creating tests for UserService.createUser

[2025-10-16T10:30:04] Generating integration tests...
Progress: 60% (6/10)
Current: Testing database operations

[2025-10-16T10:30:06] Running generated tests...
Progress: 90% (9/10)
Current: Calculating coverage

[2025-10-16T10:30:07] Complete!
✓ Test generation complete!
Generated 45 tests
Coverage: 96.2%

---

Using Streaming Tools

Test Generation with Streaming

import { TestGeneratorAgent } from 'agentic-qe';

const agent = new TestGeneratorAgent();

// Create streaming request
const stream = await agent.generateTestsStreaming({
  sourceFile: 'src/utils/validator.ts',
  framework: 'jest',
  testTypes: ['unit', 'integration'],
  targetCoverage: 90
});

// Progress bar integration
const progressBar = new ProgressBar('Generating [:bar] :percent :etas', {
  total: 100,
  width: 40
});

stream.on('progress', (update) => {
  progressBar.update(update.progress / 100);
});

stream.on('metrics', (metrics) => {
  console.log('Metrics:', metrics);
  // {
  //   testsGenerated: 23,
  //   coverage: 85.5,
  //   memoryUsage: '145 MB',
  //   duration: 12300
  // }
});

const result = await stream.complete();

Test Execution with Streaming

import { TestExecutorAgent } from 'agentic-qe';

const executor = new TestExecutorAgent();

const stream = await executor.executeTestsStreaming({
  testFiles: 'tests/**/*.test.ts',
  parallel: true,
  maxWorkers: 8
});

// Track individual test results
stream.on('test:started', (test) => {
  console.log(`Starting: ${test.name}`);
});

stream.on('test:passed', (test) => {
  console.log(`✓ ${test.name} (${test.duration}ms)`);
});

stream.on('test:failed', (test) => {
  console.log(`✗ ${test.name}`);
  console.log(`  Error: ${test.error.message}`);
});

// Summary statistics
stream.on('progress', (stats) => {
  console.log(`Tests: ${stats.passed}/${stats.total} passed (${stats.failed} failed)`);
  console.log(`Duration: ${stats.elapsed}ms`);
  console.log(`Workers: ${stats.activeWorkers}/${stats.maxWorkers}`);
});

const results = await stream.complete();

Coverage Analysis with Streaming

import { CoverageAnalyzerAgent } from 'agentic-qe';

const analyzer = new CoverageAnalyzerAgent();

const stream = await analyzer.analyzeStreaming({
  sourcePath: 'src/',
  testPath: 'tests/',
  framework: 'jest'
});

// Live coverage updates
stream.on('file:analyzed', (file) => {
  console.log(`${file.path}: ${file.coverage}%`);
});

// Gap detection
stream.on('gap:detected', (gap) => {
  console.log(`Coverage gap in ${gap.file}:`);
  console.log(`  Lines: ${gap.uncoveredLines.join(', ')}`);
  console.log(`  Functions: ${gap.uncoveredFunctions.join(', ')}`);
});

// Real-time metrics
stream.on('metrics', (metrics) => {
  console.log(`Overall coverage: ${metrics.overall}%`);
  console.log(`Files analyzed: ${metrics.filesAnalyzed}/${metrics.totalFiles}`);
});

const report = await stream.complete();

---

Integration

Integrating with Existing Code

Before (Non-Streaming)

// Old approach: No feedback until complete
async function generateTests(file: string) {
  console.log('Generating tests...');

  const result = await fleet.generateTests({ sourceFile: file });

  console.log('Done!');
  return result;
}

// User sees:
// "Generating tests..."
// [3 minutes of silence]
// "Done!"

After (Streaming)

// New approach: Real-time updates
async function generateTests(file: string) {
  const stream = await fleet.streamTestGeneration({ sourceFile: file });

  // Show live progress
  stream.on('progress', (update) => {
    console.log(`[${update.progress}%] ${update.currentOperation}`);
  });

  // Show intermediate results
  stream.on('test:generated', (test) => {
    console.log(`✓ Generated test: ${test.name}`);
  });

  const result = await stream.complete();
  return result;
}

// User sees:
// "[10%] Analyzing source file"
// "[25%] Generating unit tests"
// "✓ Generated test: should create user with valid data"
// "✓ Generated test: should reject invalid email"
// "[50%] Generating integration tests"
// ...

Migration from Non-Streaming

Step 1: Identify long-running operations

// Find operations that take >5 seconds
const operations = [
  'fleet.generateTests',
  'fleet.executeTests',
  'fleet.analyzeCoverage',
  'fleet.scanSecurity'
];

Step 2: Add streaming wrapper

// Create streaming wrapper
async function withStreaming<T>(
  operation: () => Promise<T>,
  streamFactory: () => OperationStream<T>
): Promise<T> {
  if (config.features.streamingTools) {
    const stream = streamFactory();
    attachProgressListeners(stream);
    return stream.complete();
  } else {
    // Fallback to non-streaming
    return operation();
  }
}

Step 3: Update function calls

// Before
const result = await fleet.generateTests(options);

// After
const result = await withStreaming(
  () => fleet.generateTests(options),
  () => fleet.streamTestGeneration(options)
);

CLI Integration

#!/bin/bash
# streaming-test-gen.sh

# Enable streaming in CLI
export AQE_STREAMING_ENABLED=true
export AQE_STREAMING_FORMAT=fancy  # Options: simple, fancy, json

# Run with progress bar
aqe test src/services/*.ts --stream

# Output:
# Generating tests for 5 files
# ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓░░░░ 75% | 3/5 files | user-service.ts
#
# Current: Generating integration tests
# Generated: 127 tests
# Coverage: 94.2%
# Time: 00:02:34

CI/CD Integration

# .github/workflows/test.yml
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Node.js
        uses: actions/setup-node@v3
        with:
          node-version: 18

      - name: Install dependencies
        run: npm ci

      - name: Generate tests (with streaming)
        run: |
          # Stream to GitHub Actions logs
          npx aqe test src/ --stream --format json | tee test-generation.log

      - name: Parse streaming output
        run: |
          # Extract final results from stream
          cat test-generation.log | jq -r 'select(.type == "complete") | .data'

      - name: Upload results
        uses: actions/upload-artifact@v3
        with:
          name: test-results
          path: test-generation.log

---

Best Practices

1. Buffer Management

// Configure buffer size based on operation
const config = {
  streaming: {
    bufferSize: {
      'test-generation': 100,     // Moderate buffering
      'test-execution': 1000,     // Large buffer (many events)
      'coverage-analysis': 50,    // Small buffer
      'security-scan': 200        // Moderate buffering
    }
  }
};

2. Error Recovery

const stream = await agent.generateTestsStreaming(options);

// Implement retry logic
stream.on('error', async (error) => {
  if (error.retryable) {
    console.log('Retrying...');
    await stream.retry();
  } else {
    console.error('Fatal error:', error.message);
    await stream.abort();
  }
});

// Automatic recovery
stream.on('connection:lost', async () => {
  console.log('Connection lost, reconnecting...');
  await stream.reconnect({ maxAttempts: 3, delay: 1000 });
});

3. Resource Cleanup

const stream = await agent.executeTestsStreaming(options);

// Always clean up
try {
  const result = await stream.complete();
  return result;
} catch (error) {
  await stream.abort();  // Clean up resources
  throw error;
} finally {
  stream.removeAllListeners();  // Prevent memory leaks
}

4. Progress Visualization

import chalk from 'chalk';
import ora from 'ora';

const stream = await agent.generateTestsStreaming(options);

// Spinner for operations
const spinner = ora('Analyzing source file...').start();

stream.on('progress', (update) => {
  spinner.text = `${update.currentOperation} (${update.progress}%)`;

  if (update.progress === 100) {
    spinner.succeed('Complete!');
  }
});

// Colored output for test results
stream.on('test:generated', (test) => {
  console.log(chalk.green('✓'), test.name);
});

stream.on('test:failed', (test) => {
  console.log(chalk.red('✗'), test.name);
  console.log(chalk.gray(`  ${test.error.message}`));
});

5. Performance Monitoring

const stream = await agent.executeTestsStreaming(options);

// Track performance metrics
const metrics = {
  startTime: Date.now(),
  eventsReceived: 0,
  bandwidth: 0
};

stream.on('data', (chunk) => {
  metrics.eventsReceived++;
  metrics.bandwidth += chunk.length;
});

stream.on('complete', () => {
  const duration = Date.now() - metrics.startTime;
  console.log(`Received ${metrics.eventsReceived} events in ${duration}ms`);
  console.log(`Bandwidth: ${(metrics.bandwidth / 1024).toFixed(2)} KB`);
});

---

Performance

Overhead Analysis

Operation	Non-Streaming	Streaming	Overhead	Worth It?
Test Gen (10 tests)	5s	5.1s	+2%	✗ No
Test Gen (100 tests)	45s	46s	+2%	✓ Yes
Test Exec (50 tests)	10s	10.2s	+2%	✓ Yes
Test Exec (1000 tests)	120s	122s	+1.6%	✓ Yes
Coverage (<1k LOC)	2s	2.1s	+5%	✗ No
Coverage (10k+ LOC)	30s	30.5s	+1.6%	✓ Yes

Recommendation: Use streaming for operations >5 seconds. Overhead is negligible for long operations.

Memory Considerations

// Configure memory limits
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    maxMemory: 512 * 1024 * 1024,  // 512 MB
    enableBackpressure: true,
    highWaterMark: 16384           // 16 KB
  }
});

// Monitor memory usage
stream.on('memory:warning', (usage) => {
  console.warn(`High memory usage: ${usage.current}/${usage.limit}`);
});

stream.on('memory:limit', async () => {
  console.error('Memory limit reached, pausing...');
  await stream.pause();

  // Wait for GC
  await new Promise(resolve => setTimeout(resolve, 5000));

  await stream.resume();
});

Scalability Tips

1. Connection Pooling

// Reuse connections for multiple streams
const pool = new StreamConnectionPool({
  maxConnections: 10,
  idleTimeout: 60000,
  reuseConnections: true
});

// All streams share the pool
const stream1 = await agent1.generateTestsStreaming({ pool });
const stream2 = await agent2.executeTestsStreaming({ pool });

2. Batching Events

// Batch small events to reduce overhead
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    batchEvents: true,
    batchSize: 10,
    batchInterval: 500  // 500ms
  }
});

// Receive batched events
stream.on('batch', (events) => {
  // Process 10 events at once
  events.forEach(event => processEvent(event));
});

3. Compression

// Enable compression for large payloads
const stream = await agent.analyzeCoverageStreaming({
  ...options,
  streaming: {
    compression: 'gzip',
    compressionLevel: 6
  }
});

// Reduces bandwidth by 70-80% for coverage data

---

Advanced Features

Custom Event Handlers

class CustomStreamHandler {
  onProgress(update: ProgressUpdate) {
    // Send to monitoring system
    monitoring.track('test_generation_progress', {
      progress: update.progress,
      operation: update.currentOperation
    });
  }

  onMetrics(metrics: Metrics) {
    // Send to metrics dashboard
    dashboard.update('test_metrics', metrics);
  }

  onError(error: Error) {
    // Send to error tracking
    sentry.captureException(error);
  }
}

// Register handler
stream.on('*', new CustomStreamHandler());

Stream Composition

// Compose multiple streams
const generateStream = await agent.generateTestsStreaming(options);
const executeStream = await agent.executeTestsStreaming(options);

// Chain streams
generateStream.pipe(executeStream).on('complete', (results) => {
  console.log('Both generation and execution complete!');
});

// Merge streams
const mergedStream = Stream.merge([
  generateStream,
  executeStream
]);

mergedStream.on('progress', (update) => {
  // Unified progress from both operations
  console.log(`Overall: ${update.progress}%`);
});

Stream Filtering

const stream = await agent.executeTestsStreaming(options);

// Filter events
const filteredStream = stream.filter((event) => {
  // Only show failed tests
  return event.type === 'test:failed';
});

filteredStream.on('data', (failedTest) => {
  console.error('Failed:', failedTest.name);
});

Stream Transformation

const stream = await agent.generateTestsStreaming(options);

// Transform events
const transformedStream = stream.map((event) => {
  // Convert to custom format
  return {
    timestamp: new Date().toISOString(),
    level: event.type.includes('error') ? 'error' : 'info',
    message: event.message,
    data: event.data
  };
});

// Send to logging system
transformedStream.on('data', (logEntry) => {
  logger.log(logEntry);
});

---

Troubleshooting

Issue 1: No Progress Updates

Symptoms: Stream starts but no events received

Solutions:

// 1. Check if streaming is enabled
console.log('Streaming enabled:', config.features.streamingTools);

// 2. Verify event listeners are attached BEFORE starting
stream.on('progress', handler);
await stream.start();  // Start AFTER attaching listeners

// 3. Check update interval
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    updateInterval: 1000  // Increase if too frequent
  }
});

Issue 2: Connection Timeouts

Symptoms: Stream disconnects during long operations

Solutions:

// 1. Increase timeout
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    timeout: 600000,  // 10 minutes
    keepAlive: true,
    keepAliveInterval: 30000  // 30 seconds
  }
});

// 2. Enable automatic reconnection
stream.on('connection:lost', async () => {
  await stream.reconnect({ maxAttempts: 5, backoff: true });
});

Issue 3: Memory Leaks

Symptoms: Memory usage grows continuously

Solutions:

// 1. Always remove listeners
stream.on('complete', () => {
  stream.removeAllListeners();
});

// 2. Limit buffer size
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    bufferSize: 100,  // Limit buffered events
    dropOldest: true   // Drop old events when full
  }
});

// 3. Use weak references for handlers
const handler = new WeakRef(() => {
  // Handler logic
});
stream.on('progress', handler);

Issue 4: Events Out of Order

Symptoms: Progress updates arrive in wrong order

Solutions:

// 1. Enable ordering
const stream = await agent.generateTestsStreaming({
  ...options,
  streaming: {
    ordered: true,
    sequenceNumbers: true
  }
});

// 2. Buffer and sort events
const buffer: Event[] = [];
stream.on('data', (event) => {
  buffer.push(event);
  buffer.sort((a, b) => a.sequence - b.sequence);

  // Process in order
  while (buffer.length > 0 && buffer[0].sequence === nextExpected) {
    processEvent(buffer.shift());
    nextExpected++;
  }
});

---

FAQ

Q: Can I use streaming with non-streaming MCP tools?

A: Yes, streaming is backward compatible:

// Works with both streaming and non-streaming
const result = config.features.streamingTools
  ? await agent.generateTestsStreaming(options).complete()
  : await agent.generateTests(options);

Q: Does streaming work in CI/CD?

A: Yes, but format output appropriately:

# JSON format for parsing
aqe test --stream --format json > results.json

# Simple format for logs
aqe test --stream --format simple

Q: What's the performance impact?

A: Minimal (~2% overhead) for operations >5 seconds. See Performance section.

Q: Can I cancel a streaming operation?

A: Yes:

stream.on('progress', (update) => {
  if (shouldCancel) {
    stream.abort();
  }
});

Q: How do I test streaming in unit tests?

A: Use mock streams:

import { MockStream } from 'agentic-qe/testing';

const mockStream = new MockStream();
mockStream.emit('progress', { progress: 50 });
mockStream.emit('complete', { result: 'success' });

// Test your handler
await myHandler(mockStream);

---

Next Steps

1. Enable Streaming: Update config with streamingTools: true
2. Try Examples: Run code samples from this guide
3. Monitor Performance: Check overhead in your use case
4. Integrate: Add streaming to long operations
5. Learn More: Read API Reference: Streaming API

---

Related Documentation

• Multi-Model Router Guide
• Cost Optimization Best Practices
• Migration Guide
• API Reference: Streaming API

---

Questions? Open an issue: https://github.com/proffesor-for-testing/agentic-qe/issues