Metadata-Version: 2.4
Name: tuner-tts-observer
Version: 0.1.0
Summary: TTS observability for Tuner — Cartesia adapter with barge-in and latency tracking.
Author: Tuner Team
License: MIT License
        
        Copyright (c) 2026 Tuner
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Keywords: cartesia,voice,tts,observability,tuner
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Typing :: Typed
Requires-Python: <3.14,>=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: tuner-core>=0.1.0
Requires-Dist: loguru~=0.7.2
Provides-Extra: cartesia
Requires-Dist: cartesia[websockets]>=1.0; extra == "cartesia"
Dynamic: license-file

# tuner-tts-observer

TTS observability for Tuner. Wraps your TTS provider's synthesis stream to
capture agent transcript and latency — Tuner observes synthesis, it does not
own or drive it. Removing the context managers leaves your synthesis code
completely unaffected.

## Supported adapters

| Provider | Adapter | Status |
|---|---|---|
| Cartesia | `CartesiaAdapter` | ✅ Supported |
| ElevenLabs | — | SOON |
| OpenAI TTS | — | SOON |

Need a provider that isn't listed? Extend `BaseTTSAdapter` — see
[Custom providers](#custom-providers) below.

## Installation

```bash
pip install tuner-core tuner-tts-observer
```

WebSocket support (recommended) requires the websockets extra:

```bash
pip install "tuner-tts-observer[cartesia]"
# equivalent to: pip install "cartesia[websockets]>=1.0"
```

## Usage

### WebSocket (recommended)

WebSocket is the production-standard pattern for real-time voice agents. It
supports word-level interruption detection — when the user interrupts the
agent, Tuner records only the words actually spoken, not the full intended
response.

```python
import os
from cartesia import AsyncCartesia
from tuner_core import TunerConfig, TunerSession
from tuner_tts_observer import CartesiaAdapter

cartesia_client = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])

session = TunerSession(config=TunerConfig.from_env(), call_id=call_id)
adapter = session.attach(CartesiaAdapter())

# Open one WebSocket connection per call — reused across turns
async with cartesia_client.tts.websocket_connect() as tts_ws:

    # Per turn:
    ctx = await tts_ws.context(
        model_id="sonic-2",
        voice={"id": "your-voice-id", "mode": "id"},
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
        add_timestamps=True,  # required for word-level interruption detection
    )
    await ctx.push(agent_text)
    await ctx.no_more_inputs()

    async with adapter.track_ws(agent_text) as tracked:
        async for chunk in tracked(ctx.receive()):
            if barge_in_event.is_set():
                break  # interrupted — Tuner records only what was spoken
            if chunk.audio:
                await websocket.send_bytes(chunk.audio)

await session.flush()
```

### SSE (legacy)

SSE is the simpler pattern, available for HTTP-only stacks or existing
integrations. Interruption detection is supported but spoken text cannot be
cut accurately.

```python
import os
from cartesia import Cartesia
from tuner_core import TunerConfig, TunerSession
from tuner_tts_observer import CartesiaAdapter

cartesia_client = Cartesia(api_key=os.environ["CARTESIA_API_KEY"])

session = TunerSession(config=TunerConfig.from_env(), call_id=call_id)
adapter = session.attach(CartesiaAdapter())

with adapter.track(agent_text) as tracked:
    for chunk in tracked(cartesia_client.tts.sse(
        model_id="sonic-2",
        transcript=agent_text,
        voice={"id": "your-voice-id", "mode": "id"},
        output_format={"container": "raw", "encoding": "pcm_s16le", "sample_rate": 16000},
    )):
        if chunk.audio:
            await websocket.send_bytes(chunk.audio)

await session.flush()
```

## Custom providers

For providers other than Cartesia, extend `BaseTTSAdapter` from
`tuner_tts_observer` — its docstring documents the full contract
(timestamping, `_record_agent_turn()` / `_record_tts_usage()`,
`mark_interrupted()`) with a worked example.

## What gets captured automatically

| Signal | SSE | WebSocket |
|---|---|---|
| Agent transcript text | ✓ full text | ✓ spoken words only on interruption |
| Turn start timestamp | ✓ (first audio chunk) | ✓ |
| Turn duration | ✓ (acoustic length) | ✓ |
| TTS TTFB | ✓ | ✓ |
| E2e latency | ✓ | ✓ |
| LLM latency | ✓ | ✓ |
| `interrupted: true` on barge-in | ✓ | ✓ |
| Word-accurate spoken text cut | ✗ | ✓ |

## Development

```bash
uv sync
make test
```
