# Pipeline Mode

URL: https://cogitator.app/docs/voice/pipeline

> Process voice through STT -> Agent -> TTS with VoicePipeline for batch processing and PipelineSession for continuous streaming conversations with VAD-driven turn detection.

## Overview

Pipeline mode routes audio through a three-stage loop: Speech-to-Text, Agent processing, Text-to-Speech. This works with any Cogitator agent regardless of the underlying LLM backend.

Two classes handle different use cases:

* **VoicePipeline** — batch processing: send a complete audio buffer, get back transcript + response + audio
* **PipelineSession** — streaming: push audio chunks continuously, get events as things happen

## VoicePipeline

One-shot processing for complete audio buffers.

```typescript
import { VoicePipeline, OpenAISTT, OpenAITTS, createCogitatorRunner } from '@cogitator-ai/voice';

const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  agent: createCogitatorRunner(cogitator, agent),
});
```

The `agent` is any `VoiceAgentRunner` — `{ run(input, context?) => Promise<{ content }> }`. `context` carries the `sessionId` and an abort `signal`. `createCogitatorRunner(cogitator, agent)` builds one from a Cogitator runtime and agent (memory thread `voice:<sessionId>` per session).

### `process(audio, options?)`

Transcribes the audio, runs the agent, synthesizes the response. Silent input (empty transcript) skips the agent and TTS and returns empty strings and an empty buffer; an empty agent reply skips TTS. Pass `{ sessionId }` to forward a session id to the agent.

```typescript
const result = await pipeline.process(audioBuffer);

console.log(result.transcript); // what the user said
console.log(result.response); // what the agent replied
// result.audio — synthesized response as Buffer
```

### TTS options

`ttsOptions` is passed to every `synthesize` / `streamSynthesize` call made by the pipeline and its sessions. Use it to pick the output format or a per-pipeline `voice`, `speed` or `instructions`:

```typescript
const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  agent: createCogitatorRunner(cogitator, agent),
  ttsOptions: { format: 'pcm16' },
});
```

Without `ttsOptions` the TTS provider's defaults apply (MP3 output). `format: 'pcm16'` returns raw PCM16 mono at 24kHz. `VoiceAgent` accepts the same `ttsOptions` in pipeline mode.

`openai` is an optional peer dependency: install it only when you use `OpenAISTT` / `OpenAITTS`. They load it on the first `transcribe` / `synthesize` call, which throws with an install hint if it is missing.

### `createSession(options?)`

Creates a `PipelineSession` for streaming use. Pass `{ sessionId }` to set the id the agent sees (default: a random id). See below.

## PipelineSession

Streaming mode for continuous conversations. Push audio chunks as they arrive, receive events for each stage of processing.

```typescript
const session = pipeline.createSession();

session.id; // passed to the agent as context.sessionId
session.currentState; // 'idle' | 'listening' | 'processing' | 'speaking'
```

### Pushing Audio

```typescript
// push PCM16 audio chunks as they arrive from the microphone
session.pushAudio(pcm16Chunk);
```

If the pipeline has a VAD provider, the session automatically detects when speech starts and ends. Without VAD, it streams to STT immediately and the turn is committed with `endAudio()`.

```typescript
session.endAudio(); // commit buffered audio (no-VAD mode)
await session.sendText('Hi there'); // run a turn from text, bypassing STT
```

### Events

```typescript
session.on('speech_start', () => {
  console.log('User started speaking');
});

session.on('speech_end', () => {
  console.log('User stopped speaking, transcribing...');
});

session.on('transcript', (text: string, isFinal: boolean) => {
  if (isFinal) {
    console.log('Final transcript:', text);
  } else {
    console.log('Interim:', text);
  }
});

session.on('agent_response', (text: string) => {
  console.log('Agent:', text);
});

session.on('audio', (chunk: Buffer) => {
  // streamed TTS audio — play it back to the user
  playAudio(chunk);
});

session.on('turn_end', () => {
  console.log('Agent finished speaking');
});

session.on('error', (error: Error) => {
  console.error('Session error:', error);
});
```

| Event            | Payload           | Description                              |
| ---------------- | ----------------- | ---------------------------------------- |
| `speech_start`   | —                 | VAD detected speech beginning (VAD only) |
| `speech_end`     | —                 | VAD detected speech ending (VAD only)    |
| `transcript`     | `(text, isFinal)` | STT result (interim or final)            |
| `agent_response` | `(text)`          | Agent's text response                    |
| `audio`          | `(chunk: Buffer)` | Streamed TTS audio chunk                 |
| `turn_end`       | —                 | Response fully spoken (or empty)         |
| `error`          | `(error: Error)`  | Processing error (STT, VAD, agent, TTS)  |

### Interruption

Interrupt the current processing at any stage — discards the STT stream, aborts the agent run (its `signal` fires) and stops TTS streaming. A stale turn never emits events or changes state after an interruption:

```typescript
session.interrupt();
```

After interruption, the session returns to idle and is ready for new audio input.

### Closing

```typescript
await session.close();
```

Aborts any in-progress turn (without waiting for hung providers), releases the STT stream and removes event listeners.

## Pipeline with VAD

When a VAD provider is set, the session handles turn detection automatically:

```typescript
import { VoicePipeline, OpenAISTT, OpenAITTS, EnergyVAD } from '@cogitator-ai/voice';

const pipeline = new VoicePipeline({
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  vad: new EnergyVAD({ threshold: 0.01, silenceDuration: 500 }),
  agent: myAgent,
});

const session = pipeline.createSession();

session.on('speech_start', () => {
  // VAD detected speech — session opens an STT stream
});

session.on('speech_end', () => {
  // VAD detected silence — session closes STT, runs agent, streams TTS
});

// just keep pushing audio (PCM16 at the VAD's sample rate), VAD handles the rest
microphoneStream.on('data', (chunk) => {
  session.pushAudio(chunk);
});
```

## VoiceAgent

`VoiceAgent` is the high-level class that combines pipeline/realtime processing with WebSocket transport and session management. For pipeline mode, it creates a `VoicePipeline` internally and manages per-client sessions.

```typescript
import { VoiceAgent, OpenAISTT, OpenAITTS, EnergyVAD } from '@cogitator-ai/voice';

const voiceAgent = new VoiceAgent({
  agent: createCogitatorRunner(cogitator, agent),
  mode: 'pipeline',
  stt: new OpenAISTT({ apiKey: process.env.OPENAI_API_KEY! }),
  tts: new OpenAITTS({ apiKey: process.env.OPENAI_API_KEY! }),
  vad: new EnergyVAD(),
  transport: {
    path: '/voice',
    maxConnections: 50,
    verifyClient: async (req) => isValidToken(req.headers.authorization), // false → 401; { code, message } for another status
  },
});

voiceAgent.on('session_start', (id) => console.log(`Session: ${id}`));
voiceAgent.on('session_end', (id) => console.log(`Ended: ${id}`));
voiceAgent.on('error', (err) => console.error(err));

await voiceAgent.listen(8080);

console.log(`Active sessions: ${voiceAgent.activeSessions}`);
await voiceAgent.close();
```

Use `voiceAgent.attach(httpServer)` instead of `listen()` to share an existing HTTP server; upgrades for other paths are left to other handlers (with `listen()` they get 404). `voiceAgent.port` is the bound port. Errors are only emitted when an `error` listener is attached — a misbehaving client cannot crash the process.

### Transport options

| Option           | Default    | Description                                                                                                                                                                  |
| ---------------- | ---------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `path`           | `'/voice'` | WebSocket path                                                                                                                                                               |
| `maxConnections` | `100`      | Further upgrades are rejected with `503`                                                                                                                                     |
| `verifyClient`   | —          | `(req) => boolean \| { code, message }` (may be async): `true` accepts, `false` rejects with `401`, `{ code, message }` rejects with that status; a throw rejects with `500` |

```typescript
transport: {
  verifyClient: (req) => {
    const token = new URL(req.url ?? '/', 'http://localhost').searchParams.get('token');
    if (!token) return false;
    return token === process.env.VOICE_TOKEN ? true : { code: 403, message: 'Forbidden' };
  },
},
```

Browsers cannot set headers on a WebSocket, so pass credentials in the query string or a cookie. Idle clients that stop answering pings (every 30s) are dropped. `WebSocketTransport` is exported too, for building your own server around `VoiceClient` connections (`audio` / `message` events, `sendAudio()`, `sendMessage()`).

### WebSocket protocol

| Direction        | Format | Content                                                                                             |
| ---------------- | ------ | --------------------------------------------------------------------------------------------------- |
| Client -> Server | Binary | PCM16 mono (16kHz pipeline / Gemini, 24kHz OpenAI realtime)                                         |
| Client -> Server | JSON   | `{ type: 'interrupt' }`, `{ type: 'end_of_speech' }` (pipeline), `{ type: 'text', text }`           |
| Server -> Client | Binary | Response audio (TTS output format in pipeline mode, MP3 by default; PCM16 24kHz in realtime mode)   |
| Server -> Client | JSON   | `{ type: 'transcript', text, isFinal }`, `agent_response`, `speech_start`, `speech_end`, `turn_end` |

In realtime mode `transcript` carries `role` (`'user'` or `'assistant'`) instead of `isFinal`, and there is no `agent_response` or `speech_end`. Invalid JSON and unknown control types are reported through the `error` event and otherwise ignored.

### VoicePipelineConfig

| Field        | Type               | Required | Description                                                             |
| ------------ | ------------------ | -------- | ----------------------------------------------------------------------- |
| `stt`        | `STTProvider`      | Yes      | Speech-to-text provider                                                 |
| `tts`        | `TTSProvider`      | Yes      | Text-to-speech provider                                                 |
| `vad`        | `VADProvider`      | No       | Voice activity detection                                                |
| `ttsOptions` | `TTSOptions`       | No       | Options for every TTS call (`format`, `voice`, `speed`, `instructions`) |
| `agent`      | `VoiceAgentRunner` | Yes      | `createCogitatorRunner(...)` or any `{ run(input, ctx) }`               |
| `sampleRate` | `number`           | No       | Deprecated, unused (configure the STT/VAD providers instead)            |