# RoomKit

> RoomKit is a pure async Python library for building multi-channel conversation
> systems. It provides room-based abstractions for managing conversations across
> SMS, Email, Voice, WebSocket, AI, and other channels with pluggable storage,
> identity resolution, hooks, and realtime events.
> Python 3.12+, Pydantic 2.x, fully typed, zero required dependencies beyond Pydantic.

> Minimum version: 0.38.0 | Python 3.12+

RoomKit is a pure async Python library for building multi-channel conversation systems. It provides a **room-based abstraction** where conversations happen in rooms, participants communicate through channels, and hooks let you intercept and modify the flow.

## Key Concepts

- **Room** — A container for a conversation. Holds participants, channel bindings, and an ordered event timeline.
- **Channel** — A communication endpoint (SMS, Email, WhatsApp, Telegram, Discord, Buzz/Nostr, Voice, Video, Conference/SFU, AI, ACP coding agents, WebSocket, CLI, etc. — 23 `ChannelType` values). Registered once, attached to many rooms.
- **Hook** — A function that intercepts events at specific points in the pipeline (76 triggers). Can block, modify, or observe messages.
- **Event Router** — Broadcasts events to all channels attached to a room, with content transcoding per channel capabilities.
- **Identity Pipeline** — Maps external sender IDs to known participants with challenge/response flows.
- **Realtime Events** — Ephemeral events (typing, presence, reactions) that are not stored in history.

## Architecture at a Glance

RoomKit uses a hub-and-spoke model. The `RoomKit` orchestrator sits at the center. Channels connect on the edges. Messages flow inbound through a defined pipeline, get stored, then broadcast outbound to all attached channels.

```
Inbound Message
  -> InboundRoomRouter.route()       # Find target room
  -> Channel.handle_inbound()        # Parse -> RoomEvent
  -> IdentityResolver.resolve()      # Identify sender
  -> BEFORE_BROADCAST hooks          # Can block/modify
  -> Store event
  -> EventRouter.broadcast()         # Deliver to all channels
    -> Content transcoding           # Adapt per channel capabilities
    -> Rate limiting + retry
  -> AFTER_BROADCAST hooks           # Async side effects
```

## What You Can Build

- **AI-powered support agents** across SMS, WhatsApp, and web chat
- **Voice assistants** with real-time STT/TTS and interruption handling
- **Multi-agent pipelines** where specialized agents hand off conversations
- **Notification systems** that bridge channels (SMS + Email + push)
- **Speech-to-speech AI** with Gemini Live, OpenAI Realtime, or Grok
- **AI conference participants** — a bot that joins a multi-party SFU video conference (LiveKit) with STT/TTS, transcription, and recording
- **Coding-agent frontends** — drive Claude Code or any Agent Client Protocol agent from SMS, Telegram, or the terminal via `ACPChannel`
- **Nostr workspace bots** — chat and huddle voice over Buzz relay channels

## Design Principles

- **Async-first** — All I/O is async. No synchronous blocking.
- **Pluggable everything** — Storage, identity, routing, AI providers, voice backends — all swappable via ABCs with in-memory defaults.
- **Zero required deps** — Only `pydantic>=2.9`. Everything else is optional extras.
- **Type-safe** — Strict mypy, full type hints, Pydantic models throughout.
- **Python 3.12+** — Uses modern syntax (`X | None`, not `Optional[X]`).
---

## Basic Install

```bash
pip install roomkit
# or with uv
uv add roomkit
```

The core library has a single dependency: `pydantic>=2.9`.

## Optional Extras

RoomKit uses optional extras for provider-specific dependencies. Install only what you need:

### AI Providers

```bash
pip install roomkit[anthropic]    # Anthropic (Claude)
pip install roomkit[openai]       # OpenAI (GPT)
pip install roomkit[gemini]       # Google Gemini
pip install roomkit[mistral]      # Mistral AI
pip install roomkit[vllm]         # vLLM local inference (uses openai SDK)
pip install roomkit[azure]        # Azure OpenAI
```

### Voice — Speech-to-Text

```bash
pip install roomkit[deepgram]     # Deepgram STT (cloud)
pip install roomkit[sherpa-onnx]  # SherpaOnnx STT (local, offline)
pip install roomkit[gradium]      # Gradium STT
pip install roomkit[qwen-asr]     # Qwen3 ASR
```

### Voice — Text-to-Speech

```bash
pip install roomkit[elevenlabs]   # ElevenLabs TTS (cloud)
pip install roomkit[sherpa-onnx]  # SherpaOnnx TTS (local, offline)
pip install roomkit[gradium]      # Gradium TTS
pip install roomkit[qwen-tts]     # Qwen3 TTS
pip install roomkit[neutts]       # NeuTTS
```

### Voice — Backends

```bash
pip install roomkit[local-audio]  # Local mic/speaker (sounddevice + numpy)
pip install roomkit[fastrtc]      # FastRTC WebRTC backend
pip install roomkit[rtp]          # RTP backend
pip install roomkit[sip]          # SIP backend
pip install roomkit[webtransport] # WebTransport backend
```

### Voice — Pipeline

```bash
pip install roomkit[webrtc-aec]   # WebRTC echo cancellation
pip install roomkit[aicoustics]   # ai|coustics denoiser
pip install roomkit[smart-turn]   # ML-based turn detection
```

### Realtime Voice (Speech-to-Speech)

```bash
pip install roomkit[realtime-openai]   # OpenAI Realtime API
pip install roomkit[realtime-gemini]   # Google Gemini Live API
```

### Messaging Providers

```bash
pip install roomkit[twilio]            # Twilio SMS/RCS
pip install roomkit[telegram]          # Telegram Bot API
pip install roomkit[teams]             # Microsoft Teams (Bot Framework)
pip install roomkit[whatsapp-personal] # WhatsApp Personal (neonize)
pip install roomkit[websocket]         # WebSocket source
pip install roomkit[sse]               # Server-Sent Events source
```

### Storage & Infrastructure

```bash
pip install roomkit[postgres]          # PostgreSQL persistence (asyncpg)
pip install roomkit[mcp]               # Model Context Protocol tools
pip install roomkit[opentelemetry]     # OpenTelemetry tracing
```

### Meta Extras

```bash
pip install roomkit[providers]  # All AI + transport providers
pip install roomkit[sources]    # All event-driven sources
pip install roomkit[all]        # Everything installable together
pip install roomkit[dev]        # Development (test + lint + type check)
```

### Incompatible Extras

`qwen-tts`, `qwen-asr` and `neutts` each pin an exact `transformers` version
(4.57.3, 4.57.6 and 5.1.0), so they cannot be installed together. They are also
incompatible with `smart-turn`, whose floor is `transformers>=5.5` — the first
version without GHSA-fgcw-684q-jj6r — and therefore with `all`, which includes
`smart-turn`.

```bash
pip install roomkit[all]              # includes smart-turn
pip install roomkit[qwen-tts]         # in its own environment
```

`pyproject.toml` declares these conflicts, so an unsatisfiable combination fails
at resolution naming the two extras rather than backtracking onto an older
`transformers`.

## Environment Variables

Provider-specific API keys are passed via configuration objects, not environment variables. Example:

```python
from roomkit.providers.anthropic.config import AnthropicConfig

config = AnthropicConfig(api_key="sk-ant-...", model="claude-opus-5")
```

For voice providers, use lazy loaders to avoid import-time dependency checks:

```python
from roomkit.voice import get_deepgram_provider, get_deepgram_config

DeepgramSTTProvider = get_deepgram_provider()
DeepgramConfig = get_deepgram_config()

stt = DeepgramSTTProvider(DeepgramConfig(api_key="..."))
```

## Development Setup

```bash
git clone https://github.com/roomkit-live/roomkit
cd roomkit
uv sync --extra dev    # Install all dev dependencies
make all               # Run lint + typecheck + security + tests
```
---

A complete working example: two WebSocket users chatting with an AI assistant in a moderated room.

```python
from __future__ import annotations

import asyncio

from roomkit import (
    ChannelCategory,
    HookResult,
    HookTrigger,
    InboundMessage,
    RoomContext,
    RoomEvent,
    RoomKit,
    TextContent,
    WebSocketChannel,
)
from roomkit.channels.ai import AIChannel
from roomkit.providers.ai.mock import MockAIProvider


async def main() -> None:
    # 1. Create the framework instance
    kit = RoomKit()

    # 2. Create channels
    ws_alice = WebSocketChannel("ws-alice")
    ws_bob = WebSocketChannel("ws-bob")
    ai = AIChannel("ai-assistant", provider=MockAIProvider(responses=["Got it!"]))

    # 3. Register channels with the framework
    kit.register_channel(ws_alice)
    kit.register_channel(ws_bob)
    kit.register_channel(ai)

    # 4. Wire up receive callbacks (in production, WebSocket sends to clients)
    alice_inbox: list[RoomEvent] = []
    bob_inbox: list[RoomEvent] = []

    async def alice_recv(_conn: str, event: RoomEvent) -> None:
        alice_inbox.append(event)

    async def bob_recv(_conn: str, event: RoomEvent) -> None:
        bob_inbox.append(event)

    ws_alice.register_connection("alice-conn", alice_recv)
    ws_bob.register_connection("bob-conn", bob_recv)

    # 5. Create a room and attach channels
    await kit.create_room(room_id="demo-room")
    await kit.attach_channel("demo-room", "ws-alice")
    await kit.attach_channel("demo-room", "ws-bob")
    await kit.attach_channel(
        "demo-room", "ai-assistant", category=ChannelCategory.INTELLIGENCE
    )

    # 6. Add a BEFORE_BROADCAST hook for content moderation
    @kit.hook(HookTrigger.BEFORE_BROADCAST, name="profanity_filter")
    async def profanity_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
        if isinstance(event.content, TextContent) and "badword" in event.content.body:
            return HookResult.block("Message blocked by profanity filter")
        return HookResult.allow()

    # 7. Send messages through the inbound pipeline
    result = await kit.process_inbound(
        InboundMessage(
            channel_id="ws-alice",
            sender_id="alice",
            content=TextContent(body="Hello everyone!"),
        )
    )
    print(f"Alice sent 'Hello everyone!' -> blocked={result.blocked}")

    # Both Bob and the AI receive Alice's message.
    # The AI responds with "Got it!" which is broadcast to Alice and Bob.

    # 8. Query stored conversation history
    events = await kit.store.list_events("demo-room")
    for ev in events:
        print(f"  [{ev.source.channel_id}] {ev.content.body}")


if __name__ == "__main__":
    asyncio.run(main())
```

## What Just Happened

1. **RoomKit()** created the framework with in-memory defaults (store, locks, realtime).
2. **Channels** were registered globally, then **attached** to a room.
3. AI channel was attached with `category=ChannelCategory.INTELLIGENCE` — it receives messages and generates responses.
4. A **BEFORE_BROADCAST** hook runs synchronously before every broadcast. It can `block()`, `allow()`, or `modify()` events.
5. **process_inbound()** ran the full pipeline: route -> parse -> identity -> hooks -> store -> broadcast.
6. The AI's response was automatically routed back through the same pipeline (with chain depth tracking to prevent loops).

## Core Pattern

Every RoomKit application follows this pattern:

```python
from roomkit import RoomKit

# 1. Create framework
kit = RoomKit()

# 2. Register channels
kit.register_channel(channel)

# 3. Create rooms and attach channels
await kit.create_room(room_id="my-room")
await kit.attach_channel("my-room", "channel-id")

# 4. Process inbound messages
await kit.process_inbound(InboundMessage(...))
```

## Next Steps

- Add hooks for moderation, logging, or analytics (see Hooks)
- Configure AI providers for real LLM responses (see AI Channels)
- Add voice with STT/TTS (see Voice Channels)
- Set up multi-agent orchestration (see Orchestration)
- Deploy with PostgreSQL storage (see Storage)
---

## Hub-and-Spoke Model

RoomKit uses a hub-and-spoke architecture. The `RoomKit` class is the central hub. Channels are the spokes. Messages flow in through channels, get processed by the hub, then broadcast out to all channels attached to the room.

```
                    +------------------+
                    |     RoomKit      |
                    |  (orchestrator)  |
                    +--------+---------+
                             |
         +-------------------+-------------------+
         |         |         |         |         |
     +---+---+ +---+---+ +--+---+ +---+---+ +---+---+
     |  SMS  | | Email | | Voice| |  AI   | |  WS   |
     +-------+ +-------+ +------+ +-------+ +-------+
```

## Inbound Pipeline

Every inbound message follows an immutable pipeline order:

```
1. InboundRoomRouter.route()        # Find target room by channel binding
2. Channel.handle_inbound()         # Parse external format -> RoomEvent
3. IdentityResolver.resolve()       # Map sender_id -> participant
4. Identity hooks                   # ON_IDENTITY_AMBIGUOUS / ON_IDENTITY_UNKNOWN
5. Room lock acquired               # Per-room atomic processing
6. Idempotency check                # Deduplicate by provider_message_id
7. BEFORE_BROADCAST hooks           # Sync: can block or modify the event
8. Store event + update counters    # Persist to ConversationStore
9. EventRouter.broadcast()          # Deliver to all attached channels
   -> Content transcoding           # Adapt content per channel capabilities
   -> Rate limiting                 # TokenBucketRateLimiter per binding
   -> Retry with backoff            # RetryPolicy per binding
10. AFTER_BROADCAST hooks           # Async: fire-and-forget side effects
11. Room lock released
```

This order is defined in the RFC (Section 10.1) and must not be reordered.

## Channel Categories

Channels have two categories:

- **TRANSPORT** — Push messages to users (SMS, Email, WebSocket, Voice). Default category.
- **INTELLIGENCE** — Generate responses (AI, agents). Receives broadcasts, responds through the inbound pipeline.

## Channel-to-AI Message Flow (Reentry Loop)

When you send a message from a transport channel, it flows to the AI and back automatically:

```
1. Transport channel (SMS/WS/Voice) → kit.process_inbound(InboundMessage)
2. Inbound pipeline: route → parse → identity → hooks → store
3. EventRouter.broadcast() → delivers event to ALL attached channels
   ├── Transport channels: note delivery (no response)
   └── AI channel: calls LLM provider → generates response
       └── Returns ChannelOutput(response_events=[RoomEvent])
4. REENTRY LOOP: AI response re-enters as new inbound event
   ├── BEFORE_BROADCAST hooks run again (ConversationRouter stamps routing)
   ├── Event stored in timeline
   ├── Broadcast to all channels again
   │   ├── Transport channels: DELIVER the AI response to users
   │   └── Other AI channels: see the response (may generate follow-up)
   └── If follow-up AI responses exist → loop again (chain depth checked)
5. Chain depth limit (default max=5) → stops AI-to-AI infinite loops
6. AFTER_BROADCAST hooks fire (async side effects)
```

**Key insight**: The "pipeline" is a reentry loop — not a linear chain. AI responses go back through the same BEFORE_BROADCAST hooks, content transcoding, and broadcast cycle as user messages.

### Minimal Channel→AI Example

```python
from roomkit import RoomKit, AIChannel, WebSocketChannel, ChannelCategory, InboundMessage, TextContent
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig

kit = RoomKit()

# Transport channel (user-facing)
ws = WebSocketChannel("ws-user")
ws.register_connection("conn-1", on_recv)
kit.register_channel(ws)

# Intelligence channel (AI)
ai = AIChannel("ai", provider=AnthropicAIProvider(AnthropicConfig(
    api_key="sk-ant-...", model="claude-opus-5",
)))
kit.register_channel(ai)

# Room wires them together
await kit.create_room(room_id="chat")
await kit.attach_channel("chat", "ws-user")  # TRANSPORT (default)
await kit.attach_channel("chat", "ai", category=ChannelCategory.INTELLIGENCE)

# User sends message → AI responds → response delivered to WebSocket
await kit.process_inbound(
    InboundMessage(channel_id="ws-user", sender_id="user", content=TextContent(body="Hi!"))
)
```

### Pipeline vs Pipeline

RoomKit uses "Pipeline" in two contexts:

| Term | What It Is | Where |
|------|-----------|-------|
| **Inbound processing pipeline** | The 11-step message processing flow (route → parse → identity → hooks → store → broadcast) | `core/mixins/inbound_locked.py` |
| **Pipeline orchestration strategy** | A linear agent chain (triage → handler → resolver) for multi-agent handoffs | `from roomkit import Pipeline` |

The inbound pipeline processes every message. The Pipeline strategy controls which agent handles each turn.

## Pluggable Components

Every core component follows the ABC + default pattern:

| Component | ABC | Default | Purpose |
|-----------|-----|---------|---------|
| `ConversationStore` | `store/base.py` | `InMemoryStore` | Room, event, participant persistence |
| `RoomLockManager` | `core/locks.py` | `InMemoryLockManager` | Per-room atomic processing |
| `RealtimeBackend` | `realtime/base.py` | `InMemoryRealtime` | Ephemeral events (typing, presence) |
| `IdentityResolver` | `identity/base.py` | `None` (disabled) | Sender -> participant mapping |
| `InboundRoomRouter` | `core/inbound_router.py` | `DefaultInboundRoomRouter` | Route messages to rooms |

Replace any component at construction:

```python
from roomkit import RoomKit
from roomkit.store.postgres import PostgresStore

kit = RoomKit(store=PostgresStore("postgresql://..."))
```

## Room Lifecycle

```
ACTIVE -> PAUSED -> CLOSED -> ARCHIVED
           ^          |
           +----------+  (can close from paused)
```

- **ACTIVE** — Accepting messages, all channels active.
- **PAUSED** — Messages queued, channels paused. Auto-transition via `inactive_after_seconds`.
- **CLOSED** — No new messages. Auto-transition via `closed_after_seconds`.
- **ARCHIVED** — Terminal state, read-only.

## Event Model

Every message becomes a `RoomEvent` with:

- **content** — One of 11 content types (TextContent, RichContent, MediaContent, AudioContent, VideoContent, LocationContent, CompositeContent, TemplateContent, EditContent, DeleteContent, SystemContent)
- **source** — Who sent it (channel_id, participant_id, direction)
- **index** — Sequential, monotonically increasing per room
- **metadata** — Arbitrary key-value data

## Voice Architecture

The voice subsystem has three layers:

1. **VoiceBackend** — Pure audio transport (mic, SIP, RTP, WebRTC). No speech detection.
2. **AudioPipeline** — Processes audio frames: Resampler -> Recorder -> AEC -> AGC -> Denoiser -> VAD -> Diarization.
3. **VoiceChannel** — Wires backend -> pipeline -> STT/TTS, handles interruption and turn detection.

```
Inbound:   Backend -> [Resampler] -> [Recorder] -> [AEC] -> [AGC] -> [Denoiser] -> VAD -> [Diarization] + [DTMF]
Outbound:  TTS -> [PostProcessors] -> [Recorder] -> AEC.feed_reference -> [Resampler] -> Backend
```

## Realtime Voice Architecture

Speech-to-speech AI (Gemini Live, OpenAI Realtime) bypasses STT/TTS entirely:

- **RealtimeVoiceProvider** — Handles the AI model connection (WebSocket to Gemini/OpenAI).
- **RealtimeAudioTransport** — Handles browser/client audio (WebSocket or WebRTC).
- **RealtimeVoiceChannel** — Bridges transport <-> provider with tool calling support.
---

## Creating Rooms

```python
from roomkit import RoomKit

kit = RoomKit()

# Auto-generated ID
room = await kit.create_room()

# Explicit ID with metadata
room = await kit.create_room(
    room_id="support-123",
    metadata={"topic": "billing", "priority": "high"},
)
```

## Room Lifecycle

```python
from roomkit import RoomKit

kit = RoomKit()

# Create room
room = await kit.create_room(room_id="session-1")

# Update metadata
await kit.update_room_metadata("session-1", {"status": "escalated"})

# Close a room
await kit.close_room("session-1")

# Check timers (call periodically, e.g. every 60s)
transitioned = await kit.check_all_timers()
```

## Channel Types

The `ChannelType` enum defines 23 values: `SMS`, `MMS`, `RCS`, `EMAIL`, `WHATSAPP`, `WHATSAPP_PERSONAL`, `WEBSOCKET`, `AI`, `VOICE`, `REALTIME_VOICE`, `REALTIME_AUDIO_VIDEO`, `PUSH`, `MESSENGER`, `TELEGRAM`, `TEAMS`, `DISCORD`, `BUZZ`, `WEBHOOK`, `VIDEO`, `AUDIO_VIDEO`, `CONFERENCE`, `CLI`, `SYSTEM`. RoomKit ships with these factories and channel classes:

| Channel | Factory / Class | Category | Use Case |
|---------|----------------|----------|----------|
| SMS | `SMSChannel(id, provider=...)` | TRANSPORT | Text messages via Twilio, Telnyx, Sinch, VoiceMeUp |
| RCS | `RCSChannel(id, provider=..., fallback=True)` | TRANSPORT | Rich messaging via Twilio, Telnyx |
| Email | `EmailChannel(id, provider=...)` | TRANSPORT | Email via ElasticEmail, SendGrid |
| WhatsApp | `WhatsAppChannel(id, provider=...)` | TRANSPORT | WhatsApp Business API |
| WhatsApp Personal | `WhatsAppPersonalChannel(id, provider=...)` | TRANSPORT | WhatsApp via neonize |
| Messenger | `MessengerChannel(id, provider=...)` | TRANSPORT | Facebook Messenger |
| Telegram | `TelegramChannel(id, provider=...)` | TRANSPORT | Telegram Bot API |
| Teams | `TeamsChannel(id, provider=...)` | TRANSPORT | Microsoft Teams Bot Framework |
| Discord | `DiscordChannel(id, provider=...)` | TRANSPORT | Discord bot (recipient key `discord_channel_id`) |
| Buzz | `BuzzChannel(id, provider=...)` | TRANSPORT | Nostr relay channels (recipient key `buzz_channel_id`) |
| Webhook | `HTTPChannel(id, provider=...)` | TRANSPORT | Generic HTTP webhook (`ChannelType.WEBHOOK`) |
| WebSocket | `WebSocketChannel(id)` | TRANSPORT | Real-time bidirectional |
| CLI | `CLIChannel(id)` | TRANSPORT | Interactive terminal (stdin/stdout REPL) |
| AI | `AIChannel(id, provider=...)` | INTELLIGENCE | LLM responses |
| Agent | `Agent(id, provider=...)` | INTELLIGENCE | Agent with tools + greeting |
| ACP | `ACPChannel(id, command, cwd=...)` | INTELLIGENCE | External coding agent via Agent Client Protocol (`channel_type` is `AI`; one session per room) |
| Voice | `VoiceChannel(id, stt=..., tts=..., backend=...)` | TRANSPORT | Real-time audio |
| Realtime Voice | `RealtimeVoiceChannel(id, provider=...)` | TRANSPORT | Speech-to-speech AI |
| Video | `VideoChannel(id, backend=..., vision=...)` | TRANSPORT | Video track with vision analysis |
| Audio+Video | `AudioVideoChannel(id, backend=..., stt=..., tts=...)` | TRANSPORT | Combined voice + video |
| Realtime Audio+Video | `RealtimeAudioVideoChannel(id, provider=..., transport=...)` | TRANSPORT | Speech-to-speech AI with video |
| Conference | `ConferenceChannel(id, backend=...)` | TRANSPORT | Multi-party SFU conference with AI bot (LiveKit or mock backend) |

`MMS`, `PUSH`, and `SYSTEM` have no factory: `MMS` is set automatically when an SMS channel carries media, `SYSTEM` marks framework-generated events, and `PUSH` is reserved.

## Registering and Attaching Channels

Channels are registered globally, then attached to specific rooms:

```python
from roomkit import RoomKit, SMSChannel, ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig

kit = RoomKit()

# Create and register channels
sms = SMSChannel("sms-main", provider=TwilioSMSProvider(TwilioConfig(
    account_sid="AC...",
    auth_token="...",
    from_number="+1234567890",
)))
ai = AIChannel("ai-agent", provider=AnthropicAIProvider(AnthropicConfig(
    api_key="sk-ant-...",
    model="claude-opus-5",
)))

kit.register_channel(sms)
kit.register_channel(ai)

# Create room and attach
await kit.create_room(room_id="support-room")
await kit.attach_channel("support-room", "sms-main")
await kit.attach_channel("support-room", "ai-agent", category=ChannelCategory.INTELLIGENCE)
```

## Channel Access Levels

Control what a channel can do in a room:

```python
from roomkit import Access

# Default: read and write
await kit.attach_channel("room", "channel", access=Access.READ_WRITE)

# Read only: receives messages but cannot send
await kit.attach_channel("room", "channel", access=Access.READ_ONLY)

# Write only: sends messages but doesn't receive broadcasts
await kit.attach_channel("room", "channel", access=Access.WRITE_ONLY)

# None: temporarily disabled
await kit.set_access("room", "channel", Access.NONE)
```

## Muting Channels

Muting suppresses a channel's output (AI responses) without detaching it. Side effects (tasks, observations) still fire.

```python
# Mute AI output
await kit.mute("room", "ai-agent")

# Unmute
await kit.unmute("room", "ai-agent")

# Mute only the output direction (AI won't respond, but still sees messages)
await kit.mute_output("room", "ai-agent")
await kit.unmute_output("room", "ai-agent")
```

## Per-Room Channel Configuration

Pass metadata when attaching to customize per-room behavior:

```python
await kit.attach_channel(
    "weather-room",
    "ai-agent",
    category=ChannelCategory.INTELLIGENCE,
    metadata={
        "system_prompt": "You are a weather assistant.",
        "temperature": 0.3,
        "tools": [
            {
                "name": "get_weather",
                "description": "Get current weather for a city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string"},
                    },
                    "required": ["city"],
                },
            },
        ],
    },
)
```

## Content Types

Events carry typed content. RoomKit supports 11 content types, all discriminated unions on the `type` field:

| Type | `type` Literal | Key Fields | Use Case |
|------|---------------|-----------|----------|
| `TextContent` | `"text"` | `body`, `language` | Plain text messages |
| `RichContent` | `"rich"` | `body`, `format` (html/markdown), `plain_text`, `buttons`, `cards`, `quick_replies` | Formatted messages with UI elements |
| `MediaContent` | `"media"` | `url`, `mime_type`, `filename`, `size_bytes`, `caption` | Images, documents, files |
| `AudioContent` | `"audio"` | `url`, `mime_type`, `duration_seconds`, `transcript` | Voice messages |
| `VideoContent` | `"video"` | `url`, `mime_type`, `duration_seconds`, `thumbnail_url` | Video messages |
| `LocationContent` | `"location"` | `latitude`, `longitude`, `label`, `address` | Geographic coordinates |
| `CompositeContent` | `"composite"` | `parts` (list of EventContent, max depth 5) | Multi-part messages |
| `TemplateContent` | `"template"` | `template_id`, `language`, `parameters`, `body` | WhatsApp Business / RCS templates |
| `SystemContent` | `"system"` | `body`, `code`, `data` (dict) | System-generated events |
| `EditContent` | `"edit"` | `target_event_id`, `new_content`, `edit_source` | Message edits |
| `DeleteContent` | `"delete"` | `target_event_id`, `delete_type`, `reason` | Message deletion |

```python
from roomkit.models.event import (
    TextContent, RichContent, MediaContent, AudioContent, VideoContent,
    LocationContent, CompositeContent, TemplateContent, SystemContent,
    EditContent, DeleteContent,
)

# Plain text
TextContent(body="Hello!")

# Rich content with buttons and Markdown
RichContent(body="Choose:", format="markdown", plain_text="Choose:", buttons=[{"text": "A", "payload": "a"}])

# Media, Audio, Location, Template
MediaContent(url="https://example.com/image.png", mime_type="image/png", caption="Photo")
AudioContent(url="https://example.com/voice.ogg", mime_type="audio/ogg", transcript="Hello there")
LocationContent(latitude=40.7128, longitude=-74.0060, label="NYC Office")
TemplateContent(template_id="order_confirm", language="en", parameters={"1": "ORD-123"})

# Multi-part (max depth 5)
CompositeContent(parts=[
    TextContent(body="Here's the photo:"),
    MediaContent(url="https://example.com/photo.jpg", mime_type="image/jpeg"),
])

# Edit and Delete
EditContent(target_event_id="evt-abc", new_content=TextContent(body="Corrected text"))
DeleteContent(target_event_id="evt-abc", delete_type="sender", reason="User retracted")
```

## Content Transcoding

When broadcasting events, the EventRouter automatically **transcodes** content to match each target channel's capabilities. A channel that only supports TEXT will receive a text fallback of a RichContent message.

### How Transcoding Works

1. Event broadcast starts → EventRouter iterates over all target channel bindings
2. For each target, call `ContentTranscoder.transcode(content, source_binding, target_binding)`
3. Transcoder checks `target_binding.capabilities.media_types` to decide if content passes through or needs conversion
4. If transcode returns `None` → delivery is skipped for that channel

### Channel Capabilities

Each channel binding declares what content types it supports via `ChannelCapabilities`:

```python
from roomkit.models.channel import ChannelCapabilities
from roomkit.models.enums import ChannelMediaType

# SMS channel: text only, 160 char limit
sms_caps = ChannelCapabilities(
    media_types=[ChannelMediaType.TEXT],
    max_length=160,
)

# WebSocket channel: rich content, media, audio, video
ws_caps = ChannelCapabilities(
    media_types=[
        ChannelMediaType.TEXT, ChannelMediaType.RICH, ChannelMediaType.MEDIA,
        ChannelMediaType.AUDIO, ChannelMediaType.VIDEO, ChannelMediaType.LOCATION,
    ],
    supports_edit=True,
    supports_delete=True,
)
```

`ChannelMediaType` enum: `TEXT`, `RICH`, `MEDIA`, `AUDIO`, `VIDEO`, `LOCATION`, `TEMPLATE`.

### Default Fallback Chain

| Content Type | If Target Supports It | Fallback |
|-------------|----------------------|----------|
| `TextContent` | Always passes through | — |
| `RichContent` | RICH in media_types → pass | `TextContent(plain_text or body)` |
| `MediaContent` | MEDIA in media_types → pass | `TextContent("[Media: {caption or filename or url}]")` |
| `AudioContent` | AUDIO in media_types → pass | `TextContent(transcript)` or `"[Voice message: {url}]"` |
| `VideoContent` | VIDEO in media_types → pass | `TextContent("[Video: {url}]")` |
| `LocationContent` | LOCATION in media_types → pass | `TextContent("[Location: {label} ({lat}, {lon})]")` |
| `CompositeContent` | Recursive transcode of all parts | If all parts become text → flatten to single TextContent |
| `TemplateContent` | TEMPLATE in media_types → pass | `TextContent(body or "[Template: {id}]")` |
| `EditContent` | `supports_edit` → pass | Transcode `new_content` + prefix "Correction:" |
| `DeleteContent` | `supports_delete` → pass | `TextContent("[Message deleted]")` |

### Multichannel Transcoding Example

A single event gets adapted differently for each channel:

```python
# User sends a rich message with an image from WebSocket
content = CompositeContent(parts=[
    RichContent(body="**Check this out!**", format="markdown", plain_text="Check this out!"),
    MediaContent(url="https://example.com/photo.jpg", mime_type="image/jpeg", caption="Sunset"),
])

# WebSocket channel: receives CompositeContent as-is (supports RICH + MEDIA)
# SMS channel: receives TextContent("Check this out!\n[Media: Sunset]")
# Voice channel (TTS): receives TextContent("Check this out!\n[Media: Sunset]")
```

### Custom Content Transcoder

Implement the `ContentTranscoder` ABC to customize how content is adapted:

```python
from roomkit.core.router import ContentTranscoder
from roomkit.models.channel import ChannelBinding
from roomkit.models.event import EventContent, TextContent, MediaContent
from roomkit.models.enums import ChannelMediaType

class MyTranscoder(ContentTranscoder):
    async def transcode(
        self,
        content: EventContent,
        source_binding: ChannelBinding,
        target_binding: ChannelBinding,
    ) -> EventContent | None:
        target_types = target_binding.capabilities.media_types

        # Custom: convert images to descriptive text for voice channels
        if isinstance(content, MediaContent) and content.mime_type.startswith("image/"):
            if ChannelMediaType.MEDIA not in target_types:
                caption = content.caption or "an image"
                return TextContent(body=f"[Image received: {caption}]")

        # Custom: truncate long text for SMS
        if isinstance(content, TextContent):
            max_len = target_binding.capabilities.max_length
            if max_len and len(content.body) > max_len:
                return TextContent(body=content.body[: max_len - 3] + "...")

        return content

# Override the default transcoder on the RoomKit instance
kit = RoomKit()
kit._transcoder = MyTranscoder()
```

## Detaching Channels

```python
await kit.detach_channel("room", "channel-id")
```

## Binding Metadata Updates

```python
await kit.update_binding_metadata("room", "ai-agent", {"temperature": 0.9})
```

## Querying Rooms

```python
# Get a room
room = await kit.get_room("room-id")

# List bindings
bindings = await kit.store.list_bindings("room-id")

# List participants
participants = await kit.store.list_participants("room-id")

# Query event timeline
events = await kit.get_timeline("room-id", offset=0, limit=50)
```

A participant is one shared record per `(room_id, participant_id)`, even when
`connected_via` names several channels. Its `status` is therefore global to the
record, not per-channel presence. When two channel memberships have independent
connect/disconnect lifecycles, give them distinct participant ids and correlate
them through `identity_id`.
---

Hooks intercept events at specific points in the processing pipeline. They can block messages, modify content, trigger side effects, or observe the conversation.

## Hook Basics

```python
from roomkit import RoomKit, HookTrigger, HookExecution, HookResult, RoomEvent, RoomContext

kit = RoomKit()

# Sync hook: runs BEFORE broadcast, can block or modify
@kit.hook(HookTrigger.BEFORE_BROADCAST)
async def content_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
    if "spam" in event.content.body.lower():
        return HookResult.block("Spam detected")
    return HookResult.allow()

# Async hook: runs AFTER broadcast, fire-and-forget
@kit.hook(HookTrigger.AFTER_BROADCAST, execution=HookExecution.ASYNC)
async def log_event(event: RoomEvent, ctx: RoomContext) -> None:
    await analytics.track("message", {"room": event.room_id})
```

## HookResult

Sync hooks (BEFORE_BROADCAST) must return a `HookResult`:

```python
from roomkit import HookResult, TextContent

# Allow the event to proceed
HookResult.allow()

# Block the event with a reason
HookResult.block("Contains prohibited content")

# Modify the event before broadcast
modified = event.model_copy(update={"content": TextContent(body="[REDACTED]")})
HookResult.modify(modified)
```

### What happens when a hook fails

A hook that does not produce a usable result — it raises, it exceeds its
timeout, it returns something that is not a `HookResult`, or it returns a
`modify` whose payload is the wrong type for the trigger — is treated as
**allow**, and the error is logged. A broken hook must not be able to take a
room down, so the event goes through.

**Two triggers invert that**, because their payload is content a hook may
exist to withhold:

| Trigger | On hook failure |
|---|---|
| `BEFORE_TTS` | **Blocked** — the text is not synthesised |
| `ON_TRANSCRIPTION` | **Blocked** — the transcript is not published |
| every other sync trigger | Allowed, error logged |

All four failure modes block on those two, not just exceptions: a rule that
covered exceptions but allowed timeouts would leak through the timeout.

The consequence for `BEFORE_BROADCAST` is worth being explicit about, since
it is the trigger most often used for moderation: **a moderation hook that
crashes lets the content through.** That is deliberate (RFC §9.3), not an
oversight. Two levers exist if you need more, and their exact scope matters:

- The `hook_error` framework event is emitted by the **inbound pipeline's**
  `BEFORE_BROADCAST` pass only. Other sync-hook passes (`ON_TRANSCRIPTION`,
  `BEFORE_TTS`, the re-entry path) and every async trigger report failures
  in the log, not through `hook_error` — do not build monitoring for those
  on this event.
- To make additional triggers fail closed, extend the set the engine reads
  through its instance — RoomKit builds its own `HookEngine`, so there is
  no constructor to subclass into:

  ```python
  kit.hook_engine.FAIL_CLOSED_TRIGGERS = kit.hook_engine.FAIL_CLOSED_TRIGGERS | {
      HookTrigger.BEFORE_BROADCAST,
  }
  ```

  Weigh it first: fail-closed moderation means an outage in your hook is an
  outage of the room.

## Hook Priority

Lower priority numbers run first. Default is 0.

```python
# Runs first (priority=0)
@kit.hook(HookTrigger.BEFORE_BROADCAST, name="profanity_filter", priority=0)
async def profanity_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
    blocked_words = {"badword", "spam", "scam"}
    if isinstance(event.content, TextContent):
        words = set(event.content.body.lower().split())
        if words & blocked_words:
            return HookResult.block(f"Blocked: {words & blocked_words}")
    return HookResult.allow()

# Runs second (priority=1)
@kit.hook(HookTrigger.BEFORE_BROADCAST, name="pii_redactor", priority=1)
async def pii_redactor(event: RoomEvent, ctx: RoomContext) -> HookResult:
    import re
    if isinstance(event.content, TextContent):
        redacted = re.sub(
            r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "[REDACTED]", event.content.body
        )
        if redacted != event.content.body:
            modified = event.model_copy(update={"content": TextContent(body=redacted)})
            return HookResult.modify(modified)
    return HookResult.allow()
```

## Hook Filters

Filter hooks by channel type, channel ID, or direction:

```python
from roomkit.models.enums import ChannelType, ChannelDirection

@kit.hook(
    HookTrigger.AFTER_BROADCAST,
    execution=HookExecution.ASYNC,
    channel_types={ChannelType.SMS},
    directions={ChannelDirection.INBOUND},
    priority=10,
)
async def sms_audit(event: RoomEvent, ctx: RoomContext) -> None:
    await audit_log.record(event)
```

## Room-Scoped Hooks

Add hooks to specific rooms instead of globally:

```python
from roomkit import HookExecution

await kit.add_room_hook(
    room_id="vip-room",
    trigger=HookTrigger.BEFORE_BROADCAST,
    execution=HookExecution.SYNC,
    fn=my_hook_function,
    name="vip_filter",
)

# Remove later
await kit.remove_room_hook("vip-room", "vip_filter")
```

## Complete Hook Trigger Reference

All 76 `HookTrigger` enum values (`src/roomkit/models/enums.py`). **Execution** is how
the engine invokes the trigger: SYNC triggers run through the sequential,
priority-ordered sync pipeline (can block/modify); ASYNC triggers run
concurrently, fire-and-forget, errors logged. Registration mode is forgiving in
both directions: ASYNC-registered hooks on a SYNC trigger fire as observers
after the sync pass; SYNC-registered hooks (the default) on an ASYNC trigger
fire like any other observer. Triggers marked *Reserved* exist in the enum but
are not fired by any built-in code.

### Event Pipeline

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_BROADCAST` | SYNC | `(event, ctx) -> HookResult` | Before event is stored and broadcast. Can block/modify. |
| `AFTER_BROADCAST` | ASYNC | `(event, ctx) -> None` | After event is broadcast. Fire-and-forget side effects. |

### Event Mutation (edit/delete)

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_EVENT_UPDATED` | ASYNC | `(event, ctx) -> None` | A persisted event's stored state changed — inbound `EditContent` or `kit.update_event()`. Payload is the updated event (`metadata.edited=True` on the edit path). |
| `ON_EVENT_DELETED` | ASYNC | `(event, ctx) -> None` | A persisted event was deleted — inbound `DeleteContent` (soft, `metadata.deleted=True`) or `kit.delete_event()` (hard; payload is the pre-delete snapshot). |

### Channel Lifecycle

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_CHANNEL_ATTACHED` | ASYNC | `(event, ctx) -> None` | Channel was attached to a room |
| `ON_CHANNEL_DETACHED` | ASYNC | `(event, ctx) -> None` | Channel was detached from a room |
| `ON_CHANNEL_MUTED` | ASYNC | `(event, ctx) -> None` | Channel was muted in a room |
| `ON_CHANNEL_UNMUTED` | ASYNC | `(event, ctx) -> None` | Channel was unmuted in a room |

### Room Lifecycle

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_ROOM_CREATED` | ASYNC | `(event, ctx) -> None` | Room was created |
| `ON_ROOM_PAUSED` | ASYNC | `(event, ctx) -> None` | Room was paused |
| `ON_ROOM_CLOSED` | ASYNC | `(event, ctx) -> None` | Room was closed |

### Identity

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_IDENTITY_AMBIGUOUS` | SYNC | `(event, ctx) -> IdentityHookResult` | Multiple identity matches found |
| `ON_IDENTITY_UNKNOWN` | SYNC | `(event, ctx) -> IdentityHookResult` | No identity match found |
| `ON_PARTICIPANT_IDENTIFIED` | ASYNC | `(event, ctx) -> None` | Participant was successfully identified |

### Membership

Synthetic system events (`SystemContent`, `visibility=INTERNAL`), fired by the
member management API.

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PARTICIPANT_JOINED` | ASYNC | `(event, ctx) -> None` | Member added via `kit.add_member()`. `content.data`: `participant_id`, `identity_id`. |
| `ON_PARTICIPANT_LEFT` | ASYNC | `(event, ctx) -> None` | Member removed via `kit.remove_member()` — soft status flip to LEFT/BANNED. `content.data`: `participant_id`, `status`. |
| `ON_PARTICIPANT_UPDATED` | ASYNC | `(event, ctx) -> None` | Member renamed via `kit.rename_member()` (display name only; identity never changes). |

### Delivery

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_DELIVERY_STATUS` | ASYNC | `(DeliveryStatus, ctx) -> None` | Provider delivery receipt dispatched by `kit.process_delivery_status()` / `kit.process_webhook()`. The `@kit.on_delivery_status` decorator registers a `(status)`-only callback on this trigger. |
| `BEFORE_DELIVER` | ASYNC | `(event, ctx) -> None` | Before a proactive `kit.deliver()` strategy executes (in-process and worker paths). Payload is a synthetic INTERNAL system-source event describing the delivery. Observational — invoked through the async pipeline, cannot block (RFC §9 lists it SYNC; code invokes it ASYNC). |
| `AFTER_DELIVER` | ASYNC | `(event, ctx) -> None` | After the delivery strategy completes. Same payload shape with status DELIVERED/FAILED and `metadata.error` set on failure. |

### Side Effects

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_TASK_CREATED` | ASYNC | `(event, ctx) -> None` | A `Task` returned in `HookResult.tasks` was persisted. Payload event has type `TASK_CREATED` and `metadata.task_id`/`task_title`. Fires once per persisted task. |
| `ON_ERROR` | ASYNC | `(event, ctx) -> None` | Error during processing (every provider/inference failure path funnels here). Payload event carries `metadata.error`, `error_type`, `error_category`. |

### Session Lifecycle

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_SESSION_STARTED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | A session began: voice session bound, realtime session opened, conference bot connected (`event.session` = the session/bot), or first inbound on an auto-created text room. On the inbound path, internal (`_`-prefixed) hooks are awaited (greeting gate ordering) and user hooks fire in the background. Auto-greeting is an internal hook on this trigger. |

### Voice

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_SPEECH_START` | ASYNC | `(VoiceSession, ctx) -> None` | VAD detected speech start (VoiceChannel, RealtimeVoiceChannel). Conference lanes fire it per track with a synthetic system event. |
| `ON_SPEECH_END` | ASYNC | `(VoiceSession, ctx) -> None` | VAD detected speech end. Same per-lane conference behavior. |
| `ON_TRANSCRIPTION` | SYNC | `(event, ctx) -> HookResult` | STT produced a final transcript. Can block/modify; **fails closed** — a hook that raises/times out blocks publication. Payload: `TranscriptionEvent` (VoiceChannel), `RealtimeTranscriptionEvent` (RealtimeVoiceChannel), `ConferenceTranscription` (ConferenceChannel). Modify by returning the same type (voice/realtime also accept a plain `str`) with the new text. |
| `ON_PARTIAL_TRANSCRIPTION` | ASYNC | `(PartialTranscriptionEvent, ctx) -> None` | Streaming interim STT result. Hot path — skipped entirely when no hooks are registered. |
| `BEFORE_TTS` | SYNC | `(text: str, ctx) -> HookResult` | Before text is synthesized. Payload is the text `str`; modify with a `str`. Can block; **fails closed**. |
| `AFTER_TTS` | ASYNC | `(text: str, ctx) -> None` | After TTS audio was sent. Payload is the final synthesized text. |

### Voice Pipeline

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_VAD_SILENCE` | ASYNC | `(VADSilenceEvent, ctx) -> None` | VAD detected silence |
| `ON_VAD_AUDIO_LEVEL` | ASYNC | `(VADAudioLevelEvent, ctx) -> None` | Audio level update from VAD (high-frequency; telemetry-suppressed) |
| `ON_SPEAKER_CHANGE` | ASYNC | `(SpeakerChangeEvent, ctx) -> None` | Diarization detected speaker change |
| `ON_BARGE_IN` | ASYNC | `(BargeInEvent, ctx) -> None` | User interrupted TTS playback; carries `interrupted_text`, `audio_position_ms` |
| `ON_TTS_CANCELLED` | ASYNC | `(TTSCancelledEvent, ctx) -> None` | TTS playback was cancelled (barge-in or explicit interrupt) |
| `ON_DTMF` | ASYNC | `(DTMFDetectedEvent, ctx) -> None` | DTMF tone detected |
| `ON_TURN_COMPLETE` | ASYNC | `(TurnCompleteEvent, ctx) -> None` | Turn detector says turn is complete; carries combined text + confidence |
| `ON_TURN_INCOMPLETE` | ASYNC | `(TurnIncompleteEvent, ctx) -> None` | Turn detector says turn is incomplete |
| `ON_BACKCHANNEL` | ASYNC | `(BackchannelEvent, ctx) -> None` | Backchannel detected (uh-huh, yeah) |
| `ON_RECORDING_STARTED` | ASYNC | `(RecordingStartedEvent, ctx) -> None` | Audio recording started (voice session or conference track) |
| `ON_RECORDING_STOPPED` | ASYNC | `(RecordingStoppedEvent, ctx) -> None` | Audio recording stopped, result available |
| `ON_INPUT_AUDIO_LEVEL` | ASYNC | `(AudioLevelEvent, ctx) -> None` | Inbound audio level (throttled; telemetry-suppressed) |
| `ON_OUTPUT_AUDIO_LEVEL` | ASYNC | `(AudioLevelEvent, ctx) -> None` | Outbound audio level (throttled; telemetry-suppressed) |
| `BEFORE_BRIDGE_AUDIO` | SYNC | `(BridgeAudioEvent, ctx) -> HookResult` | Before an audio frame is forwarded across an audio bridge. Can block/modify the frame. Only invoked when hooks are registered — otherwise frames bypass the event loop for latency. |

### Tool Execution

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_TOOL_USE` | SYNC | `(ToolCallEvent, ctx) -> HookResult` | Before a tool executes (AIChannel, realtime channels, external/ACP tools). Block denies the call. Fails closed: if room context cannot be built, the call is denied. |
| `ON_TOOL_CALL` | SYNC | `(ToolCallEvent, ctx) -> HookResult` | A tool call executed (AIChannel, RealtimeVoiceChannel, skills, external/ACP tools). Block returns an error result to the model; `HookResult(metadata={"result": ...})` supplies or overrides the tool result (`event.result is None` means the hook must provide it). RFC §9 lists this ASYNC; code invokes it through the sync pipeline. |
| `ON_USER_INPUT_REQUIRED` | SYNC | `(PendingInputEvent, ctx) -> HookResult` | A `HumanInputHandler`-backed tool paused awaiting human input. Sync so the notification (e.g. WebSocket push) lands before `wait()` starts blocking; block auto-rejects the pending request. |

### Realtime Voice

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_REALTIME_TEXT_INJECTED` | ASYNC | `(event, ctx) -> None` | Text was injected into realtime session |

### AI Generation

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_AI_GENERATION` | SYNC | `(AIGenerationEvent, ctx) -> HookResult` | Before the AI provider is invoked. `event.ai_context` (messages, system prompt, tools) may be mutated in place; block skips generation. |
| `ON_AI_THINKING` | — | — | Reserved — defined in the enum, not fired by any built-in code (RFC §9 marks it Implemented). |
| `ON_AI_RESPONSE` | ASYNC | `(AIResponseEvent, ctx) -> None` | AI generation completed. Carries response content, usage, latency, tool call count. |

### Protocol Observability

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PROTOCOL_TRACE` | ASYNC | `(ProtocolTrace, ctx) -> None` | Transport-level protocol trace (SIP, RTP, …) forwarded from a channel. Traces arriving before the room exists are buffered and replayed on attach. |

### Orchestration (multi-agent)

Handoff triggers fire from the `HandoffCoordinator` with a synthetic INTERNAL
system event (AI-channel source; `metadata`: `from_agent`, `to_agent`,
`accepted`, `new_phase`).

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_HANDOFF` | ASYNC | `(event, ctx) -> None` | Agent handoff accepted and executed. |
| `ON_HANDOFF_REJECTED` | ASYNC | `(event, ctx) -> None` | Agent handoff was rejected (`metadata.accepted=False`). |
| `ON_PHASE_TRANSITION` | ASYNC | `(event, ctx) -> None` | Fired alongside `ON_HANDOFF` after an accepted handoff, carrying the new phase. |
| `ON_STATUS_POSTED` | — | — | Reserved — StatusBus posts emit the `status_posted` framework event instead; no hook fires (RFC §9 marks it Implemented). |

### Delegation (background tasks)

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_TASK_DELEGATED` | ASYNC | `(event, ctx) -> None` | `kit.delegate()` dispatched a task to a child room. INTERNAL event, type `TASK_DELEGATED`; `metadata`: `task_id`, `child_room_id`, `agent_id`. |
| `ON_TASK_COMPLETED` | ASYNC | `(event, ctx) -> None` | A delegated task finished. Fires in the parent room; type `TASK_COMPLETED`, body = output or error. |

### Video

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_BRIDGE_VIDEO` | SYNC | `(BridgeVideoEvent, ctx) -> HookResult` | Before a video frame is forwarded across a bridge. Can block/modify the frame. Only invoked when hooks are registered (fast path bypasses the event loop). |
| `ON_VIDEO_SESSION_STARTED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | Video path live (VideoChannel, AVChannel, RealtimeAVChannel). `event.session` is the Video/VoiceSession. |
| `ON_VIDEO_SESSION_ENDED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | Video session ended (same payload shape). |
| `ON_VIDEO_TRACK_ADDED` | — | — | Reserved — defined in the enum, not fired by built-in channels. |
| `ON_VIDEO_TRACK_REMOVED` | — | — | Reserved — defined in the enum, not fired by built-in channels. |
| `ON_VISION_RESULT` | SYNC | `(VisionEvent, ctx) -> HookResult` | A VisionProvider analyzed a frame. Block discards the result; modify rewrites the description before event injection and AI context update. RFC §9 lists this ASYNC; code invokes it through the sync pipeline. |
| `ON_SCREEN_SHARE_STARTED` | ASYNC | `(event, ctx) -> None` | ConferenceChannel: a SCREEN_SHARE track was published. `content.data`: `track_id`, `participant_id`. |
| `ON_SCREEN_SHARE_STOPPED` | ASYNC | `(event, ctx) -> None` | ConferenceChannel: a SCREEN_SHARE track was unpublished. |
| `ON_VIDEO_DETECTION` | ASYNC | `(VideoDetectionEvent, ctx) -> None` | Video pipeline filter emitted a detection event (object, face, …). |

### Conference (SFU)

All fired by `ConferenceChannel` as synthetic system events
(`SystemContent` with a `data` dict including `channel_id`); the channel's own
bot never triggers them.

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_CONFERENCE_PARTICIPANT_JOINED` | ASYNC | `(event, ctx) -> None` | Participant joined the conference media session. `data`: `participant_id`. |
| `ON_CONFERENCE_PARTICIPANT_LEFT` | ASYNC | `(event, ctx) -> None` | Participant left the conference media session. |
| `ON_CONFERENCE_TRACK_PUBLISHED` | ASYNC | `(event, ctx) -> None` | Participant published a track. `data`: `track_id`, `participant_id`, `kind` (audio/video/screen_share). |
| `ON_CONFERENCE_TRACK_UNPUBLISHED` | ASYNC | `(event, ctx) -> None` | A track was unpublished (same `data` shape). |
| `ON_CONFERENCE_TRACK_MUTED` | ASYNC | `(event, ctx) -> None` | Publisher muted a track — "camera off" is usually a muted VIDEO track, not an unpublish. |
| `ON_CONFERENCE_TRACK_UNMUTED` | ASYNC | `(event, ctx) -> None` | Publisher unmuted a track. |
| `ON_ACTIVE_SPEAKER_CHANGED` | ASYNC | `(event, ctx) -> None` | SFU reported a dominant-speaker change. `data`: `participant_id`. |
| `ON_CONNECTION_QUALITY_CHANGED` | ASYNC | `(event, ctx) -> None` | SFU reported a participant's connection quality. `data`: `participant_id`, `quality`. |

### Planning

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PLAN_UPDATED` | — | — | Reserved — not fired by any built-in code. The `plan_tasks` tool (`enable_planning=True`) publishes an ephemeral realtime event `{"type": "plan_updated"}` instead (RFC §9 marks the hook Implemented). |

### Feedback

| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_FEEDBACK` | ASYNC | `(Observation, ctx) -> None` | User submitted quality feedback via `kit.submit_feedback()` |

## Framework Events

Framework events are lightweight lifecycle notifications (not message events):

```python
@kit.on("room_created")
async def on_room_created(event):
    print(f"Room created: {event.data['room_id']}")

@kit.on("voice_session_started")
async def on_voice(event):
    print(f"Voice session: {event.data['session_id']}")
```

Available framework event types: `room_created`, `room_closed`, `room_paused`, `room_channel_attached`, `room_channel_detached`, `channel_connected`, `channel_disconnected`, `voice_session_started`, `voice_session_ended`, `source_attached`, `source_detached`, `source_error`, `source_exhausted`.
---

AIChannel connects rooms to LLM providers. When a message is broadcast to an AI channel, it generates a response using conversation history and re-enters it through the inbound pipeline.

## Basic Setup

```python
from roomkit import RoomKit, ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig

kit = RoomKit()

ai = AIChannel(
    "ai-assistant",
    provider=AnthropicAIProvider(AnthropicConfig(
        api_key="sk-ant-...",
        model="claude-opus-5",
    )),
    system_prompt="You are a helpful customer support agent.",
    temperature=0.7,
)
kit.register_channel(ai)

await kit.create_room(room_id="support")
await kit.attach_channel("support", "ai-assistant", category=ChannelCategory.INTELLIGENCE)
```

## AI Providers

| Provider | Class | Config | Extra |
|----------|-------|--------|-------|
| Anthropic (Claude) | `AnthropicAIProvider` | `AnthropicConfig` | `roomkit[anthropic]` |
| OpenAI (GPT) | `OpenAIAIProvider` | `OpenAIConfig` | `roomkit[openai]` |
| Google Gemini | `GeminiAIProvider` | `GeminiConfig` | `roomkit[gemini]` |
| Gemini on Vertex AI | `GeminiVertexProvider` | `GeminiVertexConfig` | `roomkit[gemini]` |
| Mistral | `MistralAIProvider` | `MistralConfig` | `roomkit[mistral]` |
| Azure OpenAI | `AzureAIProvider` | `AzureAIConfig` | `roomkit[azure]` |
| OpenRouter (300+ models) | `OpenRouterAIProvider` | `OpenRouterConfig` | `roomkit[openrouter]` |
| xAI (Grok) | `XAIAIProvider` | `XAIConfig` | `roomkit[xai]` |
| PolarGrid (Canadian-hosted) | `PolarGridAIProvider` | `PolarGridConfig` | `roomkit[polargrid]` |
| vLLM (local) | `create_vllm_provider()` | `VLLMConfig` | `roomkit[vllm]` |
| Ollama (local) | `OllamaAIProvider` | `OllamaConfig` | `roomkit[ollama]` |
| Mock (testing) | `MockAIProvider` | — | built-in |

Provider notes:

- **Explicit model selection** — `model=` is required by `OpenAIConfig` and
  `AnthropicConfig`; upgrading RoomKit therefore cannot silently change cost,
  latency, or model behavior. For the selected model, `OpenAIConfig`
  automatically uses `max_completion_tokens` and omits custom temperature for
  current GPT-5 and o-series ids, while `AnthropicConfig` uses adaptive thinking
  and omits temperature for current Claude reasoning ids. Explicit flags take
  precedence, and a custom `base_url` keeps conservative legacy behavior.

- **Gemini on Vertex AI** (`roomkit.providers.gemini.vertex`) — subclass of `GeminiAIProvider` serving the same Gemini models through a Google Cloud project with a pinned region (`GeminiVertexConfig` requires `project` and `location`, e.g. `"northamerica-northeast1"`; auth is ADC, no API key). Use it when data residency matters (Québec Law 25 / PIPEDA). Generation, streaming, thinking, and the model catalog are inherited unchanged.
- **OpenRouter** (`roomkit.providers.openrouter`) — subclass of `OpenAIAIProvider` pointed at `https://openrouter.ai/api/v1`; `OpenRouterConfig` subclasses `OpenAIConfig` (adds `site_url`/`app_name` attribution headers) and `model` is a required slug like `"anthropic/claude-sonnet-4.5"`. Reasoning is forwarded to any upstream model via OpenRouter's unified `reasoning` object. Model-listing nuance: OpenRouter's `/models` items omit the `object`/`owned_by` fields the OpenAI SDK expects, so `list_models()` reads the raw JSON instead.
- **xAI (Grok)** (`roomkit.providers.xai`) — subclass of `OpenAIAIProvider` pointed at `https://api.x.ai/v1`; defaults to `max_completion_tokens` and stream usage. `XAIRealtimeProvider` (Grok speech-to-speech) is a separate import from `roomkit.providers.xai.realtime`.
- **PolarGrid** (`roomkit.providers.polargrid`) — Canadian-hosted inference network (edges in Toronto, Vancouver, Montréal) via the official `polargrid-sdk` async client; OpenAI-shaped chat-completions surface. Supports tool calling, thinking (`PolarGridConfig(thinking=True)` sets the `enable_thinking` request flag; reasoning is surfaced as `AIResponse.thinking` / `StreamThinkingDelta`), and vision (`image_url` content parts). Pin `region` in production when residency matters.

Pick **Ollama** over the OpenAI-compat shim (`OpenAIAIProvider` pointed at `http://host:11434/v1` or `create_vllm_provider()` with an Ollama URL) whenever the model is a reasoning model (DeepSeek-R1, Qwen 3 thinking variants, etc.) — only the native API exposes the `think` parameter and streams the `thinking` field separately from `content`. See `docs/c7/ollama-provider.md` for the full rundown.

```python
# OpenAI
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig

provider = OpenAIAIProvider(OpenAIConfig(api_key="sk-...", model="gpt-4o"))

# Gemini
from roomkit.providers.gemini.ai import GeminiAIProvider
from roomkit.providers.gemini.config import GeminiConfig

provider = GeminiAIProvider(GeminiConfig(api_key="...", model="gemini-2.0-flash"))

# Mock (for testing)
from roomkit.providers.ai.mock import MockAIProvider

provider = MockAIProvider(responses=["Hello!", "How can I help?"])
```

## Model Catalog and Pricing

`provider.catalog_entry()` returns the configured model's offline `ModelInfo`.
When it carries `pricing`, `pricing.cost_for(response.usage)` prices fresh
input, output, cache reads and represented cache writes. A `None` cache rate
means no separate per-token charge is represented; it contributes zero rather
than falling back implicitly. Catalogs repeat the input rate explicitly when a
vendor bills a cache counter as ordinary input.

Tiered entries also carry a long-context threshold plus input/output
multipliers. `cost_for()` applies them automatically to GPT-5.6, Gemini Pro and
current Grok usage after total input crosses the vendor's threshold.

## Agent Class

`Agent` extends `AIChannel` with role, description, greeting, and memory support — designed for multi-agent orchestration:

```python
from roomkit import Agent
from roomkit.providers.ai.mock import MockAIProvider

agent = Agent(
    "support-agent",
    provider=MockAIProvider(responses=["I can help with that."]),
    role="Customer support specialist",
    description="Handles billing and account questions",
    system_prompt="You are a support specialist. Be concise and helpful.",
    greeting="Hi! How can I help you today?",
)
```

## Tool Calling

Define tools as JSON schema and attach them to the AI channel:

```python
from roomkit import ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig

ai = AIChannel(
    "ai-assistant",
    provider=OpenAIAIProvider(OpenAIConfig(api_key="sk-...", model="gpt-4o")),
    system_prompt="You help users check the weather.",
    tools=[
        {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"},
                    "units": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                },
                "required": ["city"],
            },
        },
    ],
)
```

### Tool Handler

Register a handler to execute tool calls via the constructor:

```python
async def handle_tools(name: str, arguments: dict) -> str:
    if name == "get_weather":
        city = arguments["city"]
        return f'{{"temperature": 22, "condition": "sunny", "city": "{city}"}}'
    return '{"error": "Unknown tool"}'

ai = AIChannel(
    "ai-assistant",
    provider=provider,
    tools=[...],
    tool_handler=handle_tools,
)
```

### Tool Protocol (Tool ABC)

For structured tool definitions, use the `Tool` base class:

```python
from roomkit.tools.base import Tool

class GetWeather(Tool):
    name = "get_weather"
    description = "Get current weather for a city"
    parameters = {
        "type": "object",
        "properties": {
            "city": {"type": "string"},
        },
        "required": ["city"],
    }

    async def execute(self, arguments: dict) -> str:
        return '{"temperature": 22, "condition": "sunny"}'

ai = AIChannel("ai", provider=provider, tools=[GetWeather()])
```

### MCP Tool Provider

Integrate Model Context Protocol servers:

```python
from roomkit.tools.mcp import MCPToolProvider

mcp = MCPToolProvider(server_command=["uvx", "mcp-server-sqlite", "--db", "data.db"])
await mcp.initialize()

ai = AIChannel("ai", provider=provider, tools=mcp.tools())
```

## Tool Search (Progressive Tool Disclosure)

When an agent has dozens of tools, sending every schema to the model on
every turn burns context and makes smaller models hallucinate tool names.
**Tool Search** hides the catalogue behind two discovery tools and lets the
model reveal only what it needs:

- `find_tools(query)` — search the catalogue by natural language; the
  matches become directly invocable for the rest of the turn.
- `list_tools(category=None)` — list the catalogue (name + short description).

```python
ai = AIChannel(
    "ai",
    provider=provider,
    tool_handler=handle_tools,
    tools=big_catalogue,              # e.g. 60+ MCP tools
    tool_search=None,                 # None = auto, True/False = force
    tool_search_threshold_pct=10.0,   # auto-enable above this % of the window
    tool_search_threshold=20,         # fallback tool count when window unknown
    tool_search_pinned=["get_help"],  # always visible, never searched for
)
```

How it works:

1. The model first sees only `find_tools`/`list_tools` plus the pinned set —
   the discretionary catalogue is hidden.
2. It calls `find_tools("send a text message")`; the matches are scored and
   returned, and their names are recorded for the turn.
3. On the **next** tool-loop round the matched tools are visible and directly
   callable. The text loop re-sends its (re-filtered) tool list every round,
   so no provider reconfigure is needed — this works on **any** text/HTTP
   provider. (The realtime voice channel offers the same feature via
   `provider.reconfigure`.)

Notes:

- **Activation auto-tunes to the model.** In `auto` mode it defers when the
  *deferrable* (non-pinned) tools would cost more than `tool_search_threshold_pct`
  % of the model's context window (default 10%) — a large model is a no-op, a
  small one defers early. When the window is unknown (custom / local model ids
  absent from the provider catalog) it falls back to the `tool_search_threshold`
  tool count. Below the threshold Tool Search is a no-op and every tool is sent.
- **Pinned tools** stay visible without a search. The discovery tools always
  pass tool-policy and skill gating, so they work even under a restrictive
  `tool_policy`.
- A second `find_tools` call **swaps** the revealed window (keeping the visible
  surface small); `list_tools` reveals nothing — it is purely informational.

See `examples/ai_tool_search.py` for a runnable, no-API-key walkthrough.

## Per-Room Configuration

Override AI settings per room via binding metadata:

```python
await kit.attach_channel(
    "billing-room",
    "ai-agent",
    category=ChannelCategory.INTELLIGENCE,
    metadata={
        "system_prompt": "You are a billing specialist.",
        "temperature": 0.3,
        "tools": [...],
    },
)
```

## Streaming

AIChannel supports streaming responses to WebSocket clients:

```python
from roomkit import WebSocketChannel

ws = WebSocketChannel("ws-user")

# Register with stream support
ws.register_connection("conn-1", on_recv, stream_send_fn=on_stream)

async def on_stream(conn_id: str, msg) -> None:
    # StreamStart, StreamChunk, StreamEnd
    print(f"Stream: {msg}")
```

## AI Thinking/Reasoning

Some providers support extended thinking:

```python
ai = AIChannel(
    "ai",
    provider=AnthropicAIProvider(AnthropicConfig(
        api_key="...",
        model="claude-opus-5",
    )),
    system_prompt="Think step by step.",
    thinking_budget=4096,  # Setting a budget enables thinking mode
)
```

Per-provider mechanisms: Anthropic uses `thinking_budget` as above; OpenAI-family providers (OpenAI, Azure, OpenRouter, xAI) use the `reasoning_effort` config field — OpenRouter translates it into its unified `reasoning` object for any upstream model; Gemini (and Vertex) use `GeminiConfig.thinking_level`; Ollama exposes the native `think` parameter (see `docs/c7/ollama-provider.md`); PolarGrid uses `PolarGridConfig(thinking=True)` (the `enable_thinking` flag, surfaced as `AIResponse.thinking` / `StreamThinkingDelta`).

## Vision Support

AI providers that support vision can process images sent as `MediaContent`:

```python
from roomkit.models.event import MediaContent

await kit.process_inbound(
    InboundMessage(
        channel_id="ws-user",
        sender_id="user",
        content=MediaContent(url="https://example.com/chart.png", mime_type="image/png"),
    )
)
# AI sees the image and responds with analysis
```

OpenAI-family providers (OpenAI, Azure, OpenRouter, xAI) and PolarGrid send images as OpenAI-shaped `image_url` content parts (remote URL or `data:` URI); PolarGrid and xAI gate this on the model's `supports_vision` from their curated catalogs — whether the model actually reads the image is the deployed model's capability.

## Agentic Features

### Dangling Tool Call Recovery

When a user sends a new message while the AI is mid-tool-execution (barge-in), tool calls can be left without matching results. AIChannel automatically detects these orphaned calls and injects synthetic cancellation results before the next AI turn, preventing provider API rejections.

This is fully automatic — no configuration needed.

### Large Output Eviction

When tool results are very large (database queries, file dumps, API responses), they consume significant context budget. AIChannel can evict large results to a side buffer and replace them with a preview:

```python
ai = AIChannel(
    "ai-agent",
    provider=provider,
    system_prompt="You are a data analyst.",
    evict_threshold_tokens=5000,  # default: 5000 tokens
    tools=[QueryDatabase()],
)
```

When a tool result exceeds the threshold:
1. The full result is stored in a FIFO-bounded buffer (max 50 entries)
2. A head/tail preview (first 5 + last 5 lines) replaces the result in context
3. A `_read_tool_result` tool is auto-injected so the agent can paginate through the full output

### Planning Tools

Enable structured task planning so agents can break down complex work and track progress:

```python
ai = AIChannel(
    "ai-agent",
    provider=provider,
    system_prompt="You are a research assistant.",
    enable_planning=True,
)
```

When enabled, the AI gets a `plan_tasks` tool that accepts up to 100 tasks with a title of at most 500 characters and a `status` (`pending`, `in_progress`, `completed`, `blocked`). Undeclared task fields are discarded. The current plan is:
- Injected into the system prompt on each turn (so the AI sees its progress)
- Published as an ephemeral `CUSTOM` event with `data.type = "plan_updated"` for real-time UI rendering

Subscribe to plan updates for UI:

```python
await kit.subscribe_room("room-1", my_callback)

# Callback receives ephemeral event with:
# type: "custom", data: {"type": "plan_updated", "tasks": [...]}
```

The ephemeral event is the only plan-update signal — the `HookTrigger.ON_PLAN_UPDATED` enum value is reserved and not currently fired.

### SummarizingMemory

For long conversations, use `SummarizingMemory` to proactively manage context budget with two tiers:

```python
from roomkit.memory import SummarizingMemory, SlidingWindowMemory

ai = AIChannel(
    "ai-agent",
    provider=main_provider,
    memory=SummarizingMemory(
        inner=SlidingWindowMemory(max_events=100),
        provider=summary_provider,       # lightweight model (e.g. Haiku)
        max_context_tokens=128_000,
        tier1_ratio=0.50,                # truncate old events at 50%
        tier2_ratio=0.85,                # LLM summarization at 85%
    ),
)
```

- **Tier 1** (~50% capacity): Truncates large text bodies in older events to 2000 chars. No LLM call — cheap and fast.
- **Tier 2** (~85% capacity): Calls the summary provider to summarize older events into a concise paragraph. Keeps recent events at full fidelity. Supports chained summaries (prior summary is incorporated into the new one).

### Knowledge Retrieval (RAG)

Enrich AI context with external knowledge sources using `RetrievalMemory`:

```python
from roomkit.knowledge import KnowledgeSource, KnowledgeResult
from roomkit.memory import RetrievalMemory, SlidingWindowMemory

# Implement your own knowledge source (vector store, search engine, etc.)
class FAQSource(KnowledgeSource):
    async def search(self, query, *, room_id=None, limit=5):
        results = await my_vector_db.search(query, top_k=limit)
        return [KnowledgeResult(content=r.text, score=r.score, source="faq") for r in results]

ai = AIChannel(
    "ai-agent",
    provider=provider,
    memory=RetrievalMemory(
        sources=[FAQSource()],
        inner=SlidingWindowMemory(max_events=50),
        max_results=5,
    ),
)
```

`RetrievalMemory` searches all sources concurrently, deduplicates results, and prepends relevant knowledge as a context message. When `ingest()` is called (automatic on every inbound event), it also indexes content in all sources.

#### Built-in: PostgreSQL Full-Text Search

For production use without a vector database, use `PostgresKnowledgeSource`:

```python
from roomkit.knowledge.postgres import PostgresKnowledgeSource

source = PostgresKnowledgeSource(dsn="postgresql://localhost/mydb")
await source.init()

# Or share the pool with PostgresStore:
source = PostgresKnowledgeSource(pool=store._pool, source_name="faq")
await source.init()
```

Uses PostgreSQL `tsvector` with `ts_rank_cd` for relevance scoring. Auto-creates schema, supports room-scoped queries, and upserts on conflict.

### Response Scoring

Score AI responses automatically using the `ScoringHook`:

```python
from roomkit.scoring import ScoringHook, ConversationScorer, Score

class QualityScorer(ConversationScorer):
    async def score(self, *, response_content, query, room_id, channel_id, **kwargs):
        # Your scoring logic (LLM-as-judge, rules, heuristics)
        return [Score(value=0.9, dimension="relevance", reason="On topic")]

hook = ScoringHook(scorers=[QualityScorer()])
hook.attach(kit)

# Scores are stored as Observations and accessible via hook.recent_scores
```

### User Feedback

Collect user quality ratings:

```python
await kit.submit_feedback("room-1", rating=0.9, comment="Very helpful", dimension="helpfulness")
# Stored as Observation in ConversationStore, fires ON_FEEDBACK hook
```

## Tool Call Events

AIChannel automatically broadcasts ephemeral `TOOL_CALL_START` and `TOOL_CALL_END` events when executing tools. Subscribe to these for UI indicators:

```python
await kit.subscribe_room("room-1", my_callback)

# Callback receives:
# TOOL_CALL_START: {tool_calls: [{id, name, arguments}], round, channel_id}
# TOOL_CALL_END: {tool_calls: [{id, name, result}], round, channel_id, duration_ms}
```
---

`OllamaAIProvider` talks to the native Ollama `/api/chat` endpoint
through the official `ollama-python` SDK. Use it when you want
features the OpenAI-compatible shim hides — chiefly the `think`
parameter and streamed `thinking` deltas.

## Install

```bash
pip install roomkit[ollama]
```

## When to use this vs OpenAI-compat

Ollama also exposes an OpenAI-compatible endpoint at
`http://host:11434/v1/chat/completions`, which works fine with
`OpenAIAIProvider` (or `create_vllm_provider`). Pick `OllamaAIProvider`
when **any** of these matter:

| You want | OpenAI-compat | `OllamaAIProvider` |
|---|---|---|
| Disable reasoning on a thinking model (`think=False`) | Silently ignored | Honored |
| Force reasoning on a non-default model (`think=True`) | Silently ignored | Honored |
| Stream reasoning tokens to a UI as they arrive | Returned in a non-streamed `reasoning` field that the SDK consumer doesn't split out | Streamed as `StreamThinkingDelta` events token-by-token |
| Pass `keep_alive`, `num_ctx`, or `num_predict` cleanly | Awkward via `extra_body` | First-class config |

For plain non-reasoning models (`llama3.2`, `gemma2`, etc.) either
provider works.

## Quick start

```python
from roomkit.providers.ollama import OllamaAIProvider, OllamaConfig

provider = OllamaAIProvider(OllamaConfig(
    host="http://localhost:11434",
    model="qwen3:8b",
))

# Use exactly like any other AIProvider — pass it to AIChannel, etc.
```

## Config knobs

```python
OllamaConfig(
    host="http://localhost:11434",   # Ollama server
    model="qwen3:8b",                # any pulled model
    max_tokens=None,                 # → options.num_predict
    temperature=0.7,                 # → options.temperature
    timeout=120.0,                   # long: cold-start + reasoning is slow
    think=None,                      # None = model default, True/False = explicit
    keep_alive="5m",                 # how long the model stays loaded
    num_ctx=8192,                    # context window override
)
```

## How `think` is resolved

Per-request precedence (highest first):

1. **`AIContext.thinking_budget`** — if set, this wins:
   - `None` or `0` → `think=False`
   - any `>0` → `think=True`
2. **`OllamaConfig.think`** — fallback when `thinking_budget` is unset.
3. **Omit `think`** — let the model decide its default (reasoning
   models think, others don't).

This makes the "Leave empty to disable" semantics in higher-level
agent configs actually honor themselves: a missing/zero
`thinking_budget` at the agent layer reaches the provider as
`think=False` and the model genuinely skips the reasoning phase
instead of just having its reasoning silently discarded.

## Streamed events

`generate_structured_stream()` yields:

- `StreamThinkingDelta(thinking="...")` — one per `message.thinking`
  delta chunk from Ollama. Arrives **token-by-token**, which is the
  whole reason this provider exists.
- `StreamTextDelta(text="...")` — one per `message.content` delta.
- `StreamToolCall(id, name, arguments)` — collected from
  `message.tool_calls` and yielded after the text/thinking deltas
  for the same chunk (Ollama doesn't fragment tool-call arguments
  across chunks the way OpenAI does).
- `StreamDone(finish_reason, usage)` — terminator.

## History round-trip

The provider preserves `AIThinkingPart` in the message history by
sending it back to Ollama as a top-level `thinking` field on the
assistant message — no `<think>...</think>` tag wrapping needed.
This keeps reasoning models honest across tool-loop rounds: they
see their own prior chain-of-thought when computing the next turn.

## Tool calls

Ollama's tool-call format is essentially OpenAI's, so the standard
`AITool` definitions pass through unchanged. The one quirk: Ollama
doesn't issue stable `id` fields on tool calls, so the provider
synthesizes ones like `call_<name>_<index>` so consumers can pair
calls with their results.

## Interactive test bed

`examples/ollama_cli.py` exercises every knob in one place — `--think`,
`--no-think`, `--stream`, `--no-stream`, `--mcp <url>` for MCP tool
discovery — with a Rich-powered display that labels thinking output
in italics and answer output in bold.

```bash
# Default: model decides whether to think, streams tokens, no tools
uv run python examples/ollama_cli.py --model qwen3:8b

# Force thinking off — fast, single-pass response
uv run python examples/ollama_cli.py --model qwen3:8b --no-think

# Wire in MCP tools
uv run python examples/ollama_cli.py --model qwen3:8b \
    --mcp http://localhost:8080/mcp
```

At the prompt, `/think on|off` and `/stream on|off` toggle the
behavior mid-session; `/tools` lists what's available; `/reset`
clears history; `/quit` exits.

## Errors

`ollama.ResponseError` maps to `ProviderError(retryable=...)` where
`retryable=True` for `429`, `500`, `502`, `503`. Transport/connection
errors (timeouts, refused connections) get marked retryable so the
calling `RetryPolicy` decides whether to act.

## What this provider doesn't do

- **Image input**: `supports_vision` reports `True` and image parts
  reach the wire, but it's the server-side model that decides whether
  to honor them. Unsupported models 400 the request — RoomKit
  surfaces that as a non-retryable `ProviderError`.
- **Embeddings**: out of scope. Use `ollama.AsyncClient().embeddings()`
  directly if you need them.
---

`ACPChannel` makes RoomKit an **Agent Client Protocol client**: an external coding agent (Claude Code, Codex CLI, Gemini CLI, any registry-listed ACP agent) speaks ACP v1 — spawned here as a subprocess over stdio by default, or reached through a caller-supplied `ACPTransport`; each Room becomes one session of that agent. The reverse direction — exposing a RoomKit agent as an ACP *server* — is out of scope. `CLIChannel` is the interactive terminal transport playing the human side in local sessions.

```bash
pip install "roomkit[acp]"      # agent-client-protocol>=0.11.0,<0.12
pip install "roomkit[console]"  # rich>=13.0, for CLIChannel(markdown=True)
```

The SDK is imported lazily: `import roomkit` works without the extra; the first connection raises an actionable `ImportError`. RoomKit pins stable ACP wire protocol v1 and rejects any other negotiated `protocolVersion`.

## ACPChannel

```python
from roomkit import ACPChannel

agent = ACPChannel(
    "coding-agent",
    command=["npx", "-y", "@agentclientprotocol/claude-agent-acp@0.61.0"],
    transport=None,                    # or an ACPTransport, instead of command
    cwd="/srv/workspaces/my-project",  # required, absolute — on the AGENT's host
    additional_directories=None,       # extra absolute dirs for the session
    env=None,                          # added to the SDK's restricted env
    mcp_servers=None,                  # ACP MCP-server descriptors (SDK types)
    authentication_method=None,        # optional ACP auth method id
    external_tool_handler=None,        # permission policy; None = reject all
)
```

`command` is an argument vector executed directly, **no shell**; a bare string, empty/non-string args, or non-absolute `cwd`/`additional_directories` raise `ValueError`. Exactly one of `command` / `transport` is required (neither or both → `ValueError`), and `env`/`inherit_env` next to a `transport` raise rather than being ignored — they configure the spawn. Class attrs: `channel_type = ChannelType.AI`, `category = ChannelCategory.INTELLIGENCE`, `direction = BIDIRECTIONAL`; capabilities TEXT + RICH. `handle_inbound()` raises `NotImplementedError` — the channel reacts to Room events via `on_event()`.

### Transports

`ACPTransport` (ABC, `channels/acp_transport.py`) is the pipe, and nothing more: `open(client, *, queue) -> ClientSideConnection` (build it with `acp.connect_to_agent(client, writer, reader, queue=queue)` — the protocol only needs a reader/writer pair), `close()` (must not raise; called on teardown *and* on a failed handshake), `is_alive()` (default `True`), and a `name` property surfaced as `info["transport"]`. `StdioACPTransport(command, cwd=…, env=…, inherit_env=…)` is the default, constructed for you from `command=`, and owns the spawn: argument-vector validation, `_resolve_spawn_env`, the stderr drain, and `returncode`-based liveness. Everything protocol-level — `initialize`, version negotiation, `authenticate`, sessions, prompts, permissions, config options, event mapping — stays on the channel, so a transport inherits it.

### Process and session model

One connection per channel, opened lazily on the first prompt; one ACP **session** per Room, created on demand and tagged with extension key `roomkit.live/roomId`. Prompts are serialized per Room (per-room lock); different Rooms progress concurrently through the same connection. When the transport reports the connection dead (for stdio: the subprocess exited), the next prompt reconnects and clears all session mappings — a reconnect never resumes sessions. The client declares no fs/terminal capabilities: `fs/*`, `terminal/*`, and `session/request_input` (elicitation) requests get `method_not_found` — `ON_USER_INPUT_REQUIRED` never fires from ACP.

Methods: `session_id(room_id)` returns the process-local session id or `None`; `cancel(room_id)` cancels the active turn (`True` if a cancel was sent); `close_session(room_id)` closes and forgets one Room's session; `close()` cancels turns, closes sessions, closes the transport and stops the handler. The `info` property reports `{transport, protocol_version, sdk_version, connected, agent, session_count}`.

### How agent output enters the room

For each non-self text event, `on_event()` returns `ChannelOutput(responded=True, response_stream=...)` (TOOL_CALL_START/END events are skipped). The prompt (tagged `roomkit.live/eventId`) yields a `StreamDelta` stream consumed by the inbound-streaming pipeline:

- `agent_message_chunk` → `str` deltas → streamed to transports, persisted as the response event.
- `agent_thought_chunk` → `ThinkingDeltaMarker` in the stream, plus ephemeral `THINKING_START` / `THINKING_DELTA` (thinking truncated to 1000 chars) / `THINKING_END`.
- `tool_call` / `tool_call_update` → `ToolCallStartMarker` / `ToolCallEndMarker` in the stream (persisted as `TOOL_CALL_START`/`TOOL_CALL_END` RoomEvents) plus matching ephemeral events (`result` truncated to 500 chars, `duration_ms`); non-terminal progress → ephemeral `CUSTOM` `{"type": "acp_tool_progress"}`.
- `plan` / `plan_update` / `plan_removed` → ephemeral `CUSTOM` `{"type": "acp_plan_update", session_id, update}`. **Not** `ON_PLAN_UPDATED` — that hook belongs to AIChannel's `plan_tasks` tool.
- `usage_update` → ephemeral `CUSTOM` `{"type": "acp_usage"}`.

`register_channel()` wires `kit`'s realtime backend in automatically. A failed prompt surfaces as `ProviderError(provider="acp")`; cancellation ends the stream silently and closes any open thinking block.

A turn never outlives its tool calls. Whichever way it ends — error, cancellation, or a stop the user asked for, which returns through the ordinary end of a prompt — every tool started without a terminal `tool_call_update` is closed first: a `ToolCallEndMarker` with `status="failed"` and an `error` saying the turn ended before the tool reported, emitted into the stream so the stored `TOOL_CALL_END` exists, plus the matching ephemeral. A turn whose tools all reported emits nothing extra. One gap remains by construction: a stream closed from the outside (its consumer cancelled, a muted binding) is past yielding, so only the ephemeral goes out and the stored row stays pending.

ACP fixes the envelope and leaves the payload to the agent, so `CLIChannel(console=True)` unwraps rather than prints (`roomkit.console._tool_preview`): ACP `text`/`diff` blocks, MCP `content`, and `raw_output` wrappers (`formatted_output`+`exit_code` from Codex, `output`, `result`/`error`) all reduce to their text; a `terminal` block carries no text, so the preview falls back to `raw_output`; `image`/`audio`/`resource` blocks are named, never dumped as base64; 5 lines per result, 200 chars per line; unknown shapes render as compact JSON.

### Permission flow

Every agent `session/request_permission` goes to the `external_tool_handler` (`ExternalToolHandler` ABC from `roomkit.tools`, with `PolicyExternalToolHandler` and `ToolDecision`):

1. `process_tool_call(tool_name, tool_input, *, tool_call_id, session_id, room_id, ...)` → `ToolDecision(approved=...)`. Calling `self._fire_before_hook(...)` fires **`BEFORE_TOOL_USE`** sync hooks (callbacks injected at `register_channel`).
2. Approved → RoomKit selects the agent's `allow_once`/`allow_always` option; denied → `reject_once`/`reject_always`; no matching option → `DeniedOutcome(outcome="cancelled")`.
3. **No handler ⇒ every permission request is rejected.** `ToolDecision.modified_input`/`result` overrides cannot be applied over ACP — setting them logs a warning and rejects the call.
4. On tool completion, `on_tool_result(...)` runs; via `_fire_on_tool_hook` it fires **`ON_TOOL_CALL`** hooks.

## CLIChannel

Interactive terminal transport (`channel_type = ChannelType.CLI`, TEXT only): reads stdin, prints agent output to stdout with ANSI colors.

```python
cli = CLIChannel(
    "cli",                      # channel_id, default "cli"
    prompt="You: ",
    user_color="\033[33m",      # yellow
    agent_color="\033[36m",     # cyan
    thinking_color="\033[2;3m", # dim italic
    use_color=None,             # None = auto-detect TTY
    agent_label=None,           # channel_id -> display name ("agent-researcher" -> "Researcher")
    show_thinking=False,        # render ThinkingDeltaMarker chunks
    markdown=False,             # live Markdown rendering, requires roomkit[console]
)
```

- `sender_is_participant = True`: `run()`'s `sender_id` is a room **Participant ID**, not an address — identity resolution is skipped for typed lines.
- `deliver()` skips self-echo and prints `Label: text`. `supports_streaming_delivery` is `True`; `deliver_stream()` renders text deltas as they arrive, thinking above the answer, tool events inline (`🔧 name {args}`, `✓`/`✗ name (N ms)`).
- `run(kit, room_id, *, sender_id="user", welcome=None, content_factory=None)` — input loop in a worker thread; each line becomes `kit.process_inbound(...)` with `TextContent(body=line)`, or `content_factory(line)` (`None` skips the line). `quit`/`exit`/`q` or Ctrl+D exits.

## Example: terminal → Claude Code

Condensed from `examples/acp_claude_code.py` (requires `roomkit[acp,console]`, Node.js 22+):

```python
from roomkit import ACPChannel, ChannelCategory, CLIChannel, RoomKit
from roomkit.tools import ExternalToolHandler, ToolDecision

class TerminalPermissionHandler(ExternalToolHandler):
    async def process_tool_call(self, tool_name, tool_input, *, tool_call_id="",
                                room_id=None, **kwargs) -> ToolDecision:
        if not await self._fire_before_hook(tool_name, tool_input,
                                            tool_call_id=tool_call_id, room_id=room_id):
            return ToolDecision(approved=False, reason="Denied by BEFORE_TOOL_USE hook")
        answer = await asyncio.to_thread(input, f"Allow {tool_name}? [y/N] ")
        return ToolDecision(approved=answer.strip().casefold() in {"y", "yes"})

    async def on_tool_result(self, tool_name, tool_input, result, **kwargs) -> None:
        await self._fire_on_tool_hook(tool_name, tool_input, result)  # ON_TOOL_CALL hooks

kit = RoomKit()
cli = CLIChannel("you", show_thinking=True, markdown=True,
                 agent_label=lambda _cid: "Claude Code")
claude = ACPChannel(
    "claude-code",
    command=["npx", "-y", "@agentclientprotocol/claude-agent-acp@0.61.0"],
    cwd=workspace,  # absolute Path to the project
    env={"ANTHROPIC_API_KEY": api_key, "MAX_THINKING_TOKENS": "1024"},
    external_tool_handler=TerminalPermissionHandler(),
)
kit.register_channel(cli)
kit.register_channel(claude)
await kit.create_room(room_id="claude-code-cli")
await kit.attach_channel("claude-code-cli", "you")
await kit.attach_channel("claude-code-cli", "claude-code",
                         category=ChannelCategory.INTELLIGENCE)
await cli.run(kit, room_id="claude-code-cli")  # blocks until quit/Ctrl+D
await kit.close()
```

Flow: terminal line → `CLIChannel` inbound → Room broadcast → `ACPChannel.on_event` → ACP prompt to Claude Code; deltas, thinking, and tool activity stream back through the Room and render live in the terminal, each permission approved once at the prompt.
---

VoiceChannel handles real-time audio conversations with speech-to-text, text-to-speech, and an audio processing pipeline.

## Basic Voice Setup

```python
from roomkit import RoomKit, VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, MockVADProvider, VADEvent, VADEventType
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.backends.mock import MockVoiceBackend

kit = RoomKit()

# Create providers
backend = MockVoiceBackend()
stt = MockSTTProvider(transcripts=["Hello, how can I help?"])
tts = MockTTSProvider()
vad = MockVADProvider(events=[
    VADEvent(type=VADEventType.SPEECH_START),
    None,
    VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"audio"),
])

# Create voice channel with pipeline
voice = VoiceChannel(
    "voice-agent",
    stt=stt,
    tts=tts,
    backend=backend,
    pipeline=AudioPipelineConfig(vad=vad),
)
kit.register_channel(voice)
```

## Joining a Voice Session

```python
# Create room and attach voice channel
await kit.create_room(room_id="call-room")
await kit.attach_channel("call-room", "voice-agent")

# Join a participant to the voice session
session = await kit.join(
    room_id="call-room",
    channel_id="voice-agent",
    participant_id="caller-1",
)

# Leave when done
await kit.leave(session)
```

## STT Providers

| Provider | Class | Config | Extra |
|----------|-------|--------|-------|
| Deepgram | `DeepgramSTTProvider` | `DeepgramConfig` | `roomkit[deepgram]` |
| SherpaOnnx | `SherpaOnnxSTTProvider` | `SherpaOnnxSTTConfig` | `roomkit[sherpa-onnx]` |
| Gradium | `GradiumSTTProvider` | `GradiumSTTConfig` | `roomkit[gradium]` |
| Qwen3 ASR | `Qwen3ASRProvider` | `Qwen3ASRConfig` | `roomkit[qwen-asr]` |
| Mock | `MockSTTProvider` | — | built-in |

Use lazy loaders to avoid import-time dependency checks:

```python
from roomkit.voice import get_deepgram_provider, get_deepgram_config

DeepgramSTTProvider = get_deepgram_provider()
DeepgramConfig = get_deepgram_config()

stt = DeepgramSTTProvider(DeepgramConfig(
    api_key="...",
    model="nova-2",
    language="en",
))
```

## TTS Providers

| Provider | Class | Config | Extra |
|----------|-------|--------|-------|
| ElevenLabs | `ElevenLabsTTSProvider` | `ElevenLabsConfig` | `roomkit[elevenlabs]` |
| SherpaOnnx | `SherpaOnnxTTSProvider` | `SherpaOnnxTTSConfig` | `roomkit[sherpa-onnx]` |
| Gradium | `GradiumTTSProvider` | `GradiumTTSConfig` | `roomkit[gradium]` |
| Qwen3 | `Qwen3TTSProvider` | `Qwen3TTSConfig` | `roomkit[qwen-tts]` |
| NeuTTS | `NeuTTSProvider` | `NeuTTSConfig` | `roomkit[neutts]` |
| Grok TTS | `GrokTTSProvider` | `GrokTTSConfig` | xAI |
| Gemini TTS | `GeminiTTSProvider` | `GeminiTTSConfig` | `roomkit[gemini]` |
| Mock | `MockTTSProvider` | — | built-in |

```python
from roomkit.voice import get_elevenlabs_provider, get_elevenlabs_config

ElevenLabsTTSProvider = get_elevenlabs_provider()
ElevenLabsConfig = get_elevenlabs_config()

tts = ElevenLabsTTSProvider(ElevenLabsConfig(
    api_key="...",
    voice_id="21m00Tcm4TlvDq8ikWAM",
    model="eleven_turbo_v2",
))
```

## Voice Backends

Backends handle audio transport between the framework and participants:

| Backend | Class | Extra | Use Case |
|---------|-------|-------|----------|
| Local mic/speaker | `LocalAudioBackend` | `roomkit[local-audio]` | Development/testing |
| FastRTC (WebRTC) | `FastRTCVoiceBackend` | `roomkit[fastrtc]` | Browser-based voice |
| RTP | `RTPVoiceBackend` | `roomkit[rtp]` | VoIP integration |
| SIP | `SIPVoiceBackend` | `roomkit[sip]` | Telephony |
| WebTransport | `WebTransportBackend` | `roomkit[webtransport]` | Low-latency web |
| Mock | `MockVoiceBackend` | built-in | Testing |

```python
from roomkit.voice import get_local_audio_backend

LocalAudioBackend = get_local_audio_backend()
backend = LocalAudioBackend(sample_rate=16000, channels=1)
```

## Interruption Handling

Four strategies for handling user speech during TTS playback:

```python
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy

voice = VoiceChannel(
    "voice",
    stt=stt,
    tts=tts,
    backend=backend,
    pipeline=AudioPipelineConfig(vad=vad),
    interruption=InterruptionConfig(
        strategy=InterruptionStrategy.CONFIRMED,
        min_speech_ms=300,  # Wait 300ms of sustained speech before interrupting
    ),
)
```

| Strategy | Behavior |
|----------|----------|
| `IMMEDIATE` | Interrupt on any detected speech |
| `CONFIRMED` | Wait for sustained speech (min_speech_ms). Default. |
| `SEMANTIC` | Use BackchannelDetector to ignore "uh-huh", "yeah" |
| `DISABLED` | Never interrupt TTS playback |

## Voice Greeting

Send a greeting when a session starts:

```python
await kit.send_greeting(
    room_id="call-room",
    channel_id="voice-agent",
    greeting="Welcome! How can I help you today?",
    session=session,
)
```

Or configure on the Agent:

```python
from roomkit import Agent

agent = Agent(
    "voice-agent",
    provider=provider,
    greeting="Welcome! How can I help you today?",
    stt=stt,
    tts=tts,
    backend=backend,
    pipeline=AudioPipelineConfig(vad=vad),
)
```

## Voice Hooks

```python
from roomkit import HookTrigger, HookExecution

@kit.hook(HookTrigger.ON_SPEECH_START, execution=HookExecution.ASYNC)
async def on_speech(event, ctx):
    print("User started speaking")

@kit.hook(HookTrigger.ON_TRANSCRIPTION, execution=HookExecution.ASYNC)
async def on_transcription(event, ctx):
    print(f"Transcription: {event.content.body}")

@kit.hook(HookTrigger.BEFORE_TTS)
async def before_tts(event, ctx):
    # Can modify or block TTS text
    return HookResult.allow()

@kit.hook(HookTrigger.ON_BARGE_IN, execution=HookExecution.ASYNC)
async def on_barge_in(event, ctx):
    print("User interrupted the AI")
```

## Audio Bridging

Bridge audio between sessions for human-to-human voice calls:

```python
voice = VoiceChannel("voice", backend=backend, bridge=True)

# With bridge + STT for live transcription
voice = VoiceChannel("voice", stt=stt, backend=backend, bridge=True)
```

Audio bridge supports N-party calls with mixing and cross-rate resampling.

## DTMF

Send and detect DTMF tones:

```python
# Send DTMF
await voice.send_dtmf(session, digit="1", duration_ms=160)

# Detect DTMF via hook
@kit.hook(HookTrigger.ON_DTMF, execution=HookExecution.ASYNC)
async def on_dtmf(event, ctx):
    print(f"DTMF digit: {event.data['digit']}")
```
---

The audio pipeline sits between the voice backend and STT/TTS, processing audio through pluggable stages.

## Pipeline Architecture

```
Inbound:   Backend -> [Resampler] -> [Recorder] -> [AEC] -> [AGC] -> [Denoiser] -> VAD -> [Diarization] + [DTMF]
Outbound:  TTS -> [PostProcessors] -> [Recorder] -> AEC.feed_reference -> [Resampler] -> Backend
```

All stages are optional except VAD (required for speech detection). Stages in brackets are skipped if not configured.

## Pipeline Configuration

```python
from roomkit import VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, VADConfig

pipeline = AudioPipelineConfig(
    resampler=my_resampler,           # Sample rate conversion
    vad=my_vad_provider,              # Voice activity detection (required)
    denoiser=my_denoiser,             # Background noise removal
    diarization=my_diarizer,          # Speaker identification
    aec=my_aec,                       # Echo cancellation
    agc=my_agc,                       # Automatic gain control
    dtmf=my_dtmf_detector,            # DTMF tone detection
    recorder=my_recorder,             # Audio recording
    recording_config=my_rec_config,   # Recording settings
    turn_detector=my_turn_detector,   # Turn-taking detection
    vad_config=VADConfig(
        silence_threshold_ms=500,     # Silence before speech_end
    ),
)

voice = VoiceChannel(
    "voice",
    stt=stt,
    tts=tts,
    backend=backend,
    pipeline=pipeline,
)
```

## Capability-Aware Skipping

AEC and AGC stages automatically skip when the backend declares native capabilities:

```python
from roomkit.voice.base import VoiceCapability

# If backend has NATIVE_AEC, pipeline skips the AEC stage
# If backend has NATIVE_AGC, pipeline skips the AGC stage
```

## Pipeline Stages Reference

### Resampler

Converts audio between sample rates (e.g., 8kHz SIP to 16kHz STT).

```python
from roomkit.voice.pipeline.resampler import LinearResamplerProvider

resampler = LinearResamplerProvider()
# Or use SincResampler for higher quality
```

### VAD (Voice Activity Detection)

Detects speech start/end events. Required for the pipeline.

```python
from roomkit.voice.pipeline import MockVADProvider, VADEvent, VADEventType, VADConfig

# Mock for testing
vad = MockVADProvider(events=[
    VADEvent(type=VADEventType.SPEECH_START),
    None,  # No event for this frame
    VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech-data"),
])

# Production: SherpaOnnx VAD (local, offline)
from roomkit.voice import get_sherpa_onnx_vad_provider, get_sherpa_onnx_vad_config

SherpaVAD = get_sherpa_onnx_vad_provider()
SherpaVADConfig = get_sherpa_onnx_vad_config()
vad = SherpaVAD(SherpaVADConfig(threshold=0.5))
```

### AEC (Acoustic Echo Cancellation)

Removes echo from the microphone signal caused by speaker output.

```python
# Speex AEC
from roomkit.voice import get_speex_aec_provider

SpeexAEC = get_speex_aec_provider()
aec = SpeexAEC(sample_rate=16000, frame_size=160, filter_length=1024)

# WebRTC AEC
# pip install roomkit[webrtc-aec]
```

The pipeline feeds TTS audio as reference to the AEC via `process_outbound()`.
Reference audio is converted to the exact post-resampler capture format, and
AEC state and playback activation are isolated per voice session.
While playback is active, realtime backends keep the render timeline aligned
with capture by feeding hardware silence during jitter gaps. Ending playback
bypasses AEC but preserves its converged filter until the session ends.

### AGC (Automatic Gain Control)

Normalizes audio volume levels. AGC does not remove background noise; it makes
quiet and loud speech reach downstream denoising, VAD, and STT at a predictable
level.

```python
from roomkit.voice.pipeline import AGCConfig, AudioPipelineConfig, SimpleAGCProvider

agc = SimpleAGCProvider(AGCConfig(target_level_dbfs=-12.0))

# Or let the pipeline create the built-in provider from the config:
pipeline = AudioPipelineConfig(agc_config=AGCConfig(target_level_dbfs=-12.0))
```

Gain state is isolated per stream, near-silence is not amplified, and a peak
limiter prevents PCM clipping. Processed frames expose
`metadata.gain_applied_db`. Backends declaring `NATIVE_AGC` skip this stage.

### Denoiser

Removes background noise from audio.

```python
# Continuous WebRTC NS (same optional dependency as WebRTC AEC)
from roomkit.voice.pipeline import WebRTCNoiseSuppressorProvider

denoiser = WebRTCNoiseSuppressorProvider(sample_rate=24000)

# RNNoise (local, CPU-based)
from roomkit.voice import get_rnnoise_denoiser_provider

RNNoise = get_rnnoise_denoiser_provider()
denoiser = RNNoise()

# ai|coustics Quail (cloud API)
# pip install roomkit[aicoustics]

# SherpaOnnx denoiser (local ONNX model)
from roomkit.voice import get_sherpa_onnx_denoiser_provider, get_sherpa_onnx_denoiser_config

SherpaDenoiser = get_sherpa_onnx_denoiser_provider()
SherpaDenoiserConfig = get_sherpa_onnx_denoiser_config()
denoiser = SherpaDenoiser(SherpaDenoiserConfig())
```

WebRTC NS, RNNoise, and ai|coustics accept arbitrary caller chunk sizes without
duplicating or dropping audio; an irregular stream uses a fixed one-native-block
delay (10 ms for WebRTC NS and RNNoise). All built-in denoisers validate PCM
format before creating native state and keep model history isolated per stream.

### Diarization

Identifies different speakers in multi-speaker audio.

```python
from roomkit.voice.pipeline.diarization.mock import MockDiarizationProvider

diarizer = MockDiarizationProvider()
```

### DTMF Detection

Detects dual-tone multi-frequency signals (phone keypad tones).

```python
from roomkit.voice.pipeline.dtmf.mock import MockDTMFDetector

dtmf = MockDTMFDetector()
```

DTMF runs in parallel with other pipeline stages (before AEC/AGC/denoiser).

### Audio Recorder

Records inbound and outbound audio.

```python
from roomkit.voice.pipeline.recorder.mock import MockAudioRecorder
from roomkit.voice.pipeline.recorder import RecordingConfig

recorder = MockAudioRecorder()
config = RecordingConfig(
    format="wav",
    sample_rate=16000,
    channels=1,
)
```

### Turn Detector

Determines when a speaker's turn is complete (post-STT):

```python
from roomkit.voice.pipeline.turn.mock import MockTurnDetector

turn = MockTurnDetector()

# Production: Smart turn detection (ML-based)
# pip install roomkit[smart-turn]
```

Turn detection accumulates transcription fragments until `is_complete=True`, then routes the combined text.

### Backchannel Detector

Classifies short utterances as backchannel (e.g., "uh-huh", "yeah") to prevent false interruptions:

```python
from roomkit.voice.pipeline.backchannel.mock import MockBackchannelDetector

bc = MockBackchannelDetector()
```

Used with `InterruptionStrategy.SEMANTIC`.

### Post-Processors

Custom audio transformations on outbound TTS audio:

```python
from roomkit.voice.pipeline.postprocessor.base import AudioPostProcessor
```

## Interruption Handling

The `InterruptionHandler` manages what happens when the user speaks during TTS playback:

```python
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy

config = InterruptionConfig(
    strategy=InterruptionStrategy.CONFIRMED,
    min_speech_ms=300,
    allow_during_first_ms=600,
)
```

For `RealtimeVoiceChannel` with provider-side VAD, `allow_during_first_ms`
guards the provider input after physical playback begins. RoomKit continues to
process the real microphone signal through AEC, AGC, denoising, and recording,
but forwards equal-duration PCM silence to the provider until the guard expires.
This gives an acoustic echo canceller time to converge without letting residual
onset echo trigger a server-side interruption. The guard requires a transport
with playback callbacks; set it to `0` on a headset or another already-clean
audio path when immediate barge-in is more important.

| Strategy | When to Use |
|----------|-------------|
| `IMMEDIATE` | Fast response, accept false positives |
| `CONFIRMED` | Balanced — waits for sustained speech |
| `SEMANTIC` | Ignore backchannel ("uh-huh") using BackchannelDetector |
| `DISABLED` | Never interrupt (e.g., announcements) |

## AudioFrame

Inbound audio is represented as `AudioFrame`:

```python
from roomkit.voice.audio_frame import AudioFrame

frame = AudioFrame(
    data=b"\x00" * 320,   # Raw PCM bytes
    sample_rate=16000,     # Hz
    channels=1,            # Mono
    sample_width=2,        # 16-bit PCM
    timestamp_ms=0.0,
)
```

Pipeline stages annotate `frame.metadata` as they process: `denoiser`, `vad`, `aec`, `agc`, `diarization`, `dtmf` keys.

## Mock Providers for Testing

Every pipeline stage has a mock provider with pre-configured event sequences:

```python
from roomkit.voice.pipeline import (
    MockVADProvider,
    MockDenoiserProvider,
    MockDiarizationProvider,
    MockAGCProvider,
    MockAECProvider,
    MockDTMFDetector,
    MockAudioRecorder,
    MockTurnDetector,
    MockBackchannelDetector,
)
```

Example with mock VAD:

```python
from roomkit.voice.pipeline import MockVADProvider, VADEvent, VADEventType

vad = MockVADProvider(events=[
    VADEvent(type=VADEventType.SPEECH_START),
    None,
    VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech"),
])
```
---

RealtimeVoiceChannel connects to speech-to-speech AI models that handle audio directly — no separate STT/TTS needed. The AI model receives audio and responds with audio.

## Basic Setup

```python
from roomkit import RoomKit, ChannelCategory
from roomkit.channels.realtime_voice import RealtimeVoiceChannel

kit = RoomKit()

# Gemini Live example
from roomkit.voice import get_gemini_live_provider

GeminiLiveProvider = get_gemini_live_provider()

realtime = RealtimeVoiceChannel(
    "realtime-voice",
    provider=GeminiLiveProvider(
        api_key="...",
        model="gemini-2.0-flash-live-001",
    ),
    system_prompt="You are a helpful voice assistant. Keep responses brief.",
)
kit.register_channel(realtime)

await kit.create_room(room_id="voice-room")
await kit.attach_channel("voice-room", "realtime-voice", category=ChannelCategory.INTELLIGENCE)
```

## Providers

| Provider | Class | Extra | Description |
|----------|-------|-------|-------------|
| Google Gemini Live | `GeminiLiveProvider` | `roomkit[realtime-gemini]` | Gemini 2.0 speech-to-speech |
| OpenAI Realtime | `OpenAIRealtimeProvider` | `roomkit[realtime-openai]` | GPT Realtime audio (reasoning-capable) |
| xAI Grok | `XAIRealtimeProvider` | — | Grok speech-to-speech |
| Deepgram Voice Agent | `DeepgramAgentProvider` | `roomkit[realtime-deepgram]` | Nova listen + LLM think + Aura speak, each chosen separately |
| Mock | `MockRealtimeProvider` | built-in | Testing |

```python
# OpenAI Realtime
from roomkit.voice import get_openai_realtime_provider

OpenAIRealtime = get_openai_realtime_provider()
provider = OpenAIRealtime(api_key="sk-...")  # defaults to gpt-realtime-2.1

# xAI Grok
from roomkit.voice import get_xai_realtime_provider, get_xai_realtime_config

XAIRealtime = get_xai_realtime_provider()
XAIConfig = get_xai_realtime_config()
provider = XAIRealtime(XAIConfig(api_key="..."))
```

## Audio Transports

Transports handle the client-side audio (browser or device):

| Transport | Class | Extra | Use Case |
|-----------|-------|-------|----------|
| WebSocket | `WebSocketRealtimeTransport` | `roomkit[websocket]` | Browser via WebSocket |
| FastRTC (WebRTC) | `FastRTCRealtimeTransport` | `roomkit[fastrtc]` | Browser via WebRTC |
| Local mic/speaker | `LocalAudioBackend` | `roomkit[local-audio]` | Local development |
| Mock | `MockRealtimeTransport` | built-in | Testing |

```python
from roomkit.voice import get_websocket_realtime_transport

WSTransport = get_websocket_realtime_transport()
transport = WSTransport(host="0.0.0.0", port=8765)

realtime = RealtimeVoiceChannel(
    "realtime-voice",
    provider=provider,
    transport=transport,
    system_prompt="You are a helpful assistant.",
)
```

## Joining a Session

```python
session = await kit.join(
    room_id="voice-room",
    channel_id="realtime-voice",
    participant_id="caller-1",
)

# Leave when done
await kit.leave(session)
```

## Tool Calling

Realtime voice channels support tool calling during conversations:

```python
from roomkit.channels.realtime_voice import RealtimeVoiceChannel, ToolHandler

async def handle_tool(name: str, arguments: dict) -> str:
    if name == "get_weather":
        return '{"temperature": 22, "condition": "sunny"}'
    return '{"error": "unknown tool"}'

realtime = RealtimeVoiceChannel(
    "realtime-voice",
    provider=provider,
    system_prompt="You help with weather. Use the get_weather tool.",
    tools=[
        {
            "name": "get_weather",
            "description": "Get weather for a city",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    ],
    tool_handler=handle_tool,
)
```

## Text Injection

Inject text into an active realtime session (appears as if the AI said it):

```python
session = await kit.join("room", "realtime-voice", participant_id="user")
await realtime.inject_text(session, "Let me check that for you...")
```

## VAD Configuration

Configure voice activity detection for realtime providers:

```python
realtime = RealtimeVoiceChannel(
    "realtime-voice",
    provider=provider,
    system_prompt="...",
    vad_config={
        "threshold": 0.5,
        "silence_duration_ms": 500,
    },
)
```

## Session Resumption

Some providers support session resumption after disconnection:

```python
# OpenAI Realtime supports session resumption
provider = OpenAIRealtime(
    api_key="sk-...",
    model="gpt-realtime-2.1",
)
# Sessions automatically resume when possible
```

## Hooks

```python
from roomkit import HookTrigger, HookExecution

@kit.hook(HookTrigger.ON_TOOL_CALL, execution=HookExecution.ASYNC)
async def on_tool(event, ctx):
    print(f"Realtime tool call: {event}")

@kit.hook(HookTrigger.ON_REALTIME_TEXT_INJECTED, execution=HookExecution.ASYNC)
async def on_inject(event, ctx):
    print(f"Text injected: {event}")
```
---

Real-time video channels with pluggable processing pipelines, vision AI, bridging, recording, and avatar output.

## Video Channels

| Channel | ChannelType | Use Case |
|---------|-------------|----------|
| `VideoChannel` | `VIDEO` | Video-only transport (webcam, screen, WebSocket, WebRTC) |
| `AudioVideoChannel` | `AUDIO_VIDEO` | Combined A/V backends (SIP, RTP, FastRTC); extends `VoiceChannel` |
| `RealtimeAudioVideoChannel` | `REALTIME_AUDIO_VIDEO` | Speech-to-speech AI with video (e.g. Anam avatar); extends `RealtimeVoiceChannel` |

```python
from roomkit import RoomKit, VideoChannel
from roomkit.video import VideoPipelineConfig
from roomkit.video.backends.local import LocalVideoBackend
from roomkit.video.vision.mock import MockVisionProvider

kit = RoomKit()
video = VideoChannel(
    "video-main",
    backend=LocalVideoBackend(device=0, fps=30, width=640, height=480),
    pipeline=VideoPipelineConfig(vision=MockVisionProvider()),
    vision_interval_ms=2000,    # vision analysis throttle
    bridge=True,                # optional: or VideoBridgeConfig
)
kit.register_channel(video)

@kit.on("video_vision_result")
async def on_vision(event):
    print(event.data["description"], event.data.get("labels"))

session = await kit.join("webcam-demo", "video-main", participant_id="local-user")
await video.backend.start_capture(session)   # LocalVideoBackend-specific
# ... later
await kit.leave(session)
```

`AudioVideoChannel` adds `video_pipeline=`, `vision=`, `vision_interval_ms=`, `avatar=`, `avatar_encoder=`, `video_bridge=` to the `VoiceChannel` kwargs; its `backend` must implement both `VoiceBackend` and `VideoBackend`. `RealtimeAudioVideoChannel` adds `video_pipeline=`, `vision=`, `vision_interval_ms=`; its provider must implement `RealtimeAudioVideoProvider.on_video()`.

## VideoFrame

Inbound frames are `VideoFrame` dataclasses: `data: bytes`, `codec="h264"`, `width=640`, `height=480`, `timestamp_ms`, `keyframe=False`, `sequence`, `metadata`. Encoded codecs: `h264`, `vp8`, `vp9`, `av1`; raw: `raw_rgb24`, `raw_bgr24`, `raw_yuv420p`, `raw_nv12` (`is_encoded`/`is_raw` properties). Outbound streaming uses `VideoChunk` (encoded only).

## Video Pipeline

Inbound order: `[Decoder] -> [Resizer] -> [Transforms...] -> [Filters...] -> taps/bridge/vision`. All stages optional; decoder, resizer, and vision are singular, transforms and filters are chained lists. Vision runs async on its own throttled schedule; the rest is synchronous per-frame.

```python
from roomkit.video.pipeline.decoder.pyav import PyAVVideoDecoder
from roomkit.video.pipeline.resizer.pyav import PyAVVideoResizer

pipeline = VideoPipelineConfig(
    decoder=PyAVVideoDecoder(output_format="rgb24"),
    resizer=PyAVVideoResizer(width=640, height=480, keep_aspect=True),
    transforms=[], filters=[], vision=vision,
    recorder=recorder,
    recording_config=VideoRecordingConfig(storage="./recordings", format="mp4",
                                          codec="auto", fps=15.0),
)
```

| Stage | Role | Implementations |
|-------|------|-----------------|
| Decoder (`VideoDecoderProvider`) | Encoded → raw pixels; returns `None` until keyframe | `PyAVVideoDecoder`, mock |
| Resizer (`VideoResizerProvider`) | Downscale raw frames to fit target box | `PyAVVideoResizer`, mock |
| Transform (`VideoTransformProvider`) | Per-frame pixel modification (fast, sync) | `VideoEffectTransform(effect=...)`: grayscale, sepia, invert, blur, cartoon, edges, sketch, pixelate; mock |
| Filter (`VideoFilterProvider`) | Inspect/replace frames via `FilterContext` (latest vision result); emit `FilterEvent`s | `YOLODetectorFilter`, `WatermarkFilter`, `CensorVideoFilter`, `FaceTouchFilter` (MediaPipe), `OverlayFilter`, mocks |
| Vision (`VisionProvider`) | Async frame analysis → `VisionResult` | `OpenAIVisionProvider`, `GeminiVisionProvider`, mock |
| Recorder (`VideoRecorder`) | Tap every frame to file | `PyAVVideoRecorder`, `OpenCVVideoRecorder`, mock |
| Encoder (`VideoEncoderProvider`) | Outbound: raw → H.264 NALs (avatar path) | `PyAVVideoEncoder(width=512, height=512, fps=30, codec="libx264")` |

Key filters: `YOLODetectorFilter(model="yolo26n.pt", confidence=0.5, classes=None, every_n_frames=1, draw_boxes=False)`; `CensorVideoFilter(blocked_labels, replacement="black", grace_frames=0)` blanks frames while vision reports a blocked label; `FaceTouchFilter` emits `face_touch` detections. Overlays (`roomkit.video.pipeline.overlay`): `TextOverlayRenderer`, `ImageOverlayRenderer`, `SubtitleManager`/`subtitle_overlay` for live subtitles.

## Video Backends

`VideoBackend` ABC: `connect()`, `accept()`, `disconnect()`, `send_video()`, `send_video_sync()`, `request_keyframe()`, `set_video_passthrough()`; callbacks `on_video_received`, `on_session_ready`, `on_client_disconnected`; `capabilities` flags: `SIMULCAST`, `SVC`, `SCREEN_SHARE`, `RECORDING`, `BANDWIDTH_ESTIMATION`.

| Backend | Class | Extra |
|---------|-------|-------|
| WebSocket | `WebSocketVideoBackend` | fastapi only |
| WebRTC | `FastRTCVideoBackend` (combined A/V) | `roomkit[fastrtc]` |
| RTP | `RTPVideoBackend` (combined A/V) | `roomkit[rtp]` |
| SIP | `SIPVideoBackend` (combined A/V) | `roomkit[sip]` |
| Webcam | `LocalVideoBackend(device=0, fps=30, width=640, height=480)` | `roomkit[local-video]` |
| Screen | `ScreenCaptureBackend(monitor=1, region=None, fps=5, scale=1.0, diff_threshold=0.0)` | `roomkit[screen-capture]` |
| Mock | `MockVideoBackend` | built-in |

`WebSocketVideoBackend` mounts on FastAPI via `mount_websocket_video(app, backend, path=...)` and auto-creates sessions via `set_session_factory()`. Each backend has a `get_*` lazy loader in `roomkit.video` (e.g. `get_sip_video_backend()`).

## Vision Providers

```python
from roomkit.video import OpenAIVisionConfig, OpenAIVisionProvider

# Defaults target Ollama: base_url="http://localhost:11434/v1", model="qwen3.5"
vision = OpenAIVisionProvider(OpenAIVisionConfig(
    api_key="sk-...", base_url="https://api.openai.com/v1",
    model="gpt-4o", detail="low", max_tokens=100,
))
# GeminiVisionProvider(GeminiVisionConfig(api_key="AIza...", model="gemini-3.1-flash-lite"))
```

`await provider.analyze_frame(frame, prompt=None)` returns `VisionResult(description, labels, confidence, faces, text, metadata)` — `text` is OCR, `faces` are `FaceDetection` boxes. Results are cached per session (`channel.get_last_vision_result(session_id)`), emitted as `video_vision_result` framework events, and auto-injected into the system prompt of any `AIChannel` in the same room. For realtime voice, `setup_realtime_vision(kit, room_id, voice_channel_id)` injects via `inject_text(silent=True)`. On-demand tools: `DescribeWebcamTool`, `DescribeScreenTool`, `ListWebcamsTool`.

## Video Hook Triggers

| Trigger | Fires When | Execution |
|---------|-----------|-----------|
| `ON_VIDEO_SESSION_STARTED` | Backend signals video path live and session bound | async |
| `ON_VIDEO_SESSION_ENDED` | Session unbound / client disconnected | async |
| `ON_VISION_RESULT` | Vision analysis completed (`VisionEvent`) — can block or modify the description | sync |
| `ON_VIDEO_DETECTION` | Pipeline filter emitted a detection (`VideoDetectionEvent`: `kind`, `labels`, `confidence`, `metadata`) | async |
| `BEFORE_BRIDGE_VIDEO` | Frame about to be bridge-forwarded (`BridgeVideoEvent`); `HookResult.block()` drops it | sync |
| `ON_SCREEN_SHARE_STARTED` / `STOPPED` | ConferenceChannel: `SCREEN_SHARE`-kind track published/unpublished | async |
| `ON_VIDEO_TRACK_ADDED` / `REMOVED` | Reserved — defined in `HookTrigger`, not yet fired by built-in channels | — |

## Video Bridge

`bridge=True` (or `VideoBridgeConfig(enabled=True, max_participants=10, forwarding_strategy="forward", keyframe_interval_s=5.0)`) forwards frames between sessions in the same room for human-to-human video, requesting keyframes (PLI) from sources periodically and when receivers join. `channel.set_bridge_filter(fn)` installs a synchronous per-frame filter `(source_session, frame) -> frame | None` — the fast-path alternative to `BEFORE_BRIDGE_VIDEO`.

## Avatars

`AvatarProvider` (`roomkit.video.avatar`) generates lip-synced video from TTS audio on `AudioVideoChannel`: `MuseTalkAvatarProvider` (local GPU), `WebSocketAvatarProvider` (remote inference), `MockAvatarProvider`. Pair with `avatar_encoder=PyAVVideoEncoder(...)` for RTP/SIP output. Cloud avatar A/V (Anam) uses `RealtimeAudioVideoChannel` instead.

## Optional Extras

`roomkit[video]` (PyAV stages: av, numpy) · `[local-video]` (webcam, effects, OpenCV recorder) · `[screen-capture]` (mss) · `[screen-input]` (pyautogui) · `[yolo]` (ultralytics) · `[mediapipe]` · `[video-overlay]` (Pillow) · `[fastrtc]`/`[rtp]`/`[sip]`/`[anam]` (transports / cloud avatar).
---

ConferenceChannel bridges an external SFU conference into a RoomKit room. The SFU carries all human-to-human media; RoomKit joins as a single bot participant for transcription, AI voice, recording, and speech-to-speech — it never proxies human media (RFC §12.10).

## Constructor

```python
ConferenceChannel(
    channel_id,
    *,
    backend,                 # ConferenceBackend (required)
    stt=None,                # STTProvider — per-track transcription
    tts=None,                # TTSProvider — the bot's voice
    realtime=None,           # ConferenceRealtimeConfig — speech-to-speech (excludes tts)
    pipeline=None,           # default: 16 kHz mono contract + EnergyVADProvider
    interruption=None,       # ConferenceInterruptionConfig
    recording=None, recorder=None,   # ConferenceRecordingConfig + MediaRecorder, both or neither
    bot_identity="roomkit",
    bot_grants=None,         # explicit ConferenceGrants; default derived via for_bot()
    default_grants=None,     # what mint_access() grants humans; default ConferenceGrants()
    e2ee=False, close_room_on_detach=False,
    speak_text_events=False, # off: only AI-channel text events are spoken
    close_providers=True,
    max_queued_frames=100,   # per-track backpressure bound (lane + recording)
    identity_address_keys=None, identity_trusts_unasserted_metadata=False,
)
```

Refused at construction (and identically at plug time): `e2ee=True` with stt/recording/realtime (bot receives ciphertext); `ConferenceRecordingMode.EGRESS` (only `FRAMEWORK` is implemented); `tts` + `realtime` together (one bot track, one voice); `realtime.tools` without `tool_handler`; a `pipeline` without a VAD when stt/realtime is set.

Public surface: `mint_access()`, `plug_*/unplug_*()`, `set_bot_grants()`, `may_interrupt(participant_id)`, `active_lanes` (dict `track_id -> ConferenceLane`; `drain()`, `dropped_frames`), `info()`, `close()`.

## How the Bridge Works

The bot joins lazily, only when the channel has a need (stt/tts/recording/realtime) and someone is coming: a `mint_access()`, an arrival, an occupancy probe. With no need configured the channel is pure transport — admission gate and roster, no bot in the meeting.

- **Inbound**: each subscribed AUDIO track gets its own `ConferenceLane` (queue + task + VAD state). The VAD segments utterances; one utterance → one STT call → one `RoomEvent` attributed to the track's `participant_id` (track identity replaces diarization). A full lane drops oldest frames.
- **Outbound**: `deliver()` speaks AI-channel TextContent on the single bot track via TTS (all text events if `speak_text_events=True`); with `realtime`, text is injected into the provider's context instead.
- **Video**: never subscribed — `capabilities()` announces AUDIO only.

```python
from roomkit import ConferenceTranscription, HookResult, HookTrigger

@kit.hook(HookTrigger.ON_TRANSCRIPTION)   # sync, runs BEFORE the room; block/modify = redaction
async def on_text(payload: ConferenceTranscription, ctx) -> HookResult:
    # payload: track_id, participant_id, room_id, text
    return HookResult.allow()
```

## Conference Hooks

SYSTEM lifecycle events; payload in `event.content.data` (always includes `channel_id`).

| Trigger | Fired when | Data |
|---------|-----------|------|
| `ON_CONFERENCE_PARTICIPANT_JOINED` / `_LEFT` | SFU reports arrival/departure | `participant_id` |
| `ON_CONFERENCE_TRACK_PUBLISHED` / `_UNPUBLISHED` | Track publish/unpublish | `track_id`, `participant_id`, `kind` |
| `ON_CONFERENCE_TRACK_MUTED` / `_UNMUTED` | Publisher (un)mutes; muted VIDEO = camera off | `track_id`, `participant_id`, `kind` |
| `ON_SCREEN_SHARE_STARTED` / `_STOPPED` | SCREEN_SHARE track published/unpublished | `track_id`, `participant_id` |
| `ON_ACTIVE_SPEAKER_CHANGED` | Dominant speaker (`ACTIVE_SPEAKER` capability) | `participant_id` |
| `ON_CONNECTION_QUALITY_CHANGED` | Quality report (backend label, not normalized) | `participant_id`, `quality` |

`ON_SPEECH_START`/`ON_SPEECH_END` fire per lane at VAD edges. `ON_BARGE_IN` carries `ConferenceBargeIn(room_id, track_id, participant_id, interrupted_text, audio_position_ms)`.

## ConferenceBackend ABC (`roomkit.conference`)

- Control plane: `ensure_room(room_id, metadata=None, e2ee=False)`, `close_room()`, `mint_access(room_id, participant_id, grants, *, display_name=None) -> ConferenceAccess`, `list_participants()`, `remove_participant()`, `mute_track()`, `unmute_track()` (needs `REMOTE_UNMUTE`).
- Bot session: `join_as_bot(room_id, identity, grants) -> BotSession`, `leave(bot)`, `update_bot_grants(bot, grants)` (needs `BOT_GRANT_UPDATE`, else `ConferenceCapabilityError`), `subscribe_track(bot, track_id)`, `unsubscribe_track()`, `publish_audio(bot, chunk)` (PCM `AudioChunk`; `is_final` ends the utterance), `stop_playback(bot)` (barge-in: discard queued unplayed audio; utterance still ends on `is_final`; no-op for a gone session), `publish_video()` (needs `VIDEO_PUBLISH`), `close()`.
- Callbacks (what drives the channel): `on_participant_joined/left`, `on_track_published/unpublished/muted/unmuted`, `on_track_audio`, `on_track_video`, `on_active_speaker_changed`, `on_connection_quality`, `on_bot_session_ended` (SFU dropped the bot without a `leave()`).
- `ConferenceCapability` flags: `SCREEN_SHARE`, `EGRESS_RECORDING`, `SIP_GATEWAY`, `ACTIVE_SPEAKER`, `CONNECTION_QUALITY`, `VIDEO_PUBLISH`, `REMOTE_UNMUTE`, `BOT_GRANT_UPDATE`, `E2EE`.

## Backends

| Backend | Class | Config | Extra |
|---------|-------|--------|-------|
| LiveKit | `LiveKitConferenceBackend` | `LiveKitConfig` | `roomkit[livekit]` |
| Mock | `MockConferenceBackend` | `capabilities=` kwarg | built-in |

```python
from roomkit.conference.livekit import LiveKitConfig, LiveKitConferenceBackend

backend = LiveKitConferenceBackend(LiveKitConfig(url="wss://my-project.livekit.cloud"))
```

`LiveKitConfig`: `url`/`api_key`/`api_secret` fall back to `LIVEKIT_URL`/`LIVEKIT_API_KEY`/`LIVEKIT_API_SECRET`; `access_ttl=timedelta(minutes=15)`, `audio_sample_rate=48_000`, `audio_channels=1`, `publish_queue_ms=300`, `remote_unmute=False`, `sip_gateway=False`, `room_metadata_key="roomkit"`. Declares `SCREEN_SHARE | ACTIVE_SPEAKER | CONNECTION_QUALITY | BOT_GRANT_UPDATE` (+`REMOTE_UNMUTE`/`SIP_GATEWAY` when configured).

`MockConferenceBackend` scripts SFU events: `simulate_participant_joined/left`, `simulate_track_published(room_id, participant_id, kind=TrackKind.AUDIO)`, `simulate_track_unpublished`, `simulate_audio(track, frame)`, `simulate_track_muted/unmuted`, `simulate_active_speaker`, `simulate_connection_quality`, `simulate_bot_disconnected(bot, reason)`; fault injection `fail(method, exc)` / `delay(method, seconds)`; assertion state `calls`, `published_audio`, `utterances`, `subscriptions`, `playback_stops`.

## Models & Grants

- `ConferenceGrants(publish_audio=True, publish_video=True, publish_screen_share=True, subscribe=True, moderate=False, hidden=False)` — human defaults. `ConferenceGrants.for_bot(speaks=False, listens=True)` = least privilege; `ConferenceGrants.observer()` = subscribe-only + hidden.
- `ConferenceAccess(url, token, expires_at, provider_data)` — opaque client credential; `token` excluded from repr.
- `BotSession(id, room_id, identity, joined_at, metadata)`; `ConferenceTrack(id, room_id, participant_id, kind, muted, metadata)`; `ConferenceParticipant(participant_id, display_name, connected_at, tracks, metadata, asserted_metadata)`; `TrackKind.AUDIO/VIDEO/SCREEN_SHARE`.
- `ConferenceInterruptionConfig(strategy=InterruptionStrategy.IMMEDIATE, scope=ConferenceInterruptionScope.ANY, allowlist=[])` — scope `ANY`/`NONE`/`ALLOWLIST` decides who may barge in.
- `ConferenceRealtimeConfig(provider, system_prompt=None, voice=None, tools=None, tool_handler=None, temperature=None, input_sample_rate=24000, output_sample_rate=24000, server_vad=True, provider_config=None)` — lanes mixed N→1 into one speech-to-speech session per room; the provider speaks on the bot track.

## Hot-Plugging

Effects are in force when the plug returns: occupied conferences joined, published tracks subscribed, bot grants realigned (in place with `BOT_GRANT_UPDATE`, announced re-join otherwise). An occupied slot is refused — swap = unplug, then plug. Unplugging the last need makes the bot leave every conference.

```python
await channel.plug_stt(stt)          # optional pipeline=; joins an existing pipeline
await channel.plug_tts(tts)
await channel.plug_recording(ConferenceRecordingConfig(), recorder=recorder)
await channel.plug_realtime(config)
await channel.unplug_stt(); await channel.unplug_tts()
await channel.unplug_recording(); await channel.unplug_realtime()
await channel.set_bot_grants(ConferenceGrants.observer())  # None returns to derivation
```

## Shutdown Semantics (RFC §12.10.4)

`close()` runs exactly one shutdown per channel (`ConferenceShutdownCoordinator`): concurrent callers join the same shielded task and the terminal result is replayed on later calls. Order: close admission → stop playbacks → wait teardowns/joins → disconnect realtime sessions → bots leave every conference → close lanes → finalize recordings → close backend → close providers → settle roster writes. Every backend/provider call holds an operation lease; a resource still leased is **retained**, not closed — it closes in the background when its last lease returns, and the close **fails**: `ConferenceCloseError` aggregates structured `CloseIssue`s (`component`, `operation`, `status` in `FAILED`/`TIMED_OUT`/`ABANDONED`/`RETAINED`, `step`, `detail`) plus any bot session that could not be removed (still reported by `info()`). All waits are bounded — a slow store never holds a bot in a meeting.

`info()` answers RFC §17.7 disclosure per room: `bot_present`, `bot_hidden`, `stt_active`, `realtime_active`, `recording_active`, `recording_dropped_frames`, `active_lanes`, `collecting`, `leave_failed`.

## Minimal Example

```python
import asyncio
from roomkit import MockConferenceBackend, RoomKit
from roomkit.channels.conference import ConferenceChannel
from roomkit.voice.stt.mock import MockSTTProvider

async def main() -> None:
    backend = MockConferenceBackend()
    channel = ConferenceChannel("conf", backend=backend, stt=MockSTTProvider(transcripts=["Hi."]))
    kit = RoomKit()
    kit.register_channel(channel)
    await kit.create_room("standup")
    await kit.attach_channel("standup", "conf")

    # Mint SFU credentials for a human client; the mint also starts the lazy bot join
    await kit.ensure_participant("standup", "conf", "alice", display_name="Alice")
    access = await channel.mint_access("standup", "alice")  # hand access.url/token to the client

    # With a real backend, SFU events drive these
    await backend.simulate_participant_joined("standup", "alice", display_name="Alice")
    mic = await backend.simulate_track_published("standup", "alice")
    # feed AudioFrames: simulate_audio(mic, frame) — speech, then silence — then
    # await channel.active_lanes[mic.id].drain() and read kit.store.list_events("standup")
    await kit.close()

asyncio.run(main())
```

Runnable references: `examples/conference_quickstart.py`, `conference_ai_meeting.py`, `conference_realtime_ai.py`, `conference_livekit.py`.
---

RoomKit provides four declarative orchestration strategies for multi-agent workflows. Pass a strategy to `RoomKit(orchestration=...)` or `create_room(orchestration=...)` — agents, routing, handoff tools, and conversation state are wired automatically.

## Strategies

### Pipeline

Linear agent chain: triage -> handler -> resolver. Each agent can only hand off to the next in sequence.

```python
from roomkit import Agent, Pipeline, RoomKit, WebSocketChannel
from roomkit.providers.ai.mock import MockAIProvider

triage = Agent(
    "triage",
    provider=MockAIProvider(responses=["Transferring you..."]),
    role="Triage agent",
    description="Routes requests to the right specialist",
    system_prompt="You triage incoming requests.",
)
handler = Agent(
    "handler",
    provider=MockAIProvider(responses=["Let me help with that."]),
    role="Request handler",
    description="Handles customer requests",
    system_prompt="You handle requests.",
)
resolver = Agent(
    "resolver",
    provider=MockAIProvider(responses=["All done!"]),
    role="Resolution specialist",
    description="Confirms resolution",
    system_prompt="You resolve and close requests.",
)

kit = RoomKit(orchestration=Pipeline(agents=[triage, handler, resolver]))
```

### Swarm

Every agent can hand off to every other agent. Bidirectional routing.

```python
from roomkit import Swarm

kit = RoomKit(orchestration=Swarm(agents=[billing, shipping, returns]))
```

### Supervisor

A supervisor agent delegates tasks to worker agents in child rooms:

```python
from roomkit import Supervisor

kit = RoomKit(orchestration=Supervisor(
    supervisor=manager_agent,
    workers=[researcher, writer, reviewer],
))
```

### Loop

Producer/reviewer cycle. The reviewer has an `approve_output` tool to break the loop.

```python
from roomkit import Loop

kit = RoomKit(orchestration=Loop(
    agent=writer_agent,
    reviewer=editor_agent,
    max_iterations=3,
))
```

## Using Orchestration

```python
from roomkit import InboundMessage, TextContent, WebSocketChannel

# Register transport channel
ws = WebSocketChannel("ws-user")
kit.register_channel(ws)

# Create room — orchestration auto-registers agents, creates router, sets initial state
await kit.create_room(room_id="support")
await kit.attach_channel("support", "ws-user")

# Messages are automatically routed to the active agent
await kit.process_inbound(
    InboundMessage(
        channel_id="ws-user",
        sender_id="user",
        content=TextContent(body="I need help with billing."),
    )
)
```

## Handoff Protocol

Agents hand off conversations by calling the `handoff_conversation` tool (auto-injected by orchestration strategies):

```python
# The AI calls this tool automatically:
# handoff_conversation(target="handler", reason="Billing issue", summary="User needs invoice help")
```

The handoff:
1. Updates `ConversationState` in room metadata (phase, active agent, handoff count)
2. Mutes the outgoing agent, unmutes the incoming agent
3. Emits a system event visible to the new agent
4. Records the transition in `phase_history`

## Conversation State

`ConversationState` tracks conversation progress within a room. It's stored in `Room.metadata["_conversation_state"]` and persists across all message turns. Orchestration strategies create and update it automatically, but you can also read and modify it directly.

### ConversationState Fields

| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `phase` | `str` | `"intake"` | Current conversation phase. Can be any string. |
| `active_agent_id` | `str \| None` | `None` | Channel ID of the currently active agent. |
| `previous_agent_id` | `str \| None` | `None` | Agent before the last transition. |
| `handoff_count` | `int` | `0` | Total number of agent handoffs. |
| `phase_started_at` | `datetime` | now | When the current phase started. |
| `phase_history` | `list[PhaseTransition]` | `[]` | Immutable audit trail of all transitions. |
| `context` | `dict[str, Any]` | `{}` | **Arbitrary user data** — store custom key-value pairs across turns. |

Built-in phases: `ConversationPhase.INTAKE`, `QUALIFICATION`, `HANDLING`, `ESCALATION`, `RESOLUTION`, `FOLLOWUP`.

### Reading State

```python
from roomkit.orchestration.state import get_conversation_state

room = await kit.get_room("support")
state = get_conversation_state(room)

print(state.phase)            # "handling"
print(state.active_agent_id)  # "billing-agent"
print(state.handoff_count)    # 2
print(state.context)          # {"customer_tier": "premium", "issue_type": "refund"}

# Audit trail
for t in state.phase_history:
    print(f"{t.from_phase} -> {t.to_phase} by {t.from_agent} -> {t.to_agent} ({t.reason})")
```

### Persisting Custom Data Across Turns

Use `state.context` to store arbitrary data that survives across conversation turns:

```python
from roomkit.orchestration.state import get_conversation_state, set_conversation_state

room = await kit.get_room("support")
state = get_conversation_state(room)

# Store custom data in context (immutable update)
updated_state = state.model_copy(update={
    "context": {
        **state.context,
        "customer_tier": "premium",
        "issue_type": "refund",
        "attempts": state.context.get("attempts", 0) + 1,
    }
})

# Persist back to room
updated_room = set_conversation_state(room, updated_state)
await kit.store.update_room(updated_room)
```

### Retrieving Custom Data on Later Turns

```python
room = await kit.get_room("support")
state = get_conversation_state(room)
tier = state.context.get("customer_tier", "standard")
attempts = state.context.get("attempts", 0)
```

### Programmatic Phase Transitions

Use `state.transition()` to change phase and record an audit entry:

```python
from roomkit.orchestration.state import get_conversation_state, set_conversation_state

room = await kit.get_room("support")
state = get_conversation_state(room)

new_state = state.transition(
    to_phase="escalation",
    to_agent="supervisor-agent",
    reason="Customer requested manager",
    metadata={"escalation_priority": "high"},
)

updated_room = set_conversation_state(room, new_state)
await kit.store.update_room(updated_room)
```

### PhaseTransition Audit Record

Each transition creates an immutable `PhaseTransition`:

| Field | Type | Description |
|-------|------|-------------|
| `from_phase` | `str` | Previous phase |
| `to_phase` | `str` | New phase |
| `from_agent` | `str \| None` | Previous agent channel ID |
| `to_agent` | `str \| None` | New agent channel ID |
| `reason` | `str` | Why the transition occurred |
| `timestamp` | `datetime` | When the transition occurred (UTC) |
| `metadata` | `dict[str, Any]` | Arbitrary metadata for this transition |

### Using State in Hooks

```python
from roomkit import HookTrigger, HookResult, RoomEvent, RoomContext, TextContent
from roomkit.orchestration.state import get_conversation_state, set_conversation_state

@kit.hook(HookTrigger.BEFORE_BROADCAST)
async def track_sentiment(event: RoomEvent, ctx: RoomContext) -> HookResult:
    if not isinstance(event.content, TextContent):
        return HookResult.allow()
    state = get_conversation_state(ctx.room)
    updated_state = state.model_copy(update={
        "context": {
            **state.context,
            "message_count": state.context.get("message_count", 0) + 1,
        }
    })
    updated_room = set_conversation_state(ctx.room, updated_state)
    await kit.store.update_room(updated_room)
    return HookResult.allow()
```

## Conversation Router (Advanced)

`ConversationRouter` dynamically routes incoming messages to different agents based on conversation state, message content, origin channel, or custom logic.

### RoutingConditions Reference

All conditions are ANDed — every non-None field must match:

| Field | Type | Description |
|-------|------|-------------|
| `phases` | `set[str] \| None` | Match when `state.phase` is in this set |
| `channel_types` | `set[ChannelType] \| None` | Match when sender's channel type is in this set |
| `intents` | `set[str] \| None` | Match when `event.metadata["intent"]` is in this set |
| `source_channel_ids` | `set[str] \| None` | Match when sender's channel ID is in this set |
| `custom` | `Callable \| None` | Custom `(event, context, state) -> bool` for arbitrary logic |

### Routing by Phase

```python
from roomkit.orchestration.router import ConversationRouter, RoutingRule, RoutingConditions

router = ConversationRouter(
    rules=[
        RoutingRule(agent_id="billing-agent", conditions=RoutingConditions(phases={"billing"})),
        RoutingRule(agent_id="shipping-agent", conditions=RoutingConditions(phases={"shipping"})),
    ],
    default_agent_id="triage-agent",
)
```

### Routing by Channel Type (Origin)

```python
from roomkit.models.enums import ChannelType

router = ConversationRouter(
    rules=[
        RoutingRule(agent_id="voice-specialist", conditions=RoutingConditions(channel_types={ChannelType.VOICE})),
        RoutingRule(agent_id="sms-agent", conditions=RoutingConditions(channel_types={ChannelType.SMS, ChannelType.WHATSAPP})),
    ],
    default_agent_id="general-agent",
)
```

### Routing by Intent (Content-Based)

Set `event.metadata["intent"]` via a classification hook, then route by intent:

```python
@kit.hook(HookTrigger.BEFORE_BROADCAST, priority=-200)  # Run before router
async def classify_intent(event: RoomEvent, ctx: RoomContext) -> HookResult:
    if isinstance(event.content, TextContent):
        body = event.content.body.lower()
        intent = "billing" if "invoice" in body or "charge" in body else "general"
        modified = event.model_copy(update={"metadata": {**(event.metadata or {}), "intent": intent}})
        return HookResult.modify(modified)
    return HookResult.allow()

router = ConversationRouter(
    rules=[RoutingRule(agent_id="billing-agent", conditions=RoutingConditions(intents={"billing"}))],
    default_agent_id="general-agent",
)
```

### Routing with Custom Logic

```python
def is_high_value_customer(event, ctx, state):
    return state.context.get("customer_tier") == "premium"

def contains_urgency(event, ctx, state):
    if isinstance(event.content, TextContent):
        return any(w in event.content.body.lower() for w in ["urgent", "emergency", "asap"])
    return False

router = ConversationRouter(
    rules=[
        RoutingRule(agent_id="senior-agent", conditions=RoutingConditions(custom=is_high_value_customer), priority=-1),
        RoutingRule(agent_id="escalation-agent", conditions=RoutingConditions(custom=contains_urgency), priority=0),
    ],
    default_agent_id="general-agent",
    supervisor_id="supervisor-agent",
)
```

### Combined Conditions

Combine multiple conditions (all are ANDed):

```python
# Only route SMS messages during billing phase to the billing specialist
RoutingRule(
    agent_id="sms-billing-agent",
    conditions=RoutingConditions(
        phases={"billing"},
        channel_types={ChannelType.SMS},
    ),
    priority=0,
)
```

### Installing the Router

```python
# Option 1: Manual hook
kit.hook(HookTrigger.BEFORE_BROADCAST, execution=HookExecution.SYNC, priority=-100)(router.as_hook())

# Option 2: install() — also sets up handoff tools
handler = router.install(kit, agents=[billing_agent, shipping_agent, triage_agent])
```

### Routing Selection Priority

0. **Address** — if the event carries `addressed_to`, it decides; the router returns untouched
1. **Agent affinity** — if `state.active_agent_id` is set and attached, stick with it
2. **Rules** — evaluate in ascending `priority` order; first match wins
3. **Fallback** — return `default_agent_id`
4. **Loop prevention** — events FROM intelligence channels are never routed

## Addressing

Routing rules answer *which agent handles this kind of event*. Addressing answers *which agent am I talking to right now*, per message:

```python
await kit.process_inbound(
    InboundMessage(
        channel_id="you",
        sender_id="user",
        content=TextContent(body="review hello.py"),
        addressed_to=["codex"],       # only this agent is asked to act
    )
)
```

| `addressed_to` | Meaning |
|---|---|
| `None` | Unaddressed — every eligible agent acts, or the router decides |
| `["codex"]` | Only `codex` is asked; the others see it and stay silent |
| `[]` | Nobody is asked — a decision, not an absence |

Addressing is **not** visibility: it narrows who is *asked*, never who may *see*. Transport delivery is untouched, so the humans in the room still get the message. The address is stored on the event, so a transcript shows who was asked.

Direct injection addresses the same way — `[]` is what an application needs when it stores a message and triggers the answer itself:

```python
await kit.send_event(
    room_id=room_id,
    channel_id="system",
    content=TextContent(body=body),
    addressed_to=[],       # stored, and asking nobody
)
```

### Agent Response Policy

An agent's own output solicits the other agents by default (the chaining a pipeline needs, bounded by `max_chain_depth`). In a room of independent agents that is a hazard:

```python
# At creation, kit-wide or per room
kit = RoomKit(agent_response_policy=AgentResponsePolicy.ADDRESSED_ONLY)
await kit.create_room(room_id="dev", agent_response_policy=AgentResponsePolicy.ADDRESSED_ONLY)

# On a live room — the one that just gained a second agent
await kit.attach_channel(room_id, "codex", category=ChannelCategory.INTELLIGENCE)
await kit.set_agent_response_policy(room_id, AgentResponsePolicy.ADDRESSED_ONLY)
```

| Policy | An agent's output solicits |
|---|---|
| `AGENT_CHAIN` | every eligible intelligence channel — the default |
| `ADDRESSED_ONLY` | only the channels it addressed, if any |

A policy change applies to events processed after it; setting the policy a room already holds is a no-op. A binding that is not solicited is skipped before any work is done for it, so a roster can be attached lazily and rehydrated one agent at a time.

RoomKit takes the decision, never the syntax: `@codex`, a `/agent` command or a picker all live in the application, which passes channel ids.

## Conversation Pipeline (Advanced)

Define stages explicitly:

```python
from roomkit.orchestration.pipeline import ConversationPipeline, PipelineStage

pipeline = ConversationPipeline(stages=[
    PipelineStage(phase="triage", agent_id="triage", next="handling"),
    PipelineStage(phase="handling", agent_id="handler", next="resolution"),
    PipelineStage(phase="resolution", agent_id="resolver"),
])
```

## Status Bus

Agents can publish status updates for UI display:

```python
# Agents publish status via the status bus
# Subscribe to updates
async def on_status(update):
    print(f"Agent {update.agent_id}: {update.message} ({update.level})")

await kit.status_bus.subscribe(on_status)
```

## Delegation

Delegate tasks to background agents in child rooms:

```python
result = await kit.delegate(
    room_id="main-room",
    agent_id="researcher",
    task="Find the latest pricing for product X",
    context={"product": "X"},
)
```

## Memory Providers

Agents use memory providers to maintain conversation context across handoffs:

```python
from roomkit.memory.sliding_window import SlidingWindowMemory
from roomkit.orchestration.handoff import HandoffMemoryProvider

agent = Agent(
    "agent",
    provider=provider,
    memory=HandoffMemoryProvider(SlidingWindowMemory(max_events=50)),
)
```

`HandoffMemoryProvider` wraps any memory provider to inject handoff context (summary from the previous agent) into the conversation history.
---

Transport providers handle sending and receiving messages over external protocols. Each provider implements a channel-specific ABC.

## SMS

### Twilio

```python
from roomkit import RoomKit, SMSChannel
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig

sms = SMSChannel("sms-twilio", provider=TwilioSMSProvider(TwilioConfig(
    account_sid="AC...",
    auth_token="...",
    from_number="+15551234567",
)))

kit = RoomKit()
kit.register_channel(sms)
```

### Telnyx

```python
from roomkit.providers.telnyx.sms import TelnyxSMSProvider
from roomkit.providers.telnyx.config import TelnyxConfig

sms = SMSChannel("sms-telnyx", provider=TelnyxSMSProvider(TelnyxConfig(
    api_key="KEY...",
    from_number="+15551234567",
)))
```

### Sinch

```python
from roomkit.providers.sinch.sms import SinchSMSProvider
from roomkit.providers.sinch.config import SinchConfig

sms = SMSChannel("sms-sinch", provider=SinchSMSProvider(SinchConfig(
    service_plan_id="...",
    api_token="...",
    from_number="+15551234567",
)))
```

### VoiceMeUp

```python
from roomkit.providers.voicemeup.sms import VoiceMeUpSMSProvider
from roomkit.providers.voicemeup.config import VoiceMeUpConfig

sms = SMSChannel("sms-vmu", provider=VoiceMeUpSMSProvider(VoiceMeUpConfig(
    username="...",
    auth_token="...",
    from_number="+15551234567",
)))
```

### Webhook Parsing

Each SMS provider has a webhook parser:

```python
from roomkit.providers.twilio.sms import parse_twilio_webhook
from roomkit.providers.telnyx.sms import parse_telnyx_webhook
from roomkit.providers.sinch.sms import parse_sinch_webhook
from roomkit.providers.voicemeup.sms import VoiceMeUpSMSProvider  # use provider.parse_inbound()

# Or use the universal webhook parser
message = await kit.process_webhook(meta=request_data, channel_id="sms-twilio")
```

## RCS

```python
from roomkit import RCSChannel
from roomkit.providers.twilio.rcs import TwilioRCSProvider, TwilioRCSConfig

rcs = RCSChannel("rcs-main", provider=TwilioRCSProvider(TwilioRCSConfig(
    account_sid="AC...",
    auth_token="...",
    messaging_service_sid="MG...",  # Required for RCS (must be RCS-enabled)
)))
```

Also available via Telnyx: `TelnyxRCSProvider`, `TelnyxRCSConfig`.

## Email

### Elastic Email

```python
from roomkit import EmailChannel
from roomkit.providers.elasticemail.email import ElasticEmailProvider
from roomkit.providers.elasticemail.config import ElasticEmailConfig

email = EmailChannel("email-main", provider=ElasticEmailProvider(ElasticEmailConfig(
    api_key="...",
    from_email="support@example.com",
    from_name="Support Team",
)))
```

### SendGrid

```python
from roomkit.providers.sendgrid.email import SendGridEmailProvider
from roomkit.providers.sendgrid.config import SendGridConfig

email = EmailChannel("email-sg", provider=SendGridEmailProvider(SendGridConfig(
    api_key="SG...",
    from_email="support@example.com",
)))
```

## WhatsApp

### Business API (Cloud)

```python
from roomkit import WhatsAppChannel
from roomkit.providers.whatsapp.base import WhatsAppProvider

whatsapp = WhatsAppChannel("wa-business", provider=WhatsAppProvider(
    access_token="...",
    phone_number_id="...",
))
```

### Personal (neonize)

```python
from roomkit import WhatsAppPersonalChannel
from roomkit.providers.whatsapp.personal import WhatsAppPersonalProvider

whatsapp = WhatsAppPersonalChannel("wa-personal", provider=WhatsAppPersonalProvider())
```

Requires `pip install roomkit[whatsapp-personal]`. Uses the neonize library for multidevice protocol with typing indicators, read receipts, and media handling.

## Facebook Messenger

```python
from roomkit import MessengerChannel
from roomkit.providers.messenger.facebook import FacebookMessengerProvider
from roomkit.providers.messenger.config import MessengerConfig

messenger = MessengerChannel("messenger", provider=FacebookMessengerProvider(MessengerConfig(
    page_access_token="...",
    app_secret="...",
    verify_token="...",
)))
```

Webhook parser: `parse_messenger_webhook(request_data, channel_id="messenger")` — returns `list[InboundMessage]`.

## Telegram

```python
from roomkit import TelegramChannel
from roomkit.providers.telegram.bot import TelegramBotProvider
from roomkit.providers.telegram.config import TelegramConfig

telegram = TelegramChannel("telegram", provider=TelegramBotProvider(TelegramConfig(
    bot_token="123456:ABC-DEF...",
)))
```

Webhook parser: `parse_telegram_webhook(request_data, channel_id="telegram")` — returns `list[InboundMessage]`.

Inbound `external_id` and `idempotency_key` are `<chat_id>:<message_id>` because
Telegram message ids are chat-local. Malformed nested objects and invalid
coordinates are rejected at the webhook boundary.

- Inbound media (`photo`, `voice`, `audio`, `video_note`, `video`, `document`) parses to a `TextContent` whose body is the caption — empty when there is none, as on a voice note — plus `metadata["file_id"]` and `metadata["media_type"]`, and whichever of `duration`, `mime_type`, `file_name`, `file_size` Telegram sent.
- `parse_telegram_message(msg)` is the layer below: it reads a Telegram `message` into `TelegramMessageParts` (content, metadata, `message_id`, `sender_id`) and attributes nothing. `parse_telegram_webhook` is that function plus the ordinary attribution — the sender is `message.from.id`. Use the lower one when your identity model is not Telegram's (a one-bot-per-user deployment attributes a DM to the bot's owner), so that reading a `file_id` never costs you your identity model.
- Resolving that `file_id` to bytes belongs to the provider, which holds the bot token: `path = await provider.get_file(file_id)` then `data = await provider.download_file(path)`. Both return `None` on failure and log a warning that never carries the URL (every Bot API URL embeds the token). Telegram caps Bot API downloads at 20 MB and refuses larger files at the `getFile` step; `metadata["file_size"]` tells you before you spend the call.
- RoomKit stops at the bytes — transcription and storage are the application's call.
- `parse_telegram_message` also gives `entities` (a caption's markup comes from `caption_entities`), `reply_to_message_id` and `media_group_id`. None of them reach the `InboundMessage` — `parse_telegram_webhook`'s metadata is unchanged.

`TelegramBotProvider` is `TelegramBotAPI` plus the rendering of a `RoomEvent`. The API half is the Bot API surface an application needs around its sends, so it never writes a second HTTP client for the same token:

```python
me = await provider.get_me()                       # metadata["result"] = the bot object
if not me.success:
    ...                                            # me.error, me.metadata["description"]
await provider.set_webhook(url, secret=secret, allowed_updates=["message", "callback_query"])
await provider.answer_callback_query(cq.id, "Approved.")
await provider.edit_message_text(chat, msg_id, "Approved", reply_markup={"inline_keyboard": []})
```

Also `get_updates`, `delete_webhook`, `leave_chat`, `send_message` (plain text, no Markdown pass), `send_force_reply` (its `provider_message_id` is what a later reply is matched against), `send_chat_action` and `edit_message_reply_markup`. Every call answers with a `ProviderResult`: `telegram_<code>` / `http_<status>` / `timeout` / a safe transport exception class, and Telegram's own words under `metadata["description"]`. Transport errors never echo the token-bearing request URL.

Update reading is two levels. `parse_telegram_update(payload)` says which form arrived — a `message`, an `edited_message` (same shape, `edited=True`), or a `callback_query` parsed into a `TelegramCallback` (`id`, `data`, `sender_id`, `chat_id`, `message_id`, `message_text`). `callback_data` is posted by whoever pressed the button, so treat what it names as a claim to check.

`mentions_bot(msg, bot_username=..., bot_id=...)` says whether a group message addressed the bot — a reply to the bot, `bot_command`, `mention`, `text_mention`, or the handle as plain text. It gives the fact; whether that group answers only when addressed is your policy. `entity_text(text, entity)` slices by an entity's offsets, which count **UTF-16 code units** — a code-point slice returns the wrong substring as soon as an emoji precedes the mention.

## Discord

Discord has no webhook parser — it is a source + provider pair sharing one persistent gateway connection. `DiscordGatewaySource` owns the `discord.Client` (inbound); `DiscordBotProvider` reuses that client for outbound sends.

```python
from roomkit import DiscordChannel, RoomKit
from roomkit.providers.discord import DiscordBotProvider, DiscordConfig
from roomkit.sources.discord import DiscordGatewaySource

config = DiscordConfig(
    bot_token="...",               # SecretStr
    intents_message_content=True,  # privileged intent — enable in the Developer Portal
    ignore_bots=True,              # drop inbound messages authored by other bots
)
source = DiscordGatewaySource(config, channel_id="discord-main")
provider = DiscordBotProvider(source)  # sends through the source's client

kit = RoomKit()
kit.register_channel(DiscordChannel("discord-main", provider=provider))
await kit.attach_source("discord-main", source, auto_restart=True)  # connects the gateway
```

Requires `pip install roomkit[discord]` (discord.py). The Message Content intent is privileged: enable it under Bot > Privileged Gateway Intents in the Discord Developer Portal, or every inbound `message.content` arrives empty.

- Recipient key `discord_channel_id` resolves the target Discord channel snowflake at delivery time.
- Capabilities: text + rich + media, `max_length=2000`, threading and reactions. `RichContent` is sent as an embed; `MediaContent` with an http(s) URL rides in the message content (Discord auto-embeds), a `data:` URI is decoded and uploaded as a file.
- Threading: outbound `channel_data.thread_id` (a message snowflake) becomes a reply reference; inbound reply references become `InboundMessage.thread_id`.
- Inbound parsing: `parse_discord_message(message, channel_id, bot_user_id=..., ignore_bots=True)` returns `InboundMessage | None` — the bot's own messages (and other bots, by default) are dropped, so echo hooks never loop. Metadata carries `guild_id`, `channel_id`, `channel_name`, `author_name`, `author_bot`, `message_id`. Override via `DiscordGatewaySource(..., parser=...)`.
- Reactions: `provider.send_reaction(channel_id, message_id, emoji)` outbound; inbound reaction add/remove events reach the source's `on_event` callback as normalized dicts (`action`, `emoji`, `user_id`, `message_id`, `channel_id`) — outside the message pipeline.
- Testing: `MockDiscordProvider` records `sent` and `reactions` without the `discord` dependency. ABC: `DiscordProvider`.

Runnable example: `examples/discord_bot.py`.

## Buzz (Nostr)

Buzz (Block's Nostr-based team workspace) follows the same source + provider pairing: `BuzzRelaySource` owns a `buzzkit.BuzzClient` — NIP-42 authentication plus a realtime channel subscription — and `BuzzProvider` reuses that client for outbound sends over the relay's HTTP bridge, so one Nostr identity serves both directions.

```python
from roomkit import BuzzChannel, RoomKit
from roomkit.providers.buzz import BuzzConfig, BuzzProvider
from roomkit.sources.buzz import BuzzRelaySource

config = BuzzConfig(
    relay_url="wss://your-community.communities.buzz.xyz",
    private_key="nsec1...",   # agent's Nostr secret (nsec or hex) — signs events, authenticates (NIP-42)
    ignore_own=True,          # drop the agent's own events (echo guard)
    auto_join=True,           # NIP-29 self-join (kind 9000, role=bot) on connect
    announce_presence=True,   # kind 20001 "online" on connect + periodic heartbeat
    auth_tag=None,            # optional NIP-OA owner attestation (buzzkit.compute_auth_tag)
)
source = BuzzRelaySource(config, channel_id="buzz-main", relay_channel_id="<channel-uuid>")
provider = BuzzProvider(source)  # sends through the source's client

kit = RoomKit()
kit.register_channel(BuzzChannel("buzz-main", provider=provider))
await kit.create_room(room_id="buzz-room")
await kit.attach_channel("buzz-room", "buzz-main", metadata={"buzz_channel_id": "<channel-uuid>"})
await kit.attach_source("buzz-main", source, auto_restart=True)  # connects + subscribes
```

Requires `pip install roomkit[buzz]` (installs `buzzkit`, a compiled wheel kept out of the aggregate extras). Hosted Buzz communities are closed relays: the agent's key must be a member first — claim an invite once with `buzzkit`'s `claim_invite`, then copy the channel UUID from the Buzz app.

- Recipient key `buzz_channel_id` resolves the target Buzz relay channel UUID at delivery time. Each source subscribes to a single relay channel — register one source per Buzz channel and bind each to its room.
- Capabilities: text only, `max_length=65536`, threading and reactions advertised.
- Inbound parsing: `parse_buzz_event(event, channel_id, own_pubkey=..., ignore_own=True)` converts a kind-9 Nostr event dict to an `InboundMessage` — sender pubkey becomes `sender_id`, the Nostr event id becomes `external_id` and `idempotency_key`, metadata carries `nostr_event_id`, `nostr_kind`, `buzz_channel_id`. Subscribe to other event kinds with `BuzzRelaySource(..., kinds=[...], parser=...)`.
- `provider.send(event, to=channel_uuid)` publishes a kind-9 channel message signed with the agent's key; the returned `ProviderResult.provider_message_id` is the Nostr event id. HTTP-bridge sends succeed even while the inbound WebSocket is mid-reconnect; the source reconnects with exponential backoff (1 s doubling to a 30 s cap).
- Testing: `MockBuzzProvider` records `sent` without the `buzzkit` dependency. ABC: `BuzzRelayProvider`.
- Presence: with `announce_presence=True` the source publishes kind-20001 `"online"` on connect, heartbeats every 30 s (surviving transient publish failures), and publishes `"offline"` on a deliberate `stop()` so the agent's dot flips immediately instead of lapsing by relay TTL.
- Owner control commands (`buzzkit>=0.3.0`): with `obey_owner_commands=True` (default), a kind-9 message whose trimmed content is exactly `!shutdown` / `!cancel` / `!rotate`, mentioning the agent and authored by the **proven** owner — the NIP-OA auth tag's Schnorr-verified attester, else `BuzzConfig.owner_pubkey` — is consumed before the pipeline (the AI never answers its own stop command). `!shutdown` stops the source gracefully; all commands reach the optional `on_owner_command` callback, which takes over the response when provided. No provable owner, or a non-owner author → the message flows normally (fail-closed). Replay-safe: a command issued before the source started is stale (relays replay recent history on every subscribe) — consumed without action; one issued during a disconnection is honored when the reconnect replays it.
- Inbound metadata carries `nostr_created_at` (unix seconds, the Nostr timestamp) so apps can tell live traffic from relay-history replay — used by `examples/buzz_agent.py`'s echo guard.

Runnable example: `examples/buzz_bot.py`.

### BuzzAgent — the lifecycle runner

`BuzzAgent` (`roomkit.providers.buzz`) turns a configured RoomKit app into a first-class Buzz agent: it attaches the sources (taking over their `on_owner_command`), installs SIGTERM/SIGINT handlers, optionally arms an `exit_after_inactivity` bound (seconds; default off; reaper on its own timer), and exits every stop cause — owner `!shutdown`, signal, inactivity — through one graceful path: `kit.close()` (presence `offline`, sockets closed). `run()` is single-shot, consumes the kit, and returns a `BuzzAgentStopCause` (`owner_shutdown` / `signal` / `inactivity`); exit the process with code 0 so supervisors never restart an intentional stop.

```python
from roomkit.providers.buzz import BuzzAgent, BuzzConfig

config = BuzzConfig.from_env()   # BUZZ_PRIVATE_KEY (or NOSTR_PRIVATE_KEY) / BUZZ_RELAY_URL / BUZZ_AUTH_TAG — fail-closed
agent = BuzzAgent(kit, [source], exit_after_inactivity=7200)
cause = await agent.run()        # blocks until the owner, a signal, or idleness stops it
```

Runnable example: `examples/buzz_agent.py`.

Buzz huddles (live voice calls in ephemeral channels) are handled by the voice subsystem, not this transport: `BuzzHuddleBackend` (`roomkit.voice.backends.buzz_huddle`) is a `VoiceBackend` carrying huddle Opus audio for a `RealtimeVoiceChannel`, and `BuzzHuddleWatcher` owns the announcement-to-call lifecycle, watching the parent channel for kind-48100 announcements (`KIND_HUDDLE_STARTED` / `huddle_announcement_parser` in `roomkit.sources.buzz`) and bridging each huddle. See `examples/buzz_voice_agent.py`.

## Microsoft Teams

```python
from roomkit import TeamsChannel
from roomkit.providers.teams.bot_framework import BotFrameworkTeamsProvider
from roomkit.providers.teams.config import TeamsConfig

teams = TeamsChannel("teams", provider=BotFrameworkTeamsProvider(TeamsConfig(
    app_id="...",
    app_password="...",
)))
```

Features: proactive messaging, bot mention detection, reaction handling, conversation reference storage.

```python
from roomkit.providers.teams.webhook import parse_teams_webhook, is_bot_added

# Parse incoming Teams activity
activity = parse_teams_webhook(request_data)

# Check if bot was added to a conversation
if is_bot_added(activity):
    # Handle bot installation
    pass
```

## HTTP (Generic Webhook)

```python
from roomkit import HTTPChannel
from roomkit.providers.http.provider import WebhookHTTPProvider
from roomkit.providers.http.config import HTTPProviderConfig

http = HTTPChannel("webhook", provider=WebhookHTTPProvider(HTTPProviderConfig(
    url="https://api.example.com/messages",
    headers={"Authorization": "Bearer ..."},
)))
```

## WebSocket

WebSocket channels don't use a provider — they handle connections directly:

```python
from roomkit import WebSocketChannel

ws = WebSocketChannel("ws-client")

# Register a connection — room_id says which conversation this socket is for
ws.register_connection("conn-1", on_receive_callback, room_id="room-1")

# In production, connect to the framework
await kit.connect_websocket("ws-client", "conn-1", send_fn, room_id="room-1")
await kit.disconnect_websocket("ws-client", "conn-1")
```

One channel instance can serve several rooms, so a connection has to say which
one it belongs to: the channel delivers a room's events only to the
connections registered for it. A client holding several conversations open on
one socket subscribes to the extra rooms rather than opening more sockets:

```python
kit.subscribe_websocket("ws-client", "conn-1", "room-2")
kit.unsubscribe_websocket("ws-client", "conn-1", "room-2")
```

## Phone Number Utilities

```python
from roomkit.providers.sms.phone import is_valid_phone, normalize_phone

is_valid_phone("+15551234567")   # True
normalize_phone("555-123-4567")  # "+15551234567"
```

## Delivery Status Tracking

Track delivery status for sent messages:

```python
from roomkit import DeliveryStatus

@kit.on_delivery_status
async def track_delivery(status: DeliveryStatus) -> None:
    if status.status == "failed":
        print(f"Message {status.message_id} failed: {status.error_message}")

# Process status webhooks from providers
await kit.process_delivery_status(status)
```
---

## Identity Resolution

The identity pipeline maps external sender IDs to known participants. It runs as part of the inbound pipeline, after `handle_inbound()` and before hooks.

### How It Works

```
Inbound message arrives with sender_id
  -> IdentityResolver.resolve(sender_id, channel_type)
  -> Returns IdentityResult with status:
     IDENTIFIED      -> participant_id stamped on event, processing continues
     AMBIGUOUS       -> ON_IDENTITY_AMBIGUOUS hook fires
     PENDING         -> ON_IDENTITY_AMBIGUOUS hook fires
     UNKNOWN         -> ON_IDENTITY_UNKNOWN hook fires
     REJECTED        -> ON_IDENTITY_UNKNOWN hook fires
```

### Identity Hooks

```python
from roomkit import RoomKit, HookTrigger
from roomkit.models.identity import IdentityHookResult, Identity

kit = RoomKit()

@kit.identity_hook(HookTrigger.ON_IDENTITY_UNKNOWN)
async def handle_unknown(event, ctx):
    # Option 1: Resolve to a known identity
    return IdentityHookResult.resolved(Identity(
        id="user-123",
        display_name="Alice",
    ))

    # Option 2: Challenge the sender to identify
    return IdentityHookResult.challenge(inject=InjectedEvent(
        content=TextContent(body="Please provide your account number."),
    ))

    # Option 3: Reject the message
    return IdentityHookResult.reject("Unknown sender")

    # Option 4: Keep as pending
    return IdentityHookResult.pending(candidates=[...])
```

### Custom Identity Resolver

```python
from roomkit.identity.base import IdentityResolver
from roomkit.models.identity import IdentityResult, Identity
from roomkit.models.enums import IdentificationStatus

class DatabaseIdentityResolver(IdentityResolver):
    async def resolve(self, message: InboundMessage, context: RoomContext) -> IdentityResult:
        user = await db.find_by_phone(message.sender_id)
        if user:
            return IdentityResult(
                status=IdentificationStatus.IDENTIFIED,
                identity=Identity(id=user.id, display_name=user.name),
            )
        return IdentityResult(status=IdentificationStatus.UNKNOWN)

kit = RoomKit(identity_resolver=DatabaseIdentityResolver())
```

### Manual Resolution

```python
# Resolve a pending participant to a known identity
await kit.resolve_participant(
    room_id="room-1",
    participant_id="pending-123",
    identity_id="user-456",
)
```

## Realtime Ephemeral Events

Ephemeral events (typing, presence, reactions) are not stored in conversation history. They're delivered in real-time to subscribers.

### Publishing Events

```python
from roomkit import RoomKit

kit = RoomKit()

# Typing indicator
await kit.publish_typing("room-1", "alice", is_typing=True)
await kit.publish_typing("room-1", "alice", is_typing=False)

# Presence
await kit.publish_presence("room-1", "alice", "online")   # online/away/offline

# Reaction
await kit.publish_reaction("room-1", "alice", target_event_id="evt-123", emoji="thumbsup")

# Read receipt
await kit.publish_read_receipt("room-1", "alice", event_id="evt-123")

# Tool call events (AIChannel publishes these automatically)
from roomkit.realtime.base import EphemeralEventType

await kit.publish_tool_call("room-1", "ai-agent", [
    {"id": "tc1", "name": "search", "arguments": {"q": "test"}}
], EphemeralEventType.TOOL_CALL_START)
```

### Subscribing to Events

```python
async def on_ephemeral(event):
    print(f"Ephemeral: {event.type} from {event.user_id}")

sub_id = await kit.subscribe_room("room-1", on_ephemeral)

# Unsubscribe later
await kit.unsubscribe_room(sub_id)
```

### Event Types

| Type | Description |
|------|-------------|
| `TYPING_START` | User started typing |
| `TYPING_STOP` | User stopped typing |
| `PRESENCE_ONLINE` | User came online |
| `PRESENCE_AWAY` | User went away |
| `PRESENCE_OFFLINE` | User went offline |
| `READ_RECEIPT` | User read a message |
| `REACTION` | User reacted to a message |
| `TOOL_CALL_START` | AI started executing a tool |
| `TOOL_CALL_END` | AI finished executing a tool |
| `CUSTOM` | Custom ephemeral event |

### Read Tracking

```python
# Mark a specific event as read
await kit.mark_read("room-1", "ws-user", "evt-123")

# Mark all events as read
await kit.mark_all_read("room-1", "ws-user")
```

### Custom Realtime Backend

The default `InMemoryRealtime` works for single-process deployments. For distributed systems, implement `RealtimeBackend`:

```python
from roomkit.realtime.base import RealtimeBackend

class RedisRealtimeBackend(RealtimeBackend):
    # Implement publish, subscribe, unsubscribe using Redis Pub/Sub
    ...

kit = RoomKit(realtime=RedisRealtimeBackend())
```
---

RoomKit provides built-in resilience patterns for production deployments: rate limiting, circuit breakers, retry with backoff, chain depth limits, and room lifecycle timers.

## Rate Limiting

Apply rate limits per channel binding:

```python
from roomkit import RoomKit, RateLimit

kit = RoomKit()

await kit.attach_channel("room-1", "sms-out",
    rate_limit=RateLimit(max_per_second=1.0, max_per_minute=30.0),
)
```

Framework-level inbound rate limiting:

```python
from roomkit import RoomKit, RateLimit

kit = RoomKit(inbound_rate_limit=RateLimit(max_per_second=10.0))
```

## Circuit Breaker

Isolate provider failures with circuit breakers:

```python
from roomkit.core.circuit_breaker import CircuitBreaker

cb = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)

if cb.allow_request():
    try:
        result = await provider.send(event, to="+1234567890")
        cb.record_success()
    except Exception:
        cb.record_failure()  # Opens after 5 consecutive failures
```

States: CLOSED (normal) -> OPEN (failing, fast-reject) -> HALF_OPEN (testing recovery).

## Retry with Backoff

Retry failed operations with exponential backoff:

```python
from roomkit import RetryPolicy
from roomkit.core.retry import retry_with_backoff

policy = RetryPolicy(
    max_retries=3,
    base_delay_seconds=1.0,
    max_delay_seconds=60.0,
)

result = await retry_with_backoff(flaky_function, policy)
```

Apply per binding:

```python
await kit.attach_channel("room-1", "sms-out",
    retry_policy=RetryPolicy(max_retries=3, base_delay_seconds=1.0),
)
```

## Chain Depth Limit

Prevents infinite AI-to-AI loops. Default max depth is 5.

```python
kit = RoomKit(max_chain_depth=3)  # Stricter limit
```

When an AI response triggers another AI response, chain depth increments. Processing stops when the limit is reached.

## Room Lifecycle Timers

Auto-transition rooms based on inactivity:

```python
from roomkit import RoomKit, RoomTimers

kit = RoomKit()

room = await kit.create_room(room_id="session-1")
# Configure timers on the room:
# inactive_after_seconds -> ACTIVE to PAUSED
# closed_after_seconds -> to CLOSED

# Check timers for one room
room = await kit.check_room_timers("session-1")

# Batch check all rooms (call periodically)
transitioned = await kit.check_all_timers()
```

## Delivery Status Tracking

Track whether messages were delivered to providers:

```python
from roomkit import DeliveryStatus

@kit.on_delivery_status
async def track(status: DeliveryStatus) -> None:
    if status.status == "failed":
        logger.error("Delivery failed: %s — %s", status.message_id, status.error_message)
    elif status.status == "delivered":
        logger.info("Delivered: %s", status.message_id)

# Process status webhooks from providers
await kit.process_delivery_status(status)
```

## Production Setup Example

```python
from roomkit import RoomKit, RateLimit, RetryPolicy
from roomkit.store.postgres import PostgresStore

kit = RoomKit(
    store=PostgresStore("postgresql://user:pass@localhost/roomkit"),
    max_chain_depth=5,
    inbound_rate_limit=RateLimit(max_per_second=50.0),
    process_timeout=30.0,
)

# Per-channel resilience
await kit.attach_channel("room", "sms-out",
    rate_limit=RateLimit(max_per_second=1.0, max_per_minute=30.0),
    retry_policy=RetryPolicy(max_retries=3, base_delay_seconds=1.0),
)
```

## Framework Events for Monitoring

```python
@kit.on("source_error")
async def on_error(event):
    logger.error("Source error: %s", event.data["error"])

@kit.on("voice_session_ended")
async def on_voice_end(event):
    logger.info("Voice session ended: %s", event.data["session_id"])
```
---

RoomKit uses the `ConversationStore` ABC for persistence. The default `InMemoryStore` works out of the box. For production, use `PostgresStore`.

## InMemoryStore (Default)

```python
from roomkit import RoomKit

kit = RoomKit()  # Uses InMemoryStore automatically
```

Data lives in Python dicts — fast for development, lost on restart.

## PostgresStore

Install: `pip install roomkit[postgres]`

```python
from roomkit import RoomKit
from roomkit.store.postgres import PostgresStore

store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init()  # Creates connection pool and tables

kit = RoomKit(store=store)
```

### Connection Pooling

```python
store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init(min_size=5, max_size=20)  # Pool sizing via init()
```

### Schema

PostgresStore creates 10 tables:

| Table | Purpose |
|-------|---------|
| `rooms` | Room records with status, metadata, timers |
| `events` | Event timeline with sequential indexing |
| `participants` | Room participants with roles and status |
| `bindings` | Channel-to-room bindings with config |
| `identities` | Known identity records |
| `tasks` | AI-extracted tasks |
| `observations` | AI-extracted observations |
| `delivery_status` | Message delivery tracking |
| `read_tracking` | Per-channel read positions |
| `telemetry_spans` | Telemetry span records |

### Operations

```python
# Room operations
room = await kit.create_room(room_id="persistent", metadata={"topic": "billing"})

# Event storage and retrieval
events = await kit.store.list_events("persistent", offset=0, limit=50)

# Timeline query with filters
timeline = await kit.get_timeline("persistent", offset=0, limit=50)

# Participant management
participants = await kit.store.list_participants("persistent")

# Binding management
bindings = await kit.store.list_bindings("persistent")
```

### Full Example

```python
from __future__ import annotations

import asyncio
import os

from roomkit import InboundMessage, RoomKit, TextContent, WebSocketChannel
from roomkit.store.postgres import PostgresStore


async def main() -> None:
    store = PostgresStore(os.environ["DATABASE_URL"])
    await store.initialize()

    kit = RoomKit(store=store)

    ws = WebSocketChannel("ws-user")
    kit.register_channel(ws)

    room = await kit.create_room(room_id="support", metadata={"topic": "billing"})
    await kit.attach_channel("support", "ws-user")

    await kit.process_inbound(
        InboundMessage(
            channel_id="ws-user",
            sender_id="user",
            content=TextContent(body="I need help with my invoice"),
        )
    )

    # Data persists across restarts
    events = await kit.store.list_events("support")
    print(f"Stored {len(events)} events")

    await kit.close()


if __name__ == "__main__":
    asyncio.run(main())
```

## Custom Store

Implement `ConversationStore` for other backends (Redis, DynamoDB, etc.):

```python
from roomkit.store.base import ConversationStore
from roomkit.models.room import Room

class RedisStore(ConversationStore):
    async def create_room(self, room: Room) -> Room:
        await self.redis.set(f"room:{room.id}", room.model_dump_json())
        return room

    # Implement all abstract methods...

kit = RoomKit(store=RedisStore())
```
---

RoomKit provides mock implementations for every pluggable component: AI providers, voice backends, pipeline stages, identity resolvers, and telemetry.

## Test Setup

```bash
pip install roomkit[dev]
uv run pytest              # Run all tests
uv run pytest tests/test_framework.py -v  # Specific file
```

Configuration in `pyproject.toml`:

```toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"  # No @pytest.mark.asyncio needed
```

## Basic Test Pattern

```python
from roomkit import RoomKit, InboundMessage, TextContent, WebSocketChannel


class TestMyFeature:
    async def test_message_delivery(self) -> None:
        kit = RoomKit()

        ws = WebSocketChannel("ws-user")
        kit.register_channel(ws)

        inbox: list = []
        async def on_recv(_conn: str, event) -> None:
            inbox.append(event)

        ws.register_connection("conn", on_recv)

        await kit.create_room(room_id="test-room")
        await kit.attach_channel("test-room", "ws-user")

        result = await kit.process_inbound(
            InboundMessage(
                channel_id="ws-user",
                sender_id="user",
                content=TextContent(body="Hello"),
            )
        )

        assert not result.blocked
        assert len(inbox) == 1
        assert inbox[0].content.body == "Hello"
```

## Mock AI Provider

```python
from roomkit.providers.ai.mock import MockAIProvider
from roomkit.channels.ai import AIChannel
from roomkit import ChannelCategory

# Responds with pre-configured messages in order
provider = MockAIProvider(responses=["Response 1", "Response 2"])

ai = AIChannel("ai", provider=provider)
kit.register_channel(ai)
await kit.attach_channel("room", "ai", category=ChannelCategory.INTELLIGENCE)

# After processing, check what the AI was called with:
assert len(provider.calls) == 1
last_call = provider.calls[-1]
print(last_call.system_prompt)
print(last_call.temperature)
print([t.name for t in last_call.tools])
```

## Mock Voice Backend

```python
from roomkit.voice.backends.mock import MockVoiceBackend

backend = MockVoiceBackend()

# Simulate audio input
backend.simulate_audio_received(session, audio_frame)
```

## Mock Pipeline Providers

Every pipeline stage has a mock that accepts pre-configured event sequences:

```python
from roomkit.voice.pipeline import (
    MockVADProvider,
    VADEvent,
    VADEventType,
    MockDenoiserProvider,
    MockDiarizationProvider,
    MockAGCProvider,
    MockAECProvider,
    MockDTMFDetector,
    MockAudioRecorder,
    MockTurnDetector,
    MockBackchannelDetector,
)

# VAD with event sequence
vad = MockVADProvider(events=[
    VADEvent(type=VADEventType.SPEECH_START),
    None,  # No event for this frame
    VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech"),
])

# Other mocks
denoiser = MockDenoiserProvider()
diarizer = MockDiarizationProvider()
agc = MockAGCProvider()
aec = MockAECProvider()
dtmf = MockDTMFDetector()
recorder = MockAudioRecorder()
turn = MockTurnDetector()
backchannel = MockBackchannelDetector()
```

## Mock STT/TTS

```python
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider

stt = MockSTTProvider(transcripts=["Hello", "How are you?"])
tts = MockTTSProvider()
```

## Mock Identity Resolver

```python
from roomkit.identity.mock import MockIdentityResolver

resolver = MockIdentityResolver()
kit = RoomKit(identity_resolver=resolver)
```

## Mock Realtime Provider

```python
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport

provider = MockRealtimeProvider()
transport = MockRealtimeTransport()
```

## Testing Hooks

```python
from roomkit import HookTrigger, HookResult, HookExecution

async def test_hook_blocks_message() -> None:
    kit = RoomKit()
    ws = WebSocketChannel("ws")
    kit.register_channel(ws)

    await kit.create_room(room_id="r")
    await kit.attach_channel("r", "ws")

    @kit.hook(HookTrigger.BEFORE_BROADCAST)
    async def blocker(event, ctx):
        return HookResult.block("blocked")

    result = await kit.process_inbound(
        InboundMessage(
            channel_id="ws",
            sender_id="user",
            content=TextContent(body="test"),
        )
    )

    assert result.blocked
    assert result.reason == "blocked"
```

## Testing Voice Pipeline

```python
from roomkit import VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, MockVADProvider, VADEvent, VADEventType
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.backends.mock import MockVoiceBackend
from roomkit.voice.audio_frame import AudioFrame

async def test_voice_pipeline() -> None:
    kit = RoomKit()

    backend = MockVoiceBackend()
    stt = MockSTTProvider(transcripts=["Hello"])
    tts = MockTTSProvider()
    vad = MockVADProvider(events=[
        VADEvent(type=VADEventType.SPEECH_START),
        None,
        VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"audio"),
    ])

    voice = VoiceChannel(
        "voice", stt=stt, tts=tts, backend=backend,
        pipeline=AudioPipelineConfig(vad=vad),
    )
    kit.register_channel(voice)

    await kit.create_room(room_id="call")
    await kit.attach_channel("call", "voice")

    # Simulate audio input
    frame = AudioFrame(data=b"\x00" * 320, sample_rate=16000)
    backend.simulate_audio_received(None, frame)
```

## Testing Orchestration

```python
from roomkit import Agent, Pipeline, RoomKit, WebSocketChannel
from roomkit.providers.ai.mock import MockAIProvider

async def test_pipeline_orchestration() -> None:
    agent1 = Agent("a1", provider=MockAIProvider(responses=["Transferring..."]))
    agent2 = Agent("a2", provider=MockAIProvider(responses=["Resolved!"]))

    kit = RoomKit(orchestration=Pipeline(agents=[agent1, agent2]))

    ws = WebSocketChannel("ws")
    kit.register_channel(ws)

    await kit.create_room(room_id="test")
    await kit.attach_channel("test", "ws")

    # First message goes to agent1
    result = await kit.process_inbound(
        InboundMessage(channel_id="ws", sender_id="user", content=TextContent(body="Help"))
    )
    assert not result.blocked
```

## Test Utilities

```python
# Pydantic model updates (immutable)
modified = event.model_copy(update={"content": TextContent(body="new")})

# Never mutate models directly — always use model_copy
```
---

RoomKit exports **170 symbols** from `roomkit`. Providers and voice types import from subpackages.

## Top-Level Imports (`from roomkit import ...`)

### Framework

| Symbol | Description |
|--------|-------------|
| `RoomKit` | Central orchestrator — rooms, channels, hooks, storage |
| `RoomKitConsole` | Full-screen terminal dashboard for voice agent development (optional, requires `rich`) |
| `__version__` | Package version string |
| `content_logging_enabled` | Whether raw message content may be written to logs (default False) |
| `set_content_logging` | Enable/disable process-wide logging of raw message content |

### Channels

| Symbol | Description |
|--------|-------------|
| `ACPChannel` | Connects a room to an external ACP coding agent over stdio |
| `Agent` | AI agent with role, description, greeting, tools |
| `AIChannel` | Intelligence layer for AI responses |
| `AIChannelTurnConfig` | Per-turn generation overrides for AIChannel (None fields keep channel defaults) |
| `AudioVideoChannel` | Combined audio + video channel |
| `BuzzChannel` | Buzz (Nostr relay) transport channel factory |
| `Channel` | Base class for all channels |
| `CLIChannel` | Interactive terminal channel |
| `ConferenceChannel` | Multi-party conference channel backed by an external SFU |
| `DiscordChannel` | Discord bot transport channel factory |
| `EmailChannel` | Email transport channel factory |
| `FrameworkAwareChannel` | Channel base handed the framework it is registered with |
| `HTTPChannel` | HTTP webhook transport channel factory |
| `MessengerChannel` | Facebook Messenger transport channel factory |
| `RCSChannel` | RCS transport channel factory |
| `RealtimeAudioVideoChannel` | Realtime speech-to-speech with video |
| `RealtimeVoiceChannel` | Speech-to-speech AI channel |
| `SMSChannel` | SMS transport channel factory |
| `TeamsChannel` | Microsoft Teams transport channel factory |
| `TelegramChannel` | Telegram Bot transport channel factory |
| `TransportChannel` | Generic transport channel wrapper |
| `VideoChannel` | Video channel with vision pipeline |
| `VoiceChannel` | Real-time audio with STT/TTS/pipeline |
| `WebSocketChannel` | WebSocket bidirectional channel |
| `WhatsAppChannel` | WhatsApp Business API channel factory |
| `WhatsAppPersonalChannel` | WhatsApp Personal (neonize) channel factory |

### Conference

| Symbol | Description |
|--------|-------------|
| `BotSession` | The framework's own connection to a conference |
| `ConferenceAccess` | Credentials a client uses to join the conference directly |
| `ConferenceBackend` | ABC for SFU conference backends |
| `ConferenceBargeIn` | Event: a participant spoke over the bot and was allowed to interrupt it |
| `ConferenceCapability` | Flag enum of capabilities a ConferenceBackend can support |
| `ConferenceGrants` | Permissions encoded into a participant's conference access |
| `ConferenceInterruptionConfig` | Multi-party interruption policy |
| `ConferenceInterruptionScope` | Who may interrupt the bot while it is speaking |
| `ConferenceParticipant` | A participant's media presence in a conference |
| `ConferenceRealtimeConfig` | Composes a speech-to-speech provider with a conference |
| `ConferenceRecordingConfig` | Configuration for recording a conference |
| `ConferenceRecordingMode` | Where a conference recording is produced |
| `ConferenceRecordingStarted` | Event: a track's recording has opened |
| `ConferenceRecordingStopped` | Event: a track's recording has closed, with its destination |
| `ConferenceToolHandler` | Callable type for conference tool invocation |
| `ConferenceTrack` | A single media stream published by a conference participant |
| `ConferenceTranscription` | What a lane produced, before it enters the room |
| `TrackKind` | Kind of media carried by a conference track |
| `LiveKitConferenceBackend` | ConferenceBackend backed by a LiveKit SFU |
| `LiveKitConfig` | Connection and behaviour settings for LiveKitConferenceBackend |
| `MockConferenceBackend` | Conference backend that scripts SFU events for tests |
| `MockDelivery` | Mock media timing — how long one frame took to reach every subscriber |
| `MockFaults` | Per-operation failures and delays for the mock backend |
| `MockTrackFormat` | Audio format a participant negotiated for one track (mock) |
| `MockUtterance` | Chunks published for one utterance on one bot's track (mock) |
| `CONFERENCE_ADDRESS_KEYS` | Participant-attribute keys read as a caller's address, most specific first |
| `CONFERENCE_METADATA_KEY` | `Participant.metadata` key a conference nests provider data under (`"conference"`) |
| `CONFERENCE_UNASSERTED_METADATA_KEY` | Metadata key nesting client-claimed (unverified) participant attributes |

### Video

| Symbol | Description |
|--------|-------------|
| `VideoDetectionEvent` | Detection event emitted by video pipeline filters |
| `FaceTouchFilter` | Detects hand-to-face contact using MediaPipe landmarks |
| `FaceTouchConfig` | Configuration for face touch detection |
| `FaceTouchSensitivity` | Sensitivity presets controlling detection thresholds |
| `FaceZone` | Face zones that can be monitored for touch detection |
| `MockFaceTouchFilter` | Mock filter emitting pre-configured detection events at specific frames |

### Enums

| Symbol | Description |
|--------|-------------|
| `Access` | Channel access levels: READ_WRITE, READ_ONLY, WRITE_ONLY, NONE |
| `ChannelCategory` | TRANSPORT or INTELLIGENCE |
| `ChannelType` | 23 values: SMS, MMS, RCS, EMAIL, WHATSAPP, WHATSAPP_PERSONAL, WEBSOCKET, AI, VOICE, REALTIME_VOICE, REALTIME_AUDIO_VIDEO, PUSH, MESSENGER, TELEGRAM, TEAMS, DISCORD, BUZZ, WEBHOOK, VIDEO, AUDIO_VIDEO, CONFERENCE, CLI, SYSTEM |
| `EventStatus` | PENDING, DELIVERED, READ, FAILED, BLOCKED |
| `EventType` | 27 values (MESSAGE, SYSTEM, EDIT, DELETE, TOOL_CALL_START, DTMF, etc.) |
| `HookExecution` | SYNC or ASYNC |
| `HookTrigger` | 76 hook triggers — full list in hooks.md |
| `RoomStatus` | ACTIVE, PAUSED, CLOSED, ARCHIVED |
| `Visibility` | Scope keywords for an event's `visibility` field: ALL, NONE, TRANSPORT, INTELLIGENCE, INTERNAL |

### Models

| Symbol | Description |
|--------|-------------|
| `ChannelBinding` | Binding of a channel to a room |
| `ChannelCapabilities` | Declared capabilities of a channel |
| `ChannelOutput` | Output of a channel delivery |
| `EventSource` | Source attribution for an event |
| `FrameworkEvent` | Lightweight framework lifecycle event |
| `HookResult` | Result from sync hooks: `.allow()`, `.block(reason)`, `.modify(event)` |
| `InjectedEvent` | Event injected by a hook |
| `InboundMessage` | Incoming message from a provider |
| `InboundResult` | Result of processing an inbound message |
| `Participant` | Participant data model |
| `ProviderResult` | Result from a provider operation |
| `Room` | Room data model |
| `RoomContext` | Context passed to hooks (room, bindings, participants, events) |
| `RoomEvent` | Core event stored in the timeline |
| `RoomTimers` | Timer configuration for room inactivity |
| `SessionStartedEvent` | Event fired when a voice session starts |
| `TextContent` | Plain text content |
| `get_current_voice_session` | Get the current voice session from context |

### Tools, Callbacks & Human Input

| Symbol | Description |
|--------|-------------|
| `Tool` | Base class for tool definitions |
| `ToolHandler` | Tool handler type for realtime voice |
| `ToolPolicy` | Per-agent allow/deny rules for tool access |
| `RoleOverride` | Per-role tool policy override |
| `ToolCallCallback` | Callback type for tool call events |
| `ToolCallEvent` | Tool call event model |
| `ToolCallContent` | Content for TOOL_CALL_START and TOOL_CALL_END events |
| `AIGenerationEvent` | Payload for BEFORE_AI_GENERATION hooks, before AI provider invocation |
| `AIResponseEvent` | Payload for ON_AI_RESPONSE hooks, after AI generation completes |
| `BeforeGenerationCallback` | Async callback type receiving AIGenerationEvent |
| `AfterResponseCallback` | Async callback type receiving AIResponseEvent |
| `HumanInputHandler` | Manages pending human input requests |
| `HumanInputToolHandler` | ToolHandler wrapper that blocks on human input for specified tools |
| `PendingInput` | A pending human input request |
| `PendingInputEvent` | Event fired through ON_USER_INPUT_REQUIRED hooks |
| `PendingInputStatus` | Status of a pending human input request |

### Delivery

| Symbol | Description |
|--------|-------------|
| `DeliveryStrategy` | ABC controlling when and how content is delivered to a channel |
| `Immediate` | Deliver now; may interrupt ongoing TTS playback |
| `Queued` | Add to queue, deliver at the next idle window |
| `WaitForIdle` | Wait for TTS/speech to finish, then send |
| `DeliveryBackend` | ABC for persistent delivery queue backends |
| `DeliveryItem` | Serializable delivery request — the unit of work in the queue |
| `DeliveryItemStatus` | Lifecycle status of a delivery item |
| `InMemoryDeliveryBackend` | Asyncio-queue delivery backend (single process, no persistence) |
| `DeliveryResult` | Result of delivering a message |
| `DeliveryStatus` | Delivery status from provider webhook |

### Storage & Locking

| Symbol | Description |
|--------|-------------|
| `ConversationStore` | ABC for persistent room/event/binding/participant storage |
| `InMemoryStore` | Dict-based in-memory store for development and testing |
| `RoomLockManager` | ABC for per-room locking |
| `InMemoryLockManager` | In-process per-room asyncio locks with LRU eviction |
| `EventFilter` | Filter criteria for querying room events |
| `PersistencePolicy` | Controls which event types are persisted to the store |

### Orchestration

| Symbol | Description |
|--------|-------------|
| `Loop` | Producer/reviewer cycle strategy |
| `Orchestration` | ABC for orchestration strategies |
| `Pipeline` | Linear agent chain strategy |
| `Supervisor` | Supervisor delegates to workers strategy |
| `Swarm` | Bidirectional handoff strategy |
| `ConversationPhase` | Built-in conversation phases (StrEnum) |
| `ConversationState` | Tracks conversation progress within a room |
| `ConversationRouter` | Routes events to the appropriate agent |
| `ConversationPipeline` | Generates routing rules for sequential agent workflows |
| `PipelineStage` | A stage in a ConversationPipeline |
| `RoutingRule` | Routing rule mapping conditions to an agent |
| `RoutingConditions` | Conditions for a routing rule to match |
| `get_conversation_state` | Extract typed ConversationState from room metadata |
| `set_conversation_state` | Return a room copy with updated conversation state |
| `HandoffHandler` | Processes handoff tool calls |
| `HandoffRequest` | Parsed from an agent's handoff tool call arguments |
| `HandoffResult` | Result returned to the calling agent after a handoff |
| `HANDOFF_TOOL` | AITool definition for transferring a conversation to another agent |
| `setup_handoff` | Wires handoff into an AIChannel's tool chain |

### Memory, Skills & Sandbox

| Symbol | Description |
|--------|-------------|
| `MemoryProvider` | ABC for pluggable memory backends feeding AI context construction |
| `Skill` | Full skill definition including instructions body |
| `SkillMetadata` | Lightweight metadata parsed from SKILL.md frontmatter |
| `SkillRegistry` | Discovers, loads, and manages Agent Skills |
| `ScriptExecutor` | ABC for executing skill scripts with integrator-defined policy |
| `SandboxExecutor` | ABC for executing commands in a sandboxed environment |
| `SandboxResult` | Result of executing a sandbox command |

### Errors

| Symbol | Description |
|--------|-------------|
| `RoomKitError` | Base exception |
| `RoomNotFoundError` | Room does not exist |
| `RoomClosedError` | Room's status refuses new events (RFC §5.1) |
| `RoomNotAttachedError` | Channel acted on a room it is no longer attached to |
| `ChannelNotFoundError` | Channel not attached to room |
| `ChannelNotRegisteredError` | Channel not registered with framework |
| `ParticipantNotFoundError` | Participant not found in room |
| `ParticipantNotAdmittedError` | Participant barred from what was asked for them |
| `IdentityNotFoundError` | Identity not found |
| `SourceAlreadyAttachedError` | Source already attached |
| `SourceNotFoundError` | No source attached |
| `VoiceBackendNotConfiguredError` | Voice backend not configured |
| `VoiceNotConfiguredError` | Voice (STT/TTS) not configured |
| `ConferenceAlreadyAttachedError` | Second conference channel attached to a room |
| `ConferenceCapabilityError` | Conference operation needs a capability the backend lacks |
| `ConferenceCloseError` | Conference channel did not close all of its resources |

### AI Documentation Helpers

| Symbol | Description |
|--------|-------------|
| `get_llms_txt()` | Get llms.txt content |
| `get_llms_full_txt()` | Get llms-full.txt content (comprehensive) |
| `get_agents_md()` | Get AGENTS.md content |
| `get_ai_context()` | Get combined AI context |

## RoomKit Constructor

```python
kit = RoomKit(
    store=None,                    # ConversationStore (default: InMemoryStore)
    identity_resolver=None,        # IdentityResolver for identifying inbound senders
    identity_channel_types=None,   # Restrict identity resolution to these ChannelTypes (None = all)
    inbound_router=None,           # InboundRoomRouter (default: DefaultInboundRoomRouter)
    lock_manager=None,             # RoomLockManager (default: InMemoryLockManager)
    realtime=None,                 # RealtimeBackend for ephemeral events (default: InMemoryRealtime)
    max_chain_depth=5,             # Max reentry chain depth — AI-to-AI loop prevention
    identity_timeout=10.0,         # Identity resolution timeout (seconds)
    process_timeout=30.0,          # Locked inbound processing timeout (seconds)
    stt=None,                      # STTProvider for transcription
    tts=None,                      # TTSProvider for synthesis
    voice=None,                    # VoiceBackend for real-time audio transport
    task_runner=None,              # TaskRunner for delegated background tasks (default: InMemoryTaskRunner)
    delivery_strategy=None,        # DeliveryStrategy | str — proactive delivery of task results
    delivery_backend=None,         # DeliveryBackend — persistent queue for deliver() (None = in-process)
    status_bus=None,               # StatusBus for multi-agent coordination (default: in-memory)
    telemetry=None,                # TelemetryProvider or TelemetryConfig (default: no-op)
    inbound_rate_limit=None,       # RateLimit applied to inbound messages, keyed per channel_id
    orchestration=None,            # Default Orchestration strategy for create_room()
    persistence_policy=None,       # PersistencePolicy — which event types are persisted (None = all)
)
```

## Key RoomKit Methods

### Room Lifecycle

| Method | Description |
|--------|-------------|
| `create_room(room_id?, metadata?, orchestration?)` | Create a room |
| `get_room(room_id)` | Get room by ID |
| `close_room(room_id)` | Close a room |
| `update_room_metadata(room_id, metadata)` | Update room metadata |
| `check_room_timers(room_id)` | Check timer transitions for one room |
| `check_all_timers()` | Check all room timers |

### Channel Operations

| Method | Description |
|--------|-------------|
| `register_channel(channel)` | Register a channel |
| `attach_channel(room_id, channel_id, category?, access?, ...)` | Attach channel to room |
| `detach_channel(room_id, channel_id)` | Detach channel from room |
| `mute(room_id, channel_id)` | Mute a channel |
| `unmute(room_id, channel_id)` | Unmute a channel |
| `set_access(room_id, channel_id, access)` | Set channel access level |

### Voice/Video

| Method | Description |
|--------|-------------|
| `join(room_id, channel_id, participant_id?, ...)` | Join voice/video session |
| `leave(session)` | Leave voice/video session |
| `transcribe(audio)` | Speech-to-text |
| `synthesize(text, voice?)` | Text-to-speech |

### Inbound Pipeline

| Method | Description |
|--------|-------------|
| `process_inbound(message, room_id?)` | Process an inbound message |

### Hooks

| Method | Description |
|--------|-------------|
| `hook(trigger, execution?, priority?, ...)` | Decorator to register a hook |
| `on(event_type)` | Decorator for framework events |
| `identity_hook(trigger, ...)` | Decorator for identity hooks |
| `on_delivery_status(fn)` | Decorator for delivery status |
| `add_room_hook(room_id, trigger, execution, fn, ...)` | Add room-scoped hook |
| `remove_room_hook(room_id, name)` | Remove room-scoped hook |

### Realtime

| Method | Description |
|--------|-------------|
| `publish_typing(room_id, user_id, is_typing?)` | Typing indicator |
| `publish_presence(room_id, user_id, status)` | Presence update |
| `publish_reaction(room_id, user_id, target_event_id, emoji)` | Reaction |
| `publish_read_receipt(room_id, user_id, event_id)` | Read receipt |
| `subscribe_room(room_id, callback)` | Subscribe to ephemeral events |
| `unsubscribe_room(subscription_id)` | Unsubscribe |

### Sources

| Method | Description |
|--------|-------------|
| `attach_source(channel_id, source, auto_restart?, ...)` | Attach event source |
| `detach_source(channel_id)` | Detach event source |
| `source_health(channel_id)` | Get source health |

### Other

| Method | Description |
|--------|-------------|
| `delegate(room_id, agent_id, task, ...)` | Delegate to background agent |
| `send_greeting(room_id, channel_id?, greeting?, ...)` | Send greeting |
| `send_event(room_id, channel_id, content, ...)` | Send event directly |
| `get_timeline(room_id, offset?, limit?)` | Query event timeline |
| `close()` | Shutdown framework |

## Provider Subpackage Imports

### AI Providers

```python
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig
from roomkit.providers.gemini.ai import GeminiAIProvider
from roomkit.providers.gemini.config import GeminiConfig
from roomkit.providers.mistral.ai import MistralAIProvider
from roomkit.providers.mistral.config import MistralConfig
from roomkit.providers.ai.mock import MockAIProvider
from roomkit.providers.ai.base import AIProvider, AIContext, AIResponse, AITool, AIToolCall
```

### SMS Providers

```python
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig
from roomkit.providers.telnyx.sms import TelnyxSMSProvider
from roomkit.providers.telnyx.config import TelnyxConfig
from roomkit.providers.sinch.sms import SinchSMSProvider
from roomkit.providers.sinch.config import SinchConfig
from roomkit.providers.sms.mock import MockSMSProvider
```

### Voice (Lazy Loaders)

```python
from roomkit.voice import (
    get_deepgram_provider, get_deepgram_config,
    get_elevenlabs_provider, get_elevenlabs_config,
    get_gemini_tts_provider, get_gemini_tts_config,
    get_sherpa_onnx_stt_provider, get_sherpa_onnx_tts_provider,
    get_local_audio_backend,
    get_fastrtc_backend,
    get_rtp_backend,
    get_sip_backend,
    get_gemini_live_provider,
    get_openai_realtime_provider,
    get_xai_realtime_provider,
    get_websocket_realtime_transport,
    get_speex_aec_provider,
    get_rnnoise_denoiser_provider,
)
```

### Voice Mocks

```python
from roomkit.voice.backends.mock import MockVoiceBackend
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport
```

### Pipeline

```python
from roomkit.voice.pipeline import (
    AudioPipelineConfig, VADConfig,
    MockVADProvider, VADEvent, VADEventType,
    MockDenoiserProvider, MockDiarizationProvider,
    MockAGCProvider, MockAECProvider, MockDTMFDetector,
    MockAudioRecorder, MockTurnDetector, MockBackchannelDetector,
)
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
from roomkit.voice.audio_frame import AudioFrame
```

### Orchestration

```python
from roomkit.orchestration.state import get_conversation_state, ConversationState
from roomkit.orchestration.router import ConversationRouter, RoutingRule
from roomkit.orchestration.pipeline import ConversationPipeline, PipelineStage
from roomkit.orchestration.handoff import HandoffHandler, HandoffMemoryProvider
```

### Storage

```python
from roomkit.store.base import ConversationStore
from roomkit.store.memory import InMemoryStore
from roomkit.store.postgres import PostgresStore
```

### Content Types

```python
from roomkit.models.event import (
    TextContent, RichContent, MediaContent, AudioContent, VideoContent,
    LocationContent, CompositeContent, TemplateContent, SystemContent,
    EditContent, DeleteContent,
)
```
