From 379d11429bae04f6ff3cfd0e827b514aba98b0b8 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:04:17 +0530 Subject: [PATCH 1/7] fix(bugs): natural prompt-version sort, Offer.short None guard, stale docstrings - list_versions sorted lexically by filename so v10 ordered before v2; now sorts numerically by the declared version field - Offer.short() rendered '/Noned' when expiry_days was unset - streaming.py + place_streaming_voice_call.py docstrings still said OpenRouter LLM / MarkerStripperProcessor; actual wiring is OpenAI gpt-4o-mini + the end_call tool Co-Authored-By: Claude Fable 5 --- agents/prompt_loader.py | 5 ++- agents/schemas.py | 3 +- apps/voice/streaming.py | 4 +- scripts/place_streaming_voice_call.py | 2 +- tests/test_prompt_loader.py | 56 +++++++++++++++++++++++++++ tests/test_schemas.py | 16 ++++++++ 6 files changed, 81 insertions(+), 5 deletions(-) create mode 100644 tests/test_prompt_loader.py create mode 100644 tests/test_schemas.py diff --git a/agents/prompt_loader.py b/agents/prompt_loader.py index 890ae45..a55d644 100644 --- a/agents/prompt_loader.py +++ b/agents/prompt_loader.py @@ -53,9 +53,12 @@ def list_versions(agent: AgentName) -> list[PromptVersion]: agent_dir = _agent_dir(agent) if not agent_dir.exists(): return versions - for path in sorted(agent_dir.glob("v*.yaml")): + for path in agent_dir.glob("v*.yaml"): raw = yaml.safe_load(path.read_text()) versions.append(PromptVersion.model_validate(raw)) + # Sort numerically by the declared version, not by filename — lexical + # glob order puts v10 before v2. + versions.sort(key=lambda v: v.version) return versions diff --git a/agents/schemas.py b/agents/schemas.py index 7a22467..fbd52db 100644 --- a/agents/schemas.py +++ b/agents/schemas.py @@ -30,7 +30,8 @@ class Offer(BaseModel): conditions: str = "" def short(self) -> str: - return f"{self.type.value}:${self.amount:.0f}/{self.expiry_days}d" + base = f"{self.type.value}:${self.amount:.0f}" + return f"{base}/{self.expiry_days}d" if self.expiry_days is not None else base class Objection(BaseModel): diff --git a/apps/voice/streaming.py b/apps/voice/streaming.py index e0ad598..fccd957 100644 --- a/apps/voice/streaming.py +++ b/apps/voice/streaming.py @@ -13,7 +13,7 @@ │ bidirectional audio over a single WebSocket (no polling, no Pause)│ ▼ - Deepgram STT ──interim transcripts──> OpenRouter LLM + Deepgram STT ──interim transcripts──> OpenAI LLM (gpt-4o-mini) │ │ tokens stream ▼ @@ -450,7 +450,7 @@ async def streaming_ws(websocket: WebSocket, session_id: str) -> None: # Frame serializer — translates Twilio's mu-law/8kHz JSON envelopes into # Pipecat's internal frame format and back. # auto_hang_up=True: when the pipeline emits an EndFrame (triggered by - # MarkerStripperProcessor on DEAL_AGREED / NO_DEAL / HARDSHIP_REFERRED), + # the end_call tool handler pushing EndTaskFrame upstream), # the serializer calls Twilio's REST API to terminate the call. Without # this, the agent says goodbye but the line stays open until the borrower # hangs up — which during real testing meant the agent kept fielding diff --git a/scripts/place_streaming_voice_call.py b/scripts/place_streaming_voice_call.py index a8a693a..3740047 100644 --- a/scripts/place_streaming_voice_call.py +++ b/scripts/place_streaming_voice_call.py @@ -3,7 +3,7 @@ Twilio dials BORROWER_PHONE; the TwiML returned by /voice/streaming/twiml/{sid} contains a that bridges the call audio to our FastAPI WebSocket at /voice/streaming/ws/{sid}, where Pipecat handles the -bidirectional pipeline (Deepgram STT -> OpenRouter LLM -> Rime TTS). +bidirectional pipeline (Deepgram STT -> OpenAI gpt-4o-mini LLM -> Rime TTS). Pre-requisites — same as scripts/place_voice_call.py plus: * pipecat-ai installed (`pip install -e '.[voice]'`) diff --git a/tests/test_prompt_loader.py b/tests/test_prompt_loader.py new file mode 100644 index 0000000..cfe5080 --- /dev/null +++ b/tests/test_prompt_loader.py @@ -0,0 +1,56 @@ +"""list_versions must order numerically by declared version, not by filename +(lexical glob order puts v10 before v2).""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from agents import prompt_loader + + +def _write_version(agent_dir: Path, version: int, status: str = "candidate") -> None: + agent_dir.mkdir(parents=True, exist_ok=True) + (agent_dir / f"v{version}.yaml").write_text( + "\n".join( + [ + "agent: resolution", + f"version: {version}", + "created_at: 2026-01-01T00:00:00Z", + f"status: {status}", + "system_prompt: |", + f" prompt body v{version}", + ] + ) + ) + + +@pytest.fixture +def prompts_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setattr(prompt_loader, "PROMPTS_ROOT", tmp_path) + return tmp_path + + +def test_versions_sorted_numerically_not_lexically(prompts_root: Path) -> None: + agent_dir = prompts_root / "resolution" + for v in [10, 2, 1, 11, 3]: + _write_version(agent_dir, v) + + versions = prompt_loader.list_versions("resolution") + + assert [v.version for v in versions] == [1, 2, 3, 10, 11] + + +def test_champion_found_regardless_of_filename_order(prompts_root: Path) -> None: + agent_dir = prompts_root / "resolution" + _write_version(agent_dir, 10, status="champion") + for v in [1, 2, 3]: + _write_version(agent_dir, v, status="retired") + + champion = prompt_loader.load_champion("resolution") + + assert champion.version == 10 + + +def test_missing_agent_dir_returns_empty(prompts_root: Path) -> None: + assert prompt_loader.list_versions("final_notice") == [] diff --git a/tests/test_schemas.py b/tests/test_schemas.py new file mode 100644 index 0000000..de3075e --- /dev/null +++ b/tests/test_schemas.py @@ -0,0 +1,16 @@ +"""Offer.short() must render cleanly with and without an expiry.""" +from __future__ import annotations + +from agents.schemas import Offer, OfferType + + +def test_offer_short_with_expiry() -> None: + offer = Offer(type=OfferType.LUMP_SUM, amount=8750, expiry_days=3) + assert offer.short() == "lump_sum:$8750/3d" + + +def test_offer_short_without_expiry_has_no_none() -> None: + offer = Offer(type=OfferType.PAYMENT_PLAN, amount=12500) + rendered = offer.short() + assert rendered == "payment_plan:$12500" + assert "None" not in rendered From 9b4e4287b20cbce8a6e1c3ec11817241b4760668 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:05:25 +0530 Subject: [PATCH 2/7] fix(secrets): Secrets-based config in pipecat_provider, GROQ_API_KEY in .env.example - pipecat_provider read six secrets via bare os.environ[...] which crashed with KeyError; now uses the existing Secrets settings object with a _require() helper raising actionable RuntimeErrors - secrets are validated before the pipecat imports so misconfiguration fails deterministically with the most useful message first - .env.example gains GROQ_API_KEY (required by settings.yaml llm.agent_voice but previously undocumented), TWILIO_WEBHOOK_AUTH, and WEBHOOK_BASE_URL Co-Authored-By: Claude Fable 5 --- .env.example | 9 +++++ apps/voice/pipecat_provider.py | 46 +++++++++++++++++++++----- tests/test_pipecat_provider_secrets.py | 44 ++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 9 deletions(-) create mode 100644 tests/test_pipecat_provider_secrets.py diff --git a/.env.example b/.env.example index 3c48c0a..504619b 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,15 @@ RIME_API_KEY= TWILIO_ACCOUNT_SID= TWILIO_AUTH_TOKEN= TWILIO_PHONE_NUMBER=+1... +# Groq — required by settings.yaml llm.agent_voice (low-latency voice agent role) +GROQ_API_KEY=gsk_... +# Twilio webhook auth for /voice/* routes: auto (default; enforced when +# TWILIO_AUTH_TOKEN is set) | on | off +TWILIO_WEBHOOK_AUTH=auto +# Public base URL Twilio calls back to (cloudflared/ngrok tunnel or stable +# host). Required for signature validation behind a tunnel — Twilio signs +# the public URL, not the internal uvicorn one. +WEBHOOK_BASE_URL=https://.trycloudflare.com # Infra TEMPORAL_HOST=temporal:7233 diff --git a/apps/voice/pipecat_provider.py b/apps/voice/pipecat_provider.py index 10e0b45..1530870 100644 --- a/apps/voice/pipecat_provider.py +++ b/apps/voice/pipecat_provider.py @@ -15,20 +15,49 @@ from agents.handoff_renderer import render_handoff_for_prompt from agents.prompt_loader import load_champion from agents.schemas import HandoffContext, Message, VoiceOutcome +from agents.settings import get_secrets from apps.voice.handoff_capture import StructuredEvents, extract_events logger = logging.getLogger(__name__) +def _require(value: str, name: str, hint: str) -> str: + if not value: + raise RuntimeError(f"{name} is not set. {hint} Add it to .env — see .env.example.") + return value + + class PipecatProvider: """Outbound voice call provider built on Pipecat. - Requires environment variables: - DEEPGRAM_API_KEY, RIME_API_KEY, TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, - TWILIO_PHONE_NUMBER, ANTHROPIC_API_KEY. + Reads configuration from ``agents.settings.Secrets`` (populated from + .env): DEEPGRAM_API_KEY, RIME_API_KEY, TWILIO_ACCOUNT_SID, + TWILIO_AUTH_TOKEN, TWILIO_PHONE_NUMBER, ANTHROPIC_API_KEY. """ async def resolve(self, handoff: HandoffContext) -> VoiceOutcome: + # Validate configuration before the heavyweight pipecat imports so a + # missing key fails with an actionable message instead of a KeyError + # (or an unrelated import error) mid-call-setup. + secrets = get_secrets() + from_number = _require( + secrets.TWILIO_PHONE_NUMBER, "TWILIO_PHONE_NUMBER", + "Twilio 'from' number for outbound dials.", + ) + deepgram_key = _require( + secrets.DEEPGRAM_API_KEY, "DEEPGRAM_API_KEY", "Deepgram STT key.", + ) + anthropic_key = _require( + secrets.ANTHROPIC_API_KEY, "ANTHROPIC_API_KEY", "Anthropic LLM key.", + ) + rime_key = _require(secrets.RIME_API_KEY, "RIME_API_KEY", "Rime TTS key.") + twilio_sid = _require( + secrets.TWILIO_ACCOUNT_SID, "TWILIO_ACCOUNT_SID", "Twilio account SID.", + ) + twilio_token = _require( + secrets.TWILIO_AUTH_TOKEN, "TWILIO_AUTH_TOKEN", "Twilio auth token.", + ) + try: from pipecat.frames.frames import EndFrame, TranscriptionFrame from pipecat.pipeline.pipeline import Pipeline @@ -47,7 +76,6 @@ async def resolve(self, handoff: HandoffContext) -> VoiceOutcome: # 1. Resolve target phone number for this borrower (placeholder lookup). to_number = _lookup_phone_number(handoff.borrower_id) - from_number = os.environ["TWILIO_PHONE_NUMBER"] # 2. Build the system prompt with handoff context inlined. prompt = load_champion("resolution") @@ -57,16 +85,16 @@ async def resolve(self, handoff: HandoffContext) -> VoiceOutcome: ) # 3. Wire the pipeline. - stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"]) + stt = DeepgramSTTService(api_key=deepgram_key) llm = AnthropicLLMService( - api_key=os.environ["ANTHROPIC_API_KEY"], + api_key=anthropic_key, model="claude-sonnet-4-6", system_prompt=system_text, ) - tts = RimeTTSService(api_key=os.environ["RIME_API_KEY"], voice_id="luna") + tts = RimeTTSService(api_key=rime_key, voice_id="luna") transport = TwilioTransport( - account_sid=os.environ["TWILIO_ACCOUNT_SID"], - auth_token=os.environ["TWILIO_AUTH_TOKEN"], + account_sid=twilio_sid, + auth_token=twilio_token, from_number=from_number, to_number=to_number, recording_enabled=True, diff --git a/tests/test_pipecat_provider_secrets.py b/tests/test_pipecat_provider_secrets.py new file mode 100644 index 0000000..b822db1 --- /dev/null +++ b/tests/test_pipecat_provider_secrets.py @@ -0,0 +1,44 @@ +"""PipecatProvider must fail fast with an actionable error when secrets are +missing — not a bare KeyError mid-call-setup.""" +from __future__ import annotations + +import pytest + +from agents.schemas import HandoffContext +from agents.settings import get_secrets +from apps.voice.pipecat_provider import PipecatProvider + + +@pytest.fixture(autouse=True) +def _clear_secrets_cache(monkeypatch: pytest.MonkeyPatch): + # get_secrets() is lru_cached; ensure each test sees its own env. Also + # neutralize any real .env values for the vars under test. + for var in [ + "TWILIO_PHONE_NUMBER", "DEEPGRAM_API_KEY", "ANTHROPIC_API_KEY", + "RIME_API_KEY", "TWILIO_ACCOUNT_SID", "TWILIO_AUTH_TOKEN", + ]: + monkeypatch.setenv(var, "") + get_secrets.cache_clear() + yield + get_secrets.cache_clear() + + +def _handoff() -> HandoffContext: + return HandoffContext(borrower_id="b-test", stage="resolution") + + +async def test_missing_phone_number_raises_actionable_error() -> None: + with pytest.raises(RuntimeError, match="TWILIO_PHONE_NUMBER"): + await PipecatProvider().resolve(_handoff()) + + +async def test_missing_deepgram_key_named_in_error(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TWILIO_PHONE_NUMBER", "+15550001111") + get_secrets.cache_clear() + with pytest.raises(RuntimeError, match="DEEPGRAM_API_KEY"): + await PipecatProvider().resolve(_handoff()) + + +async def test_error_mentions_env_example() -> None: + with pytest.raises(RuntimeError, match=r"\.env\.example"): + await PipecatProvider().resolve(_handoff()) From 57cfae82c36c79dd31dc85b54144b207492b97a6 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:06:32 +0530 Subject: [PATCH 3/7] test: silence basedpyright unused-fixture warnings Co-Authored-By: Claude Fable 5 --- tests/test_pipecat_provider_secrets.py | 2 +- tests/test_prompt_loader.py | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_pipecat_provider_secrets.py b/tests/test_pipecat_provider_secrets.py index b822db1..7262347 100644 --- a/tests/test_pipecat_provider_secrets.py +++ b/tests/test_pipecat_provider_secrets.py @@ -10,7 +10,7 @@ @pytest.fixture(autouse=True) -def _clear_secrets_cache(monkeypatch: pytest.MonkeyPatch): +def clear_secrets_cache(monkeypatch: pytest.MonkeyPatch): # get_secrets() is lru_cached; ensure each test sees its own env. Also # neutralize any real .env values for the vars under test. for var in [ diff --git a/tests/test_prompt_loader.py b/tests/test_prompt_loader.py index cfe5080..55bb0f4 100644 --- a/tests/test_prompt_loader.py +++ b/tests/test_prompt_loader.py @@ -52,5 +52,6 @@ def test_champion_found_regardless_of_filename_order(prompts_root: Path) -> None assert champion.version == 10 -def test_missing_agent_dir_returns_empty(prompts_root: Path) -> None: +@pytest.mark.usefixtures("prompts_root") +def test_missing_agent_dir_returns_empty() -> None: assert prompt_loader.list_versions("final_notice") == [] From 98518efffa20de0e0126372ee0a1baedb4787d95 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:09:49 +0530 Subject: [PATCH 4/7] feat(security): Twilio webhook signature validation + Media Streams WS token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously every /voice/* webhook and the streaming WebSocket were unauthenticated — anyone who found the tunnel URL could drive calls or attach a pipeline. - apps/api/twilio_security.py: X-Twilio-Signature validation as a FastAPI dependency (validates against the public WEBHOOK_BASE_URL Twilio actually signed) + stateless HMAC stream token for the WS - enforcement on Twilio-called routes only (twiml/gather/wait/status/ streaming-twiml); operator endpoints (dial-init/state) remain open — operator auth tracked separately - WS endpoint rejects bad tokens pre-accept (403 handshake) before any pipeline resources are built; carries ?token=... - TWILIO_WEBHOOK_AUTH=auto|on|off (auto: enforced iff TWILIO_AUTH_TOKEN is set) keeps local dev and CI hermetic - simulators keep working under enforcement: the streaming simulator uses the ws_token from dial-init; the TwiML simulator signs its requests with the documented HMAC-SHA1 algorithm Co-Authored-By: Claude Fable 5 --- apps/api/twilio_security.py | 100 +++++++++++++++++++ apps/api/voice_twiml.py | 32 ++++-- apps/voice/streaming.py | 11 +++ scripts/simulate_twilio_stream.py | 10 +- scripts/simulate_twiml_call.py | 34 ++++++- tests/test_twilio_security.py | 159 ++++++++++++++++++++++++++++++ 6 files changed, 335 insertions(+), 11 deletions(-) create mode 100644 apps/api/twilio_security.py create mode 100644 tests/test_twilio_security.py diff --git a/apps/api/twilio_security.py b/apps/api/twilio_security.py new file mode 100644 index 0000000..ebd6202 --- /dev/null +++ b/apps/api/twilio_security.py @@ -0,0 +1,100 @@ +"""Twilio request authentication for the /voice/* surface. + +Two mechanisms: + +1. HTTP webhooks (X-Twilio-Signature). Twilio signs every webhook with + HMAC-SHA1 over the exact public URL it requested plus the sorted POST + form params, keyed by the account's auth token. Behind a tunnel the + internal ``request.url`` differs from what Twilio signed, so the check + reconstructs the public URL from WEBHOOK_BASE_URL (same convention as + ``voice_twiml._base_url``). + +2. Media Streams WebSocket. Twilio's WS handshake carries no signature, + so the embeds a short-lived-enough token: a stateless + HMAC-SHA256 over the session id, keyed by TWILIO_AUTH_TOKEN. Both the + TwiML endpoint and the WS endpoint derive it independently — no shared + state, survives restarts and multiple workers. + +Enforcement is controlled by TWILIO_WEBHOOK_AUTH: + "auto" (default) — enforced iff TWILIO_AUTH_TOKEN is configured + "on" / "off" — explicit override (simulators/tests use "off", or + sign their requests like the TwiML simulator does) +""" +from __future__ import annotations + +import hashlib +import hmac +import os + +from fastapi import HTTPException, Request + +from agents.settings import get_secrets + + +def auth_enabled() -> bool: + flag = os.environ.get("TWILIO_WEBHOOK_AUTH", "auto").lower() + if flag in ("off", "false", "0"): + return False + if flag in ("on", "true", "1"): + return True + return bool(get_secrets().TWILIO_AUTH_TOKEN) + + +def _public_url(request: Request) -> str: + """The URL Twilio actually signed — public base + path + query.""" + base = os.environ.get("WEBHOOK_BASE_URL", "").rstrip("/") + if base: + url = f"{base}{request.url.path}" + else: + url = f"{request.url.scheme}://{request.url.netloc}{request.url.path}" + if request.url.query: + url += f"?{request.url.query}" + return url + + +async def verify_twilio_webhook(request: Request) -> None: + """FastAPI dependency for Twilio-called HTTP webhook routes. + + Raises 403 unless the request carries a valid X-Twilio-Signature. + Reading the form here is safe — Starlette caches it, so endpoint + Form(...) parameters still parse. + """ + if not auth_enabled(): + return + token = get_secrets().TWILIO_AUTH_TOKEN + if not token: + raise HTTPException( + status_code=403, + detail="Twilio webhook auth is on but TWILIO_AUTH_TOKEN is not configured", + ) + try: + from twilio.request_validator import RequestValidator + except ImportError as exc: # twilio ships with the voice extra only + raise RuntimeError( + "Twilio webhook auth is enabled but the twilio SDK is not " + "installed. Install with `pip install -e '.[voice]'` or set " + "TWILIO_WEBHOOK_AUTH=off." + ) from exc + + params: dict[str, str] = {} + if request.method == "POST": + content_type = request.headers.get("content-type", "") + if "application/x-www-form-urlencoded" in content_type or "multipart/form-data" in content_type: + params = {k: str(v) for k, v in (await request.form()).items()} + + validator = RequestValidator(token) + signature = request.headers.get("X-Twilio-Signature", "") + if not validator.validate(_public_url(request), params, signature): + raise HTTPException(status_code=403, detail="invalid Twilio signature") + + +def stream_token(session_id: str) -> str: + """Stateless auth token for the Media Streams WS endpoint.""" + secret = get_secrets().TWILIO_AUTH_TOKEN + return hmac.new( + secret.encode(), f"stream:{session_id}".encode(), hashlib.sha256, + ).hexdigest()[:32] + + +def verify_stream_token(session_id: str, token: str) -> bool: + return hmac.compare_digest(stream_token(session_id), token or "") diff --git a/apps/api/voice_twiml.py b/apps/api/voice_twiml.py index 0006ca4..789ea78 100644 --- a/apps/api/voice_twiml.py +++ b/apps/api/voice_twiml.py @@ -44,7 +44,7 @@ import os import re -from fastapi import APIRouter, Form, HTTPException, Request +from fastapi import APIRouter, Depends, Form, HTTPException, Request from fastapi.responses import PlainTextResponse from agents.schemas import EmotionalState, HandoffContext @@ -53,6 +53,12 @@ TurnResult, _SESSIONS, ) +from apps.api.twilio_security import auth_enabled, stream_token, verify_twilio_webhook + +# Applied to every route Twilio itself calls back. Operator endpoints +# (/dial-init, /streaming/dial-init, /state) are called by our own scripts +# and stay open — operator auth is tracked separately. +_TWILIO_AUTH = [Depends(verify_twilio_webhook)] logger = logging.getLogger(__name__) @@ -207,7 +213,7 @@ async def dial_init(borrower_id: str = "voice-demo", account_last_four: str = "1 } -@router.api_route("/twiml/{session_id}", methods=["GET", "POST"], response_class=PlainTextResponse) +@router.api_route("/twiml/{session_id}", methods=["GET", "POST"], response_class=PlainTextResponse, dependencies=_TWILIO_AUTH) async def twiml_initial(session_id: str, request: Request) -> PlainTextResponse: session = _SESSIONS.get(session_id) if session is None: @@ -227,7 +233,7 @@ async def twiml_initial(session_id: str, request: Request) -> PlainTextResponse: return PlainTextResponse(body, media_type="application/xml") -@router.post("/gather/{session_id}", response_class=PlainTextResponse) +@router.post("/gather/{session_id}", response_class=PlainTextResponse, dependencies=_TWILIO_AUTH) async def twiml_gather( session_id: str, request: Request, @@ -280,7 +286,7 @@ async def twiml_gather( ) -@router.api_route("/wait/{session_id}/{turn_n}/{cycle}", methods=["GET", "POST"], response_class=PlainTextResponse) +@router.api_route("/wait/{session_id}/{turn_n}/{cycle}", methods=["GET", "POST"], response_class=PlainTextResponse, dependencies=_TWILIO_AUTH) async def twiml_wait(session_id: str, turn_n: int, cycle: int, request: Request) -> PlainTextResponse: """Twilio polls this endpoint waiting for the LLM to finish. Each call finishes in <100ms — either returns the real reply if the task is done, @@ -348,7 +354,7 @@ async def twiml_wait(session_id: str, turn_n: int, cycle: int, request: Request) ) -@router.post("/status/{session_id}") +@router.post("/status/{session_id}", dependencies=_TWILIO_AUTH) async def call_status( session_id: str, CallStatus: str = Form(default=""), @@ -406,10 +412,15 @@ async def streaming_dial_init(borrower_id: str = "voice-stream", account_last_fo """ import uuid sid = uuid.uuid4().hex[:10] - return {"session_id": sid, "stage": "resolution", "mode": "streaming"} + response = {"session_id": sid, "stage": "resolution", "mode": "streaming"} + if auth_enabled(): + # Hand the dialer/simulator the WS token so it can connect without + # knowing the HMAC scheme (Twilio gets it via the ). + response["ws_token"] = stream_token(sid) + return response -@router.api_route("/streaming/twiml/{session_id}", methods=["GET", "POST"], response_class=PlainTextResponse) +@router.api_route("/streaming/twiml/{session_id}", methods=["GET", "POST"], response_class=PlainTextResponse, dependencies=_TWILIO_AUTH) async def streaming_twiml(session_id: str, request: Request) -> PlainTextResponse: """TwiML that bridges the call to our Pipecat WebSocket endpoint. @@ -422,10 +433,15 @@ async def streaming_twiml(session_id: str, request: Request) -> PlainTextRespons # Convert https:// -> wss:// (Twilio media-streams require wss in prod; # ws:// works for plain HTTP local dev but Twilio rejects it for outbound). ws_base = base.replace("https://", "wss://").replace("http://", "ws://") + stream_url = f"{ws_base}/voice/streaming/ws/{session_id}" + if auth_enabled(): + # Twilio preserves the query string on in the WS + # handshake, so the endpoint can reject bad tokens pre-accept. + stream_url += f"?token={stream_token(session_id)}" body = f""" - + """ return PlainTextResponse(body, media_type="application/xml") diff --git a/apps/voice/streaming.py b/apps/voice/streaming.py index fccd957..2525003 100644 --- a/apps/voice/streaming.py +++ b/apps/voice/streaming.py @@ -415,6 +415,17 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: @router.websocket("/ws/{session_id}") async def streaming_ws(websocket: WebSocket, session_id: str) -> None: """Twilio Media Streams WebSocket endpoint.""" + from apps.api.twilio_security import auth_enabled, verify_stream_token + + if auth_enabled() and not verify_stream_token( + session_id, websocket.query_params.get("token", ""), + ): + # Pre-accept close rejects the handshake (HTTP 403) before any + # pipeline resources are constructed. + logger.warning("Streaming WS rejected for session_id=%s: bad token", session_id) + await websocket.close(code=1008) + return + await websocket.accept() logger.info("Streaming WS accepted for session_id=%s", session_id) diff --git a/scripts/simulate_twilio_stream.py b/scripts/simulate_twilio_stream.py index daf97ae..0e4655c 100644 --- a/scripts/simulate_twilio_stream.py +++ b/scripts/simulate_twilio_stream.py @@ -109,12 +109,18 @@ async def run_simulation(args: argparse.Namespace) -> None: }, ) r.raise_for_status() - session_id = r.json()["session_id"] + dial_init = r.json() + session_id = dial_init["session_id"] + ws_token = dial_init.get("ws_token", "") print(f" session_id: {session_id}") print() - # 2) Connect to the WebSocket as if we were Twilio + # 2) Connect to the WebSocket as if we were Twilio. When webhook auth is + # enforced server-side, dial-init hands us the same token the TwiML + # endpoint would embed in . ws_url = f"{ws_base}/voice/streaming/ws/{session_id}" + if ws_token: + ws_url += f"?token={ws_token}" print(f"--- 2. Connecting to {ws_url} ---") fake_stream_sid = "MZ" + uuid.uuid4().hex fake_call_sid = "CA" + uuid.uuid4().hex diff --git a/scripts/simulate_twiml_call.py b/scripts/simulate_twiml_call.py index 456a25e..014ad89 100644 --- a/scripts/simulate_twiml_call.py +++ b/scripts/simulate_twiml_call.py @@ -32,15 +32,45 @@ import argparse import asyncio +import base64 +import hashlib +import hmac +import os import re import sys import time +import urllib.parse import xml.etree.ElementTree as ET from dataclasses import dataclass import httpx +def _twilio_signing_hook(token: str): + """httpx request hook that attaches X-Twilio-Signature the way Twilio + does (HMAC-SHA1 over URL + sorted form params), so the simulator keeps + working when the server enforces webhook auth (TWILIO_WEBHOOK_AUTH). + + The signature covers the exact URL the simulator requests, which is what + the server validates when WEBHOOK_BASE_URL is unset. If your local server + sets WEBHOOK_BASE_URL, unset it (or TWILIO_WEBHOOK_AUTH=off) for sim runs. + """ + + async def sign(request: httpx.Request) -> None: + payload = str(request.url) + body = request.content.decode() if request.content else "" + if body and request.headers.get("content-type", "").startswith( + "application/x-www-form-urlencoded" + ): + params = urllib.parse.parse_qsl(body, keep_blank_values=True) + for key, value in sorted(params): + payload += key + value + digest = hmac.new(token.encode(), payload.encode(), hashlib.sha1).digest() + request.headers["X-Twilio-Signature"] = base64.b64encode(digest).decode() + + return sign + + # Default 5-turn borrower script — same as the canonical take-4 phone call. # Each line is a borrower utterance; the simulator submits them as # `SpeechResult=` POST-form params on /voice/gather. @@ -184,7 +214,9 @@ async def run_simulation(args: argparse.Namespace) -> int: turns: list[TurnRecord] = [] failures: list[str] = [] - async with httpx.AsyncClient(timeout=120.0) as client: + auth_token = os.environ.get("TWILIO_AUTH_TOKEN", "") + event_hooks = {"request": [_twilio_signing_hook(auth_token)]} if auth_token else {} + async with httpx.AsyncClient(timeout=120.0, event_hooks=event_hooks) as client: # 1) Allocate session r = await client.post( f"{api_base}/voice/dial-init", diff --git a/tests/test_twilio_security.py b/tests/test_twilio_security.py new file mode 100644 index 0000000..83b0b4f --- /dev/null +++ b/tests/test_twilio_security.py @@ -0,0 +1,159 @@ +"""Twilio webhook signature validation + Media Streams WS token. + +All hermetic — no live keys, no network. Signatures are computed with the +same stdlib HMAC-SHA1+base64 algorithm Twilio documents, so the twilio SDK +is exercised only on the validation side. +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac + +import pytest +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect + +from agents.settings import get_secrets +from apps.api import twilio_security +from apps.api.main import app + +TEST_TOKEN = "test-auth-token-12345" + + +@pytest.fixture(autouse=True) +def secrets_env(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("TWILIO_AUTH_TOKEN", TEST_TOKEN) + monkeypatch.setenv("TWILIO_WEBHOOK_AUTH", "on") + monkeypatch.delenv("WEBHOOK_BASE_URL", raising=False) + get_secrets.cache_clear() + yield + get_secrets.cache_clear() + + +@pytest.fixture +def client() -> TestClient: + return TestClient(app) + + +def _sign(url: str, params: dict[str, str] | None = None) -> str: + payload = url + for key, value in sorted((params or {}).items()): + payload += key + value + digest = hmac.new(TEST_TOKEN.encode(), payload.encode(), hashlib.sha1).digest() + return base64.b64encode(digest).decode() + + +# --- auth_enabled matrix ---------------------------------------------------- + + +@pytest.mark.parametrize( + ("flag", "token", "expected"), + [ + ("on", "", True), + ("off", TEST_TOKEN, False), + ("auto", TEST_TOKEN, True), + ("auto", "", False), + ("false", TEST_TOKEN, False), + ], +) +def test_auth_enabled_matrix( + monkeypatch: pytest.MonkeyPatch, flag: str, token: str, expected: bool, +) -> None: + monkeypatch.setenv("TWILIO_WEBHOOK_AUTH", flag) + monkeypatch.setenv("TWILIO_AUTH_TOKEN", token) + get_secrets.cache_clear() + assert twilio_security.auth_enabled() is expected + + +# --- HTTP webhook signature ------------------------------------------------- + + +def test_unsigned_webhook_rejected(client: TestClient) -> None: + r = client.post("/voice/gather/nosuchsession", data={"SpeechResult": "hi"}) + assert r.status_code == 403 + + +def test_garbage_signature_rejected(client: TestClient) -> None: + r = client.post( + "/voice/gather/nosuchsession", + data={"SpeechResult": "hi"}, + headers={"X-Twilio-Signature": "bogus"}, + ) + assert r.status_code == 403 + + +def test_valid_signature_passes_dependency(client: TestClient) -> None: + url = "http://testserver/voice/gather/nosuchsession" + params = {"SpeechResult": "hi", "Confidence": "0.9"} + r = client.post( + "/voice/gather/nosuchsession", + data=params, + headers={"X-Twilio-Signature": _sign(url, params)}, + ) + # Signature accepted; the endpoint itself then 404s on the unknown session. + assert r.status_code == 404 + + +def test_get_route_signature_covers_url_only(client: TestClient) -> None: + url = "http://testserver/voice/twiml/nosuchsession" + r = client.get(url, headers={"X-Twilio-Signature": _sign(url)}) + assert r.status_code == 404 # past the 403 gate + + +def test_auth_off_lets_unsigned_through( + monkeypatch: pytest.MonkeyPatch, client: TestClient, +) -> None: + monkeypatch.setenv("TWILIO_WEBHOOK_AUTH", "off") + r = client.post("/voice/gather/nosuchsession", data={"SpeechResult": "hi"}) + assert r.status_code == 404 + + +def test_operator_dial_init_stays_open(client: TestClient) -> None: + # /voice/streaming/dial-init is operator-facing, not Twilio-called: no + # signature required, and it returns the WS token when auth is on. + r = client.post("/voice/streaming/dial-init") + assert r.status_code == 200 + body = r.json() + assert body["ws_token"] == twilio_security.stream_token(body["session_id"]) + + +def test_streaming_twiml_embeds_ws_token(client: TestClient) -> None: + sid = "sess42" + url = f"http://testserver/voice/streaming/twiml/{sid}" + r = client.post(url, headers={"X-Twilio-Signature": _sign(url, {})}) + assert r.status_code == 200 + assert f"/voice/streaming/ws/{sid}?token={twilio_security.stream_token(sid)}" in r.text + + +# --- Media Streams WS token ------------------------------------------------- + + +def test_stream_token_deterministic_and_verifiable() -> None: + token = twilio_security.stream_token("abc") + assert token == twilio_security.stream_token("abc") + assert twilio_security.verify_stream_token("abc", token) + assert not twilio_security.verify_stream_token("abc", "wrong") + assert not twilio_security.verify_stream_token("abc", "") + assert token != twilio_security.stream_token("abd") + + +def test_ws_bad_token_rejected_pre_accept(client: TestClient) -> None: + with pytest.raises(WebSocketDisconnect): + with client.websocket_connect("/voice/streaming/ws/sess1?token=bad"): + pass + + +def test_ws_missing_token_rejected(client: TestClient) -> None: + with pytest.raises(WebSocketDisconnect): + with client.websocket_connect("/voice/streaming/ws/sess1"): + pass + + +def test_ws_good_token_accepted(client: TestClient) -> None: + token = twilio_security.stream_token("sess1") + with client.websocket_connect(f"/voice/streaming/ws/sess1?token={token}") as ws: + # Handshake accepted. Feed junk so the Twilio-envelope parse fails + # and the endpoint closes cleanly without building the pipeline. + ws.send_text("not-json") + ws.send_text("also-not-json") From bc04aad08388811d945021d47fbc31e702b994e5 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:12:26 +0530 Subject: [PATCH 5/7] feat(metrics): voice-call cost/latency observer, CostMeter.record_usage, COST_LOG fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The streaming pipeline generated Pipecat MetricsFrames (enable_metrics/enable_usage_metrics) and dropped them; streaming calls were also invisible to the cost ledger. - apps/voice/call_metrics.py: VoiceCallMetricsObserver collects per-stage TTFB/processing stats + LLM token and TTS character usage (deduped on frame.id — observers see a frame once per pipeline edge) and writes a per-call summary to data/eval_runs/voice_call_metrics.jsonl - finalize() mirrors voice_llm/voice_stt/voice_tts costs into CostMeter so scripts/cost_report.py covers voice calls; STT priced by wall-clock audio-minutes (documented approximation), TTS by characters - CostMeter.record_usage(): generic entry point for non-LLMResponse usage; PRICES_USD gains gpt-4o-mini; new VOICE_PRICES_USD table (Deepgram nova-3 $0.0077/min, Rime mistv2 $0.03/1k chars) - fix: COST_LOG was resolved at import into a ClassVar, so configuring it later silently wrote to the default path; now resolved per call Co-Authored-By: Claude Fable 5 --- agents/cost_meter.py | 77 +++++++++++-- apps/voice/call_metrics.py | 167 ++++++++++++++++++++++++++++ apps/voice/streaming.py | 12 ++ tests/test_call_metrics_observer.py | 122 ++++++++++++++++++++ tests/test_cost_meter.py | 50 +++++++++ 5 files changed, 420 insertions(+), 8 deletions(-) create mode 100644 apps/voice/call_metrics.py create mode 100644 tests/test_call_metrics_observer.py create mode 100644 tests/test_cost_meter.py diff --git a/agents/cost_meter.py b/agents/cost_meter.py index 7caa0cc..2194f5d 100644 --- a/agents/cost_meter.py +++ b/agents/cost_meter.py @@ -36,6 +36,17 @@ "meta-llama/llama-3.3-70b-instruct": (0.59, 0.0, 0.79), "google/gemini-2.0-flash": (0.10, 0.0, 0.40), "deepseek/deepseek-r1": (0.55, 0.0, 2.19), + + # OpenAI direct — the streaming voice path's LLM (apps/voice/streaming.py) + "gpt-4o-mini": (0.15, 0.075, 0.60), +} + +# Non-token voice pricing (July 2026 published pay-as-you-go rates). +VOICE_PRICES_USD: dict[str, float] = { + # Deepgram streaming STT — USD per audio-minute + "deepgram/nova-3-general": 0.0077, + # Rime Mist v2 TTS — USD per character ($0.03 / 1k chars) + "rime/mistv2": 0.03 / 1000, } @@ -89,29 +100,79 @@ class CostMeter: """JSONL-append cost log. Thread-safe.""" _lock: ClassVar[threading.Lock] = threading.Lock() - _path: ClassVar[Path] = Path(os.environ.get("COST_LOG", "data/eval_runs/cost_log.jsonl")) + + @classmethod + def _log_path(cls) -> Path: + # Resolved at call time so setting COST_LOG after import works + # (the previous ClassVar froze the path at import). + return Path(os.environ.get("COST_LOG", "data/eval_runs/cost_log.jsonl")) + + @classmethod + def _append(cls, rec: CostRecord) -> None: + path = cls._log_path() + path.parent.mkdir(parents=True, exist_ok=True) + with cls._lock, path.open("a") as f: + f.write(json.dumps(asdict(rec)) + "\n") @classmethod def record(cls, role: str, resp: LLMResponse, label: str = "") -> CostRecord: rec = _price_response(role, resp) rec.label = label - cls._path.parent.mkdir(parents=True, exist_ok=True) - with cls._lock, cls._path.open("a") as f: - f.write(json.dumps(asdict(rec)) + "\n") + cls._append(rec) + return rec + + @classmethod + def record_usage( + cls, + *, + role: str, + model: str, + input_tokens: int = 0, + cached_input_tokens: int = 0, + output_tokens: int = 0, + cost_usd: float | None = None, + label: str = "", + ) -> CostRecord: + """Generic entry point for usage that doesn't arrive as an LLMResponse + (the streaming voice pipeline's LLM/STT/TTS). When ``cost_usd`` is + given it is recorded as-is (STT minutes / TTS characters priced by + the caller); otherwise it is priced from PRICES_USD token rates. + """ + if cost_usd is None: + in_p, cached_p, out_p = _lookup_prices(model) + uncached = max(0, input_tokens - cached_input_tokens) + cost_usd = ( + uncached * in_p / 1_000_000 + + cached_input_tokens * cached_p / 1_000_000 + + output_tokens * out_p / 1_000_000 + ) + rec = CostRecord( + ts=time.time(), + role=role, + model=model, + input_tokens=input_tokens, + cached_input_tokens=cached_input_tokens, + output_tokens=output_tokens, + cost_usd=cost_usd, + label=label, + ) + cls._append(rec) return rec @classmethod def total_usd(cls) -> float: - if not cls._path.exists(): + path = cls._log_path() + if not path.exists(): return 0.0 - return sum(json.loads(line)["cost_usd"] for line in cls._path.read_text().splitlines()) + return sum(json.loads(line)["cost_usd"] for line in path.read_text().splitlines()) @classmethod def breakdown(cls) -> dict[str, float]: - if not cls._path.exists(): + path = cls._log_path() + if not path.exists(): return {} totals: dict[str, float] = {} - for line in cls._path.read_text().splitlines(): + for line in path.read_text().splitlines(): r = json.loads(line) key = f"{r['role']}/{r['model']}" totals[key] = totals.get(key, 0.0) + r["cost_usd"] diff --git a/apps/voice/call_metrics.py b/apps/voice/call_metrics.py new file mode 100644 index 0000000..d0d2b6f --- /dev/null +++ b/apps/voice/call_metrics.py @@ -0,0 +1,167 @@ +"""Per-call latency + cost metering for the streaming voice pipeline. + +The pipeline already generates Pipecat MetricsFrames +(``PipelineParams(enable_metrics=True, enable_usage_metrics=True)``) — +before this module they were generated and dropped. The observer collects: + +- TTFB per processor (STT / LLM / TTS time-to-first-byte per turn) +- processing time per processor +- LLM token usage and TTS character usage + +``finalize()`` (called from ``streaming_ws``'s ``finally`` block) writes a +one-line-per-call JSONL summary to VOICE_CALL_METRICS_LOG (default +``data/eval_runs/voice_call_metrics.jsonl``) and mirrors the three cost +components into the shared CostMeter log so ``scripts/cost_report.py`` +includes voice calls. + +STT cost approximation: Pipecat emits no Deepgram usage metric, and Twilio +streams call audio continuously, so billed STT audio-minutes ≈ wall-clock +call duration. Documented trade-off, revisit if Deepgram usage data lands. +""" +from __future__ import annotations + +import json +import logging +import os +import time +from pathlib import Path + +from pipecat.frames.frames import MetricsFrame +from pipecat.metrics.metrics import ( + LLMUsageMetricsData, + ProcessingMetricsData, + TTFBMetricsData, + TTSUsageMetricsData, +) +from pipecat.observers.base_observer import BaseObserver, FramePushed + +from agents.cost_meter import VOICE_PRICES_USD, CostMeter + +logger = logging.getLogger(__name__) + +# Models the streaming pipeline is wired with (apps/voice/streaming.py). +# Used to price STT/TTS usage; the LLM model arrives on the metrics data. +STT_PRICE_KEY = "deepgram/nova-3-general" +TTS_PRICE_KEY = "rime/mistv2" + + +def _stats(values: list[float]) -> dict: + ordered = sorted(values) + p95_index = max(0, round(0.95 * (len(ordered) - 1))) + return { + "avg": round(sum(ordered) / len(ordered), 4), + "p95": round(ordered[p95_index], 4), + "max": round(ordered[-1], 4), + "n": len(ordered), + } + + +class VoiceCallMetricsObserver(BaseObserver): + """Consumes MetricsFrames: per-stage TTFB + LLM/TTS usage accumulation. + + ``on_push_frame`` fires once per pipeline *edge* a frame traverses, so a + single MetricsFrame is observed many times — dedup on ``frame.id`` is + mandatory or every stat gets multiply counted. + """ + + def __init__(self, session_id: str) -> None: + super().__init__() + self.session_id = session_id + self.call_sid: str | None = None + self._seen_frames: set[int] = set() + self._ttfb: dict[str, list[float]] = {} + self._processing: dict[str, list[float]] = {} + self._llm_model = "" + self._prompt_tokens = 0 + self._cached_prompt_tokens = 0 + self._completion_tokens = 0 + self._tts_chars = 0 + self._started = time.monotonic() + self._finalized = False + + async def on_push_frame(self, data: FramePushed) -> None: + frame = data.frame + if not isinstance(frame, MetricsFrame) or frame.id in self._seen_frames: + return + self._seen_frames.add(frame.id) + for item in frame.data: + if isinstance(item, TTFBMetricsData): + self._ttfb.setdefault(item.processor, []).append(item.value) + elif isinstance(item, ProcessingMetricsData): + self._processing.setdefault(item.processor, []).append(item.value) + elif isinstance(item, LLMUsageMetricsData): + self._llm_model = item.model or self._llm_model + self._prompt_tokens += item.value.prompt_tokens + self._cached_prompt_tokens += item.value.cache_read_input_tokens or 0 + self._completion_tokens += item.value.completion_tokens + elif isinstance(item, TTSUsageMetricsData): + self._tts_chars += item.value + + def finalize(self) -> dict | None: + """Writes the per-call summary + cost records. Idempotent; returns + the summary dict (or None on repeat calls).""" + if self._finalized: + return None + self._finalized = True + + duration_s = time.monotonic() - self._started + label = f"streaming/{self.session_id}" + + llm_rec = CostMeter.record_usage( + role="voice_llm", + model=self._llm_model or "gpt-4o-mini", + input_tokens=self._prompt_tokens, + cached_input_tokens=self._cached_prompt_tokens, + output_tokens=self._completion_tokens, + label=label, + ) + stt_minutes = duration_s / 60.0 + stt_cost = stt_minutes * VOICE_PRICES_USD[STT_PRICE_KEY] + CostMeter.record_usage( + role="voice_stt", model=STT_PRICE_KEY, cost_usd=stt_cost, label=label, + ) + tts_cost = self._tts_chars * VOICE_PRICES_USD[TTS_PRICE_KEY] + CostMeter.record_usage( + role="voice_tts", model=TTS_PRICE_KEY, cost_usd=tts_cost, label=label, + ) + + summary = { + "ts": time.time(), + "session_id": self.session_id, + "call_sid": self.call_sid, + "duration_s": round(duration_s, 2), + "llm": { + "model": llm_rec.model, + "prompt_tokens": self._prompt_tokens, + "cached_prompt_tokens": self._cached_prompt_tokens, + "completion_tokens": self._completion_tokens, + "cost_usd": round(llm_rec.cost_usd, 6), + }, + "stt": { + "model": STT_PRICE_KEY, + "audio_min": round(stt_minutes, 3), + "cost_usd": round(stt_cost, 6), + }, + "tts": { + "model": TTS_PRICE_KEY, + "chars": self._tts_chars, + "cost_usd": round(tts_cost, 6), + }, + "ttfb_s": {name: _stats(vals) for name, vals in self._ttfb.items()}, + "processing_s": {name: _stats(vals) for name, vals in self._processing.items()}, + "total_cost_usd": round(llm_rec.cost_usd + stt_cost + tts_cost, 6), + } + + path = Path( + os.environ.get("VOICE_CALL_METRICS_LOG", "data/eval_runs/voice_call_metrics.jsonl") + ) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a") as f: + f.write(json.dumps(summary) + "\n") + + logger.info( + "voice call metrics: session=%s duration=%.1fs cost=$%.4f ttfb=%s", + self.session_id, duration_s, summary["total_cost_usd"], + {k: v["avg"] for k, v in summary["ttfb_s"].items()}, + ) + return summary diff --git a/apps/voice/streaming.py b/apps/voice/streaming.py index 2525003..5a1baac 100644 --- a/apps/voice/streaming.py +++ b/apps/voice/streaming.py @@ -69,6 +69,7 @@ from agents.prompt_loader import load_champion from agents.schemas import EmotionalState, HandoffContext from agents.settings import get_secrets +from apps.voice.call_metrics import VoiceCallMetricsObserver logger = logging.getLogger(__name__) @@ -649,6 +650,12 @@ async def handle_propose_offer(params: FunctionCallParams) -> None: # speech, no agent speech) for 60s, the task auto-cancels and the # call hangs up. Defends against the agent-stuck-in-goodbye-loop # behavior we hit on the Sociopath stress test (incident #16). + # Per-call latency/cost observer — consumes the MetricsFrames that + # enable_metrics/enable_usage_metrics generate (previously dropped). + # Summary lands in data/eval_runs/voice_call_metrics.jsonl + CostMeter. + metrics_observer = VoiceCallMetricsObserver(session_id) + metrics_observer.call_sid = call_sid + task = PipelineTask( pipeline, params=PipelineParams( @@ -656,6 +663,7 @@ async def handle_propose_offer(params: FunctionCallParams) -> None: enable_metrics=True, enable_usage_metrics=True, ), + observers=[metrics_observer], idle_timeout_secs=60.0, cancel_on_idle_timeout=True, ) @@ -682,6 +690,10 @@ async def on_client_disconnected(transport_, client) -> None: # noqa: ARG001 except Exception: logger.exception("Streaming pipeline failed for session=%s", session_id) finally: + try: + metrics_observer.finalize() + except Exception: + logger.exception("Metrics finalize failed for session=%s", session_id) try: await aiohttp_session.close() except Exception: diff --git a/tests/test_call_metrics_observer.py b/tests/test_call_metrics_observer.py new file mode 100644 index 0000000..ea786eb --- /dev/null +++ b/tests/test_call_metrics_observer.py @@ -0,0 +1,122 @@ +"""VoiceCallMetricsObserver: MetricsFrame accumulation, per-edge dedup, and +finalize() output (summary JSONL + CostMeter records). Hermetic — synthetic +frames, tmp_path logs.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from pipecat.frames.frames import MetricsFrame, TextFrame +from pipecat.metrics.metrics import ( + LLMTokenUsage, + LLMUsageMetricsData, + TTFBMetricsData, + TTSUsageMetricsData, +) +from pipecat.observers.base_observer import FramePushed +from pipecat.processors.frame_processor import FrameDirection + +from agents.cost_meter import VOICE_PRICES_USD +from apps.voice.call_metrics import VoiceCallMetricsObserver + + +@pytest.fixture +def log_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[Path, Path]: + metrics_log = tmp_path / "voice_call_metrics.jsonl" + cost_log = tmp_path / "cost_log.jsonl" + monkeypatch.setenv("VOICE_CALL_METRICS_LOG", str(metrics_log)) + monkeypatch.setenv("COST_LOG", str(cost_log)) + return metrics_log, cost_log + + +def _pushed(frame) -> FramePushed: + return FramePushed( + source=None, destination=None, frame=frame, + direction=FrameDirection.DOWNSTREAM, timestamp=0, + ) + + +def _metrics_frame() -> MetricsFrame: + return MetricsFrame( + data=[ + TTFBMetricsData(processor="OpenAILLMService#0", model="gpt-4o-mini", value=0.42), + TTFBMetricsData(processor="RimeHttpTTSService#0", model="mistv2", value=0.2), + LLMUsageMetricsData( + processor="OpenAILLMService#0", + model="gpt-4o-mini", + value=LLMTokenUsage( + prompt_tokens=1000, completion_tokens=500, total_tokens=1500, + ), + ), + TTSUsageMetricsData( + processor="RimeHttpTTSService#0", model="mistv2", value=200, + ), + ] + ) + + +async def test_same_frame_seen_on_multiple_edges_counted_once(log_paths) -> None: + observer = VoiceCallMetricsObserver("sess1") + frame = _metrics_frame() + # An observer sees the same frame once per pipeline edge it traverses. + await observer.on_push_frame(_pushed(frame)) + await observer.on_push_frame(_pushed(frame)) + await observer.on_push_frame(_pushed(frame)) + + summary = observer.finalize() + + assert summary is not None + assert summary["llm"]["prompt_tokens"] == 1000 + assert summary["llm"]["completion_tokens"] == 500 + assert summary["tts"]["chars"] == 200 + assert summary["ttfb_s"]["OpenAILLMService#0"]["n"] == 1 + + +async def test_non_metrics_frames_ignored(log_paths) -> None: + observer = VoiceCallMetricsObserver("sess2") + await observer.on_push_frame(_pushed(TextFrame("hello"))) + summary = observer.finalize() + assert summary["llm"]["prompt_tokens"] == 0 + assert summary["ttfb_s"] == {} + + +async def test_finalize_writes_summary_and_cost_records(log_paths) -> None: + metrics_log, cost_log = log_paths + observer = VoiceCallMetricsObserver("sess3") + observer.call_sid = "CA123" + await observer.on_push_frame(_pushed(_metrics_frame())) + await observer.on_push_frame(_pushed(_metrics_frame())) # second turn + + summary = observer.finalize() + + # Summary JSONL line + lines = metrics_log.read_text().splitlines() + assert len(lines) == 1 + written = json.loads(lines[0]) + assert written["session_id"] == "sess3" + assert written["call_sid"] == "CA123" + assert written["llm"]["prompt_tokens"] == 2000 + assert written["ttfb_s"]["OpenAILLMService#0"]["n"] == 2 + + # LLM cost math: 2000 in @ $0.15/M + 1000 out @ $0.60/M + expected_llm = 2000 * 0.15 / 1e6 + 1000 * 0.60 / 1e6 + assert written["llm"]["cost_usd"] == pytest.approx(expected_llm, abs=1e-9) + # TTS cost math: 400 chars at the Rime per-char rate + assert written["tts"]["cost_usd"] == pytest.approx( + 400 * VOICE_PRICES_USD["rime/mistv2"], abs=1e-9, + ) + assert written["total_cost_usd"] == pytest.approx( + written["llm"]["cost_usd"] + written["stt"]["cost_usd"] + written["tts"]["cost_usd"], + abs=1e-6, + ) + + # Three mirrored CostMeter records with the session label + cost_lines = [json.loads(line) for line in cost_log.read_text().splitlines()] + assert {r["role"] for r in cost_lines} == {"voice_llm", "voice_stt", "voice_tts"} + assert all(r["label"] == "streaming/sess3" for r in cost_lines) + + # finalize is idempotent + assert observer.finalize() is None + assert len(metrics_log.read_text().splitlines()) == 1 + assert summary is not None diff --git a/tests/test_cost_meter.py b/tests/test_cost_meter.py new file mode 100644 index 0000000..2db15e4 --- /dev/null +++ b/tests/test_cost_meter.py @@ -0,0 +1,50 @@ +"""CostMeter.record_usage pricing + the COST_LOG-resolved-at-import regression.""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from agents.cost_meter import CostMeter + + +def test_record_usage_prices_gpt4o_mini_tokens( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + log = tmp_path / "cost.jsonl" + monkeypatch.setenv("COST_LOG", str(log)) + + rec = CostMeter.record_usage( + role="voice_llm", model="gpt-4o-mini", + input_tokens=1_000_000, cached_input_tokens=200_000, output_tokens=100_000, + ) + + # 800k uncached @ $0.15/M + 200k cached @ $0.075/M + 100k out @ $0.60/M + assert rec.cost_usd == pytest.approx(0.8 * 0.15 + 0.2 * 0.075 + 0.1 * 0.60) + + +def test_record_usage_explicit_cost_passthrough( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("COST_LOG", str(tmp_path / "cost.jsonl")) + rec = CostMeter.record_usage( + role="voice_stt", model="deepgram/nova-3-general", cost_usd=0.0123, + ) + assert rec.cost_usd == 0.0123 + + +def test_cost_log_env_respected_after_import( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + # Regression: _path used to be a ClassVar resolved at import time, so + # setting COST_LOG afterwards silently wrote to the default location. + log = tmp_path / "late-configured.jsonl" + monkeypatch.setenv("COST_LOG", str(log)) + + CostMeter.record_usage(role="test", model="gpt-4o-mini", output_tokens=10) + + assert log.exists() + row = json.loads(log.read_text().splitlines()[0]) + assert row["role"] == "test" + assert CostMeter.total_usd() == pytest.approx(row["cost_usd"]) From 395070d68fefe7600da6a18ae7ef835621efeccc Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:14:25 +0530 Subject: [PATCH 6/7] fix(docker): copy sources before install and add voice extras to api/worker images MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous order (COPY pyproject.toml -> uv pip install . -> COPY sources) failed the hatchling wheel build outright — packages = [agents, apps, learning] didn't exist at install time — and even a successful base install lacked pipecat/twilio, so apps.api.main (which imports the streaming voice router) could not start in-container. Verified: both images build; `import apps.api.main` (full pipecat + Smart Turn chain) and PipecatProvider import succeed in-container. Co-Authored-By: Claude Fable 5 --- Dockerfile.api | 8 ++++++-- Dockerfile.worker | 8 ++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/Dockerfile.api b/Dockerfile.api index 55fcd81..2c6d91f 100644 --- a/Dockerfile.api +++ b/Dockerfile.api @@ -4,12 +4,16 @@ WORKDIR /app RUN pip install --no-cache-dir uv +# Sources must be present before the install: hatchling builds a wheel from +# packages = ["agents", "apps", "learning"] and fails if they're missing. +# The [voice] extra pulls pipecat + twilio, which apps.api.main imports via +# the streaming voice router. COPY pyproject.toml ./ -RUN uv pip install --system --no-cache . - COPY agents ./agents COPY apps ./apps COPY learning ./learning +RUN uv pip install --system --no-cache '.[voice]' + COPY data ./data COPY settings.yaml ./ diff --git a/Dockerfile.worker b/Dockerfile.worker index be9a55d..a08f371 100644 --- a/Dockerfile.worker +++ b/Dockerfile.worker @@ -4,12 +4,16 @@ WORKDIR /app RUN pip install --no-cache-dir uv +# Sources must be present before the install: hatchling builds a wheel from +# packages = ["agents", "apps", "learning"] and fails if they're missing. +# The worker needs [voice] too — resolve_via_voice imports PipecatProvider +# when settings.yaml voice.provider is "pipecat". COPY pyproject.toml ./ -RUN uv pip install --system --no-cache . - COPY agents ./agents COPY apps ./apps COPY learning ./learning +RUN uv pip install --system --no-cache '.[voice]' + COPY data ./data COPY settings.yaml ./ From 33cb5bb248d7565f818108bede069bf95847ba76 Mon Sep 17 00:00:00 2001 From: Kaustav Ghosh Date: Fri, 10 Jul 2026 18:15:31 +0530 Subject: [PATCH 7/7] ci: run unit tests on push/PR; keep live smokes behind workflow_dispatch voice-smoke.yml was a fully disabled stub (if: false). Renamed to ci.yml: - unit job runs on every push/PR with no secrets (installs .[voice,dev] since the unit suite imports apps.voice.streaming) - smoke job (live STT/LLM/TTS, ~$0.06/run) unchanged but now reachable via manual dispatch, gated on unit passing Co-Authored-By: Claude Fable 5 --- .github/workflows/{voice-smoke.yml => ci.yml} | 58 +++++++++++++------ 1 file changed, 41 insertions(+), 17 deletions(-) rename .github/workflows/{voice-smoke.yml => ci.yml} (50%) diff --git a/.github/workflows/voice-smoke.yml b/.github/workflows/ci.yml similarity index 50% rename from .github/workflows/voice-smoke.yml rename to .github/workflows/ci.yml index 0963401..c701828 100644 --- a/.github/workflows/voice-smoke.yml +++ b/.github/workflows/ci.yml @@ -1,33 +1,35 @@ -# Voice-pipeline smoke tests — disabled by default. +# CI — unit tests on every push/PR (no secrets needed), plus the +# voice-pipeline integration smokes behind manual dispatch. # -# Runs both simulator-based smoke tests against live external APIs (OpenAI, -# OpenRouter, Rime, Deepgram) without ever dialing Twilio. Catches drift in -# the upstream APIs (e.g. Rime renaming voice IDs, Pipecat moving classes) -# that unit tests would miss. +# The smoke job runs both simulator-based smoke tests against live external +# APIs (OpenAI, OpenRouter, Rime, Deepgram) without ever dialing Twilio. +# Catches drift in the upstream APIs (e.g. Rime renaming voice IDs, Pipecat +# moving classes) that unit tests would miss. # -# To enable: +# To run the smokes: # 1. Add these repo secrets at github.com///settings/secrets/actions: # OPENROUTER_API_KEY, OPENAI_API_KEY, RIME_API_KEY, DEEPGRAM_API_KEY -# 2. Change the `on:` trigger below from `workflow_dispatch:` (manual only) -# to `schedule: - cron: '0 4 * * *'` for nightly automatic runs. -# 3. Remove the top-level `if: false` guard. +# 2. Trigger the workflow manually (Actions -> ci -> Run workflow), or +# add `schedule: - cron: '0 4 * * *'` for nightly automatic runs. # -# Cost per run: ~$0.06 (OpenAI gpt-4o-mini for streaming, OpenRouter free -# tier for TwiML if available). Free Deepgram + Rime quotas cover the rest. +# Cost per smoke run: ~$0.06 (OpenAI gpt-4o-mini for streaming, OpenRouter +# free tier for TwiML if available). Free Deepgram + Rime quotas cover the rest. -name: voice-smoke +name: ci on: - workflow_dispatch: # manual-only by default + push: + branches: [main] + pull_request: + workflow_dispatch: # enables the smoke job permissions: contents: read jobs: - smoke: - if: false # safety: remove this line to enable + unit: runs-on: ubuntu-latest - timeout-minutes: 15 + timeout-minutes: 20 steps: - uses: actions/checkout@v4 @@ -38,13 +40,35 @@ jobs: python-version: "3.12" cache: pip + # voice extras are required by the unit suite too: test_propose_offer / + # test_tool_call_filter / test_twilio_security import apps.voice.streaming. - name: Install runtime + voice extras run: | python -m pip install --upgrade pip pip install -e '.[voice,dev]' - name: Run unit tests - run: pytest -m "not integration" -q + run: pytest -q # addopts already excludes -m integration + + smoke: + if: github.event_name == 'workflow_dispatch' + needs: unit + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python 3.12 + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install runtime + voice extras + run: | + python -m pip install --upgrade pip + pip install -e '.[voice,dev]' - name: Boot API env: