proffesor-for-testing__agentic-qe/v2/docs/api/STREAMING-API.md

Streaming API Reference

Version: 1.0.5 Module: agentic-qe/streaming

---

Overview

The Streaming API provides real-time progress updates for long-running QE operations. It enables live feedback, cancellation support, and better resource management for test generation, execution, and analysis.

---

Core Classes

OperationStream

Base class for all streaming operations.

import { OperationStream } from 'agentic-qe/streaming';

class OperationStream<T> extends EventEmitter {
  constructor(config: StreamConfig);

  // Lifecycle
  start(): Promise<void>;
  pause(): Promise<void>;
  resume(): Promise<void>;
  abort(): Promise<void>;
  complete(): Promise<T>;

  // State
  get state(): StreamState;
  get progress(): number;
  get isPaused(): boolean;
  get isAborted(): boolean;

  // Events
  on(event: string, handler: Function): this;
  once(event: string, handler: Function): this;
  removeListener(event: string, handler: Function): this;
  removeAllListeners(event?: string): this;

  // Piping
  pipe<U>(destination: OperationStream<U>): OperationStream<U>;
  unpipe(destination?: OperationStream<any>): this;

  // Filtering
  filter(predicate: (event: StreamEvent) => boolean): OperationStream<T>;
  map<U>(transform: (event: StreamEvent) => U): OperationStream<U>;
}

---

Specialized Streams

TestGenerationStream

Stream for test generation operations.

class TestGenerationStream extends OperationStream<TestGenerationResult> {
  // Additional events:
  // - 'test:generated': Emitted when a test is created
  // - 'file:analyzed': Emitted when source file is analyzed
  // - 'coverage:calculated': Emitted when coverage is computed
}

Example:

import { TestGenerationStream } from 'agentic-qe/streaming';

const stream = new TestGenerationStream({
  sourceFile: 'src/services/user-service.ts',
  framework: 'jest',
  targetCoverage: 95
});

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

stream.on('progress', (update) => {
  console.log(`Progress: ${update.progress}%`);
});

const result = await stream.complete();

---

TestExecutionStream

Stream for test execution operations.

class TestExecutionStream extends OperationStream<TestExecutionResult> {
  // Additional events:
  // - 'test:started': Test execution started
  // - 'test:passed': Test passed
  // - 'test:failed': Test failed
  // - 'test:skipped': Test skipped
  // - 'suite:started': Test suite started
  // - 'suite:completed': Test suite completed
}

Example:

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

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

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

const result = await stream.complete();

---

CoverageAnalysisStream

Stream for coverage analysis operations.

class CoverageAnalysisStream extends OperationStream<CoverageReport> {
  // Additional events:
  // - 'file:analyzed': File coverage analyzed
  // - 'gap:detected': Coverage gap found
  // - 'metrics:updated': Coverage metrics updated
}

Example:

const stream = new CoverageAnalysisStream({
  sourcePath: 'src/',
  testPath: 'tests/',
  threshold: 80
});

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

stream.on('gap:detected', (gap) => {
  console.log(`Gap in ${gap.file}: lines ${gap.lines.join(', ')}`);
});

const report = await stream.complete();

---

Interfaces

StreamConfig

interface StreamConfig {
  // Buffer settings
  bufferSize?: number;
  dropOldest?: boolean;

  // Update settings
  updateInterval?: number;
  batchEvents?: boolean;
  batchSize?: number;

  // Connection settings
  timeout?: number;
  keepAlive?: boolean;
  keepAliveInterval?: number;

  // Ordering
  ordered?: boolean;
  sequenceNumbers?: boolean;

  // Memory management
  maxMemory?: number;
  enableBackpressure?: boolean;
  highWaterMark?: number;

  // Compression
  compression?: 'none' | 'gzip' | 'brotli';
  compressionLevel?: number;
}

---

StreamEvent

interface StreamEvent {
  // Identity
  type: string;
  timestamp: number;
  sequence?: number;

  // Data
  data: any;

  // Metadata
  metadata?: {
    duration?: number;
    progress?: number;
    [key: string]: any;
  };
}

---

StreamState

type StreamState =
  | 'pending'
  | 'running'
  | 'paused'
  | 'completed'
  | 'aborted'
  | 'error';

interface StreamStateInfo {
  state: StreamState;
  startTime?: number;
  endTime?: number;
  duration?: number;
  eventsEmitted: number;
  bytesTransferred: number;
}

---

ProgressUpdate

interface ProgressUpdate {
  // Progress
  progress: number;  // 0-100
  completed: number;
  total: number;

  // Status
  status: string;
  currentOperation: string;

  // Timing
  timestamp: number;
  elapsed: number;
  remaining?: number;

  // Metrics
  metrics?: {
    testsGenerated?: number;
    testsPassed?: number;
    testsFailed?: number;
    coverage?: number;
    [key: string]: any;
  };
}

---

Event Types

Common Events

All streams emit these events:

progress

Emitted periodically with progress updates.

stream.on('progress', (update: ProgressUpdate) => {
  console.log(`${update.progress}% - ${update.currentOperation}`);
});

---

metrics

Emitted when metrics are updated.

stream.on('metrics', (metrics: Metrics) => {
  console.log('Metrics:', metrics);
});

---

complete

Emitted when operation completes successfully.

stream.on('complete', (result: T) => {
  console.log('Operation complete:', result);
});

---

error

Emitted when an error occurs.

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

---

abort

Emitted when operation is aborted.

stream.on('abort', (reason?: string) => {
  console.log('Operation aborted:', reason);
});

---

Connection Events

connection:open

Connection established.

stream.on('connection:open', () => {
  console.log('Connection opened');
});

---

connection:lost

Connection lost.

stream.on('connection:lost', () => {
  console.log('Connection lost, reconnecting...');
});

---

connection:restored

Connection restored.

stream.on('connection:restored', () => {
  console.log('Connection restored');
});

---

Memory Events

memory:warning

Memory usage high.

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

---

memory:limit

Memory limit reached.

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

---

Methods

start()

Start the streaming operation.

async start(): Promise<void>

Example:

const stream = new TestGenerationStream(options);

// Attach listeners first
stream.on('progress', handleProgress);
stream.on('test:generated', handleTest);

// Then start
await stream.start();

---

pause()

Pause the operation.

async pause(): Promise<void>

Example:

stream.on('memory:warning', async () => {
  await stream.pause();
  await gc(); // Wait for garbage collection
  await stream.resume();
});

---

resume()

Resume a paused operation.

async resume(): Promise<void>

Example:

await stream.pause();
// ... do something ...
await stream.resume();

---

abort()

Abort the operation.

async abort(reason?: string): Promise<void>

Example:

// Cancel button clicked
cancelButton.addEventListener('click', async () => {
  await stream.abort('User cancelled');
});

// Timeout
setTimeout(async () => {
  if (stream.state === 'running') {
    await stream.abort('Timeout exceeded');
  }
}, 300000); // 5 minutes

---

complete()

Wait for operation to complete and get result.

async complete(): Promise<T>

Example:

try {
  const result = await stream.complete();
  console.log('Success:', result);
} catch (error) {
  console.error('Failed:', error);
} finally {
  stream.removeAllListeners();
}

---

pipe()

Pipe events to another stream.

pipe<U>(destination: OperationStream<U>): OperationStream<U>

Example:

const genStream = new TestGenerationStream(options);
const execStream = new TestExecutionStream(options);

// Pipe generation to execution
genStream.pipe(execStream);

// When generation completes, execution starts
const finalResult = await execStream.complete();

---

filter()

Filter events.

filter(predicate: (event: StreamEvent) => boolean): OperationStream<T>

Example:

// Only show failed tests
const failedOnly = stream.filter(event =>
  event.type === 'test:failed'
);

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

---

map()

Transform events.

map<U>(transform: (event: StreamEvent) => U): OperationStream<U>

Example:

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

formatted.on('data', (log) => {
  logger.log(log);
});

---

Error Types

StreamConnectionError

class StreamConnectionError extends Error {
  constructor(message: string, public retryable: boolean);
}

---

StreamAbortedError

class StreamAbortedError extends Error {
  constructor(message: string, public reason?: string);
}

---

StreamTimeoutError

class StreamTimeoutError extends Error {
  constructor(message: string, public timeout: number);
}

---

StreamMemoryError

class StreamMemoryError extends Error {
  constructor(message: string, public usage: MemoryUsage);
}

---

Advanced Features

Reconnection

Automatic reconnection on connection loss.

const stream = new TestGenerationStream({
  ...options,
  reconnection: {
    enabled: true,
    maxAttempts: 5,
    delay: 1000,
    backoff: true,
    maxDelay: 30000
  }
});

stream.on('connection:lost', () => {
  console.log('Connection lost, will retry...');
});

stream.on('connection:restored', () => {
  console.log('Connection restored');
});

---

Buffering

Control event buffering.

const stream = new TestGenerationStream({
  ...options,
  bufferSize: 1000,
  dropOldest: true,  // Drop old events when buffer full
  batchEvents: true,  // Batch events to reduce overhead
  batchSize: 10,
  batchInterval: 500
});

// Receive batched events
stream.on('batch', (events: StreamEvent[]) => {
  events.forEach(event => processEvent(event));
});

---

Compression

Enable compression for large payloads.

const stream = new CoverageAnalysisStream({
  ...options,
  compression: 'gzip',
  compressionLevel: 6
});

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

---

Backpressure

Handle slow consumers.

const stream = new TestExecutionStream({
  ...options,
  enableBackpressure: true,
  highWaterMark: 16384
});

stream.on('backpressure', async () => {
  console.log('Consumer slow, pausing...');
  await stream.pause();

  // Wait for consumer to catch up
  await processBacklog();

  await stream.resume();
});

---

Integration Examples

With Progress Bar

import ProgressBar from 'progress';

const stream = new TestGenerationStream(options);

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

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

const result = await stream.complete();
bar.terminate();

---

With React

import { useState, useEffect } from 'react';

function TestGenerator({ options }) {
  const [progress, setProgress] = useState(0);
  const [status, setStatus] = useState('');
  const [tests, setTests] = useState([]);

  useEffect(() => {
    const stream = new TestGenerationStream(options);

    stream.on('progress', (update) => {
      setProgress(update.progress);
      setStatus(update.currentOperation);
    });

    stream.on('test:generated', (test) => {
      setTests(prev => [...prev, test]);
    });

    stream.start();

    return () => {
      stream.abort();
      stream.removeAllListeners();
    };
  }, [options]);

  return (
    <div>
      <progress value={progress} max={100} />
      <div>{status}</div>
      <ul>
        {tests.map(test => <li key={test.id}>{test.name}</li>)}
      </ul>
    </div>
  );
}

---

With Node.js Streams

import { Writable } from 'stream';

const stream = new TestExecutionStream(options);

// Write to Node.js stream
const writable = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  }
});

stream.on('data', (event) => {
  writable.write(JSON.stringify(event) + '\n');
});

await stream.complete();
writable.end();

---

Performance

Memory Usage

// Monitor memory usage
const stream = new TestGenerationStream({
  ...options,
  maxMemory: 512 * 1024 * 1024,  // 512 MB
  enableBackpressure: true
});

stream.on('metrics', (metrics) => {
  console.log(`Memory: ${metrics.memory.used}/${metrics.memory.limit}`);
});

---

Throughput

// Optimize for throughput
const stream = new TestExecutionStream({
  ...options,
  bufferSize: 10000,
  batchEvents: true,
  batchSize: 100,
  compression: 'gzip'
});

// Can handle 1000+ events/second

---

Best Practices

1. Always Clean Up

async function executeWithCleanup() {
  const stream = new TestGenerationStream(options);

  try {
    stream.on('progress', handleProgress);
    const result = await stream.complete();
    return result;
  } finally {
    stream.removeAllListeners();
  }
}

---

2. Handle Errors Gracefully

stream.on('error', async (error) => {
  if (error instanceof StreamConnectionError && error.retryable) {
    console.log('Retrying...');
    await stream.retry();
  } else {
    console.error('Fatal error:', error);
    await stream.abort();
  }
});

---

3. Monitor Resource Usage

stream.on('memory:warning', async () => {
  console.warn('High memory, pausing...');
  await stream.pause();
  await gc();
  await stream.resume();
});

stream.on('connection:lost', () => {
  console.log('Connection lost, will retry...');
});

---

4. Use Appropriate Buffer Sizes

// Small operations: small buffer
const smallStream = new TestGenerationStream({
  ...options,
  bufferSize: 10
});

// Large operations: large buffer
const largeStream = new TestExecutionStream({
  ...options,
  bufferSize: 1000
});

---

Migration Guide

From Non-Streaming

Before:

console.log('Generating tests...');
const result = await fleet.generateTests(options);
console.log('Done!');

After:

const stream = await fleet.streamTestGeneration(options);

stream.on('progress', (update) => {
  console.log(`${update.progress}% - ${update.currentOperation}`);
});

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

const result = await stream.complete();
console.log('Done!');

---

Related Documentation

• Streaming API Tutorial
• Multi-Model Router Guide
• Migration Guide
• Routing API Reference

---

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