---
name: voice-ai-development
description: Expert in building voice AI applications - from real-time voice
  agents to voice-enabled apps. Covers OpenAI Realtime API, Vapi for voice
  agents, Deepgram for transcription, ElevenLabs for synthesis, LiveKit for
  real-time infrastructure, and WebRTC fundamentals.
risk: critical
source: vibeship-spawner-skills (Apache 2.0)
date_added: 2026-02-27
---

# Voice AI Development

Expert in building voice AI applications - from real-time voice agents to voice-enabled apps.
Covers OpenAI Realtime API, Vapi for voice agents, Deepgram for transcription, ElevenLabs
for synthesis, LiveKit for real-time infrastructure, and WebRTC fundamentals. Knows how to
build low-latency, production-ready voice experiences.

**Role**: Voice AI Architect

You are an expert in building real-time voice applications. You think in terms of
latency budgets, audio quality, and user experience. You know that voice apps feel
magical when fast and broken when slow. You choose the right combination of providers
for each use case and optimize relentlessly for perceived responsiveness.

### Expertise

- Real-time audio streaming
- Voice agent architecture
- Provider selection
- Latency optimization
- Audio quality tuning

## Capabilities

- OpenAI Realtime API
- Vapi voice agents
- Deepgram STT/TTS
- ElevenLabs voice synthesis
- LiveKit real-time infrastructure
- WebRTC audio handling
- Voice agent design
- Latency optimization

## Prerequisites

- 0: Async programming
- 1: WebSocket basics
- 2: Audio concepts (sample rate, codec)
- Required skills: Python or Node.js, API keys for providers, Audio handling knowledge

## Scope

- 0: Latency varies by provider
- 1: Cost per minute adds up
- 2: Quality depends on network
- 3: Complex debugging

## Ecosystem

### Primary

- OpenAI Realtime API
- Vapi
- Deepgram
- ElevenLabs

### Infrastructure

- LiveKit
- Daily.co
- Twilio

### Common_integrations

- WebRTC
- WebSockets
- Telephony (SIP/PSTN)

### Platforms

- Web applications
- Mobile apps
- Call centers
- Voice assistants

## Patterns

### OpenAI Realtime API

Native voice-to-voice with GPT-4o

**When to use**: When you want integrated voice AI without separate STT/TTS

import asyncio
import websockets
import json
import base64
import os

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

async def voice_session():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",  # alloy, echo, fable, onyx, nova, shimmer
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",  # Voice activity detection
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500
                },
                "tools": [
                    {
                        "type": "function",
                        "name": "get_weather",
                        "description": "Get weather for a location",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "location": {"type": "string"}
                            }
                        }
                    }
                ]
            }
        }))

        # Send audio (PCM16, 24kHz, mono)
        async def send_audio(audio_bytes):
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_bytes).decode()
            }))

        # Receive events
        async for message in ws:
            event = json.loads(message)

            if event["type"] == "response.audio.delta":
                # Play audio chunk
                audio = base64.b64decode(event["delta"])
                play_audio(audio)

            elif event["type"] == "response.audio_transcript.done":
                print(f"Assistant said: {event['transcript']}")

            elif event["type"] == "input_audio_buffer.speech_started":
                print("User started speaking")

            elif event["type"] == "response.function_call_arguments.done":
                # Handle tool call
                name = event["name"]
                args = json.loads(event["arguments"])
                result = call_function(name, args)
                await ws.send(json.dumps({
                    "type": "conversation.item.create",
                    "item": {
                        "type": "function_call_output",
                        "call_id": event["call_id"],
                        "output": json.dumps(result)
                    }
                }))

### Vapi Voice Agent

Build voice agents with Vapi platform

**When to use**: Phone-based agents, quick deployment

# Vapi provides hosted voice agents with webhooks

from flask import Flask, request, jsonify
import vapi

app = Flask(__name__)
client = vapi.Vapi(api_key="...")

# Create an assistant
assistant = client.assistants.create(
    name="Support Agent",
    model={
        "provider": "openai",
        "model": "gpt-4o",
        "messages": [
            {
                "role": "system",
                "content": "You are a helpful support agent..."
            }
        ]
    },
    voice={
        "provider": "11labs",
        "voiceId": "21m00Tcm4TlvDq8ikWAM"  # Rachel
    },
    firstMessage="Hi! How can I help you today?",
    transcriber={
        "provider": "deepgram",
        "model": "nova-2"
    }
)

# Webhook for conversation events
@app.route("/vapi/webhook", methods=["POST"])
def vapi_webhook():
    event = request.json

    if event["type"] == "function-call":
        # Handle tool call
        name = event["functionCall"]["name"]
        args = event["functionCall"]["parameters"]

        if name == "check_order":
            result = check_order(args["order_id"])
            return jsonify({"result": result})

    elif event["type"] == "end-of-call-report":
        # Call ended - save transcript
        transcript = event["transcript"]
        save_transcript(event["call"]["id"], transcript)

    return jsonify({"ok": True})

# Start outbound call
call = client.calls.create(
    assistant_id=assistant.id,
    customer={
        "number": "+1234567890"
    },
    phoneNumber={
        "twilioPhoneNumber": "+0987654321"
    }
)

# Or create web call
web_call = client.calls.create(
    assistant_id=assistant.id,
    type="web"
)
# Returns URL for WebRTC connection

### Deepgram STT + ElevenLabs TTS

Best-in-class transcription and synthesis

**When to use**: High quality voice, custom pipeline

import asyncio
from deepgram import DeepgramClient, LiveTranscriptionEvents
from elevenlabs import ElevenLabs

# Deepgram real-time transcription
deepgram = DeepgramClient(api_key="...")

async def transcribe_stream(audio_stream):
    connection = deepgram.listen.live.v("1")

    async def on_transcript(result):
        transcript = result.channel.alternatives[0].transcript
        if transcript:
            print(f"Heard: {transcript}")
            if result.is_final:
                # Process final transcript
                await handle_user_input(transcript)

    connection.on(LiveTranscriptionEvents.Transcript, on_transcript)

    await connection.start({
        "model": "nova-2",  # Best quality
        "language": "en",
        "smart_format": True,
        "interim_results": True,  # Get partial results
        "utterance_end_ms": 1000,
        "vad_events": True,  # Voice activity detection
        "encoding": "linear16",
        "sample_rate": 16000
    })

    # Stream audio
    async for chunk in audio_stream:
        await connection.send(chunk)

    await connection.finish()

# ElevenLabs streaming synthesis
eleven = ElevenLabs(api_key="...")

def text_to_speech_stream(text: str):
    """Stream TTS audio chunks."""
    audio_stream = eleven.text_to_speech.convert_as_stream(
        voice_id="21m00Tcm4TlvDq8ikWAM",  # Rachel
        model_id="eleven_turbo_v2_5",  # Fastest
        text=text,
        output_format="pcm_24000"  # Raw PCM for low latency
    )

    for chunk in audio_stream:
        yield chunk

# Or with WebSocket for lowest latency
async def tts_websocket(text_stream):
    async with eleven.text_to_speech.stream_async(
        voice_id="21m00Tcm4TlvDq8ikWAM",
        model_id="eleven_turbo_v2_5"
    ) as tts:
        async for text_chunk in text_stream:
            audio = await tts.send(text_chunk)
            yield audio

        # Flush remaining audio
        final_audio = await tts.flush()
        yield final_audio

### LiveKit Real-time Infrastructure

WebRTC infrastructure for voice apps

**When to use**: Building custom real-time voice apps

from livekit import api, rtc
import asyncio

# Server-side: Create room and tokens
lk_api = api.LiveKitAPI(
    url="wss://your-livekit.livekit.cloud",
    api_key="...",
    api_secret="..."
)

async def create_room(room_name: str):
    room = await lk_api.room.create_room(
        api.CreateRoomRequest(name=room_name)
    )
    return room

def create_token(room_name: str, participant_name: str):
    token = api.AccessToken(
        api_key="...",
        api_secret="..."
    )
    token.with_identity(participant_name)
    token.with_grants(api.VideoGrants(
        room_join=True,
        room=room_name
    ))
    return token.to_jwt()

# Agent-side: Connect and process audio
async def voice_agent(room_name: str):
    room = rtc.Room()

    @room.on("track_subscribed")
    def on_track(track, publication, participant):
        if track.kind == rtc.TrackKind.KIND_AUDIO:
            # Process incoming audio
            audio_stream = rtc.AudioStream(track)
            asyncio.create_task(process_audio(audio_stream))

    token = create_token(room_name, "agent")
    await room.connect("wss://your-livekit.livekit.cloud", token)

    # Publish agent's audio
    source = rtc.AudioSource(sample_rate=24000, num_channels=1)
    track = rtc.LocalAudioTrack.create_audio_track("agent-voice", source)
    await room.local_participant.publish_track(track)

    # Send audio from TTS
    async def speak(text: str):
        for audio_chunk in text_to_speech(text):
            await source.capture_frame(rtc.AudioFrame(
                data=audio_chunk,
                sample_rate=24000,
                num_channels=1,
                samples_per_channel=len(audio_chunk) // 2
            ))

    return room, speak

# Process audio with STT
async def process_audio(audio_stream):
    async for frame in audio_stream:
        # Send to Deepgram or other STT
        await transcriber.send(frame.data)

### Full Voice Agent Pipeline

Complete voice agent with all components

**When to use**: Custom production voice agent

import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

@dataclass
class VoiceAgentConfig:
    stt_provider: str = "deepgram"
    tts_provider: str = "elevenlabs"
    llm_provider: str = "openai"
    vad_enabled: bool = True
    interrupt_enabled: bool = True

class VoiceAgent:
    def __init__(self, config: VoiceAgentConfig):
        self.config = config
        self.is_speaking = False
        self.conversation_history = []

    async def process_audio_stream(
        self,
        audio_in: AsyncIterator[bytes],
        audio_out: asyncio.Queue
    ):
        """Main audio processing loop."""

        # STT streaming
        async def transcribe():
            transcript_buffer = ""
            async for audio_chunk in audio_in:
                # Check for interruption
                if self.is_speaking and self.config.interrupt_enabled:
                    if await self.detect_speech(audio_chunk):
                        await self.stop_speaking()

                result = await self.stt.transcribe(audio_chunk)
                if result.is_final:
                    yield result.transcript

        # Process transcripts
        async for user_text in transcribe():
            if not user_text.strip():
                continue

            self.conversation_history.append({
                "role": "user",
                "content": user_text
            })

            # Generate response with streaming
            self.is_speaking = True
            async for audio_chunk in self.generate_response(user_text):
                await audio_out.put(audio_chunk)
            self.is_speaking = False

    async def generate_response(self, text: str) -> AsyncIterator[bytes]:
        """Stream LLM response through TTS."""

        # Stream LLM tokens
        llm_stream = self.llm.stream_chat(self.conversation_history)

        # Buffer for TTS (need ~50 chars for good prosody)
        text_buffer = ""
        full_response = ""

        async for token in llm_stream:
            text_buffer += token
            full_response += token

            # Send to TTS when we have enough text
            if len(text_buffer) > 50 or token in ".!?":
                async for audio in self.tts.synthesize_stream(text_buffer):
                    yield audio
                text_buffer = ""

        # Flush remaining
        if text_buffer:
            async for audio in self.tts.synthesize_stream(text_buffer):
                yield audio

        self.conversation_history.append({
            "role": "assistant",
            "content": full_response
        })

    async def detect_speech(self, audio: bytes) -> bool:
        """Voice activity detection."""
        # Use WebRTC VAD or Silero VAD
        return self.vad.is_speech(audio)

    async def stop_speaking(self):
        """Handle interruption."""
        self.is_speaking = False
        # Clear audio queue
        # Stop TTS generation

# Latency optimization tips:
# 1. Use streaming everywhere (STT, LLM, TTS)
# 2. Start TTS before LLM finishes (~50 char buffer)
# 3. Use PCM audio format (no encoding overhead)
# 4. Keep WebSocket connections alive
# 5. Use regional endpoints close to users

## Validation Checks

### Non-Streaming TTS

Severity: HIGH

Message: Non-streaming TTS adds significant latency.

Fix action: Use tts.synthesize_stream() or tts.convert_as_stream()

### Hardcoded Sample Rate

Severity: MEDIUM

Message: Hardcoded sample rate may cause format mismatches.

Fix action: Define sample rates as constants, document expected formats

### WebSocket Without Reconnection

Severity: HIGH

Message: WebSocket connections need reconnection logic.

Fix action: Add retry loop with exponential backoff

### Missing VAD Configuration

Severity: MEDIUM

Message: VAD needs tuning for good user experience.

Fix action: Configure threshold and silence_duration_ms

### Blocking Audio Processing

Severity: HIGH

Message: Audio processing should be async to avoid blocking.

Fix action: Use async def and await for audio operations

### Missing Interruption Handling

Severity: MEDIUM

Message: Voice agents should handle user interruptions.

Fix action: Add barge-in detection and cancel current response

### Audio Queue Without Clear

Severity: LOW

Message: Audio queues should be clearable for interruptions.

Fix action: Add method to clear queue on interruption

### WebSocket Without Error Handling

Severity: HIGH

Message: WebSocket operations need error handling.

Fix action: Wrap in try/except for ConnectionClosed

## Collaboration

### Delegation Triggers

- agent graph|workflow|state -> langgraph (Need complex agent logic behind voice)
- extract|structured|json -> structured-output (Need to extract structured data from voice)
- observability|tracing|monitoring -> langfuse (Need to monitor voice agent quality)
- frontend|web|react -> nextjs-app-router (Need web interface for voice agent)

### Intelligent Voice Agent

Skills: voice-ai-development, langgraph, structured-output

Workflow:

```
1. Design agent graph with tools
2. Add voice interface layer
3. Use structured output for tool responses
4. Optimize for voice latency
```

### Monitored Voice Agent

Skills: voice-ai-development, langfuse

Workflow:

```
1. Build voice agent with provider of choice
2. Add Langfuse callbacks
3. Track latency, quality, conversation flow
4. Iterate based on metrics
```

### Phone-based Agent

Skills: voice-ai-development, twilio

Workflow:

```
1. Set up Vapi or custom agent
2. Connect to Twilio for PSTN
3. Handle inbound/outbound calls
4. Implement call routing logic
```

## Related Skills

Works well with: `langgraph`, `structured-output`, `langfuse`

## When to Use
- User mentions or implies: voice ai
- User mentions or implies: voice agent
- User mentions or implies: speech to text
- User mentions or implies: text to speech
- User mentions or implies: realtime voice
- User mentions or implies: vapi
- User mentions or implies: deepgram
- User mentions or implies: elevenlabs
- User mentions or implies: livekit
- User mentions or implies: openai realtime

## Limitations
- Use this skill only when the task clearly matches the scope described above.
- Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.


---
name: voice-ai-engine-development
description: "Build real-time conversational AI voice engines using async worker pipelines, streaming transcription, LLM agents, and TTS synthesis with interrupt handling and multi-provider support"
risk: critical
source: community
date_added: "2026-02-27"
---

# Voice AI Engine Development

## Overview

This skill guides you through building production-ready voice AI engines with real-time conversation capabilities. Voice AI engines enable natural, bidirectional conversations between users and AI agents through streaming audio processing, speech-to-text transcription, LLM-powered responses, and text-to-speech synthesis.

The core architecture uses an async queue-based worker pipeline where each component runs independently and communicates via `asyncio.Queue` objects, enabling concurrent processing, interrupt handling, and real-time streaming at every stage.

## When to Use This Skill

Use this skill when:
- Building real-time voice conversation systems
- Implementing voice assistants or chatbots
- Creating voice-enabled customer service agents
- Developing voice AI applications with interrupt capabilities
- Integrating multiple transcription, LLM, or TTS providers
- Working with streaming audio processing pipelines
- The user mentions Vocode, voice engines, or conversational AI

## Core Architecture Principles

### The Worker Pipeline Pattern

Every voice AI engine follows this pipeline:

```
Audio In → Transcriber → Agent → Synthesizer → Audio Out
           (Worker 1)   (Worker 2)  (Worker 3)
```

**Key Benefits:**
- **Decoupling**: Workers only know about their input/output queues
- **Concurrency**: All workers run simultaneously via asyncio
- **Backpressure**: Queues automatically handle rate differences
- **Interruptibility**: Everything can be stopped mid-stream

### Base Worker Pattern

Every worker follows this pattern:

```python
class BaseWorker:
    def __init__(self, input_queue, output_queue):
        self.input_queue = input_queue   # asyncio.Queue to consume from
        self.output_queue = output_queue # asyncio.Queue to produce to
        self.active = False

    def start(self):
        """Start the worker's processing loop"""
        self.active = True
        asyncio.create_task(self._run_loop())

    async def _run_loop(self):
        """Main processing loop - runs forever until terminated"""
        while self.active:
            item = await self.input_queue.get()  # Block until item arrives
            await self.process(item)              # Process the item

    async def process(self, item):
        """Override this - does the actual work"""
        raise NotImplementedError

    def terminate(self):
        """Stop the worker"""
        self.active = False
```

## Component Implementation Guide

### 1. Transcriber (Audio → Text)

**Purpose**: Converts incoming audio chunks to text transcriptions

**Interface Requirements**:
```python
class BaseTranscriber:
    def __init__(self, transcriber_config):
        self.input_queue = asyncio.Queue()   # Audio chunks (bytes)
        self.output_queue = asyncio.Queue()  # Transcriptions
        self.is_muted = False

    def send_audio(self, chunk: bytes):
        """Client calls this to send audio"""
        if not self.is_muted:
            self.input_queue.put_nowait(chunk)
        else:
            # Send silence instead (prevents echo during bot speech)
            self.input_queue.put_nowait(self.create_silent_chunk(len(chunk)))

    def mute(self):
        """Called when bot starts speaking (prevents echo)"""
        self.is_muted = True

    def unmute(self):
        """Called when bot stops speaking"""
        self.is_muted = False
```

**Output Format**:
```python
class Transcription:
    message: str          # "Hello, how are you?"
    confidence: float     # 0.95
    is_final: bool        # True = complete sentence, False = partial
    is_interrupt: bool    # Set by TranscriptionsWorker
```

**Supported Providers**:
- **Deepgram** - Fast, accurate, streaming
- **AssemblyAI** - High accuracy, good for accents
- **Azure Speech** - Enterprise-grade
- **Google Cloud Speech** - Multi-language support

**Critical Implementation Details**:
- Use WebSocket for bidirectional streaming
- Run sender and receiver tasks concurrently with `asyncio.gather()`
- Mute transcriber when bot speaks to prevent echo/feedback loops
- Handle both final and partial transcriptions

### 2. Agent (Text → Response)

**Purpose**: Processes user input and generates conversational responses

**Interface Requirements**:
```python
class BaseAgent:
    def __init__(self, agent_config):
        self.input_queue = asyncio.Queue()   # TranscriptionAgentInput
        self.output_queue = asyncio.Queue()  # AgentResponse
        self.transcript = None               # Conversation history

    async def generate_response(self, human_input, is_interrupt, conversation_id):
        """Override this - returns AsyncGenerator of responses"""
        raise NotImplementedError
```

**Why Streaming Responses?**
- **Lower latency**: Start speaking as soon as first sentence is ready
- **Better interrupts**: Can stop mid-response
- **Sentence-by-sentence**: More natural conversation flow

**Supported Providers**:
- **OpenAI** (GPT-4, GPT-3.5) - High quality, fast
- **Google Gemini** - Multimodal, cost-effective
- **Anthropic Claude** - Long context, nuanced responses

**Critical Implementation Details**:
- Maintain conversation history in `Transcript` object
- Stream responses using `AsyncGenerator`
- **IMPORTANT**: Buffer entire LLM response before yielding to synthesizer (prevents audio jumping)
- Handle interrupts by canceling current generation task
- Update conversation history with partial messages on interrupt

### 3. Synthesizer (Text → Audio)

**Purpose**: Converts agent text responses to speech audio

**Interface Requirements**:
```python
class BaseSynthesizer:
    async def create_speech(self, message: BaseMessage, chunk_size: int) -> SynthesisResult:
        """
        Returns a SynthesisResult containing:
        - chunk_generator: AsyncGenerator that yields audio chunks
        - get_message_up_to: Function to get partial text (for interrupts)
        """
        raise NotImplementedError
```

**SynthesisResult Structure**:
```python
class SynthesisResult:
    chunk_generator: AsyncGenerator[ChunkResult, None]
    get_message_up_to: Callable[[float], str]  # seconds → partial text

    class ChunkResult:
        chunk: bytes          # Raw PCM audio
        is_last_chunk: bool
```

**Supported Providers**:
- **ElevenLabs** - Most natural voices, streaming
- **Azure TTS** - Enterprise-grade, many languages
- **Google Cloud TTS** - Cost-effective, good quality
- **Amazon Polly** - AWS integration
- **Play.ht** - Voice cloning

**Critical Implementation Details**:
- Stream audio chunks as they're generated
- Convert audio to LINEAR16 PCM format (16kHz sample rate)
- Implement `get_message_up_to()` for interrupt handling
- Handle audio format conversion (MP3 → PCM)

### 4. Output Device (Audio → Client)

**Purpose**: Sends synthesized audio back to the client

**CRITICAL: Rate Limiting for Interrupts**

```python
async def send_speech_to_output(self, message, synthesis_result,
                                stop_event, seconds_per_chunk):
    chunk_idx = 0
    async for chunk_result in synthesis_result.chunk_generator:
        # Check for interrupt
        if stop_event.is_set():
            logger.debug(f"Interrupted after {chunk_idx} chunks")
            message_sent = synthesis_result.get_message_up_to(
                chunk_idx * seconds_per_chunk
            )
            return message_sent, True  # cut_off = True

        start_time = time.time()

        # Send chunk to output device
        self.output_device.consume_nonblocking(chunk_result.chunk)

        # CRITICAL: Wait for chunk to play before sending next one
        # This is what makes interrupts work!
        speech_length = seconds_per_chunk
        processing_time = time.time() - start_time
        await asyncio.sleep(max(speech_length - processing_time, 0))

        chunk_idx += 1

    return message, False  # cut_off = False
```

**Why Rate Limiting?**
Without rate limiting, all audio chunks would be sent immediately, which would:
- Buffer entire message on client side
- Make interrupts impossible (all audio already sent)
- Cause timing issues

By sending one chunk every N seconds:
- Real-time playback is maintained
- Interrupts can stop mid-sentence
- Natural conversation flow is preserved

## The Interrupt System

The interrupt system is critical for natural conversations.

### How Interrupts Work

**Scenario**: Bot is saying "I think the weather will be nice today and tomorrow and—" when user interrupts with "Stop".

**Step 1: User starts speaking**
```python
# TranscriptionsWorker detects new transcription while bot speaking
async def process(self, transcription):
    if not self.conversation.is_human_speaking:  # Bot was speaking!
        # Broadcast interrupt to all in-flight events
        interrupted = self.conversation.broadcast_interrupt()
        transcription.is_interrupt = interrupted
```

**Step 2: broadcast_interrupt() stops everything**
```python
def broadcast_interrupt(self):
    num_interrupts = 0
    # Interrupt all queued events
    while True:
        try:
            interruptible_event = self.interruptible_events.get_nowait()
            if interruptible_event.interrupt():  # Sets interruption_event
                num_interrupts += 1
        except queue.Empty:
            break

    # Cancel current tasks
    self.agent.cancel_current_task()              # Stop generating text
    self.agent_responses_worker.cancel_current_task()  # Stop synthesizing
    return num_interrupts > 0
```

**Step 3: SynthesisResultsWorker detects interrupt**
```python
async def send_speech_to_output(self, synthesis_result, stop_event, ...):
    async for chunk_result in synthesis_result.chunk_generator:
        # Check stop_event (this is the interruption_event)
        if stop_event.is_set():
            logger.debug("Interrupted! Stopping speech.")
            # Calculate what was actually spoken
            seconds_spoken = chunk_idx * seconds_per_chunk
            partial_message = synthesis_result.get_message_up_to(seconds_spoken)
            # e.g., "I think the weather will be nice today"
            return partial_message, True  # cut_off = True
```

**Step 4: Agent updates history**
```python
if cut_off:
    # Update conversation history with partial message
    self.agent.update_last_bot_message_on_cut_off(message_sent)
    # History now shows:
    # Bot: "I think the weather will be nice today" (incomplete)
```

### InterruptibleEvent Pattern

Every event in the pipeline is wrapped in an `InterruptibleEvent`:

```python
class InterruptibleEvent:
    def __init__(self, payload, is_interruptible=True):
        self.payload = payload
        self.is_interruptible = is_interruptible
        self.interruption_event = threading.Event()  # Initially not set
        self.interrupted = False

    def interrupt(self) -> bool:
        """Interrupt this event"""
        if not self.is_interruptible:
            return False
        if not self.interrupted:
            self.interruption_event.set()  # Signal to stop!
            self.interrupted = True
            return True
        return False

    def is_interrupted(self) -> bool:
        return self.interruption_event.is_set()
```

## Multi-Provider Factory Pattern

Support multiple providers with a factory pattern:

```python
class VoiceHandler:
    """Multi-provider factory for voice components"""

    def create_transcriber(self, agent_config: Dict):
        """Create transcriber based on transcriberProvider"""
        provider = agent_config.get("transcriberProvider", "deepgram")

        if provider == "deepgram":
            return self._create_deepgram_transcriber(agent_config)
        elif provider == "assemblyai":
            return self._create_assemblyai_transcriber(agent_config)
        elif provider == "azure":
            return self._create_azure_transcriber(agent_config)
        elif provider == "google":
            return self._create_google_transcriber(agent_config)
        else:
            raise ValueError(f"Unknown transcriber provider: {provider}")

    def create_agent(self, agent_config: Dict):
        """Create LLM agent based on llmProvider"""
        provider = agent_config.get("llmProvider", "openai")

        if provider == "openai":
            return self._create_openai_agent(agent_config)
        elif provider == "gemini":
            return self._create_gemini_agent(agent_config)
        else:
            raise ValueError(f"Unknown LLM provider: {provider}")

    def create_synthesizer(self, agent_config: Dict):
        """Create voice synthesizer based on voiceProvider"""
        provider = agent_config.get("voiceProvider", "elevenlabs")

        if provider == "elevenlabs":
            return self._create_elevenlabs_synthesizer(agent_config)
        elif provider == "azure":
            return self._create_azure_synthesizer(agent_config)
        elif provider == "google":
            return self._create_google_synthesizer(agent_config)
        elif provider == "polly":
            return self._create_polly_synthesizer(agent_config)
        elif provider == "playht":
            return self._create_playht_synthesizer(agent_config)
        else:
            raise ValueError(f"Unknown voice provider: {provider}")
```

## WebSocket Integration

Voice AI engines typically use WebSocket for bidirectional audio streaming:

```python
@app.websocket("/conversation")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    # Create voice components
    voice_handler = VoiceHandler()
    transcriber = voice_handler.create_transcriber(agent_config)
    agent = voice_handler.create_agent(agent_config)
    synthesizer = voice_handler.create_synthesizer(agent_config)

    # Create output device
    output_device = WebsocketOutputDevice(
        ws=websocket,
        sampling_rate=16000,
        audio_encoding=AudioEncoding.LINEAR16
    )

    # Create conversation orchestrator
    conversation = StreamingConversation(
        output_device=output_device,
        transcriber=transcriber,
        agent=agent,
        synthesizer=synthesizer
    )

    # Start all workers
    await conversation.start()

    try:
        # Receive audio from client
        async for message in websocket.iter_bytes():
            conversation.receive_audio(message)
    except WebSocketDisconnect:
        logger.info("Client disconnected")
    finally:
        await conversation.terminate()
```

## Common Pitfalls and Solutions

### 1. Audio Jumping/Cutting Off

**Problem**: Bot's audio jumps or cuts off mid-response.

**Cause**: Sending text to synthesizer in small chunks causes multiple TTS calls.

**Solution**: Buffer the entire LLM response before sending to synthesizer:

```python
# ❌ Bad: Yields sentence-by-sentence
async for sentence in llm_stream:
    yield GeneratedResponse(message=BaseMessage(text=sentence))

# ✅ Good: Buffer entire response
full_response = ""
async for chunk in llm_stream:
    full_response += chunk
yield GeneratedResponse(message=BaseMessage(text=full_response))
```

### 2. Echo/Feedback Loop

**Problem**: Bot hears itself speaking and responds to its own audio.

**Cause**: Transcriber not muted during bot speech.

**Solution**: Mute transcriber when bot starts speaking:

```python
# Before sending audio to output
self.transcriber.mute()
# After audio playback complete
self.transcriber.unmute()
```

### 3. Interrupts Not Working

**Problem**: User can't interrupt bot mid-sentence.

**Cause**: All audio chunks sent at once instead of rate-limited.

**Solution**: Rate-limit audio chunks to match real-time playback:

```python
async for chunk in synthesis_result.chunk_generator:
    start_time = time.time()

    # Send chunk
    output_device.consume_nonblocking(chunk)

    # Wait for chunk duration before sending next
    processing_time = time.time() - start_time
    await asyncio.sleep(max(seconds_per_chunk - processing_time, 0))
```

### 4. Memory Leaks from Unclosed Streams

**Problem**: Memory usage grows over time.

**Cause**: WebSocket connections or API streams not properly closed.

**Solution**: Always use context managers and cleanup:

```python
try:
    async with websockets.connect(url) as ws:
        # Use websocket
        pass
finally:
    # Cleanup
    await conversation.terminate()
    await transcriber.terminate()
```

## Production Considerations

### 1. Error Handling

```python
async def _run_loop(self):
    while self.active:
        try:
            item = await self.input_queue.get()
            await self.process(item)
        except Exception as e:
            logger.error(f"Worker error: {e}", exc_info=True)
            # Don't crash the worker, continue processing
```

### 2. Graceful Shutdown

```python
async def terminate(self):
    """Gracefully shut down all workers"""
    self.active = False

    # Stop all workers
    self.transcriber.terminate()
    self.agent.terminate()
    self.synthesizer.terminate()

    # Wait for queues to drain
    await asyncio.sleep(0.5)

    # Close connections
    if self.websocket:
        await self.websocket.close()
```

### 3. Monitoring and Logging

```python
# Log key events
logger.info(f"🎤 [TRANSCRIBER] Received: '{transcription.message}'")
logger.info(f"🤖 [AGENT] Generating response...")
logger.info(f"🔊 [SYNTHESIZER] Synthesizing {len(text)} characters")
logger.info(f"⚠ [INTERRUPT] User interrupted bot")

# Track metrics
metrics.increment("transcriptions.count")
metrics.timing("agent.response_time", duration)
metrics.gauge("active_conversations", count)
```

### 4. Rate Limiting and Quotas

```python
# Implement rate limiting for API calls
from aiolimiter import AsyncLimiter

rate_limiter = AsyncLimiter(max_rate=10, time_period=1)  # 10 calls/second

async def call_api(self, data):
    async with rate_limiter:
        return await self.client.post(data)
```

## Key Design Patterns

### 1. Producer-Consumer with Queues

```python
# Producer
async def producer(queue):
    while True:
        item = await generate_item()
        queue.put_nowait(item)

# Consumer
async def consumer(queue):
    while True:
        item = await queue.get()
        await process_item(item)
```

### 2. Streaming Generators

Instead of returning complete results:

```python
# ❌ Bad: Wait for entire response
async def generate_response(prompt):
    response = await openai.complete(prompt)  # 5 seconds
    return response

# ✅ Good: Stream chunks as they arrive
async def generate_response(prompt):
    async for chunk in openai.complete(prompt, stream=True):
        yield chunk  # Yield after 0.1s, 0.2s, etc.
```

### 3. Conversation State Management

Maintain conversation history for context:

```python
class Transcript:
    event_logs: List[Message] = []

    def add_human_message(self, text):
        self.event_logs.append(Message(sender=Sender.HUMAN, text=text))

    def add_bot_message(self, text):
        self.event_logs.append(Message(sender=Sender.BOT, text=text))

    def to_openai_messages(self):
        return [
            {"role": "user" if msg.sender == Sender.HUMAN else "assistant",
             "content": msg.text}
            for msg in self.event_logs
        ]
```

## Testing Strategies

### 1. Unit Test Workers in Isolation

```python
async def test_transcriber():
    transcriber = DeepgramTranscriber(config)

    # Mock audio input
    audio_chunk = b'\x00\x01\x02...'
    transcriber.send_audio(audio_chunk)

    # Check output
    transcription = await transcriber.output_queue.get()
    assert transcription.message == "expected text"
```

### 2. Integration Test Pipeline

```python
async def test_full_pipeline():
    # Create all components
    conversation = create_test_conversation()

    # Send test audio
    conversation.receive_audio(test_audio_chunk)

    # Wait for response
    response = await wait_for_audio_output(timeout=5)

    assert response is not None
```

### 3. Test Interrupts

```python
async def test_interrupt():
    conversation = create_test_conversation()

    # Start bot speaking
    await conversation.agent.generate_response("Tell me a long story")

    # Interrupt mid-response
    await asyncio.sleep(1)  # Let it speak for 1 second
    conversation.broadcast_interrupt()

    # Verify partial message in transcript
    last_message = conversation.transcript.event_logs[-1]
    assert last_message.text != full_expected_message
```

## Implementation Workflow

When implementing a voice AI engine:

1. **Start with Base Workers**: Implement the base worker pattern first
2. **Add Transcriber**: Choose a provider and implement streaming transcription
3. **Add Agent**: Implement LLM integration with streaming responses
4. **Add Synthesizer**: Implement TTS with audio streaming
5. **Connect Pipeline**: Wire all workers together with queues
6. **Add Interrupts**: Implement the interrupt system
7. **Add WebSocket**: Create WebSocket endpoint for client communication
8. **Test Components**: Unit test each worker in isolation
9. **Test Integration**: Test the full pipeline end-to-end
10. **Add Error Handling**: Implement robust error handling and logging
11. **Optimize**: Add rate limiting, monitoring, and performance optimizations

## Related Skills

- `@websocket-patterns` - For WebSocket implementation details
- `@async-python` - For asyncio and async patterns
- `@streaming-apis` - For streaming API integration
- `@audio-processing` - For audio format conversion and processing
- `@systematic-debugging` - For debugging complex async pipelines

## Resources

**Libraries**:
- `asyncio` - Async programming
- `websockets` - WebSocket client/server
- `FastAPI` - WebSocket server framework
- `pydub` - Audio manipulation
- `numpy` - Audio data processing

**API Providers**:
- Transcription: Deepgram, AssemblyAI, Azure Speech, Google Cloud Speech
- LLM: OpenAI, Google Gemini, Anthropic Claude
- TTS: ElevenLabs, Azure TTS, Google Cloud TTS, Amazon Polly, Play.ht

## Summary

Building a voice AI engine requires:
- ✅ Async worker pipeline for concurrent processing
- ✅ Queue-based communication between components
- ✅ Streaming at every stage (transcription, LLM, synthesis)
- ✅ Interrupt system for natural conversations
- ✅ Rate limiting for real-time audio playback
- ✅ Multi-provider support for flexibility
- ✅ Proper error handling and graceful shutdown

**The key insight**: Everything must stream and everything must be interruptible for natural, real-time conversations.

## Limitations
- Use this skill only when the task clearly matches the scope described above.
- Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.


---
name: vps-server-management
description: "Manage authorized VPS hosts and server-side agents through cautious SSH and operations workflows."
category: operations
risk: critical
source: community
source_repo: davidondrej/skills
source_type: community
date_added: "2026-07-07"
author: davidondrej
tags: [vps, ssh, server-management]
tools: [claude, codex]
license: "MIT"
license_source: "https://github.com/davidondrej/skills/blob/main/LICENSE"
---

# VPS Server Management

## When to Use

- Use when the user asks to operate an authorized VPS or agent running on a remote host.
- Use when SSH, deployment, restart, status, or log inspection is needed with explicit permission.

Source of truth: `library/infrastructure.md` (read it for the latest — IPs/expirations change).

## Servers (Hostinger VPS) — 3 total

| Hostname | IP | OS | Purpose | Expires |
|---|---|---|---|---|
| openclaw-server | <IP> | Ubuntu 24.04 (Dokploy) | OpenClaw — personal instance | <expiry> |
| n8n-server | <IP> | Ubuntu 24.04 (n8n) | All n8n workflow automations (primary) | <expiry> |
| hermes-server | <IP> | Ubuntu 24.04 | Hermes Agent — Discord gateway (Vilnius, LT) | <expiry> |

SSH as `root@<IP>`.

## Access levels (never share higher than needed)

1. **App login** — e.g. `app.example.hstgr.cloud`. Build/edit workflows, no server access. Safest to share.
2. **VPS SSH** — `root@<IP>`. Docker, files, system config. Trusted technical people only.
3. **Hostinger hPanel** — `hpanel.hostinger.com`. Billing, reboot, OS reinstall. Exposes SSH creds + browser terminal, so it grants server access too. The user only.

## Managing a VPS via an agent

For multi-step or exploratory work, **SSH into the box first and launch the agent ON the VPS** (e.g. `codex --yolo`), then talk to that local-on-server agent — it has full filesystem/process context and avoids fragile SSH round-trips. For short command sequences (update, config change, restart), driving an existing SSH session directly (e.g. via a cmux pane) is fine.

When checking on a remote/on-box agent, send the user one concise status line each time: what it is doing and whether it is on track.

Claude Code cmux note: after Claude finishes, it may prefill a predicted next user message; that draft is Claude, not the user speaking.

## Agents on servers

- **OpenClaw** → openclaw-server (managed via Dokploy).
- **Hermes** → hermes-server (Discord gateway). Setup/config docs in `library/hermes/`.
- **n8n** → n8n-server.

## Hermes ops (on hermes-server)

```bash
hermes --version            # shows version + commits behind
hermes update               # auto-snapshots, updates deps, rebuilds web UI, restarts gateway itself
hermes gateway status|restart
journalctl --user -u hermes-gateway --since '5 min ago' --no-pager   # gateway logs (systemd USER service)
```

- **Default model** lives in `~/.hermes/config.yaml` under `model.provider` + `model.default` — NOT in `.env`. Change via `hermes model` (interactive) or edit the yaml directly, then `hermes gateway restart` to propagate to gateways.
- npm `EBADENGINE` warnings during update (deps want Node >=24, box runs v22) are non-blocking — do not "fix" them.
- Deeper docs (Discord/Slack/WhatsApp setup, file structure, vision config): `library/hermes/`.

## Limitations

- Adapted from `davidondrej/skills`; verify local paths, tools, credentials, and agent features before acting.
- For commands, remote access, scheduling, browser automation, or file-changing workflows, get explicit user approval and confirm the target environment first.


---
name: vr-ar
description: "VR/AR development principles. Comfort, interaction, performance requirements."
risk: none
source: community
date_added: "2026-02-27"
---

# VR/AR Development

> Immersive experience principles.

---

## 1. Platform Selection

### VR Platforms

| Platform | Use Case |
|----------|----------|
| **Quest** | Standalone, wireless |
| **PCVR** | High fidelity |
| **PSVR** | Console market |
| **WebXR** | Browser-based |

### AR Platforms

| Platform | Use Case |
|----------|----------|
| **ARKit** | iOS devices |
| **ARCore** | Android devices |
| **WebXR** | Browser AR |
| **HoloLens** | Enterprise |

---

## 2. Comfort Principles

### Motion Sickness Prevention

| Cause | Solution |
|-------|----------|
| **Locomotion** | Teleport, snap turn |
| **Low FPS** | Maintain 90 FPS |
| **Camera shake** | Avoid or minimize |
| **Rapid acceleration** | Gradual movement |

### Comfort Settings

- Vignette during movement
- Snap vs smooth turning
- Seated vs standing modes
- Height calibration

---

## 3. Performance Requirements

### Target Metrics

| Platform | FPS | Resolution |
|----------|-----|------------|
| Quest 2 | 72-90 | 1832x1920 |
| Quest 3 | 90-120 | 2064x2208 |
| PCVR | 90 | 2160x2160+ |
| PSVR2 | 90-120 | 2000x2040 |

### Frame Budget

- VR requires consistent frame times
- Single dropped frame = visible judder
- 90 FPS = 11.11ms budget

---

## 4. Interaction Principles

### Controller Interaction

| Type | Use |
|------|-----|
| **Point + click** | UI, distant objects |
| **Grab** | Manipulation |
| **Gesture** | Magic, special actions |
| **Physical** | Throwing, swinging |

### Hand Tracking

- More immersive but less precise
- Good for: social, casual
- Challenging for: action, precision

---

## 5. Spatial Design

### World Scale

- 1 unit = 1 meter (critical)
- Objects must feel right size
- Test with real measurements

### Depth Cues

| Cue | Importance |
|-----|------------|
| Stereo | Primary depth |
| Motion parallax | Secondary |
| Shadows | Grounding |
| Occlusion | Layering |

---

## 6. Anti-Patterns

| ❌ Don't | ✅ Do |
|----------|-------|
| Move camera without player | Player controls camera |
| Drop below 90 FPS | Maintain frame rate |
| Use tiny UI text | Large, readable text |
| Ignore arm length | Scale to player reach |

---

> **Remember:** Comfort is not optional. Sick players don't play.

## When to Use
This skill is applicable to execute the workflow or actions described in the overview.

## Limitations
- Use this skill only when the task clearly matches the scope described above.
- Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.


---
name: vscode-extension-guide-en
description: "Guide for VS Code extension development from scaffolding to Marketplace publication"
category: core-dev
risk: safe
source: community
source_repo: lewiswigmore/agent-skills
source_type: community
date_added: "2026-04-12"
author: lewiswigmore
tags: [vscode, extension, ide, typescript, marketplace]
tools: [claude, cursor, copilot, codex, gemini]
---

# VS Code Extension Guide (English)

## Overview

An English guide for building VS Code extensions, covering the full lifecycle from scaffolding to Marketplace publication. Includes reference material on webview patterns, CSP security, TreeView, testing, packaging and troubleshooting. Updated for VS Code 1.74+ APIs.

Adapted from aktsmm/agent-skills (CC BY-NC-SA 4.0), translated to English with corrections for current VS Code APIs.

## When to Use This Skill

- Use when creating a new VS Code extension from scratch
- Use when adding commands, keybindings or settings to an extension
- Use when building TreeView or Webview UI in an extension
- Use when publishing an extension to the VS Code Marketplace
- Use when troubleshooting extension activation or packaging issues

## How It Works

### Quick Start

```bash
npm install -g yo generator-code
yo code
```

### Project Structure

```
my-extension/
├── package.json          # Extension manifest
├── src/extension.ts      # Entry point
├── out/                  # Compiled JS
├── images/icon.png       # 128x128 PNG for Marketplace
└── .vscodeignore         # Exclude files from VSIX
```

### Building and Packaging

```bash
npm run compile           # Build once
npm run watch             # Watch mode (F5 to launch debug)
npx @vscode/vsce package  # Creates .vsix
```

## Reference Topics

The full skill includes detailed reference documents on:

- **Webview patterns** with CSP security and message passing
- **TreeView** data providers and drag-and-drop
- **Testing** setup with @vscode/test-electron
- **Publishing** to the VS Code Marketplace
- **AI customization** for extension projects
- **Code review prompts** for extension code
- **Troubleshooting** common extension issues

## Install the Full Skill

For the complete guide with all reference documents:

```bash
npx skills add lewiswigmore/agent-skills --skill vscode-extension-guide-en
```

## Best Practices

- Unify package name, setting keys, command IDs and view IDs before publishing
- Keep package size under 5MB using `.vscodeignore`
- Since VS Code 1.74, `activationEvents` are auto-detected for contributed commands and views
- Always test with the Extension Development Host (F5) before packaging

## Common Pitfalls

- **Problem:** Extension not loading
  **Solution:** Check `activationEvents`. Since VS Code 1.74, these are auto-detected for contributed commands/views.

- **Problem:** Command not found
  **Solution:** Match the command ID exactly between package.json and your code.

- **Problem:** Webview content not displaying
  **Solution:** Check your Content Security Policy. Use the webview's `cspSource` property.

## Related Skills

- `@test-driven-development` - Write tests before implementing extension features
- `@debugging-strategies` - Systematic troubleshooting for extension issues

## Limitations
- Use this skill only when the task clearly matches the scope described above.
- Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.


---
name: vulnerability-scanner
description: "Advanced vulnerability analysis principles. OWASP 2025, Supply Chain Security, attack surface mapping, risk prioritization."
risk: critical
source: community
date_added: "2026-02-27"
---

# Vulnerability Scanner

> Think like an attacker, defend like an expert. 2025 threat landscape awareness.

## 🔧 Runtime Scripts

**Execute for automated validation:**

| Script | Purpose | Usage |
|--------|---------|-------|
| `scripts/security_scan.py` | Validate security principles applied | `python scripts/security_scan.py <project_path>` |

## 📋 Reference Files

| File | Purpose |
|------|---------|
| [checklists.md](checklists.md) | OWASP Top 10, Auth, API, Data protection checklists |

---

## 1. Security Expert Mindset

### Core Principles

| Principle | Application |
|-----------|-------------|
| **Assume Breach** | Design as if attacker already inside |
| **Zero Trust** | Never trust, always verify |
| **Defense in Depth** | Multiple layers, no single point |
| **Least Privilege** | Minimum required access only |
| **Fail Secure** | On error, deny access |

### Threat Modeling Questions

Before scanning, ask:
1. What are we protecting? (Assets)
2. Who would attack? (Threat actors)
3. How would they attack? (Attack vectors)
4. What's the impact? (Business risk)

---

## 2. OWASP Top 10:2025

### Risk Categories

| Rank | Category | Think About |
|------|----------|-------------|
| **A01** | Broken Access Control | Who can access what? IDOR, SSRF |
| **A02** | Security Misconfiguration | Defaults, headers, exposed services |
| **A03** | Software Supply Chain 🆕 | Dependencies, CI/CD, build integrity |
| **A04** | Cryptographic Failures | Weak crypto, exposed secrets |
| **A05** | Injection | User input → system commands |
| **A06** | Insecure Design | Flawed architecture |
| **A07** | Authentication Failures | Session, credential management |
| **A08** | Integrity Failures | Unsigned updates, tampered data |
| **A09** | Logging & Alerting | Blind spots, no monitoring |
| **A10** | Exceptional Conditions 🆕 | Error handling, fail-open states |

### 2025 Key Changes

```
2021 → 2025 Shifts:
├── SSRF merged into A01 (Access Control)
├── A02 elevated (Cloud/Container configs)
├── A03 NEW: Supply Chain (major focus)
├── A10 NEW: Exceptional Conditions
└── Focus shift: Root causes > Symptoms
```

---

## 3. Supply Chain Security (A03)

### Attack Surface

| Vector | Risk | Question to Ask |
|--------|------|-----------------|
| **Dependencies** | Malicious packages | Do we audit new deps? |
| **Lock files** | Integrity attacks | Are they committed? |
| **Build pipeline** | CI/CD compromise | Who can modify? |
| **Registry** | Typosquatting | Verified sources? |

### Defense Principles

- Verify package integrity (checksums)
- Pin versions, audit updates
- Use private registries for critical deps
- Sign and verify artifacts

---

## 4. Attack Surface Mapping

### What to Map

| Category | Elements |
|----------|----------|
| **Entry Points** | APIs, forms, file uploads |
| **Data Flows** | Input → Process → Output |
| **Trust Boundaries** | Where auth/authz checked |
| **Assets** | Secrets, PII, business data |

### Prioritization Matrix

```
Risk = Likelihood × Impact

High Impact + High Likelihood → CRITICAL
High Impact + Low Likelihood  → HIGH
Low Impact + High Likelihood  → MEDIUM
Low Impact + Low Likelihood   → LOW
```

---

## 5. Risk Prioritization

### CVSS + Context

| Factor | Weight | Question |
|--------|--------|----------|
| **CVSS Score** | Base severity | How severe is the vuln? |
| **EPSS Score** | Exploit likelihood | Is it being exploited? |
| **Asset Value** | Business context | What's at risk? |
| **Exposure** | Attack surface | Internet-facing? |

### Prioritization Decision Tree

```
Is it actively exploited (EPSS >0.5)?
├── YES → CRITICAL: Immediate action
└── NO → Check CVSS
         ├── CVSS ≥9.0 → HIGH
         ├── CVSS 7.0-8.9 → Consider asset value
         └── CVSS <7.0 → Schedule for later
```

---

## 6. Exceptional Conditions (A10 - New)

### Fail-Open vs Fail-Closed

| Scenario | Fail-Open (BAD) | Fail-Closed (GOOD) |
|----------|-----------------|---------------------|
| Auth error | Allow access | Deny access |
| Parsing fails | Accept input | Reject input |
| Timeout | Retry forever | Limit + abort |

### What to Check

- Exception handlers that catch-all and ignore
- Missing error handling on security operations
- Race conditions in auth/authz
- Resource exhaustion scenarios

---

## 7. Scanning Methodology

### Phase-Based Approach

```
1. RECONNAISSANCE
   └── Understand the target
       ├── Technology stack
       ├── Entry points
       └── Data flows

2. DISCOVERY
   └── Identify potential issues
       ├── Configuration review
       ├── Dependency analysis
       └── Code pattern search

3. ANALYSIS
   └── Validate and prioritize
       ├── False positive elimination
       ├── Risk scoring
       └── Attack chain mapping

4. REPORTING
   └── Actionable findings
       ├── Clear reproduction steps
       ├── Business impact
       └── Remediation guidance
```

---

## 8. Code Pattern Analysis

### High-Risk Patterns

| Pattern | Risk | Look For |
|---------|------|----------|
| **String concat in queries** | Injection | `"SELECT * FROM " + user_input` |
| **Dynamic code execution** | RCE | `eval()`, `exec()`, `Function()` | <!-- security-allowlist: defensive vulnerability taxonomy -->
| **Unsafe deserialization** | RCE | `pickle.loads()`, `unserialize()` |
| **Path manipulation** | Traversal | User input in file paths |
| **Disabled security** | Various | `verify=False`, `--insecure` |

### Secret Patterns

| Type | Indicators |
|------|-----------|
| API Keys | `api_key`, `apikey`, high entropy |
| Tokens | `token`, `bearer`, `jwt` |
| Credentials | `password`, `secret`, `key` |
| Cloud | `AWS_`, `AZURE_`, `GCP_` prefixes |

---

## 9. Cloud Security Considerations

### Shared Responsibility

| Layer | You Own | Provider Owns |
|-------|---------|---------------|
| Data | ✅ | ❌ |
| Application | ✅ | ❌ |
| OS/Runtime | Depends | Depends |
| Infrastructure | ❌ | ✅ |

### Cloud-Specific Checks

- IAM: Least privilege applied?
- Storage: Public buckets?
- Network: Security groups tightened?
- Secrets: Using secrets manager?

---

## 10. Anti-Patterns

| ❌ Don't | ✅ Do |
|----------|-------|
| Scan without understanding | Map attack surface first |
| Alert on every CVE | Prioritize by exploitability + asset |
| Ignore false positives | Maintain verified baseline |
| Fix symptoms only | Address root causes |
| Scan once before deploy | Continuous scanning |
| Trust third-party deps blindly | Verify integrity, audit code |

---

## 11. Reporting Principles

### Finding Structure

Each finding should answer:
1. **What?** - Clear vulnerability description
2. **Where?** - Exact location (file, line, endpoint)
3. **Why?** - Root cause explanation
4. **Impact?** - Business consequence
5. **How to fix?** - Specific remediation

### Severity Classification

| Severity | Criteria |
|----------|----------|
| **Critical** | RCE, auth bypass, mass data exposure |
| **High** | Data exposure, privilege escalation |
| **Medium** | Limited scope, requires conditions |
| **Low** | Informational, best practice |

---

> **Remember:** Vulnerability scanning finds issues. Expert thinking prioritizes what matters. Always ask: "What would an attacker do with this?"

## When to Use
This skill is applicable to execute the workflow or actions described in the overview.

## Limitations
- Use this skill only when the task clearly matches the scope described above.
- Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
- Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.


---
name: warehouse
description: "Plan and review read-only data warehouse analysis with explicit scope, privacy, provenance, and validation checks."
category: data
risk: critical
source: self
source_type: self
date_added: "2026-07-12"
author: Rudra-G-23
tags: [analytics, data-warehouse, sql, data-quality]
tools: [claude, cursor, gemini]
---

# Warehouse Analysis

## Overview

Use this skill to turn a business question into a careful, reproducible warehouse-analysis plan. It is vendor-neutral and assumes no particular schema, semantic layer, connector, or command-line tool.

The skill defaults to read-only work. It helps identify the data needed, review a proposed query, and communicate results without overstating what the evidence supports.

## When to Use

- The user wants to answer a business question using an authorized data warehouse.
- A proposed SQL query needs a review for grain, joins, filters, privacy, or interpretation risks.
- An analysis needs a clear record of scope, freshness, assumptions, and source tables.

Do not use this skill for warehouse administration, pipeline repair, access escalation, schema mutation, or production data changes.

## Required Inputs

Before proposing a query, establish:

1. The decision or question the analysis should inform.
2. The population, metric, dimensions, and time window.
3. The authorized warehouse or query interface available to the user.
4. The relevant schema documentation or table metadata.
5. Any privacy, retention, regional, or minimum-group-size constraints.

If a required input is missing, ask a focused question. Never invent table names, column names, business definitions, credentials, or query results.

## Workflow

### 1. Define the analytical contract

Restate the request as:

- **Question:** what is being measured or compared.
- **Population:** which entities are included and excluded.
- **Metric:** numerator, denominator, aggregation, and unit.
- **Window:** dates, timezone, and whether the period is complete.
- **Decision:** how the result will be used.

Call out ambiguous terms such as “active,” “customer,” “revenue,” or “last month.” Resolve ambiguity before querying.

### 2. Find governed sources

Prefer documented metrics, curated models, and governed tables over raw event streams. Use only schema information supplied by the user or available through an authorized interface.

For each proposed source, record:

- table or model name;
- expected grain and primary key;
- freshness or maximum available date;
- owner or documentation reference;
- known exclusions and quality warnings.

If the source cannot be verified, label the plan as provisional and stop before presenting numerical conclusions.

### 3. Draft a read-only query

Create a query only when the real schema is known. The query should:

- select only the columns needed for the stated question;
- filter the requested time window explicitly;
- use qualified column names and deterministic joins;
- guard division by zero and null-sensitive calculations;
- avoid row-level personal data when an aggregate answers the question;
- include a conservative row limit for exploratory output when appropriate.

Do not emit guessed SQL with fictional identifiers. If no authorized execution tool is available, provide the reviewed query for the user to run rather than claiming it was executed.

### 4. Review before execution

Check the proposed query against this list:

- Does every join preserve the intended grain?
- Can a one-to-many join duplicate the numerator or denominator?
- Are test, deleted, internal, or incomplete records handled deliberately?
- Are timezone boundaries and partial periods explicit?
- Does the query expose identifiers or small groups unnecessarily?
- Would a simpler aggregate answer reduce data access?
- Are metric definitions consistent with the documented source?

Revise any failed check before execution. For sensitive or high-impact decisions, ask for review by the data owner or another qualified analyst.

### 5. Execute only with authorization

Run a query only through a user-authorized, read-only interface. Do not request credentials in chat, bypass access controls, broaden permissions, or turn a read-only task into a write operation.

Stop if the interface is unavailable, the scope exceeds the user's authorization, or the result would reveal restricted personal or confidential data.

### 6. Validate the result

Before interpreting output:

- compare row counts and totals with a trusted reference when one exists;
- inspect null rates, duplicates, and unexpected categories;
- test whether conclusions change under reasonable window or filter choices;
- separate observed values from hypotheses about their causes.

Do not infer causality from a descriptive query. Do not hide contradictory or incomplete evidence.

### 7. Report with provenance

Use a compact result structure:

```text
Finding: [what the data shows]
Scope: [population and period]
Method: [metric and source summary]
Confidence: [high, medium, or low, with reason]
Caveats: [freshness, exclusions, quality, or privacy limits]
Next step: [optional validation or decision input]
```

Include the query or a reproducible query summary when disclosure is appropriate. Redact secrets, credentials, and unnecessary row-level data.

## Example

**Request:** “Did weekly activated accounts improve after the onboarding change?”

**Safe response plan:**

1. Clarify the activation definition, rollout date, eligible population, timezone, and comparison window.
2. Locate the governed activation metric and account cohort source.
3. Aggregate weekly counts or rates without selecting account-level identifiers.
4. Review cohort overlap, partial weeks, seasonality, and join duplication.
5. Report the observed change as an association, with confidence and caveats, not as proof of causation.

## Security & Safety Notes

- Treat warehouse contents and schema metadata as confidential unless the user establishes otherwise.
- Use least privilege and read-only access; never modify tables, permissions, pipelines, or production configuration.
- Minimize personal data and aggregate results whenever possible.
- Never place credentials, tokens, connection strings, or raw sensitive records in prompts or reports.
- Stop and escalate to the data owner when policy, authorization, or disclosure boundaries are unclear.

## Limitations

- This skill cannot discover an undocumented schema or verify a result without an authorized data source.
- It does not replace organization-specific metric definitions, privacy policy, or expert review.
- It does not diagnose pipelines, administer warehouses, or make product and business decisions.
- Conclusions remain limited by source quality, freshness, sampling, and the analytical design.


---
name: warp-delegate
description: Delegate coding tasks to the Warp Agent CLI (`oz`) only when the user
  explicitly requests it, while the orchestrator retains review and landing responsibility.
risk: safe
category: agent-orchestration
source: https://github.com/amElnagdy/delegate-skills
source_repo: amElnagdy/delegate-skills
source_type: community
date_added: '2026-08-26'
license: MIT
license_source: https://github.com/amElnagdy/delegate-skills/blob/master/LICENSE
compatibility: Requires the `oz` CLI (Warp Agent CLI) installed and authenticated
  (`oz login`, or `WARP_API_KEY` for a headless host; Warp AI features need an eligible
  Warp plan or your own provider key), Node 18+, and git. The orchestrating agent
  must be able to run shell commands and read files. Shell examples assume bash/zsh
  (macOS/Linux, or Git Bash/WSL on Windows).
metadata:
  version: 0.5.0
---
# Warp Delegate

## When to Use

- You want to delegate a bounded coding task to a separate `warp` implementer (`Warp Agent CLI`) and then review its diff yourself.
- The user explicitly asked for delegation to this implementer.

You are the **orchestrator**. Delegate a bounded coding task to a separate **implementer** - the
Warp Agent CLI - then review what it produced and land it yourself. You write the brief and own the
judgment; the implementer makes changes in its own conversation; you verify and commit.

The loop needs only a shell command and file access, so any comparable orchestrator can drive it.

## The binary is `oz`, not `warp`

Warp ships two different programs, and only one of them can be delegated to:

- **`oz`** - the Warp Agent CLI. Headless and scriptable; `oz agent run` executes an agent against a
  local directory. **This is what the relay drives.**
- **`warp`** - the interactive Warp TUI. It requires a terminal device, has no prompt or print flag
  (its only options are `--resume`, `--auto-approve`, `--api-key`, and the provider-key commands),
  and exits with `Device not configured` when stdin is a pipe. It cannot be relayed.

If `oz` is missing but `warp` is installed, you have the TUI, not the CLI.

## When NOT to use this

- The task is small enough to do inline; delegation overhead is not worth it.
- The `oz` CLI is not installed or authenticated.
- You need a sandboxed or read-only implementer. `oz agent run` has **no sandbox, no permission
  mode, and no read-only run** - see [Autonomy and permissions](#autonomy-and-permissions).
- The work must stay off Warp's servers. `oz agent run` uploads an end-of-run workspace snapshot
  unless `--no-snapshot` is passed, and conversations live server-side.

## Prerequisites (check once)

1. Install the Warp Agent CLI - see <https://docs.warp.dev/cli/>.
2. Authenticate: `oz login`, or set `WARP_API_KEY` for CI, a container, or any headless host.
3. Confirm the account has AI quota. **A working login is not enough** - unlike the other CLIs in
   this package. `oz whoami` can succeed while every dispatch fails with `In order to use Warp's AI
   features, subscribe to a Warp plan, or bring your own inference.` Warp records this internally as
   `QuotaLimit` / "lack of AI quota", so it is a credit condition on the account rather than a
   CLI-specific entitlement: `oz` runs the same agent harness as the Warp app and draws on the same
   account, plan, and credits. Check that `oz whoami` names the account holding the plan - if it
   does not, `oz logout && oz login` fixes it. Otherwise confirm the plan's AI credits are not
   spent, or store your own provider key -
   `warp --set-provider-api-key <openai|anthropic|google|grok>`, or `/api-keys` inside the TUI.
   Bring-your-own-key needs no paid Warp plan.
4. Confirm `oz --version` succeeds and `oz whoami` prints your user.
5. Work in, or point `--cd` at, the target git repository.

On macOS the CLI is distributed as a signed Developer ID binary; a first run may be held by
Gatekeeper until it is approved.

## Choose the model (optional)

Omit `--model` to use Warp's configured default. To pick another, choose an id from `oz model list`
and pass it verbatim. The relay accepts letters, digits, and `. _ : / -` only, so a value cannot be
mistaken for another `oz` flag.

## The loop

Run these five steps per task. Steps 1, 4, and 5 require judgment; 2 and 3 are mechanical.

### 1. Write the brief

Warp sees only the text you send plus what it can inspect in the workspace - no chat history or
shared context. Include the goal, current state, what to change, what to leave untouched, the
project's **actual** gates, and a report contract. Tell it not to commit. Keep one task per brief.
The brief is delivered as the `--prompt` value on argv, so it is visible in the host process list -
keep secrets out of it and reference workspace files instead. See
[references/writing-the-brief.md](references/writing-the-brief.md).

### 2. Dispatch

Use the bundled relay. It runs `oz agent run --output-format ndjson`, captures the event stream, and
writes `result.json`. (`<skill-dir>` is the installed folder containing this `SKILL.md`.)

```bash
node "<skill-dir>/scripts/relay.mjs" --brief brief.txt --cd /path/to/repo
# choose a model:                          add --model <id from oz model list>
# use an agent profile:                    add --profile <id>
# label the run:                           add --name <label>
# continue an existing conversation:       add --conversation <id> (delta brief only)
# base the run on a Warp skill:            add --skill <name|repo:name|org/repo:name>
# start MCP servers:                       add --mcp <path-or-inline-json>  (repeatable)
# suppress the workspace snapshot upload:  add --no-snapshot
# hard time limit (watchdog):              add --timeout 2h  (the 30m default suits short runs; implementation briefs routinely need 1-2h)
# see all options:                         node .../relay.mjs --help
```

The relay pins the workspace with both the child process's cwd and Warp's own `--cwd`. It writes
artifacts under the system temp dir by default and never commits. See
[references/dispatch-and-poll.md](references/dispatch-and-poll.md).

### 3. Wait for completion

The relay blocks until `oz` finishes. Run it with the orchestrator's background-command facility, or
background it in the shell and poll for `result.json`. A pre-run usage error exits 2 and writes no
result; a missing `oz` exits 127 and writes `status: "warp_unavailable"`.

Trust process state and the working tree over a progress display. Completion means the process
exited and `result.json` exists. Warp's report is the `finalMessage` field in `result.json` (also
printed on stdout between the report markers); the raw event stream is always in `events.jsonl`.

### 4. Review - do not trust the self-report

Treat Warp's final message and gate claims as claims:

- Re-run the project's gates yourself.
- Read the diff against the brief, starting with `touchedFiles`.
- Run relevant guard skills if installed.
- Round-trip migrations and grep for dangling references after removals or renames.

Because there is no read-only mode to fall back on, the diff is the **only** record you get - and it
records what git can see in the workspace afterward, not everything the run did. Dispatch from a
clean tree so the two are as close as they can be. See
[references/review-and-land.md](references/review-and-land.md).

### 5. Land it

The implementer edits the working tree; **the orchestrator commits.** Commit only after the gates
pass and the diff holds. If rework is needed, send a delta brief with `--conversation <id>` using the
`conversationId` from `result.json`, then review again.

## Autonomy and permissions

`oz agent run` has **no sandbox, no permission mode, and no read-only mode**. A headless run reads,
writes, edits, and executes commands with your own user permissions and never prompts. There is
nothing in the CLI to restrict that surface, so this relay ships no `--read-only` flag - offering one
would imply an enforcement that does not exist. The controls you actually have are:

1. **Scope by directory.** `--cd` pins the workspace, and the relay passes it to Warp's own `--cwd`.
   Treat this as *aim*, not a fence: on oz 0.2026.05.27 shell commands did run in the pinned
   workspace, but the agent's file tool resolved bare relative paths against `$HOME`. Name absolute
   paths in the brief - see [references/writing-the-brief.md](references/writing-the-brief.md).
2. **Review the diff.** `touchedFiles` is `git status --porcelain` taken after the run - post-run,
   git-visible worktree state, not a log of what the agent did. It cannot show an ignored file, an
   edit the run made and then reverted, or a write outside the repository (see item 1), and it
   carries anything that was already dirty before dispatch. Dispatch from a clean tree so those are
   the same set, and treat the diff as the best available record, not a complete one.
3. **Snapshot egress.** `--no-snapshot` forwards Warp's flag so the end-of-run workspace snapshot is
   not uploaded. Without it, the upload is Warp's default.

`--auto-approve` belongs to the interactive `warp` TUI and has no bearing on `oz agent run`.

## Authorization model

Delegation is something the human opts into. Once they have ("run this queue", "proceed"),
committing verified, gate-passing work is the agreed contract. Two limits remain: **surface, don't
absorb** (report Warp's design decisions, defensible-but-unasked turns, and non-blocking nitpicks)
and **stop for scope changes** (if correct completion needs going beyond the brief, ask instead of
expanding the mandate). See [references/review-and-land.md](references/review-and-land.md).

## References

- [references/writing-the-brief.md](references/writing-the-brief.md) - structure, report contract,
  real gates, argv delivery, and delta briefs.
- [references/dispatch-and-poll.md](references/dispatch-and-poll.md) - flags, artifacts,
  `result.json`, polling, and failure recovery.
- [references/review-and-land.md](references/review-and-land.md) - review checklist, commit
  boundary, and rework through Warp conversations.
- [references/multi-task-queues.md](references/multi-task-queues.md) - sequential queues,
  constraint carry-forward, progress tracking, and the final coherence pass.

## Limitations

- Docs-only import — executable `scripts/relay.mjs` not included; see upstream for full runtime. Requires `warp` CLI, Node 18+, git.
- Relay never commits — it only returns structured result JSON; you review and land the commit.

> Adapted from [amElnagdy/delegate-skills](https://github.com/amElnagdy/delegate-skills) (MIT) — docs-only, runtime not bundled.