+ {Object.entries(costs).map(([provider, models]) => (
+
+
+ {titleCase(provider)} {providerSuffix}
+
+
+ {Object.entries(models).map(([model, cost]) => {
+ const isHighest = hasSpread && globalHighest?.provider === provider && globalHighest?.model === model;
+ const isLowest = hasSpread && globalLowest?.provider === provider && globalLowest?.model === model;
return (
-
- {Object.entries(selectedRecord.session_model_costs_usd).map(([provider, models]) => (
-
-
{titleCase(provider)} STT
-
- {Object.entries(models).map(([model, cost]) => {
- const isHighest = hasSpread && globalHighest?.provider === provider && globalHighest?.model === model;
- const isLowest = hasSpread && globalLowest?.provider === provider && globalLowest?.model === model;
- return (
-
-
-
{model}
-
- {isHighest ? (
- Highest
- ) : null}
- {isLowest ? (
- Lowest
- ) : null}
-
-
-
~${cost.toFixed(6)}
-
- );
- })}
-
+
+
+
{model}
+
+ {isHighest ? (
+ Highest
+ ) : null}
+ {isLowest ? (
+ Lowest
+ ) : null}
- ))}
+
+
~${cost.toFixed(6)}
);
- })()}
+ })}
-
- ) : null}
-
+
+ ))}
+
);
}
+function SectionHeader({ label }: { label: string }) {
+ return (
+
{label}
+ );
+}
+
function MetricCard({ label, value, sub }: { label: string; value: string; sub: string }) {
return (
diff --git a/client/src/lib/types.ts b/client/src/lib/types.ts
index 311dd22..d8c74af 100644
--- a/client/src/lib/types.ts
+++ b/client/src/lib/types.ts
@@ -73,8 +73,12 @@ export interface AudioEvaluationRecord {
created_at: string;
turn_count: number;
session_stt_duration_sec: number;
+ streamed_seconds: number;
+ stt_cost_usd: number | null;
session_model_costs_usd: Record>;
provider_session_metrics: Record;
file_paths: string[];
evaluate_mode: boolean;
+ session_tts_sent_characters?: number | null;
+ session_tts_model_costs_usd?: Record>;
}
diff --git a/docs/cost-metering-design.md b/docs/cost-metering-design.md
new file mode 100644
index 0000000..3e98452
--- /dev/null
+++ b/docs/cost-metering-design.md
@@ -0,0 +1,385 @@
+# Cross-Provider Cost Comparison: Design & Metering Plan
+
+**Status:** Design proposal (not yet implemented)
+**Date:** 2026-07-15
+**Scope:** STT + TTS cost metering and cross-provider cost comparison for voice sessions.
+
+---
+
+## 1. Intent
+
+VoiceLab runs each voice session on one configured STT provider and one TTS
+provider (currently Deepgram or ElevenLabs). At the end of a session, the user
+should see a **cost comparison**: what this exact session cost on the
+configured providers, and what it *would have cost* on each alternative
+provider/model.
+
+## 2. Guiding principle
+
+> **Cost is not a thing you measure — usage is. Cost = usage × rate.**
+
+If usage is captured once, in provider-neutral billing units, then comparing
+N providers is pure arithmetic against a rate table. The billing units are:
+
+| Service | Billing unit | Comparable across providers? |
+|---------|-------------|------------------------------|
+| STT | Seconds of **audio streamed** to the provider (silence included) | Yes — same audio would be streamed to any provider |
+| TTS | Characters **sent** to the provider (including text cleared on interruption, see §6) | Yes — the pipeline sends the same text to any provider; interruption timing, not the provider, decides what gets sent |
+| LLM | Input/output **tokens** | Approximately (tokenizers differ) — out of scope for now |
+
+A direct corollary: **do not run shadow sessions on other providers to learn
+cost.** Real parallel calls are the right tool for comparing *quality and
+latency* (which the existing STT-evaluation feature does), but for cost they
+are pure waste — you would pay every provider to learn a number that is
+computable from published rates. Extrapolation from measured usage is the
+correct method for cost.
+
+---
+
+## 3. Audit of the current implementation
+
+Findings from a full sweep of the codebase (2026-07-15):
+
+### 3.1 Two contradictory pricing tables
+
+- `server/app/services/pricing.py` — LLM + STT + TTS rates. **Dead code**:
+ imported nowhere in `app/` (only its own test). Consumes `usage.llm` /
+ `usage.stt` / `usage.tts` trace events that are no longer emitted.
+- `server/app/services/stt_evaluation_pricing.py` — STT-only, the live one.
+ Deepgram, ElevenLabs, Sarvam rates per audio-minute.
+- The two tables **disagree** (e.g. Deepgram STT $0.0043/min vs $0.0048/min)
+ and use different model naming.
+
+### 3.2 Only STT usage is measured, and it measures the wrong thing
+
+- `AudioBufferProcessor(enable_turn_audio=True)`
+ (`pipecat_streaming_runtime.py:524`) captures per-user-turn audio.
+- `compute_all_model_costs(duration_sec)`
+ (`stt_evaluation_pricing.py:71`) multiplies that single turn duration by
+ every provider's rate.
+- **Problem:** streaming STT providers bill on audio *streamed* to them —
+ silence included — not on VAD-detected speech turns. Turn-only counting
+ systematically underestimates real cost.
+
+### 3.3 TTS and LLM are not metered at all
+
+- `PipelineParams(enable_usage_metrics=False)`
+ (`pipecat_streaming_runtime.py:565`) disables Pipecat's usage metrics.
+- No TTS character counting exists in the active path; no LLM token counting.
+- The TTS half of the cost-comparison intent is currently 100% missing.
+
+### 3.4 Results are stored on disk, ranked on the frontend
+
+- Costs live in `recordings//metrics.jsonl`
+ (`stt_evaluation_store.py`), not Postgres — not transactional, not
+ queryable; losing the recordings dir loses cost history even though the
+ runs still exist in the DB.
+- `GET /api/audio-evaluations/agent/{agent_id}` re-reads JSONL files per
+ request.
+- The highest/lowest ranking is recomputed client-side on every render
+ (`client/src/components/audio/AudioView.tsx:159-166`).
+
+### 3.5 Other gaps
+
+- **Pricing alias gap:** a live agent using Deepgram `nova-3` misses the
+ pricing keys `nova-3-monolingual` / `nova-3-multilingual` and silently
+ computes **$0** (`stt_evaluation_pricing.py:64-66`).
+- **Sarvam mismatch:** Sarvam is priced and shadow-evaluated but is not a
+ selectable live provider (`SUPPORTED_STT_PROVIDERS` excludes it,
+ `schemas/agent.py:10`).
+- Shadow STT calls to ElevenLabs + Sarvam run on every turn regardless of
+ the agent's provider — justified for quality comparison only, and already
+ gated by `enable_stt_evaluation`; keep it that way.
+
+---
+
+## 4. Target architecture (four layers)
+
+### Layer 1 — Meter usage in neutral units, per session, into `trace_events`
+
+Emit immutable usage facts alongside existing transcript events. **No prices
+anywhere in these events** — prices change, facts don't.
+
+- `usage.stt` → `{ provider, model, streamed_seconds, speech_seconds }`
+- `usage.tts` → `{ provider, model, voice, sent_characters }`
+ (one per session, emitted at teardown like `usage.stt`; see §6.
+ `heard_characters` is a future analytics addition)
+- `usage.llm` → `{ model, input_tokens, output_tokens }` (future)
+
+### Layer 2 — One pricing catalog, server-side
+
+A single source of truth (delete the dead `pricing.py`). Entries keyed by
+`(service, provider, model)` with:
+
+- `unit` — per audio-minute / per 1M chars / per 1M tokens
+- `rate`
+- `effective_from` — so rate history is explicit
+- **billing adjusters** — per-provider minimum billable duration,
+ second-rounding, character-block rounding. These make the comparison
+ honest; naive `duration × rate` is only an approximation.
+
+Start as a Python module; keep the door open to a `pricing_rates` DB table so
+rates can change without deploys. Fix the `nova-3` alias gap.
+
+### Layer 3 — Compute + snapshot the comparison server-side, at session end
+
+When the run closes: aggregate the run's usage events → apply the catalog →
+produce one cost object:
+
+- actual cost for the configured providers,
+- hypothetical cost per alternative provider/model,
+- **the rate used embedded in each line** (snapshot — when a provider changes
+ prices next quarter, old runs must still show what was true at run time).
+
+Store it in `runs.summary` (existing, unused JSONB column) or a small
+`run_costs` table — **not** JSONL files on disk. The server also computes
+ranking and deltas; the frontend only renders.
+
+### Layer 4 — Presentation
+
+End-of-session summary in the test-call panel and the Runs view: a table of
+provider/model → estimated cost, the actually-used provider highlighted,
+delta vs. actual ("ElevenLabs would have been +$0.0021, +38%"). Label
+alternatives explicitly as *estimates*.
+
+---
+
+## 5. STT metering: what we can do, do, and should do
+
+**The question:** how many seconds does the provider bill for a session?
+Providers bill on **audio sent to them**, silence included.
+
+### Three possible measuring points
+
+| Option | How | Verdict |
+|--------|-----|---------|
+| 1. Speech turns | Capture audio only while VAD says the user is speaking | What we do today. Right for analytics, **wrong for billing** — most of a session is silence/agent speech that still streams to the provider |
+| 2. Connection wall-time | Websocket open duration × sample rate | Crude approximation; ignores mutes, reconnect gaps |
+| 3. **Bytes actually sent** | Count bytes crossing the wire to the provider; seconds = bytes ÷ (sample_rate × 2) for 16-bit mono PCM | **Correct** — this is exactly the billed quantity |
+
+### Does Pipecat provide a tool for this?
+
+**No.** Verified against the installed Pipecat 1.3.0: there is no
+`STTUsageMetricsData` and no STT usage metering of any kind (STT services
+emit only TTFB/processing metrics). We must count it ourselves.
+
+### The one right seam
+
+Pipecat's `STTService.process_audio_frame`
+(`pipecat/services/stt_service.py:349`) applies all "will this audio actually
+be sent" guards — reconnect buffering, mute-drop, empty-frame skip — and only
+then calls `run_stt(frame.audio)`, which pushes the bytes over the provider
+websocket. So **`run_stt` receives exactly the bytes the provider bills
+for.**
+
+We already subclass both STT services in
+`pipecat_streaming_runtime.py` (`InstrumentedDeepgramSTTService:143` and the
+ElevenLabs equivalent), so:
+
+```python
+class InstrumentedDeepgramSTTService(ProviderRequestTraceMixin, DeepgramSTTService):
+ def __init__(self, ...):
+ ...
+ self._streamed_bytes = 0
+
+ async def run_stt(self, audio: bytes):
+ self._streamed_bytes += len(audio)
+ async for frame in super().run_stt(audio):
+ yield frame
+
+ @property
+ def streamed_seconds(self) -> float:
+ # 16-bit mono PCM: 2 bytes per sample
+ return self._streamed_bytes / (self.sample_rate * 2)
+```
+
+(`self.sample_rate` is set once the `StartFrame` arrives; input is mono per
+the websocket serializer at `pipecat_streaming_runtime.py:71`.)
+
+At session teardown (after `runner.run(task)` returns), emit one event:
+
+```python
+await record_trace("usage.stt", {
+ "provider": ...,
+ "model": ...,
+ "streamed_seconds": stt.streamed_seconds,
+})
+```
+
+### Keep both numbers
+
+Do **not** delete the `AudioBufferProcessor` turn capture. Keep:
+
+- `streamed_seconds` — the billing truth (feeds cost),
+- `speech_seconds` — the analytics number (feeds quality/latency work).
+
+They answer different questions.
+
+### Free calibration signal
+
+`InstrumentedDeepgramSTTService._on_message` already intercepts Deepgram's
+`ListenV1Metadata` for `request_id`
+(`pipecat_streaming_runtime.py:158-160`). That metadata also carries a
+`duration` field — the seconds **Deepgram itself** says it processed. Log it
+next to our byte-counted `streamed_seconds`: if they agree within rounding,
+the meter is provably accurate, which is what makes extrapolated costs for
+*other* providers trustworthy. No extra HTTP calls needed. (A fuller
+usage-API reconciliation was attempted in commit `07f2dd9` and reverted;
+revisit once metering is solid.)
+
+---
+
+## 6. TTS metering: bill = characters *sent*, and Pipecat's built-in metric can't measure that
+
+*(Rewritten 2026-07-16 — the earlier version of this section was wrong on two
+counts, marked below.)*
+
+### 6.1 How Deepgram actually bills
+
+- Aura is billed per input character: Aura-2 $0.030/1k chars pay-as-you-go
+ ($0.027 Growth), Aura-1 $0.015/1k.
+- We use the **websocket** TTS API. Deepgram counts characters **sent to the
+ websocket** — its own throughput limit is documented as "measured by the
+ number of characters sent to the websocket." Nothing in the `Clear` docs
+ promises a refund for buffered text cleared before synthesis; assume
+ **sent = billed** until reconciled against the console (§6.4).
+- **Interruption effect:** Pipecat pushes each sentence to the socket as soon
+ as the LLM streams it — usually several sentences ahead of audio playback.
+ On user interruption, Pipecat sends `Clear`
+ (`pipecat/services/deepgram/tts.py:274`); audio stops, but every character
+ already submitted via `Speak` messages was sent and is billed. So
+ **Deepgram's bill > characters the user actually heard**, and > any count
+ derived from the spoken transcript. Same lesson as STT: the billing unit is
+ what crosses the wire (characters sent), not what the user experiences
+ (characters heard).
+
+### 6.2 Why `enable_usage_metrics=True` is NOT sufficient (correction)
+
+The earlier draft said flipping the flag suffices. Verified against the
+installed package, it does not:
+
+- Our `AdkDeepgramTTSService` extends the **websocket** `DeepgramTTSService`,
+ whose `run_tts` never calls `start_tts_usage_metrics`. The call the earlier
+ draft cited (`deepgram/tts.py:497`) is in `DeepgramHttpTTSService`, which we
+ don't use. In our default SENTENCE aggregation mode, Deepgram WS TTS would
+ report **zero** usage.
+- The base-class fallback (`_streamed_text` accumulated and emitted at
+ `LLMFullResponseEndFrame`, `tts_service.py:723-726`) only operates in TOKEN
+ aggregation mode — and even there, `_handle_interruption`
+ (`tts_service.py:902-910`) wipes `_streamed_text` **without emitting**, so
+ interrupted turns are never counted. Interrupted turns are exactly where
+ billing diverges from the transcript, so this is the worst possible gap.
+- ElevenLabs WS `run_tts` *does* call `start_tts_usage_metrics(text)` right
+ after sending (`elevenlabs/tts.py:1026`), i.e. the built-in metric is
+ per-provider inconsistent: roughly right for ElevenLabs, absent for
+ Deepgram.
+
+### 6.3 The fix: meter at the send seam (mirror of `SttUsageMeterMixin`)
+
+Count characters ourselves at the exact point text leaves for the provider —
+`run_tts(text, context_id)`, which receives the final prepared/transformed
+text (post `normalize_tts_text`) and immediately sends it (`Speak` for
+Deepgram, `_send_text` for ElevenLabs):
+
+```python
+class TtsUsageMeterMixin:
+ """Count characters at the run_tts seam — the exact text sent to the provider."""
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self._sent_characters = 0
+
+ async def run_tts(self, text: str, context_id: str):
+ self._sent_characters += len(text)
+ async for frame in super().run_tts(text, context_id):
+ yield frame
+```
+
+Apply to both `AdkDeepgramTTSService` and `AdkElevenLabsTTSService`. At
+session teardown (next to `usage.stt`), emit:
+
+- `usage.tts` → `{ provider, model, voice, sent_characters }`
+ - `sent_characters` — billing truth.
+ - `heard_characters` (future, analytics) — characters actually played to
+ the user (derivable from the `agent.text` transcript events). The delta
+ quantifies interruption waste — worth surfacing in the UI as its own
+ insight ("23% of TTS spend was interrupted audio").
+
+### 6.4 Calibration
+
+- **Deepgram:** the Management/Usage API reports per-request character counts;
+ the console usage page shows the same. Run one deliberately-interrupted
+ session, compare our `sent_characters` against Deepgram's reported count —
+ this both validates the meter and settles the cleared-text question
+ empirically (same role `ListenV1Metadata.duration` plays for STT).
+- **ElevenLabs:** the subscription/user API exposes a cumulative character
+ counter; snapshot before/after a session.
+
+### 6.5 Cross-provider comparison remains valid
+
+`sent_characters` is **pipeline-determined** (LLM sentence pacing +
+interruption timing), not provider-determined — the same conversation with
+the same interruptions pushes the same text to whichever TTS service sits in
+the pipeline, and both Deepgram and ElevenLabs bill on characters received.
+So `sent_characters × rate` extrapolates fairly. Catalog detail: ElevenLabs
+bills in credits with per-model multipliers (e.g. Flash v2.5 ≈ 0.5
+credits/char vs Multilingual v2 at 1 credit/char) — encode as model-specific
+rates, and label its effective $/char from a subscription tier explicitly,
+since ElevenLabs pricing is plan-based rather than pure pay-as-you-go.
+
+**Per-provider interruption semantics differ — in the catalog, not the
+meter.** ElevenLabs' stated policy is that credits are deducted on
+*successful audio generation* (failed requests aren't charged), and on
+interruption Pipecat sends `close_context` (`elevenlabs/tts.py:843-845`), so
+text buffered but never synthesized *may* not be billed — unlike Deepgram,
+where we must assume sent = billed. Neither vendor documents the
+interrupted-buffer case precisely, and since ElevenLabs synthesizes eagerly
+per chunk, sent ≈ generated in practice. Treat `sent_characters` as exact for
+Deepgram and as a tight upper bound for ElevenLabs; per-provider calibration
+(§6.4) is what turns these assumptions into measured facts. The meter itself
+never changes per provider — one `TtsUsageMeterMixin` on the shared `run_tts`
+seam covers every Pipecat TTS service, current and future.
+
+**Caveat (unchanged):** if we also flip `enable_usage_metrics=True` for other
+reasons, `LLMUsageMetricsData` frames may reach `MetricsSink` — handle or
+ignore them explicitly. LLM tokens can otherwise be read from ADK/Gemini
+response `usage_metadata`.
+
+---
+
+## 7. Implementation checklist (in order)
+
+1. **Pricing catalog** — consolidate to one table with units, effective
+ dates, and billing adjusters; delete dead `pricing.py`; fix the `nova-3`
+ alias gap (currently silently $0).
+2. **STT metering** — ✅ implemented (2026-07-15). `SttUsageMeterMixin` counts
+ bytes in `run_stt` on both instrumented STT subclasses; a `usage.stt`
+ trace event (streamed + speech + provider-reported seconds) is emitted at
+ session teardown; Deepgram metadata `duration` is accumulated for
+ calibration. Tests in `server/tests/test_stt_usage_meter.py`.
+3. **TTS metering** — `TtsUsageMeterMixin` counting `len(text)` in `run_tts`
+ on both instrumented TTS subclasses (do NOT rely on
+ `enable_usage_metrics` — see §6.2); emit `usage.tts` with
+ `sent_characters` (+ optional `heard_characters`) at session teardown;
+ calibrate one interrupted session against Deepgram console usage.
+4. **Aggregation** — at run close, aggregate `usage.*` events → compute
+ actual + hypothetical costs with rate snapshots → store in
+ `runs.summary` (or `run_costs` table). Stop treating `metrics.jsonl` as
+ the source of truth for cost.
+5. **Presentation** — server-computed ranking/deltas; render in the
+ test-call end screen and Runs view; label alternatives as estimates.
+
+## 8. Key file reference
+
+| Concern | Path |
+|---------|------|
+| Live pricing (STT-only) | `server/app/services/stt_evaluation_pricing.py` |
+| Dead pricing (to delete) | `server/app/services/pricing.py` |
+| Pipeline + instrumented services | `server/app/services/pipecat_streaming_runtime.py` |
+| Metrics sink (pipeline tail) | `server/app/services/pipeline_metrics.py` |
+| STT shadow evaluation | `server/app/services/stt_evaluation_service.py` |
+| JSONL metrics store (to retire for cost) | `server/app/services/stt_evaluation_store.py` |
+| Evaluation API | `server/app/api/routes/audio_evaluations.py` |
+| Agent provider config schema | `server/app/schemas/agent.py` |
+| Cost comparison UI | `client/src/components/audio/AudioView.tsx` |
+| Pipecat STT base (the `run_stt` seam) | `.venv/.../pipecat/services/stt_service.py:349` |
+| Pipecat usage metric types | `.venv/.../pipecat/metrics/metrics.py` |
diff --git a/server/app/api/routes/audio_evaluations.py b/server/app/api/routes/audio_evaluations.py
index ff53ed8..05e11e0 100644
--- a/server/app/api/routes/audio_evaluations.py
+++ b/server/app/api/routes/audio_evaluations.py
@@ -3,12 +3,13 @@
from typing import Annotated, Any
from fastapi import APIRouter, Depends
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import get_settings
from app.core.db import get_db_session
from app.repositories.run_repository import RunRepository
+from app.services.stt_evaluation_pricing import compute_all_model_costs
from app.services.stt_evaluation_store import resolve_recordings_root
router = APIRouter(prefix="/audio-evaluations", tags=["audio-evaluations"])
@@ -31,10 +32,14 @@ class AudioEvaluationRead(BaseModel):
created_at: str
turn_count: int
session_stt_duration_sec: float
+ streamed_seconds: float
+ stt_cost_usd: float | None = None
session_model_costs_usd: dict[str, dict[str, float]]
provider_session_metrics: dict[str, AudioProviderMetricsRead]
file_paths: list[str]
evaluate_mode: bool
+ session_tts_sent_characters: int | None = None
+ session_tts_model_costs_usd: dict[str, dict[str, float]] = Field(default_factory=dict)
@router.get("/agent/{agent_id}", response_model=list[AudioEvaluationRead])
@@ -51,6 +56,7 @@ async def list_audio_evaluations(agent_id: str, session: SessionDep) -> list[Aud
payload = _load_metrics_summary(metrics_path)
if payload is None:
continue
+ usage = _extract_usage_stt(run.trace_events)
records.append(
AudioEvaluationRead(
session_id=run.adk_session_id,
@@ -58,11 +64,18 @@ async def list_audio_evaluations(agent_id: str, session: SessionDep) -> list[Aud
adk_session_id=run.adk_session_id,
created_at=run.created_at.isoformat(),
turn_count=payload["turn_count"],
- session_stt_duration_sec=payload["session_stt_duration_sec"],
- session_model_costs_usd=payload["session_model_costs_usd"],
+ session_stt_duration_sec=usage["streamed_seconds"]
+ or payload["session_stt_duration_sec"],
+ streamed_seconds=usage["streamed_seconds"],
+ stt_cost_usd=usage["cost_usd"],
+ session_model_costs_usd=compute_all_model_costs(
+ usage["streamed_seconds"] or payload["session_stt_duration_sec"]
+ ),
provider_session_metrics=payload["provider_session_metrics"],
file_paths=payload["file_paths"],
evaluate_mode=payload["evaluate_mode"],
+ session_tts_sent_characters=payload["session_tts_sent_characters"],
+ session_tts_model_costs_usd=payload["session_tts_model_costs_usd"],
)
)
return records
@@ -95,4 +108,31 @@ def _load_metrics_summary(metrics_path: Path) -> dict[str, Any] | None:
"session_model_costs_usd": session_summary.get("session_model_costs_usd") or {},
"provider_session_metrics": session_summary.get("provider_session_metrics") or {},
"evaluate_mode": bool(session_summary.get("evaluate_mode", False)),
+ "session_tts_sent_characters": (
+ int(session_summary["session_tts_sent_characters"])
+ if session_summary.get("session_tts_sent_characters") is not None
+ else None
+ ),
+ "session_tts_model_costs_usd": session_summary.get("session_tts_model_costs_usd") or {},
}
+
+
+def _extract_usage_stt(trace_events: list[Any]) -> dict[str, float | None]:
+ for event in reversed(trace_events):
+ if getattr(event, "event_type", None) != "usage.stt":
+ continue
+ payload = getattr(event, "payload", {}) or {}
+ return {
+ "streamed_seconds": float(payload.get("streamed_seconds") or 0.0),
+ "cost_usd": _to_float(payload.get("cost_usd")),
+ }
+ return {"streamed_seconds": 0.0, "cost_usd": None}
+
+
+def _to_float(value: Any) -> float | None:
+ if value is None:
+ return None
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return None
diff --git a/server/app/repositories/run_repository.py b/server/app/repositories/run_repository.py
index 3c37a13..84740c2 100644
--- a/server/app/repositories/run_repository.py
+++ b/server/app/repositories/run_repository.py
@@ -39,6 +39,7 @@ async def list_by_agent(self, agent_id: str) -> list[RunRecord]:
result = await self.session.execute(
select(RunRecord)
.where(RunRecord.agent_id == agent_id)
+ .options(selectinload(RunRecord.trace_events))
.order_by(RunRecord.created_at.desc())
)
return list(result.scalars())
diff --git a/server/app/services/pipecat_streaming_runtime.py b/server/app/services/pipecat_streaming_runtime.py
index 5812685..61c1b05 100644
--- a/server/app/services/pipecat_streaming_runtime.py
+++ b/server/app/services/pipecat_streaming_runtime.py
@@ -1,9 +1,10 @@
import json
import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable
from typing import Any
from uuid import uuid4
+from deepgram.listen.v1.types import ListenV1Metadata, ListenV1Results
from fastapi import WebSocket
from pipecat.frames.frames import (
BotStartedSpeakingFrame,
@@ -20,14 +21,14 @@
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.serializers.base_serializer import FrameSerializer
-from pipecat.services.settings import LLMSettings
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.deepgram.tts import DeepgramTTSService
from pipecat.services.elevenlabs.stt import CommitStrategy, ElevenLabsRealtimeSTTService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
+from pipecat.services.settings import LLMSettings
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams, FastAPIWebsocketTransport
from pipecat_adk import AdkLLMService, SessionParams, VqlTTSMixin
from pipecat_adk.frames import (
@@ -35,15 +36,15 @@
VqlLLMFullResponseStartFrame,
VqlLLMTextFrame,
)
-from deepgram.listen.v1.types import ListenV1Metadata, ListenV1Results
from app.core.config import get_settings
-from app.schemas.agent import AgentConfig, DEFAULT_TTS_MODEL_BY_PROVIDER
+from app.schemas.agent import DEFAULT_TTS_MODEL_BY_PROVIDER, AgentConfig
from app.services.adk_session_service import create_adk_session_service, ensure_adk_session
from app.services.pipecat_adk_runtime import PipecatAdkRuntime
from app.services.pipeline_metrics import (
MetricsSink,
)
+from app.services.pricing import compute_cost
from app.services.stt_evaluation_service import SttEvaluationSession
logger = logging.getLogger("uvicorn.error")
@@ -140,7 +141,37 @@ async def _record_provider_request_from_mapping(
return
-class InstrumentedDeepgramSTTService(ProviderRequestTraceMixin, DeepgramSTTService):
+class SttUsageMeterMixin:
+ """Meter raw audio bytes actually sent to the STT provider.
+
+ `STTService.process_audio_frame` applies the mute/reconnect/empty-frame
+ guards and only then calls `run_stt`, so every byte counted here is audio
+ the provider bills for (silence included). VAD turn audio undercounts.
+ """
+
+ _streamed_audio_bytes: int = 0
+ _provider_reported_audio_seconds: float = 0.0
+
+ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]:
+ self._streamed_audio_bytes += len(audio)
+ async for frame in super().run_stt(audio):
+ yield frame
+
+ @property
+ def streamed_audio_seconds(self) -> float:
+ if not self.sample_rate:
+ return 0.0
+ # 16-bit mono PCM: 2 bytes per sample.
+ return self._streamed_audio_bytes / (self.sample_rate * 2)
+
+ @property
+ def provider_reported_audio_seconds(self) -> float | None:
+ return self._provider_reported_audio_seconds or None
+
+
+class InstrumentedDeepgramSTTService(
+ SttUsageMeterMixin, ProviderRequestTraceMixin, DeepgramSTTService
+):
def __init__(
self, *, record_trace: TraceRecorder, provider_model: str, run_tag: str, **kwargs: Any
) -> None:
@@ -158,13 +189,46 @@ def __init__(
async def _on_message(self, message: Any) -> None:
if isinstance(message, ListenV1Metadata):
await self._record_provider_request(message.request_id)
+ # Deepgram reports processed audio duration per connection on
+ # close; accumulate across reconnects for meter calibration.
+ duration = getattr(message, "duration", None)
+ if duration:
+ self._provider_reported_audio_seconds += float(duration)
elif isinstance(message, ListenV1Results):
metadata = getattr(message, "metadata", None)
await self._record_provider_request(getattr(metadata, "request_id", None))
await super()._on_message(message)
-class AdkDeepgramTTSService(ProviderRequestTraceMixin, VqlTTSMixin, DeepgramTTSService):
+class TtsUsageMeterMixin:
+ """Meter characters actually sent to the TTS provider.
+
+ `run_tts` receives the final prepared text (post text-transforms) and
+ immediately sends it to the provider (`Speak` message for Deepgram,
+ context text for ElevenLabs). Providers bill on characters sent —
+ including text later cleared/closed by an interruption — so this counts
+ the billed quantity. Transcript-derived counts undercount whenever the
+ user interrupts, and Pipecat's built-in `TTSUsageMetricsData` metric is
+ never emitted by the Deepgram websocket service and is dropped on
+ interruption, so we count at the send seam ourselves (same approach as
+ `SttUsageMeterMixin`).
+ """
+
+ _sent_characters: int = 0
+
+ async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]:
+ self._sent_characters += len(text)
+ async for frame in super().run_tts(text, context_id):
+ yield frame
+
+ @property
+ def sent_characters(self) -> int:
+ return self._sent_characters
+
+
+class AdkDeepgramTTSService(
+ TtsUsageMeterMixin, ProviderRequestTraceMixin, VqlTTSMixin, DeepgramTTSService
+):
def __init__(
self,
*,
@@ -193,7 +257,9 @@ async def _connect_websocket(self):
await self._record_provider_request_from_headers(response_headers, "dg-request-id")
-class InstrumentedElevenLabsSTTService(ProviderRequestTraceMixin, ElevenLabsRealtimeSTTService):
+class InstrumentedElevenLabsSTTService(
+ SttUsageMeterMixin, ProviderRequestTraceMixin, ElevenLabsRealtimeSTTService
+):
def __init__(
self, *, record_trace: TraceRecorder, provider_model: str, run_tag: str, **kwargs: Any
) -> None:
@@ -224,7 +290,9 @@ async def _process_response(self, data: dict):
await super()._process_response(data)
-class AdkElevenLabsTTSService(ProviderRequestTraceMixin, VqlTTSMixin, ElevenLabsTTSService):
+class AdkElevenLabsTTSService(
+ TtsUsageMeterMixin, ProviderRequestTraceMixin, VqlTTSMixin, ElevenLabsTTSService
+):
def __init__(
self,
*,
@@ -609,5 +677,28 @@ async def on_session_timeout(_transport, _websocket):
try:
await runner.run(task)
finally:
- await stt_evaluation.finalize()
+ try:
+ stt_usage_payload = {
+ "provider": config.stt_provider,
+ "model": config.stt_model,
+ "streamed_seconds": round(stt.streamed_audio_seconds, 3),
+ "speech_seconds": stt_evaluation.session_duration_sec,
+ "provider_reported_seconds": stt.provider_reported_audio_seconds,
+ }
+ stt_usage_payload["cost_usd"] = float(compute_cost("usage.stt", stt_usage_payload))
+ await record_trace(
+ "usage.stt",
+ stt_usage_payload,
+ )
+ await record_trace(
+ "usage.tts",
+ {
+ "provider": config.tts_provider,
+ "model": metrics_tts_model,
+ "voice": config.tts_voice,
+ "sent_characters": tts.sent_characters,
+ },
+ )
+ finally:
+ await stt_evaluation.finalize(tts_sent_characters=tts.sent_characters)
logger.info("pipecat streaming test-call ended run_id=%s", run_id)
diff --git a/server/app/services/pricing.py b/server/app/services/pricing.py
index 2e20ac7..ae6c702 100644
--- a/server/app/services/pricing.py
+++ b/server/app/services/pricing.py
@@ -23,9 +23,14 @@
# STT rates: USD per audio-minute submitted.
STT_RATES: dict[str, Decimal] = {
- "deepgram:nova-3": Decimal("0.0043"),
+ "deepgram:nova-3": Decimal("0.0077"),
+ "deepgram:nova-3-streaming": Decimal("0.0077"),
+ "deepgram:nova-3-multilingual": Decimal("0.0043"),
"deepgram:nova-2": Decimal("0.0043"),
- "deepgram:base": Decimal("0.0125"),
+ "elevenlabs:scribe-v2": Decimal("0.00367"),
+ "elevenlabs:scribe-v2-realtime": Decimal("0.0065"),
+ "sarvam:saarika": Decimal("0.005263"),
+ "sarvam:saarika-diarization": Decimal("0.007895"),
}
# TTS rates: USD per 1M characters synthesized.
@@ -55,7 +60,7 @@ def _stt_cost(payload: dict[str, Any]) -> Decimal:
rate = STT_RATES.get(key)
if rate is None:
return Decimal("0")
- seconds = Decimal(str(payload.get("audio_seconds") or 0))
+ seconds = _stt_seconds(payload)
return (seconds / Decimal("60")) * rate
@@ -107,7 +112,7 @@ def session_totals(events: list[Any]) -> dict[str, Any]:
llm_latencies_ms.append(float(inline_latency))
elif event.event_type == "usage.stt":
payload = event.payload or {}
- local_stt_seconds += Decimal(str(payload.get("audio_seconds") or 0))
+ local_stt_seconds += _stt_seconds(payload)
local_stt_cost += _stt_cost(payload)
elif event.event_type == "usage.tts":
payload = event.payload or {}
@@ -141,6 +146,7 @@ def session_totals(events: list[Any]) -> dict[str, Any]:
},
"stt": {
"audio_seconds": _q(stt_seconds, "0.001"),
+ "streamed_seconds": _q(stt_seconds, "0.001"),
"cost_usd": _q(stt_cost),
"source": "runtime",
},
@@ -160,5 +166,11 @@ def _avg(values: list[float]) -> float:
return round(sum(values) / len(values), 1)
+def _stt_seconds(payload: dict[str, Any]) -> Decimal:
+ return Decimal(
+ str(payload.get("streamed_seconds") or payload.get("audio_seconds") or 0)
+ )
+
+
def _q(value: Decimal, quant: str = "0.000001") -> float:
return float(value.quantize(Decimal(quant)))
diff --git a/server/app/services/stt_evaluation_pricing.py b/server/app/services/stt_evaluation_pricing.py
index c10606d..a5eb126 100644
--- a/server/app/services/stt_evaluation_pricing.py
+++ b/server/app/services/stt_evaluation_pricing.py
@@ -4,27 +4,31 @@
RATE_PER_MINUTE_USD: dict[str, dict[str, Decimal]] = {
"deepgram": {
- "nova-3-monolingual": Decimal("0.0048"),
- "nova-3-multilingual": Decimal("0.0058"),
+ "nova-3": Decimal("0.0043"),
+ "nova-3-streaming": Decimal("0.0077"),
+ "nova-3-multilingual": Decimal("0.0043"),
},
"elevenlabs": {
"scribe-v2": Decimal("0.00367"),
"scribe-v2-realtime": Decimal("0.0065"),
},
"sarvam": {
- "saarika": Decimal("0.0060"),
+ "saarika": Decimal("0.005263"),
+ "saarika-diarization": Decimal("0.007895"),
},
}
PRODUCTION_MODEL_BY_PROVIDER = {
- "deepgram": "nova-3-monolingual",
+ "deepgram": "nova-3",
"elevenlabs": "scribe_v2",
"sarvam": "saarika:v2",
}
PRICING_MODEL_ALIASES = {
"deepgram": {
- "nova-3-monolingual": "nova-3-monolingual",
+ "nova-3": "nova-3",
+ "nova-3-monolingual": "nova-3",
+ "nova-3-streaming": "nova-3-streaming",
"nova-3-multilingual": "nova-3-multilingual",
},
"elevenlabs": {
@@ -39,6 +43,9 @@
"saarika:v2": "saarika",
"saarika:v2.5": "saarika",
"saarika:flash": "saarika",
+ "saarika:diarization": "saarika-diarization",
+ "saarika-v2-diarization": "saarika-diarization",
+ "saarika-diarization": "saarika-diarization",
"saarika-v1": "saarika",
"saarika-v2": "saarika",
"saarika": "saarika",
diff --git a/server/app/services/stt_evaluation_service.py b/server/app/services/stt_evaluation_service.py
index c603417..a563a6a 100644
--- a/server/app/services/stt_evaluation_service.py
+++ b/server/app/services/stt_evaluation_service.py
@@ -21,6 +21,9 @@
pricing_model_name,
)
from app.services.stt_evaluation_store import SavedTurnAudio, SttEvaluationStore
+from app.services.tts_evaluation_pricing import (
+ compute_all_model_costs as compute_all_tts_model_costs,
+)
logger = logging.getLogger("uvicorn.error")
TraceRecorder = Callable[[str, dict[str, object]], Awaitable[None]]
@@ -64,6 +67,10 @@ def __init__(
self._background_tasks: set[asyncio.Task[None]] = set()
self._provider_results_by_name: dict[str, list[ProviderEvaluationResult]] = {}
+ @property
+ def session_duration_sec(self) -> float:
+ return round(self._session_duration_sec, 3)
+
async def handle_user_turn_audio(
self, audio: bytes, sample_rate: int, num_channels: int
) -> None:
@@ -152,22 +159,25 @@ async def handle_user_turn_audio(
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
- async def finalize(self) -> None:
+ async def finalize(self, *, tts_sent_characters: int | None = None) -> None:
if self._background_tasks:
await asyncio.gather(*self._background_tasks, return_exceptions=True)
- await self._store.append_metrics(
- self._session_id,
- {
- "type": "session.summary",
- "session_id": self._session_id,
- "run_id": self._run_id,
- "session_stt_duration_sec": round(self._session_duration_sec, 3),
- "session_model_costs_usd": compute_session_model_costs(self._session_duration_sec),
- "provider_session_metrics": self._provider_session_metrics(),
- "evaluate_mode": self._evaluate_mode,
- },
- )
+ summary: dict[str, Any] = {
+ "type": "session.summary",
+ "session_id": self._session_id,
+ "run_id": self._run_id,
+ "session_stt_duration_sec": round(self._session_duration_sec, 3),
+ "session_model_costs_usd": compute_session_model_costs(self._session_duration_sec),
+ "provider_session_metrics": self._provider_session_metrics(),
+ "evaluate_mode": self._evaluate_mode,
+ }
+ if tts_sent_characters is not None:
+ summary["session_tts_sent_characters"] = int(tts_sent_characters)
+ summary["session_tts_model_costs_usd"] = compute_all_tts_model_costs(
+ int(tts_sent_characters)
+ )
+ await self._store.append_metrics(self._session_id, summary)
await self._record_trace(
"evaluation.stt.session_summary",
{
diff --git a/server/app/services/tts_evaluation_pricing.py b/server/app/services/tts_evaluation_pricing.py
new file mode 100644
index 0000000..7b725a0
--- /dev/null
+++ b/server/app/services/tts_evaluation_pricing.py
@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from decimal import Decimal
+
+# USD per 1,000,000 characters sent to the TTS provider.
+#
+# Rates are pay-as-you-go list prices. Enterprise / committed-spend rates
+# can be materially lower. Sources:
+# - Deepgram TTS pricing: https://deepgram.com/pricing
+# - ElevenLabs pricing: https://elevenlabs.io/pricing
+#
+# TTS providers price per *model*, not per *voice*. All Aura-2 voices
+# (thalia, asteria, luna, orion, ...) share the aura-2 rate. All ElevenLabs
+# voices used with eleven_multilingual_v2 share the same rate; the same
+# voice on eleven_flash_v2_5 uses the flash rate instead.
+#
+# ElevenLabs bills in "credits per character" (Flash/Turbo = 0.5×,
+# Multilingual v2 / v1 / Monolingual v1 = 1×, v3 alpha = 1× for now).
+# The dollar-per-M-char figures below convert those credit ratios against
+# the $165/M anchor used in the legacy pricing.py module.
+RATE_PER_MILLION_CHARS_USD: dict[str, dict[str, Decimal]] = {
+ "deepgram": {
+ "aura-2": Decimal("30.00"),
+ "aura": Decimal("15.00"),
+ },
+ "elevenlabs": {
+ # UI has no ElevenLabs model dropdown; the app drives eleven_turbo_v2_5
+ # as the default (see server/app/schemas/agent.py). Add more entries here
+ # only when the UI actually exposes them.
+ "eleven_turbo_v2_5": Decimal("82.50"),
+ },
+}
+
+# Deepgram Aura model IDs are voice-suffixed at runtime (e.g. `aura-2-thalia-en`,
+# `aura-2-asteria-en`). We resolve any string starting with a known prefix to the
+# parent model rate, so new voices don't need explicit entries.
+DEEPGRAM_MODEL_PREFIXES: tuple[str, ...] = ("aura-2", "aura")
+
+# ElevenLabs model IDs are stable exact strings; users occasionally pass the
+# dash-hyphenated variant. Map those to the underscore form used above.
+ELEVENLABS_MODEL_ALIASES: dict[str, str] = {
+ "v3": "eleven_v3",
+ "eleven-v3": "eleven_v3",
+ "multilingual-v2": "eleven_multilingual_v2",
+ "eleven-multilingual-v2": "eleven_multilingual_v2",
+ "multilingual-v1": "eleven_multilingual_v1",
+ "monolingual-v1": "eleven_monolingual_v1",
+ "flash-v2-5": "eleven_flash_v2_5",
+ "flash-v2": "eleven_flash_v2",
+ "turbo-v2-5": "eleven_turbo_v2_5",
+ "turbo-v2": "eleven_turbo_v2",
+}
+
+
+def pricing_model_name(*, provider: str, model: str) -> str:
+ if provider == "deepgram":
+ for prefix in DEEPGRAM_MODEL_PREFIXES:
+ if model == prefix or model.startswith(f"{prefix}-") or model.startswith(f"{prefix}_"):
+ return prefix
+ return model
+ if provider == "elevenlabs":
+ if model in RATE_PER_MILLION_CHARS_USD["elevenlabs"]:
+ return model
+ return ELEVENLABS_MODEL_ALIASES.get(model, model)
+ return model
+
+
+def compute_model_cost(sent_characters: int, *, provider: str, model: str) -> float:
+ pricing_model = pricing_model_name(provider=provider, model=model)
+ rate = RATE_PER_MILLION_CHARS_USD.get(provider, {}).get(pricing_model)
+ if rate is None:
+ return 0.0
+ chars = Decimal(str(max(sent_characters, 0))) / Decimal("1000000")
+ return _q(chars * rate)
+
+
+def compute_all_model_costs(sent_characters: int) -> dict[str, dict[str, float]]:
+ chars = Decimal(str(max(sent_characters, 0))) / Decimal("1000000")
+ return {
+ provider: {model: _q(chars * rate) for model, rate in models.items()}
+ for provider, models in RATE_PER_MILLION_CHARS_USD.items()
+ }
+
+
+def _q(value: Decimal) -> float:
+ return float(value.quantize(Decimal("0.000001")))
diff --git a/server/tests/test_pricing.py b/server/tests/test_pricing.py
index 87c592b..34aaa87 100644
--- a/server/tests/test_pricing.py
+++ b/server/tests/test_pricing.py
@@ -22,12 +22,20 @@ def test_llm_cost_unknown_model_is_zero() -> None:
def test_stt_cost_deepgram_nova_3() -> None:
- # $0.0043 per audio minute.
+ # $0.0077 per audio minute for streaming.
+ cost = compute_cost(
+ "usage.stt",
+ {"streamed_seconds": 60, "provider": "deepgram", "model": "nova-3"},
+ )
+ assert float(cost) == 0.0077
+
+
+def test_stt_cost_falls_back_to_audio_seconds_for_legacy_payloads() -> None:
cost = compute_cost(
"usage.stt",
{"audio_seconds": 60, "provider": "deepgram", "model": "nova-3"},
)
- assert float(cost) == 0.0043
+ assert float(cost) == 0.0077
def test_tts_cost_deepgram_aura_2() -> None:
@@ -58,7 +66,7 @@ def test_session_totals_sums_across_events() -> None:
),
SimpleNamespace(
event_type="usage.stt",
- payload={"audio_seconds": 30, "provider": "deepgram", "model": "nova-3"},
+ payload={"streamed_seconds": 30, "provider": "deepgram", "model": "nova-3"},
),
SimpleNamespace(
event_type="usage.tts",
@@ -70,13 +78,14 @@ def test_session_totals_sums_across_events() -> None:
assert totals["llm"]["total_tokens"] == 2500
# 2000*0.30/1e6 + 500*2.50/1e6 = 0.0006 + 0.00125 = 0.00185
assert totals["llm"]["cost_usd"] == 0.00185
- # 30/60 * 0.0043 = 0.00215
+ # 30/60 * 0.0077 = 0.00385
assert totals["stt"]["audio_seconds"] == 30.0
- assert totals["stt"]["cost_usd"] == 0.00215
+ assert totals["stt"]["streamed_seconds"] == 30.0
+ assert totals["stt"]["cost_usd"] == 0.00385
# 500/1e6 * 30.0 = 0.015
assert totals["tts"]["characters"] == 500
assert totals["tts"]["cost_usd"] == 0.015
- assert totals["total_cost_usd"] == 0.019
+ assert totals["total_cost_usd"] == 0.0207
assert totals["stt"]["source"] == "runtime"
assert totals["tts"]["source"] == "runtime"
@@ -125,6 +134,7 @@ def test_session_totals_ignores_legacy_provider_usage_reconciliation_events() ->
totals = session_totals(events)
assert totals["stt"]["audio_seconds"] == 30.0
+ assert totals["stt"]["streamed_seconds"] == 30.0
assert totals["stt"]["cost_usd"] == 0.00215
assert totals["stt"]["source"] == "runtime"
assert totals["tts"]["characters"] == 500
diff --git a/server/tests/test_stt_evaluation_pricing.py b/server/tests/test_stt_evaluation_pricing.py
index bf97d28..eeec2e4 100644
--- a/server/tests/test_stt_evaluation_pricing.py
+++ b/server/tests/test_stt_evaluation_pricing.py
@@ -18,9 +18,11 @@ def test_compute_duration_seconds_for_stereo_pcm16() -> None:
def test_compute_all_model_costs_reuses_same_duration() -> None:
costs = compute_all_model_costs(60.0)
- assert costs["deepgram"]["nova-3-monolingual"] == 0.0048
+ assert costs["deepgram"]["nova-3"] == 0.0043
+ assert costs["deepgram"]["nova-3-streaming"] == 0.0077
assert costs["elevenlabs"]["scribe-v2"] == 0.00367
- assert costs["sarvam"]["saarika"] == 0.006
+ assert costs["sarvam"]["saarika"] == 0.005263
+ assert costs["sarvam"]["saarika-diarization"] == 0.007895
def test_compute_model_cost_returns_zero_for_unknown_model() -> None:
diff --git a/server/tests/test_stt_usage_meter.py b/server/tests/test_stt_usage_meter.py
new file mode 100644
index 0000000..62b2eb5
--- /dev/null
+++ b/server/tests/test_stt_usage_meter.py
@@ -0,0 +1,64 @@
+from collections.abc import AsyncGenerator
+
+import pytest
+from pipecat.frames.frames import Frame
+
+from app.services.pipecat_streaming_runtime import SttUsageMeterMixin
+
+
+class _FakeSTTService:
+ def __init__(self, sample_rate: int) -> None:
+ self._sample_rate = sample_rate
+ self.received: list[bytes] = []
+
+ @property
+ def sample_rate(self) -> int:
+ return self._sample_rate
+
+ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]:
+ self.received.append(audio)
+ yield None
+
+
+class _MeteredService(SttUsageMeterMixin, _FakeSTTService):
+ pass
+
+
+async def _drain(gen: AsyncGenerator) -> None:
+ async for _ in gen:
+ pass
+
+
+@pytest.mark.asyncio
+async def test_streamed_seconds_counts_all_forwarded_audio() -> None:
+ service = _MeteredService(sample_rate=16000)
+ # 16-bit mono at 16 kHz: 32000 bytes per second.
+ await _drain(service.run_stt(b"\x00" * 32000))
+ await _drain(service.run_stt(b"\x00" * 16000))
+ assert service.streamed_audio_seconds == pytest.approx(1.5)
+ assert service.received == [b"\x00" * 32000, b"\x00" * 16000]
+
+
+@pytest.mark.asyncio
+async def test_streamed_seconds_zero_without_audio_or_sample_rate() -> None:
+ service = _MeteredService(sample_rate=16000)
+ assert service.streamed_audio_seconds == 0.0
+
+ unstarted = _MeteredService(sample_rate=0)
+ await _drain(unstarted.run_stt(b"\x00" * 3200))
+ assert unstarted.streamed_audio_seconds == 0.0
+
+
+@pytest.mark.asyncio
+async def test_meter_state_is_per_instance() -> None:
+ first = _MeteredService(sample_rate=16000)
+ second = _MeteredService(sample_rate=16000)
+ await _drain(first.run_stt(b"\x00" * 32000))
+ assert second.streamed_audio_seconds == 0.0
+
+
+def test_provider_reported_seconds_defaults_to_none() -> None:
+ service = _MeteredService(sample_rate=16000)
+ assert service.provider_reported_audio_seconds is None
+ service._provider_reported_audio_seconds += 2.5
+ assert service.provider_reported_audio_seconds == pytest.approx(2.5)
diff --git a/server/tests/test_tts_evaluation_pricing.py b/server/tests/test_tts_evaluation_pricing.py
new file mode 100644
index 0000000..a169240
--- /dev/null
+++ b/server/tests/test_tts_evaluation_pricing.py
@@ -0,0 +1,60 @@
+from app.services.tts_evaluation_pricing import (
+ compute_all_model_costs,
+ compute_model_cost,
+ pricing_model_name,
+)
+
+
+def test_deepgram_voice_suffixed_models_resolve_to_aura_2() -> None:
+ for voice_model in (
+ "aura-2",
+ "aura-2-thalia-en",
+ "aura-2-asteria-en",
+ "aura-2-luna-en",
+ "aura-2-orion-en",
+ ):
+ assert pricing_model_name(provider="deepgram", model=voice_model) == "aura-2"
+
+
+def test_deepgram_legacy_aura_voice_suffixes_resolve_to_aura() -> None:
+ for voice_model in ("aura", "aura-luna-en", "aura-orion-en"):
+ assert pricing_model_name(provider="deepgram", model=voice_model) == "aura"
+
+
+def test_elevenlabs_dash_variants_alias_to_underscore_model_ids() -> None:
+ assert (
+ pricing_model_name(provider="elevenlabs", model="turbo-v2-5")
+ == "eleven_turbo_v2_5"
+ )
+
+
+def test_elevenlabs_exact_model_ids_pass_through() -> None:
+ assert (
+ pricing_model_name(provider="elevenlabs", model="eleven_turbo_v2_5")
+ == "eleven_turbo_v2_5"
+ )
+
+
+def test_compute_model_cost_uses_resolved_rate() -> None:
+ # 1M chars on aura-2 should be $30
+ assert compute_model_cost(1_000_000, provider="deepgram", model="aura-2-thalia-en") == 30.0
+ # 1M chars on eleven_turbo_v2_5 should be $82.50
+ assert (
+ compute_model_cost(1_000_000, provider="elevenlabs", model="turbo-v2-5")
+ == 82.5
+ )
+
+
+def test_compute_model_cost_returns_zero_for_unknown_model() -> None:
+ assert compute_model_cost(1_000_000, provider="deepgram", model="unknown") == 0.0
+ assert compute_model_cost(1_000_000, provider="unknown", model="whatever") == 0.0
+
+
+def test_compute_all_model_costs_covers_full_model_list() -> None:
+ costs = compute_all_model_costs(1_000_000)
+ assert costs["deepgram"] == {"aura-2": 30.0, "aura": 15.0}
+ assert costs["elevenlabs"] == {"eleven_turbo_v2_5": 82.5}
+
+
+def test_compute_costs_clamp_negative_characters_to_zero() -> None:
+ assert compute_model_cost(-100, provider="deepgram", model="aura-2") == 0.0
diff --git a/server/tests/test_tts_usage_meter.py b/server/tests/test_tts_usage_meter.py
new file mode 100644
index 0000000..479d668
--- /dev/null
+++ b/server/tests/test_tts_usage_meter.py
@@ -0,0 +1,56 @@
+from collections.abc import AsyncGenerator
+
+import pytest
+from pipecat.frames.frames import Frame
+
+from app.services.pipecat_streaming_runtime import TtsUsageMeterMixin
+
+
+class _FakeTTSService:
+ def __init__(self) -> None:
+ self.received: list[str] = []
+
+ async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]:
+ self.received.append(text)
+ yield None
+
+
+class _MeteredService(TtsUsageMeterMixin, _FakeTTSService):
+ pass
+
+
+async def _drain(gen: AsyncGenerator) -> None:
+ async for _ in gen:
+ pass
+
+
+@pytest.mark.asyncio
+async def test_sent_characters_counts_all_text_sent_to_provider() -> None:
+ service = _MeteredService()
+ await _drain(service.run_tts("Hello there.", "ctx-1"))
+ await _drain(service.run_tts("How can I help?", "ctx-1"))
+ assert service.sent_characters == len("Hello there.") + len("How can I help?")
+ assert service.received == ["Hello there.", "How can I help?"]
+
+
+@pytest.mark.asyncio
+async def test_sent_characters_counted_even_when_generator_not_drained() -> None:
+ # Characters are billed once sent; an interruption that abandons the
+ # generator must not un-count text that already left for the provider.
+ service = _MeteredService()
+ gen = service.run_tts("This sentence gets interrupted mid-playback.", "ctx-1")
+ await gen.__anext__()
+ await gen.aclose()
+ assert service.sent_characters == len("This sentence gets interrupted mid-playback.")
+
+
+@pytest.mark.asyncio
+async def test_meter_state_is_per_instance() -> None:
+ first = _MeteredService()
+ second = _MeteredService()
+ await _drain(first.run_tts("only counted on first", "ctx-1"))
+ assert second.sent_characters == 0
+
+
+def test_sent_characters_defaults_to_zero() -> None:
+ assert _MeteredService().sent_characters == 0