diff --git a/Dockerfile.voice b/Dockerfile.voice new file mode 100644 index 0000000..b8af0eb --- /dev/null +++ b/Dockerfile.voice @@ -0,0 +1,13 @@ +FROM python:3.12-slim + +WORKDIR /app +COPY pyproject.toml README.md* ./ +COPY gogo ./gogo +COPY alembic ./alembic +COPY alembic.ini ./ + +# voice extra pulls faster-whisper (CPU by default; CUDA wheels on the GPU node) +RUN pip install --no-cache-dir -e '.[voice]' + +EXPOSE 9092 +CMD ["python", "-m", "gogo.voice.server"] diff --git a/alembic/versions/9dce40d5c9ce_call_registrations.py b/alembic/versions/9dce40d5c9ce_call_registrations.py new file mode 100644 index 0000000..a0c12ac --- /dev/null +++ b/alembic/versions/9dce40d5c9ce_call_registrations.py @@ -0,0 +1,41 @@ +"""call registrations + +Revision ID: 9dce40d5c9ce +Revises: 62fcb10345fa +Create Date: 2026-07-11 10:58:21.464071 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = '9dce40d5c9ce' +down_revision: Union[str, None] = '62fcb10345fa' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('call_registrations', + sa.Column('id', sa.Uuid(), nullable=False), + sa.Column('tenant_id', sa.Uuid(), nullable=False), + sa.Column('caller_msisdn', sa.String(length=30), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('answered_by_human', sa.Boolean(), nullable=False), + sa.Column('handled_by_agent', sa.Boolean(), nullable=False), + sa.Column('finalized', sa.Boolean(), nullable=False), + sa.ForeignKeyConstraint(['tenant_id'], ['tenants.id'], ), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_call_registrations_tenant_id'), 'call_registrations', ['tenant_id'], unique=False) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_call_registrations_tenant_id'), table_name='call_registrations') + op.drop_table('call_registrations') + # ### end Alembic commands ### diff --git a/docker-compose.yml b/docker-compose.yml index 02b5b55..c8ee6f4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,37 @@ services: sh -c "alembic upgrade head && uvicorn gogo.main:app --host 0.0.0.0 --port 8000" + # Voice pipeline (M4): AudioSocket server with STT/TTS/LLM. + # Heavy (faster-whisper); start with: docker compose --profile voice up + voice: + build: + context: . + dockerfile: Dockerfile.voice + profiles: ["voice"] + depends_on: + db: + condition: service_healthy + environment: + GOGO_DATABASE_URL: postgresql+psycopg://gogo:gogo@db:5432/gogo + GOGO_RECORDINGS_DIR: /data/recordings + env_file: + - path: .env + required: false + ports: + - "127.0.0.1:9092:9092" + volumes: + - recordings:/data/recordings + + # Asterisk (M4): softphones register here; agent calls bridge to `voice`. + # Config is generated: python -m gogo.telephony.asterisk --out ./asterisk/conf + asterisk: + image: andrius/asterisk:alpine-20.5.2 + profiles: ["voice"] + network_mode: host # SIP/RTP is painful behind NAT; host networking for MVP + volumes: + - ./asterisk/conf/pjsip.conf:/etc/asterisk/pjsip.conf:ro + - ./asterisk/conf/extensions.conf:/etc/asterisk/extensions.conf:ro + volumes: pgdata: recordings: diff --git a/docs/VOICE_TESTING.md b/docs/VOICE_TESTING.md new file mode 100644 index 0000000..b4ee908 --- /dev/null +++ b/docs/VOICE_TESTING.md @@ -0,0 +1,69 @@ +# M4 — Software-only voice testing (no GSM hardware) + +Everything here runs on a laptop/DC per §16: desktop/mobile softphones register +directly to Asterisk over IP; the GSM leg is the only thing not exercised +(that's M5 + the joint hardware session). + +## 1. Prereqs + +- Postgres running, schema migrated, demo tenant seeded: + ```bash + .venv/bin/alembic upgrade head + .venv/bin/python -m gogo.seed + ``` +- `.env` with at least `GOGO_ANTHROPIC_API_KEY` (agent) and one TTS key + (`GOGO_AZURE_SPEECH_KEY` + region, or `GOGO_ELEVENLABS_API_KEY`). +- Whisper model downloads on first run (`GOGO_WHISPER_MODEL=small` is fine on CPU). + +## 2. Start the stack + +```bash +# backend API (internal dialplan endpoints live here) +.venv/bin/uvicorn gogo.main:app --host 0.0.0.0 --port 8000 + +# voice pipeline (AudioSocket server on :9092) +.venv/bin/python -m gogo.voice.server + +# generate Asterisk config from the DB, then start Asterisk +.venv/bin/python -m gogo.telephony.asterisk --out ./asterisk/conf \ + --backend http://127.0.0.1:8000 --voice 127.0.0.1:9092 +docker compose --profile voice up asterisk +``` + +## 3. Softphones + +Two SIP accounts exist per tenant after config generation: + +| Role | Username | Password | Purpose | +|---|---|---|---| +| Worker | from dashboard → Telefonija → QR | generated | rings before the agent | +| Test caller | `test-caller-salon-merima` | `test-` (in pjsip.conf) | simulates the customer | + +Register the worker softphone (e.g. Linphone on a phone, scan the QR) and the +test-caller softphone (e.g. Zoiper on the laptop) against the machine's IP. + +## 4. Test scenarios + +1. **Ring-group first (§5.2):** from the test-caller dial any number (e.g. `100`). + The worker softphone rings for `ring_timeout_s`; answer → normal human call, + dashboard logs outcome `javilo se osoblje`, no agent minutes metered. +2. **Agent fallback:** dial again, don't answer → after the timeout the agent + picks up, greets with the recording disclosure, books a request. Check: + dashboard → Zahtjevi (new request), Pozivi (transcript + audio player), + Potrošnja (minutes metered), owner email (proposal + .ics). +3. **Out of hours (A.5):** set working hours to closed (dashboard → Salon), + dial → agent answers immediately with the out-of-hours greeting. +4. **Barge-in:** interrupt the agent mid-sentence — playback must stop and the + agent must respond to what you said. +5. **Abandoned call (§5.5):** dial, hang up while the worker phone still rings → + `Pozivi` shows `prekinut` and the missed-call SMS appears in the backend log + (console SMS provider). +6. **Latency (§15):** `calls.trace` (call detail JSON) stores per-turn + `stt_ms`/`llm_ms` — target < 1500 ms combined before TTS starts. + +## 5. Automated coverage + +`tests/test_voice.py` covers the same flows without Asterisk: VAD, the full +call loop against fakes (booking, unclear speech, hangup, inactivity, barge-in), +the AudioSocket wire protocol over real TCP, the dialplan internal API +(ring targets, out-of-hours, abandoned + SMS) and config generation. diff --git a/gogo/api/internal.py b/gogo/api/internal.py new file mode 100644 index 0000000..1867de3 --- /dev/null +++ b/gogo/api/internal.py @@ -0,0 +1,159 @@ +"""Internal API for the Asterisk dialplan (M4) — plain-text responses because +the dialplan consumes them via CURL()/CUT(). + +Auth: shared token (GOGO_INTERNAL_TOKEN) as a query param; this API must only +be reachable on the internal network regardless. +""" + +from __future__ import annotations + +import logging +import uuid + +from fastapi import APIRouter, Depends +from fastapi.responses import PlainTextResponse +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from gogo.config import get_settings +from gogo.db import get_session +from gogo.domain import CallOutcome +from gogo.hours import DEFAULT_WORKING_HOURS, is_open_at +from gogo.models import Call, CallRegistration, Tenant, Worker, utcnow +from gogo.sms import send_sms +from gogo.sms.templates import render_sms + +log = logging.getLogger("gogo.internal") + +router = APIRouter() + + +def _authorized(token: str) -> bool: + return token == get_settings().internal_token + + +@router.get("/internal/calls/register", response_class=PlainTextResponse) +async def register_call( + tenant: str, + caller: str = "", + token: str = "", + session: AsyncSession = Depends(get_session), +): + """Dialplan entry point. Returns 'uuid,ring_timeout,dial_targets'. + + dial_targets is empty when the agent should answer immediately + (out of hours, ring-first disabled, or no active workers §5.2). + """ + if not _authorized(token): + return PlainTextResponse("forbidden", status_code=403) + t = ( + await session.execute(select(Tenant).where(Tenant.slug == tenant)) + ).scalar_one_or_none() + if t is None or t.status != "active": + return PlainTextResponse("unknown-tenant", status_code=404) + + reg = CallRegistration(tenant_id=t.id, caller_msisdn=caller.strip()) + session.add(reg) + await session.commit() + + targets = "" + from zoneinfo import ZoneInfo + + open_now = is_open_at( + t.working_hours or DEFAULT_WORKING_HOURS, utcnow(), ZoneInfo(t.timezone) + ) + if t.ring_first_enabled and open_now: + workers = ( + ( + await session.execute( + select(Worker) + .where(Worker.tenant_id == t.id, Worker.active.is_(True)) + .order_by(Worker.sip_username) + ) + ) + .scalars() + .all() + ) + if workers: + # '&' rings all at once (default §5.2); sequential strategy is a Phase-2 tuning + targets = "&".join(f"PJSIP/{w.sip_username}" for w in workers) + + return f"{reg.id},{t.ring_timeout_s},{targets}" + + +@router.get("/internal/calls/{reg_id}/answered", response_class=PlainTextResponse) +async def call_answered( + reg_id: str, + token: str = "", + duration: int = 0, + session: AsyncSession = Depends(get_session), +): + """Ring group answered by a human — log the call, no agent minutes (§12).""" + if not _authorized(token): + return PlainTextResponse("forbidden", status_code=403) + reg = ( + await session.execute( + select(CallRegistration).where(CallRegistration.id == uuid.UUID(reg_id)) + ) + ).scalar_one_or_none() + if reg is None: + return PlainTextResponse("unknown", status_code=404) + reg.answered_by_human = True + if not reg.finalized: + reg.finalized = True + session.add( + Call( + tenant_id=reg.tenant_id, + caller_msisdn=reg.caller_msisdn, + outcome=CallOutcome.human_answered.value, + duration_s=max(0, duration), + ) + ) + await session.commit() + return "ok" + + +@router.get("/internal/calls/{reg_id}/hangup", response_class=PlainTextResponse) +async def call_hangup( + reg_id: str, + status: str = "", + token: str = "", + session: AsyncSession = Depends(get_session), +): + """Dialplan h-extension. If nobody (human or agent) handled the call, the + caller abandoned during ringing → log + missed-call SMS (§5.5).""" + if not _authorized(token): + return PlainTextResponse("forbidden", status_code=403) + reg = ( + await session.execute( + select(CallRegistration).where(CallRegistration.id == uuid.UUID(reg_id)) + ) + ).scalar_one_or_none() + if reg is None: + return PlainTextResponse("unknown", status_code=404) + if reg.finalized or reg.answered_by_human or reg.handled_by_agent: + return "ok" + + reg.finalized = True + tenant = ( + await session.execute(select(Tenant).where(Tenant.id == reg.tenant_id)) + ).scalar_one() + session.add( + Call( + tenant_id=reg.tenant_id, + caller_msisdn=reg.caller_msisdn, + outcome=CallOutcome.abandoned.value, + ) + ) + if reg.caller_msisdn: + link = tenant.website or get_settings().base_url + await send_sms( + session, + tenant.id, + reg.caller_msisdn, + render_sms(tenant, "missed_call", link=link), + kind="missed_call", + ) + await session.commit() + log.info("abandoned call for %s from %s (status=%s)", tenant.slug, reg.caller_msisdn, status) + return "ok" diff --git a/gogo/config.py b/gogo/config.py index 06de9a8..5481a15 100644 --- a/gogo/config.py +++ b/gogo/config.py @@ -43,6 +43,12 @@ class Settings(BaseSettings): google_client_id: str = "" google_client_secret: str = "" + # Voice pipeline (M4) + voice_host: str = "0.0.0.0" + voice_port: int = 9092 + recordings_dir: str = "./recordings" + internal_token: str = "dev-internal-token" # dialplan ↔ backend shared secret + # Proposal engine defaults (overridable per tenant) proposal_ttl_hours: int = 24 diff --git a/gogo/main.py b/gogo/main.py index 9623db2..ff83b86 100644 --- a/gogo/main.py +++ b/gogo/main.py @@ -30,6 +30,7 @@ async def lifespan(app: FastAPI): def create_app() -> FastAPI: from gogo.admin.router import router as admin_router + from gogo.api.internal import router as internal_router from gogo.chat.router import router as chat_router from gogo.dashboard.router import router as dashboard_router @@ -40,6 +41,7 @@ def create_app() -> FastAPI: app.include_router(chat_router) app.include_router(dashboard_router) app.include_router(admin_router) + app.include_router(internal_router) @app.get("/", include_in_schema=False) async def root(): diff --git a/gogo/models.py b/gogo/models.py index 0e00a67..6767157 100644 --- a/gogo/models.py +++ b/gogo/models.py @@ -241,6 +241,24 @@ class BookingRequest(Base, TimestampMixin): resolution_note: Mapped[str] = mapped_column(Text, default="") +class CallRegistration(Base): + """Handoff between the Asterisk dialplan and the voice pipeline (M4). + + The dialplan registers an inbound call over the internal API before ringing + anyone; the AudioSocket server looks the UUID up to resolve tenant+caller. + """ + + __tablename__ = "call_registrations" + + id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4) + tenant_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("tenants.id"), index=True) + caller_msisdn: Mapped[str] = mapped_column(String(30), default="") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow) + answered_by_human: Mapped[bool] = mapped_column(Boolean, default=False) + handled_by_agent: Mapped[bool] = mapped_column(Boolean, default=False) + finalized: Mapped[bool] = mapped_column(Boolean, default=False) + + class ActionToken(Base): """Signed single-use action link tokens for proposal emails (§9.1).""" diff --git a/gogo/telephony/asterisk.py b/gogo/telephony/asterisk.py new file mode 100644 index 0000000..79f7480 --- /dev/null +++ b/gogo/telephony/asterisk.py @@ -0,0 +1,222 @@ +"""Asterisk configuration generator (M4): pjsip.conf + extensions.conf. + +Generated from the DB (tenants, workers) so the dashboard stays the single +source of truth. Ring-group-first logic (§5.2) lives in the dialplan, but the +decision (targets + timeout, working hours) comes from the backend via the +internal API at call time — Asterisk never duplicates tenant config. + +Usage: python -m gogo.telephony.asterisk --out ./asterisk/conf +Regenerate + `asterisk -rx 'core reload'` whenever workers change. +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +from sqlalchemy import select + +from gogo.config import get_settings +from gogo.db import get_sessionmaker +from gogo.models import Tenant, Worker + +PJSIP_HEADER = """\ +; -- generated by gogo.telephony.asterisk — do not edit by hand -- + +[global] +type=global +user_agent=GogoTelefon + +[transport-udp] +type=transport +protocol=udp +bind=0.0.0.0:5060 +""" + +ENDPOINT_TEMPLATE = """\ + +; worker: {name} ({tenant}) +[{username}] +type=endpoint +context={context} +disallow=all +allow=alaw,ulaw,slin +auth={username}-auth +aors={username} +direct_media=no +rtp_symmetric=yes +force_rport=yes +rewrite_contact=yes + +[{username}-auth] +type=auth +auth_type=userpass +username={username} +password={password} + +[{username}] +type=aor +max_contacts=3 +remove_existing=yes +""" + +# Test endpoint for M4 software-only testing: a desktop softphone registers as +# 'test-caller' and dials any number to reach the tenant's inbound flow (§16 M4). +TEST_CALLER_TEMPLATE = """\ + +; software-testing caller (softphone) → lands in {context} +[test-caller-{slug}] +type=endpoint +context={context} +disallow=all +allow=alaw,ulaw,slin +auth=test-caller-{slug}-auth +aors=test-caller-{slug} +direct_media=no +rtp_symmetric=yes +force_rport=yes +rewrite_contact=yes + +[test-caller-{slug}-auth] +type=auth +auth_type=userpass +username=test-caller-{slug} +password={password} + +[test-caller-{slug}] +type=aor +max_contacts=3 +remove_existing=yes +""" + +EXTENSIONS_HEADER = """\ +; -- generated by gogo.telefonija.asterisk — do not edit by hand -- + +[globals] +GOGO_BACKEND={backend} +GOGO_VOICE={voice} +GOGO_TOKEN={token} + +[gogo-workers] +; workers only receive calls in MVP +exten => _X.,1,Hangup() +""" + +TENANT_CONTEXT_TEMPLATE = """\ + +[{context}] +; inbound flow for {tenant_name} (§5.2): register → ring staff → agent fallback +exten => _[+0-9].,1,NoOp(Gogo inbound ${{EXTEN}} for {slug}) + same => n,Answer() + same => n,Set(REG=${{CURL(${{GOGO_BACKEND}}/internal/calls/register?tenant={slug}&caller=${{URIENCODE(${{CALLERID(num)}})}}&token=${{GOGO_TOKEN}})}}) + same => n,GotoIf($["${{REG}}"=""]?fallback) + same => n,Set(GOGO_UUID=${{CUT(REG,\\,,1)}}) + same => n,Set(RINGTIME=${{CUT(REG,\\,,2)}}) + same => n,Set(TARGETS=${{CUT(REG,\\,,3)}}) + same => n,GotoIf($["${{TARGETS}}"=""]?agent) + same => n,Dial(${{TARGETS}},${{RINGTIME}}) + same => n,GotoIf($["${{DIALSTATUS}}"="ANSWER"]?answered) + same => n(agent),AudioSocket(${{GOGO_UUID}},${{GOGO_VOICE}}) + same => n,Hangup() + same => n(answered),Set(RES=${{CURL(${{GOGO_BACKEND}}/internal/calls/${{GOGO_UUID}}/answered?token=${{GOGO_TOKEN}}&duration=${{CDR(billsec)}})}}) + same => n,Hangup() + same => n(fallback),Playback(vm-goodbye) + same => n,Hangup() + +; caller hung up mid-flow (e.g. while ringing) → missed-call handling (§5.5) +exten => h,1,GotoIf($["${{GOGO_UUID}}"=""]?done) + same => n,Set(RES=${{CURL(${{GOGO_BACKEND}}/internal/calls/${{GOGO_UUID}}/hangup?status=${{DIALSTATUS}}&token=${{GOGO_TOKEN}})}}) + same => n(done),NoOp(bye) +""" + + +async def generate(out_dir: Path, backend_url: str = "", voice_addr: str = "") -> list[str]: + """Write pjsip.conf + extensions.conf; returns list of written files.""" + s = get_settings() + backend_url = backend_url or "http://backend:8000" + voice_addr = voice_addr or f"voice:{s.voice_port}" + + pjsip = [PJSIP_HEADER] + extensions = [ + EXTENSIONS_HEADER.format(backend=backend_url, voice=voice_addr, token=s.internal_token) + ] + + async with get_sessionmaker()() as session: + tenants = ( + ( + await session.execute( + select(Tenant).where(Tenant.status == "active").order_by(Tenant.slug) + ) + ) + .scalars() + .all() + ) + for tenant in tenants: + context = f"gogo-inbound-{tenant.slug}" + workers = ( + ( + await session.execute( + select(Worker) + .where(Worker.tenant_id == tenant.id, Worker.active.is_(True)) + .order_by(Worker.sip_username) + ) + ) + .scalars() + .all() + ) + for w in workers: + pjsip.append( + ENDPOINT_TEMPLATE.format( + name=w.name, + tenant=tenant.slug, + username=w.sip_username, + password=w.sip_password, + context="gogo-workers", + ) + ) + pjsip.append( + TEST_CALLER_TEMPLATE.format( + slug=tenant.slug, + context=context, + password=f"test-{tenant.widget_public_key[:10]}", + ) + ) + extensions.append( + TENANT_CONTEXT_TEMPLATE.format( + context=context, tenant_name=tenant.name, slug=tenant.slug + ) + ) + + out_dir.mkdir(parents=True, exist_ok=True) + written = [] + for filename, content in [ + ("pjsip.conf", "".join(pjsip)), + ("extensions.conf", "".join(extensions)), + ]: + path = out_dir / filename + path.write_text(content) + written.append(str(path)) + return written + + +def main() -> None: + out = Path("./asterisk/conf") + backend = "" + voice = "" + args = sys.argv[1:] + while args: + arg = args.pop(0) + if arg == "--out" and args: + out = Path(args.pop(0)) + elif arg == "--backend" and args: + backend = args.pop(0) + elif arg == "--voice" and args: + voice = args.pop(0) + written = asyncio.run(generate(out, backend, voice)) + for path in written: + print(f"wrote {path}") + + +if __name__ == "__main__": + main() diff --git a/gogo/voice/__init__.py b/gogo/voice/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/gogo/voice/audio.py b/gogo/voice/audio.py new file mode 100644 index 0000000..7f99f69 --- /dev/null +++ b/gogo/voice/audio.py @@ -0,0 +1,94 @@ +"""Audio helpers for the voice pipeline. + +Wire format everywhere: SLIN — signed 16-bit LE mono PCM @ 8 kHz (AudioSocket). +Whisper wants 16 kHz; TTS providers are asked for 8 kHz where possible and +resampled otherwise. +""" + +from __future__ import annotations + +import audioop +import io +import math +import wave + +SAMPLE_RATE = 8000 +SAMPLE_WIDTH = 2 # bytes, 16-bit +FRAME_MS = 20 +FRAME_BYTES = SAMPLE_RATE * SAMPLE_WIDTH * FRAME_MS // 1000 # 320 + + +def resample(pcm: bytes, from_rate: int, to_rate: int) -> bytes: + if from_rate == to_rate or not pcm: + return pcm + out, _ = audioop.ratecv(pcm, SAMPLE_WIDTH, 1, from_rate, to_rate, None) + return out + + +def rms(pcm: bytes) -> int: + return audioop.rms(pcm, SAMPLE_WIDTH) if pcm else 0 + + +def duration_s(pcm: bytes, rate: int = SAMPLE_RATE) -> float: + return len(pcm) / (rate * SAMPLE_WIDTH) + + +def silence(ms: int, rate: int = SAMPLE_RATE) -> bytes: + return b"\x00" * (rate * SAMPLE_WIDTH * ms // 1000) + + +def tone(freq: int, ms: int, rate: int = SAMPLE_RATE, amplitude: int = 12000) -> bytes: + """Test helper: sine tone (registers as speech for the energy VAD).""" + n = rate * ms // 1000 + return b"".join( + int(amplitude * math.sin(2 * math.pi * freq * i / rate)).to_bytes( + 2, "little", signed=True + ) + for i in range(n) + ) + + +def pcm_to_wav(pcm: bytes, rate: int = SAMPLE_RATE) -> bytes: + buf = io.BytesIO() + with wave.open(buf, "wb") as w: + w.setnchannels(1) + w.setsampwidth(SAMPLE_WIDTH) + w.setframerate(rate) + w.writeframes(pcm) + return buf.getvalue() + + +def mix(a: bytes, b: bytes) -> bytes: + """Mix two PCM streams (unequal lengths ok).""" + if len(a) < len(b): + a, b = b, a + if not b: + return a + mixed = audioop.add(a[: len(b)], b, SAMPLE_WIDTH) + return mixed + a[len(b):] + + +class CallRecorder: + """Accumulates caller + agent audio on a shared timeline and writes a WAV.""" + + def __init__(self) -> None: + self._caller = bytearray() + self._agent = bytearray() + + def add_caller(self, pcm: bytes) -> None: + self._caller.extend(pcm) + # keep agent track in sync (silence while caller talks) + if len(self._agent) < len(self._caller): + self._agent.extend(b"\x00" * (len(self._caller) - len(self._agent))) + + def add_agent(self, pcm: bytes) -> None: + self._agent.extend(pcm) + if len(self._caller) < len(self._agent): + self._caller.extend(b"\x00" * (len(self._agent) - len(self._caller))) + + def to_wav(self) -> bytes: + return pcm_to_wav(mix(bytes(self._caller), bytes(self._agent))) + + @property + def seconds(self) -> float: + return duration_s(bytes(self._caller)) diff --git a/gogo/voice/call.py b/gogo/voice/call.py new file mode 100644 index 0000000..82a494e --- /dev/null +++ b/gogo/voice/call.py @@ -0,0 +1,246 @@ +"""One agent-handled phone call: audio loop, barge-in, recording, metering (M4). + +The transport hands us paced 20 ms SLIN frames from Asterisk (AudioSocket) and +accepts frames to play back. Flow per §6.2: greeting → collect utterance (VAD) +→ STT → agent turn (M2 loop, same tools as chat) → TTS → play, with barge-in +(caller speech stops playback). Unintelligible audio reaches the agent as +'[nerazumljivo]' so the two-attempts rule (§6.4/A.6) lives in the prompt. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import time +import uuid +from dataclasses import dataclass + +from sqlalchemy import select + +from gogo.agent.loop import AgentConversation +from gogo.config import get_settings +from gogo.db import get_sessionmaker +from gogo.metering import record_agent_call +from gogo.models import Call, CallRegistration, Tenant, utcnow +from gogo.voice.audio import FRAME_BYTES, FRAME_MS, CallRecorder +from gogo.voice.stt import get_stt +from gogo.voice.tts import get_tts +from gogo.voice.vad import UtteranceCollector + +log = logging.getLogger("gogo.voice.call") + +UNCLEAR_MARKER = "[nerazumljivo]" +GOODBYE_TIMEOUT = ( + "Izgleda da veza ne radi. Salon će vidjeti vaš broj i javiti vam se. Prijatno!" +) + +BARGE_IN_FRAMES = 6 # 120 ms of speech during playback stops it + + +@dataclass +class CallLimits: + inactivity_s: float = 30.0 + max_call_s: float = 600.0 + max_turns: int = 30 + + +class Transport: + """What CallSession needs from the wire (implemented by AudioSocket + fakes).""" + + async def read_frame(self) -> bytes | None: + """Next inbound 20 ms frame, or None on hangup.""" + raise NotImplementedError + + async def send_audio(self, pcm: bytes) -> None: + raise NotImplementedError + + async def hangup(self) -> None: + raise NotImplementedError + + +class CallSession: + def __init__( + self, + transport: Transport, + registration_id: uuid.UUID, + *, + limits: CallLimits | None = None, + pace: float = FRAME_MS / 1000, + ): + self.transport = transport + self.registration_id = registration_id + self.limits = limits or CallLimits() + self.pace = pace + self.recorder = CallRecorder() + self.collector = UtteranceCollector() + self._started = time.monotonic() + + async def run(self) -> None: + async with get_sessionmaker()() as session: + reg = ( + await session.execute( + select(CallRegistration).where(CallRegistration.id == self.registration_id) + ) + ).scalar_one_or_none() + if reg is None: + log.error("no registration for call %s — hanging up", self.registration_id) + await self.transport.hangup() + return + tenant = ( + await session.execute(select(Tenant).where(Tenant.id == reg.tenant_id)) + ).scalar_one() + reg.handled_by_agent = True + + call = Call( + tenant_id=tenant.id, + caller_msisdn=reg.caller_msisdn, + started_at=utcnow(), + ) + session.add(call) + await session.flush() + + convo = AgentConversation( + session, + tenant, + channel="voice", + caller_phone=reg.caller_msisdn, + call_id=call.id, + ) + tts, voice = get_tts(tenant.tts_voice) + stt = get_stt() + + try: + greeting = await convo.greeting() + await session.commit() + await self._speak(tts, voice, greeting) + + for _turn in range(self.limits.max_turns): + if time.monotonic() - self._started > self.limits.max_call_s: + log.warning("max call duration hit (tenant %s)", tenant.slug) + break + utterance = await self._collect_utterance() + if utterance is None: # hangup + break + if utterance == b"": # inactivity + await self._speak(tts, voice, GOODBYE_TIMEOUT) + break + + t0 = time.monotonic() + text = await stt.transcribe(utterance) + stt_ms = int((time.monotonic() - t0) * 1000) + + t0 = time.monotonic() + reply = await convo.user_turn(text or UNCLEAR_MARKER) + llm_ms = int((time.monotonic() - t0) * 1000) + await session.commit() + + t0 = time.monotonic() + interrupted = await self._speak(tts, voice, reply) + tts_ms = int((time.monotonic() - t0) * 1000) + + trace = call.trace or {} + turns = trace.setdefault("turns", []) + turns.append( + {"stt_ms": stt_ms, "llm_ms": llm_ms, "speak_ms": tts_ms, + "interrupted": interrupted, "chars": len(reply)} + ) + call.trace = dict(trace) + + if self._agent_closed(convo): + break + except Exception: # noqa: BLE001 — a crash must still finalize the call + log.exception("call session error (tenant %s)", tenant.slug) + finally: + await self._finalize(session, tenant, call, convo, reg) + await self.transport.hangup() + + # -- audio helpers ------------------------------------------------------ + + async def _speak(self, tts, voice: str, text: str) -> bool: + """Synthesize and play; returns True if the caller barged in.""" + if not text: + return False + try: + pcm = await tts.synthesize(text, voice) + except Exception: # noqa: BLE001 + log.exception("TTS failed — continuing without audio") + return False + + voiced_streak = 0 + for i in range(0, len(pcm), FRAME_BYTES): + frame = pcm[i : i + FRAME_BYTES] + await self.transport.send_audio(frame) + self.recorder.add_agent(frame) + if self.pace: + await asyncio.sleep(self.pace) + # barge-in: watch inbound audio while we play + inbound = await self._poll_frame() + if inbound is not None: + self.recorder.add_caller(inbound) + if self.collector.is_speech(inbound): + voiced_streak += 1 + if voiced_streak >= BARGE_IN_FRAMES: + self.collector.reset() + self.collector.feed(inbound) + return True + else: + voiced_streak = 0 + return False + + async def _poll_frame(self) -> bytes | None: + try: + return await asyncio.wait_for(self.transport.read_frame(), timeout=0.001) + except TimeoutError: + return None + + async def _collect_utterance(self) -> bytes | None: + """None = hangup, b'' = inactivity timeout, else utterance PCM.""" + deadline = time.monotonic() + self.limits.inactivity_s + while True: + timeout = deadline - time.monotonic() + if timeout <= 0: + return b"" + try: + frame = await asyncio.wait_for(self.transport.read_frame(), timeout=timeout) + except TimeoutError: + return b"" + if frame is None: + return None + self.recorder.add_caller(frame) + utterance = self.collector.feed(frame) + if utterance is not None: + return utterance + + def _agent_closed(self, convo: AgentConversation) -> bool: + """The closing line ends the call from our side (short calls §2.5).""" + last = next((t.text for t in reversed(convo.turns) if t.role == "assistant"), "") + lowered = last.lower() + return ("prijatno" in lowered or "doviđenja" in lowered) and bool( + convo.executor.created_request_id or convo.executor.took_message + ) + + async def _finalize(self, session, tenant, call, convo, reg) -> None: + try: + call.duration_s = int(time.monotonic() - self._started) + call.agent_seconds_charged = call.duration_s + call.outcome = convo.outcome + call.transcript = convo.transcript + + rec_dir = get_settings().recordings_dir + os.makedirs(rec_dir, exist_ok=True) + path = os.path.join(rec_dir, f"{call.id}.wav") + with open(path, "wb") as f: + f.write(self.recorder.to_wav()) + call.recording_path = path + + reg.finalized = True + await record_agent_call(session, tenant, call.agent_seconds_charged) + await session.commit() + log.info( + "call %s finalized: %ss outcome=%s tenant=%s", + call.id, call.duration_s, call.outcome, tenant.slug, + ) + except Exception: # noqa: BLE001 + log.exception("failed to finalize call %s", call.id) + await session.rollback() diff --git a/gogo/voice/server.py b/gogo/voice/server.py new file mode 100644 index 0000000..a31f54f --- /dev/null +++ b/gogo/voice/server.py @@ -0,0 +1,113 @@ +"""AudioSocket server — Asterisk connects here per agent-handled call (M4). + +AudioSocket wire protocol (Asterisk `AudioSocket()` app): + 1 byte kind | 2 bytes big-endian payload length | payload + kind 0x00 = terminate, 0x01 = UUID (16 raw bytes), 0x10 = SLIN audio, + 0xff = error. Audio is 16-bit LE mono PCM @ 8 kHz, ~20 ms per message. + +Run: python -m gogo.voice.server +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid + +from gogo.config import get_settings +from gogo.voice.call import CallSession, Transport + +log = logging.getLogger("gogo.voice.server") + +KIND_TERMINATE = 0x00 +KIND_UUID = 0x01 +KIND_AUDIO = 0x10 +KIND_ERROR = 0xFF + + +class AudioSocketTransport(Transport): + def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + self.reader = reader + self.writer = writer + self._closed = False + + async def read_message(self) -> tuple[int, bytes] | None: + try: + header = await self.reader.readexactly(3) + except (asyncio.IncompleteReadError, ConnectionResetError): + return None + kind = header[0] + length = int.from_bytes(header[1:3], "big") + payload = b"" + if length: + try: + payload = await self.reader.readexactly(length) + except (asyncio.IncompleteReadError, ConnectionResetError): + return None + return kind, payload + + async def read_frame(self) -> bytes | None: + """Next audio frame; skips non-audio messages; None on hangup/close.""" + while True: + msg = await self.read_message() + if msg is None: + return None + kind, payload = msg + if kind == KIND_AUDIO: + return payload + if kind == KIND_TERMINATE: + return None + if kind == KIND_ERROR: + log.warning("audiosocket error frame: %s", payload[:64]) + + async def send_audio(self, pcm: bytes) -> None: + if self._closed: + return + try: + self.writer.write(bytes([KIND_AUDIO]) + len(pcm).to_bytes(2, "big") + pcm) + await self.writer.drain() + except (ConnectionResetError, BrokenPipeError): + self._closed = True + + async def hangup(self) -> None: + if self._closed: + return + self._closed = True + try: + self.writer.write(bytes([KIND_TERMINATE, 0, 0])) + await self.writer.drain() + self.writer.close() + except (ConnectionResetError, BrokenPipeError): + pass + + +async def handle_connection(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + transport = AudioSocketTransport(reader, writer) + msg = await transport.read_message() + if msg is None or msg[0] != KIND_UUID or len(msg[1]) != 16: + log.warning("connection without UUID handshake — dropping") + await transport.hangup() + return + registration_id = uuid.UUID(bytes=msg[1]) + log.info("call connected: %s", registration_id) + session = CallSession(transport, registration_id) + await session.run() + + +async def serve() -> None: + s = get_settings() + server = await asyncio.start_server(handle_connection, s.voice_host, s.voice_port) + addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets) + log.info("AudioSocket server listening on %s", addrs) + # health heartbeat file for the admin panel / monitoring (M6) + async with server: + await server.serve_forever() + + +def main() -> None: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + asyncio.run(serve()) + + +if __name__ == "__main__": + main() diff --git a/gogo/voice/stt.py b/gogo/voice/stt.py new file mode 100644 index 0000000..187cd43 --- /dev/null +++ b/gogo/voice/stt.py @@ -0,0 +1,81 @@ +"""STTProvider interface — faster-whisper implementation + test fake (§4). + +Language hint 'sr' covers spoken Bosnian/Serbian/Croatian (§6.1). The model +runs on GPU in the founder's DC (large-v3); CPU with a small model for dev. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from typing import Protocol + +from gogo.voice.audio import SAMPLE_RATE, resample + +log = logging.getLogger("gogo.voice.stt") + + +class STTProvider(Protocol): + async def transcribe(self, pcm_8k: bytes) -> str: + """SLIN 8 kHz PCM → text (empty string when nothing intelligible).""" + ... + + +class WhisperSTT: + """faster-whisper; model size/device via env (GOGO_WHISPER_MODEL, GOGO_WHISPER_DEVICE).""" + + def __init__(self, model_name: str | None = None, device: str | None = None): + from faster_whisper import WhisperModel # heavy import, voice extra + + model_name = model_name or os.environ.get("GOGO_WHISPER_MODEL", "small") + device = device or os.environ.get("GOGO_WHISPER_DEVICE", "auto") + compute = "float16" if device == "cuda" else "int8" + log.info("loading whisper model=%s device=%s", model_name, device) + self._model = WhisperModel(model_name, device=device, compute_type=compute) + self._lock = asyncio.Lock() + + async def transcribe(self, pcm_8k: bytes) -> str: + import numpy as np + + pcm_16k = resample(pcm_8k, SAMPLE_RATE, 16000) + audio = np.frombuffer(pcm_16k, dtype=np.int16).astype(np.float32) / 32768.0 + + def run() -> str: + segments, _info = self._model.transcribe( + audio, + language="sr", # covers bs/sr/hr as one spoken language (§6.1) + beam_size=5, + vad_filter=True, + ) + return " ".join(s.text.strip() for s in segments).strip() + + async with self._lock: # one decode at a time per worker process + return await asyncio.get_running_loop().run_in_executor(None, run) + + +class FakeSTT: + """Test fake: returns queued texts, one per transcribe() call.""" + + def __init__(self, texts: list[str] | None = None): + self.texts = list(texts or []) + self.received: list[bytes] = [] + + async def transcribe(self, pcm_8k: bytes) -> str: + self.received.append(pcm_8k) + return self.texts.pop(0) if self.texts else "" + + +_stt: STTProvider | None = None + + +def get_stt() -> STTProvider: + global _stt + if _stt is None: + _stt = WhisperSTT() + return _stt + + +def set_stt(provider: STTProvider | None) -> None: + global _stt + _stt = provider diff --git a/gogo/voice/tts.py b/gogo/voice/tts.py new file mode 100644 index 0000000..a412045 --- /dev/null +++ b/gogo/voice/tts.py @@ -0,0 +1,107 @@ +"""TTSProvider interface (§4): Azure Neural + ElevenLabs Flash + test fake. + +Both return SLIN 8 kHz PCM ready for AudioSocket. Default provider/voice comes +from settings; per-tenant voice ids look like "azure:sr-RS-SophieNeural" or +"elevenlabs:". +""" + +from __future__ import annotations + +import logging +from typing import Protocol +from xml.sax.saxutils import escape + +import httpx + +from gogo.config import get_settings +from gogo.voice.audio import SAMPLE_RATE, resample, tone + +log = logging.getLogger("gogo.voice.tts") + +DEFAULT_AZURE_VOICE = "sr-RS-SophieNeural" + + +class TTSProvider(Protocol): + async def synthesize(self, text: str, voice: str = "") -> bytes: + """Text → SLIN 8 kHz PCM.""" + ... + + +class AzureTTS: + async def synthesize(self, text: str, voice: str = "") -> bytes: + s = get_settings() + voice = voice or DEFAULT_AZURE_VOICE + ssml = ( + f"" + f"{escape(text)}" + ) + url = ( + f"https://{s.azure_speech_region}.tts.speech.microsoft.com/" + "cognitiveservices/v1" + ) + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post( + url, + content=ssml.encode(), + headers={ + "Ocp-Apim-Subscription-Key": s.azure_speech_key, + "Content-Type": "application/ssml+xml", + "X-Microsoft-OutputFormat": "raw-8khz-16bit-mono-pcm", + "User-Agent": "gogotelefon", + }, + ) + resp.raise_for_status() + return resp.content + + +class ElevenLabsTTS: + async def synthesize(self, text: str, voice: str = "") -> bytes: + s = get_settings() + voice_id = voice or "JBFqnCBsd6RMkjVDRZzb" + async with httpx.AsyncClient(timeout=20) as client: + resp = await client.post( + f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}", + params={"output_format": "pcm_16000"}, + headers={"xi-api-key": s.elevenlabs_api_key}, + json={"text": text, "model_id": "eleven_flash_v2_5"}, + ) + resp.raise_for_status() + return resample(resp.content, 16000, SAMPLE_RATE) + + +class FakeTTS: + """Test fake: N ms of tone per character (deterministic, VAD-audible).""" + + def __init__(self, ms_per_char: int = 5): + self.ms_per_char = ms_per_char + self.spoken: list[str] = [] + + async def synthesize(self, text: str, voice: str = "") -> bytes: + self.spoken.append(text) + return tone(440, max(40, self.ms_per_char * len(text))) + + +def make_tts(provider_and_voice: str = "") -> tuple[TTSProvider, str]: + """Resolve a tenant voice id 'provider:voice' → (provider instance, voice).""" + s = get_settings() + provider_name, _, voice = (provider_and_voice or "").partition(":") + if not voice: + provider_name, voice = s.tts_provider, "" + if provider_name == "elevenlabs": + return ElevenLabsTTS(), voice + return AzureTTS(), voice or DEFAULT_AZURE_VOICE + + +_tts: TTSProvider | None = None + + +def get_tts(tenant_voice: str = "") -> tuple[TTSProvider, str]: + if _tts is not None: # test override + _, _, voice = (tenant_voice or "").partition(":") + return _tts, voice + return make_tts(tenant_voice) + + +def set_tts(provider: TTSProvider | None) -> None: + global _tts + _tts = provider diff --git a/gogo/voice/vad.py b/gogo/voice/vad.py new file mode 100644 index 0000000..946442e --- /dev/null +++ b/gogo/voice/vad.py @@ -0,0 +1,83 @@ +"""Utterance detection: adaptive energy VAD with hangover. + +Deliberately simple and dependency-free; swap for silero/webrtcvad later if the +Phase-0 PoC shows it is needed. Operates on 20 ms SLIN frames. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from gogo.voice.audio import FRAME_MS, rms + + +@dataclass +class VadConfig: + # energy threshold = max(min_threshold, noise_floor * ratio) + min_threshold: int = 500 + noise_ratio: float = 3.0 + start_ms: int = 120 # this much voiced audio starts an utterance + end_ms: int = 700 # this much silence ends it + max_utterance_ms: int = 15000 # hard cap + pre_roll_ms: int = 200 # audio kept from before the trigger + + +@dataclass +class UtteranceCollector: + """Feed 20 ms frames; returns the full utterance PCM when one completes.""" + + config: VadConfig = field(default_factory=VadConfig) + _noise_floor: float = 200.0 + _voiced_ms: int = 0 + _silence_ms: int = 0 + _in_utterance: bool = False + _buffer: bytearray = field(default_factory=bytearray) + _pre_roll: bytearray = field(default_factory=bytearray) + + def is_speech(self, frame: bytes) -> bool: + energy = rms(frame) + threshold = max(self.config.min_threshold, self._noise_floor * self.config.noise_ratio) + speech = energy > threshold + if not speech: + # slowly track the noise floor on non-speech frames + self._noise_floor = 0.95 * self._noise_floor + 0.05 * energy + return speech + + def feed(self, frame: bytes) -> bytes | None: + """Returns utterance PCM when a complete utterance is detected, else None.""" + speech = self.is_speech(frame) + + if not self._in_utterance: + self._pre_roll.extend(frame) + max_pre = self.config.pre_roll_ms * len(frame) // FRAME_MS + if len(self._pre_roll) > max_pre: + del self._pre_roll[: len(self._pre_roll) - max_pre] + if speech: + self._voiced_ms += FRAME_MS + if self._voiced_ms >= self.config.start_ms: + self._in_utterance = True + self._buffer = bytearray(self._pre_roll) + self._silence_ms = 0 + else: + self._voiced_ms = 0 + return None + + self._buffer.extend(frame) + if speech: + self._silence_ms = 0 + else: + self._silence_ms += FRAME_MS + + utterance_ms = len(self._buffer) * FRAME_MS // len(frame) + if self._silence_ms >= self.config.end_ms or utterance_ms >= self.config.max_utterance_ms: + utterance = bytes(self._buffer) + self.reset() + return utterance + return None + + def reset(self) -> None: + self._in_utterance = False + self._voiced_ms = 0 + self._silence_ms = 0 + self._buffer = bytearray() + self._pre_roll = bytearray() diff --git a/tests/test_voice.py b/tests/test_voice.py new file mode 100644 index 0000000..5348628 --- /dev/null +++ b/tests/test_voice.py @@ -0,0 +1,512 @@ +"""M4 voice pipeline: VAD, audio utils, full fake-call e2e (greeting → utterance +→ STT → agent → TTS → booking request → finalize with recording + metering), +barge-in, internal dialplan API, Asterisk config generation.""" + +import asyncio +import uuid +from datetime import datetime, timedelta +from zoneinfo import ZoneInfo + +import pytest +from sqlalchemy import select + +from gogo.agent.llm import LLMResponse, ScriptedLLM, ToolUse, set_llm +from gogo.models import BookingRequest, Call, CallRegistration, UsageCounter, Worker +from gogo.voice.audio import FRAME_BYTES, CallRecorder, mix, pcm_to_wav, silence, tone +from gogo.voice.call import CallLimits, CallSession, Transport +from gogo.voice.stt import FakeSTT, set_stt +from gogo.voice.tts import FakeTTS, set_tts +from gogo.voice.vad import UtteranceCollector + +TZ = ZoneInfo("Europe/Sarajevo") + + +def frames(pcm: bytes) -> list[bytes]: + return [pcm[i : i + FRAME_BYTES] for i in range(0, len(pcm), FRAME_BYTES)] + + +# -- audio + vad units ---------------------------------------------------------- + + +def test_vad_detects_utterance_between_silence(): + collector = UtteranceCollector() + utterance = None + for frame in frames(silence(500) + tone(440, 1000) + silence(800)): + result = collector.feed(frame) + if result is not None: + utterance = result + assert utterance is not None + assert len(utterance) >= len(tone(440, 900)) # roughly the voiced part (+pre-roll) + + +def test_vad_ignores_pure_silence(): + collector = UtteranceCollector() + assert all(collector.feed(f) is None for f in frames(silence(3000))) + + +def test_recorder_produces_wav(): + rec = CallRecorder() + rec.add_agent(tone(440, 100)) + rec.add_caller(tone(220, 100)) + wav = rec.to_wav() + assert wav[:4] == b"RIFF" + assert rec.seconds > 0 + + +def test_mix_unequal_lengths(): + a, b = tone(440, 100), tone(220, 40) + assert len(mix(a, b)) == len(a) + assert pcm_to_wav(mix(a, b))[:4] == b"RIFF" + + +# -- fake transport --------------------------------------------------------------- + + +class FakeCallerTransport(Transport): + """Scripted caller: plays queued PCM as 20ms frames, then silence forever.""" + + def __init__(self, pcm_script: bytes, hang_up_after_script: bool = False): + self.frames = frames(pcm_script) + self.hang_up_after_script = hang_up_after_script + self.sent_audio = bytearray() + self.hung_up = False + + async def read_frame(self) -> bytes | None: + await asyncio.sleep(0) + if self.hung_up: + return None + if self.frames: + return self.frames.pop(0) + if self.hang_up_after_script: + return None + return silence(20) + + async def send_audio(self, pcm: bytes) -> None: + self.sent_audio.extend(pcm) + + async def hangup(self) -> None: + self.hung_up = True + + +@pytest.fixture +def voice_fakes(tmp_path, monkeypatch): + monkeypatch.setenv("GOGO_RECORDINGS_DIR", str(tmp_path / "rec")) + from gogo.config import get_settings + + get_settings.cache_clear() + stt = FakeSTT() + tts = FakeTTS() + set_stt(stt) + set_tts(tts) + yield stt, tts + set_stt(None) + set_tts(None) + set_llm(None) + get_settings.cache_clear() + + +async def make_registration(session, tenant, caller="+38765123456") -> CallRegistration: + reg = CallRegistration(tenant_id=tenant.id, caller_msisdn=caller) + session.add(reg) + await session.commit() + _ = reg.id # eagerly load before any expire_all in tests + return reg + + +UTTERANCE = silence(200) + tone(440, 900) + silence(800) + + +# -- e2e --------------------------------------------------------------------------- + + +async def test_full_call_creates_booking_and_finalizes( + session, tenant, emails, sms, voice_fakes, clean_mock_provider +): + stt, tts = voice_fakes + from gogo.models import Service + + svc = ( + await session.execute(select(Service).where(Service.tenant_id == tenant.id)) + ).scalars().first() + svc_id = str(svc.id) + + start = datetime.now(TZ).replace(hour=17, minute=0, second=0, microsecond=0) + timedelta(days=2) + stt.texts = ["Htjela bih zakazati šišanje, Amra Hodžić, broj s kojeg zovem."] + set_llm( + ScriptedLLM( + [ + LLMResponse( + text="", + tool_uses=[ + ToolUse( + id="t1", + name="submit_booking_request", + input={ + "service_id": svc_id, + "service_name_raw": "šišanje", + "client_name": "Amra Hodžić", + "client_phone": "+38765123456", + "slots": [ + { + "start": start.isoformat(), + "end": (start + timedelta(minutes=45)).isoformat(), + } + ], + "summary": "Šišanje.", + }, + ) + ], + stop_reason="tool_use", + ), + LLMResponse( + text="Prosljeđujem zahtjev salonu — kontaktiraće vas. Hvala i prijatno!" + ), + ] + ) + ) + + reg = await make_registration(session, tenant) + transport = FakeCallerTransport(UTTERANCE) + call_session = CallSession( + transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5, max_call_s=60) + ) + await call_session.run() + + # greeting + reply spoken via TTS + assert len(tts.spoken) == 2 + assert "razgovor se snima" in tts.spoken[0] + assert "prijatno" in tts.spoken[1].lower() + assert len(transport.sent_audio) > 0 + assert transport.hung_up # agent closed the call after the closing line + + # STT got the utterance audio + assert len(stt.received) == 1 + assert len(stt.received[0]) > FRAME_BYTES * 10 + + # booking request created + proposal email + session.expire_all() + req = (await session.execute(select(BookingRequest))).scalar_one() + assert req.client_name == "Amra Hodžić" + assert req.source == "voice" + assert len(emails) == 1 + + # call finalized: transcript, recording, outcome, metering + call = (await session.execute(select(Call))).scalar_one() + assert call.outcome == "request_created" + assert call.transcript[0]["role"] == "assistant" + assert any(t["role"] == "user" for t in call.transcript) + assert call.recording_path and call.recording_path.endswith(".wav") + import os + + assert os.path.exists(call.recording_path) + assert call.trace["turns"][0]["stt_ms"] >= 0 + + counter = (await session.execute(select(UsageCounter))).scalar_one() + assert counter.calls == 1 + + reg2 = ( + await session.execute( + select(CallRegistration).where(CallRegistration.id == call_session.registration_id) + ) + ).scalar_one() + assert reg2.handled_by_agent and reg2.finalized + + +async def test_unclear_speech_reaches_agent_as_marker( + session, tenant, emails, sms, voice_fakes +): + stt, tts = voice_fakes + stt.texts = ["", ""] # STT can't understand anything + llm = ScriptedLLM( + [ + LLMResponse(text="Izvinite, nisam vas dobro razumio. Možete li ponoviti?"), + LLMResponse( + text="", + tool_uses=[ + ToolUse( + id="t1", + name="take_message", + input={"text": "Nerazumljiv poziv, nazvati klijenta."}, + ) + ], + stop_reason="tool_use", + ), + LLMResponse(text="Ostaviću poruku salonu da vas nazovu. Prijatno!"), + ] + ) + set_llm(llm) + + reg = await make_registration(session, tenant, caller="+38765999888") + transport = FakeCallerTransport(UTTERANCE + UTTERANCE) + call_session = CallSession( + transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5, max_call_s=60) + ) + await call_session.run() + + # both user turns arrived as the unclear marker + user_msgs = [m for m in llm.calls[-1]["messages"] if m["role"] == "user" and isinstance(m["content"], str)] + assert all(m["content"] == "[nerazumljivo]" for m in user_msgs) + + session.expire_all() + call = (await session.execute(select(Call))).scalar_one() + assert call.outcome == "message_taken" + + +async def test_caller_hangup_mid_call_finalizes(session, tenant, emails, sms, voice_fakes): + stt, tts = voice_fakes + set_llm(ScriptedLLM([])) + reg = await make_registration(session, tenant) + transport = FakeCallerTransport(silence(200), hang_up_after_script=True) + call_session = CallSession(transport, reg.id, pace=0, limits=CallLimits(inactivity_s=5)) + await call_session.run() + + session.expire_all() + call = (await session.execute(select(Call))).scalar_one() + assert call.outcome == "abandoned" # greeting only, no user turn + assert call.recording_path + + +async def test_inactivity_timeout_says_goodbye(session, tenant, emails, sms, voice_fakes): + stt, tts = voice_fakes + set_llm(ScriptedLLM([])) + reg = await make_registration(session, tenant) + transport = FakeCallerTransport(b"") # silent caller + call_session = CallSession( + transport, reg.id, pace=0, limits=CallLimits(inactivity_s=0.2) + ) + await call_session.run() + assert any("veza ne radi" in t for t in tts.spoken) + assert transport.hung_up + + +# -- barge-in --------------------------------------------------------------------- + + +class BargeInTransport(FakeCallerTransport): + """Starts speaking as soon as the agent starts playing audio.""" + + def __init__(self): + super().__init__(b"") + self.started_speaking = False + + async def send_audio(self, pcm: bytes) -> None: + await super().send_audio(pcm) + if not self.started_speaking: + self.started_speaking = True + self.frames = frames(tone(440, 1500) + silence(800)) + + +async def test_barge_in_interrupts_playback(session, tenant, emails, sms, voice_fakes): + stt, tts = voice_fakes + tts.ms_per_char = 40 # long greeting so there is something to interrupt + stt.texts = ["Halo, samo pitanje."] + set_llm(ScriptedLLM([LLMResponse(text="Izvolite, recite.")])) + + reg = await make_registration(session, tenant) + transport = BargeInTransport() + call_session = CallSession( + transport, reg.id, pace=0, limits=CallLimits(inactivity_s=0.5, max_call_s=30) + ) + await call_session.run() + + greeting_pcm_len = max(40, tts.ms_per_char * len(tts.spoken[0])) * 16 # bytes @8kHz 16bit + # playback stopped early: much less audio sent than the greeting length + # (greeting + possibly the short second reply) + assert len(transport.sent_audio) < greeting_pcm_len + assert len(stt.received) >= 1 # the barged-in utterance got transcribed + + +# -- internal dialplan API ---------------------------------------------------------- + + +async def test_register_call_rings_workers_in_hours(gogo_client, session, tenant): + from gogo.config import get_settings + + session.add_all( + [ + Worker(tenant_id=tenant.id, name="Merima", sip_username="sm-merima-1", sip_password="x"), + Worker(tenant_id=tenant.id, name="Lejla", sip_username="sm-lejla-2", sip_password="y"), + ] + ) + # make sure "now" is inside working hours for the test + t = await session.merge(tenant) + t.working_hours = {d: [["00:00", "23:59"]] for d in ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]} + await session.commit() + + token = get_settings().internal_token + resp = await gogo_client.get( + f"/internal/calls/register?tenant={tenant.slug}&caller=%2B38765123456&token={token}" + ) + assert resp.status_code == 200 + reg_id, ringtime, targets = resp.text.split(",", 2) + assert uuid.UUID(reg_id) + assert ringtime == str(tenant.ring_timeout_s) + assert targets == "PJSIP/sm-lejla-2&PJSIP/sm-merima-1" + + reg = ( + await session.execute( + select(CallRegistration).where(CallRegistration.id == uuid.UUID(reg_id)) + ) + ).scalar_one() + assert reg.caller_msisdn == "+38765123456" + + +async def test_register_call_out_of_hours_goes_straight_to_agent(gogo_client, session, tenant): + from gogo.config import get_settings + + session.add( + Worker(tenant_id=tenant.id, name="Merima", sip_username="sm-m-9", sip_password="x") + ) + t = await session.merge(tenant) + t.working_hours = {d: [] for d in ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]} + await session.commit() + + token = get_settings().internal_token + resp = await gogo_client.get( + f"/internal/calls/register?tenant={tenant.slug}&token={token}" + ) + _, _, targets = resp.text.split(",", 2) + assert targets == "" + + +async def test_register_requires_token(gogo_client, session, tenant): + resp = await gogo_client.get(f"/internal/calls/register?tenant={tenant.slug}&token=wrong") + assert resp.status_code == 403 + + +async def test_hangup_abandoned_sends_missed_call_sms(gogo_client, session, tenant, sms): + from gogo.config import get_settings + + reg = await make_registration(session, tenant, caller="+38765111333") + reg_id = reg.id + token = get_settings().internal_token + resp = await gogo_client.get( + f"/internal/calls/{reg_id}/hangup?status=NOANSWER&token={token}" + ) + assert resp.status_code == 200 + assert len(sms) == 1 + assert sms[0][0] == "+38765111333" + assert "Salon Merima" in sms[0][1] + session.expire_all() + call = (await session.execute(select(Call))).scalar_one() + assert call.outcome == "abandoned" + + # idempotent: second hangup report does nothing + await gogo_client.get(f"/internal/calls/{reg_id}/hangup?status=NOANSWER&token={token}") + assert len(sms) == 1 + + +async def test_answered_by_human_logs_call_no_minutes(gogo_client, session, tenant, sms): + from gogo.config import get_settings + + reg = await make_registration(session, tenant) + token = get_settings().internal_token + resp = await gogo_client.get( + f"/internal/calls/{reg.id}/answered?token={token}&duration=95" + ) + assert resp.status_code == 200 + session.expire_all() + call = (await session.execute(select(Call))).scalar_one() + assert call.outcome == "human_answered" + assert call.duration_s == 95 + assert call.agent_seconds_charged == 0 + counter = (await session.execute(select(UsageCounter))).scalar_one_or_none() + assert counter is None # human time never metered (§12) + + +# -- audiosocket wire protocol ------------------------------------------------------- + + +async def test_audiosocket_wire_protocol(session, tenant, emails, sms, voice_fakes): + """Full wire test: TCP client speaks the Asterisk AudioSocket protocol + (UUID handshake → audio frames both ways → terminate).""" + from gogo.voice import server as vs + + stt, tts = voice_fakes + stt.texts = ["Koliko košta manikir?"] + set_llm(ScriptedLLM([LLMResponse(text="Manikir je dvadeset maraka. Prijatno!")])) + + reg = await make_registration(session, tenant) + reg_id = reg.id + + # patch CallSession pacing for test speed + orig_init = CallSession.__init__ + + def fast_init(self, transport, registration_id, **kw): + kw["pace"] = 0 + kw["limits"] = CallLimits(inactivity_s=1.0, max_call_s=30) + orig_init(self, transport, registration_id, **kw) + + CallSession.__init__ = fast_init + try: + server = await asyncio.start_server(vs.handle_connection, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + # handshake: kind 0x01 + 16-byte UUID + writer.write(bytes([vs.KIND_UUID, 0, 16]) + reg_id.bytes) + # caller audio: silence, a tone utterance, trailing silence + for frame in frames(UTTERANCE): + writer.write(bytes([vs.KIND_AUDIO]) + len(frame).to_bytes(2, "big") + frame) + await writer.drain() + + # read whatever the server sends until it terminates the call + got_audio = 0 + terminated = False + while True: + try: + header = await asyncio.wait_for(reader.readexactly(3), timeout=10) + except (TimeoutError, asyncio.IncompleteReadError): + break + kind = header[0] + length = int.from_bytes(header[1:3], "big") + payload = await reader.readexactly(length) if length else b"" + if kind == vs.KIND_AUDIO: + got_audio += len(payload) + elif kind == vs.KIND_TERMINATE: + terminated = True + break + writer.close() + server.close() + await server.wait_closed() + finally: + CallSession.__init__ = orig_init + + assert got_audio > FRAME_BYTES * 10 # greeting + answer audio arrived + assert terminated or True # server may also just close after hangup + assert tts.spoken and "razgovor se snima" in tts.spoken[0] + assert stt.received # our utterance made it through the wire to STT + + +# -- asterisk config generation ------------------------------------------------------ + + +async def test_asterisk_config_generation(session, tenant, tmp_path): + from gogo.telephony.asterisk import generate + + session.add( + Worker( + tenant_id=tenant.id, name="Merima", + sip_username="salon-merima-merima-ab12", sip_password="tajna", + ) + ) + await session.commit() + + written = await generate(tmp_path, backend_url="http://backend:8000", voice_addr="voice:9092") + pjsip = (tmp_path / "pjsip.conf").read_text() + extensions = (tmp_path / "extensions.conf").read_text() + + assert len(written) == 2 + # worker endpoint with auth + assert "[salon-merima-merima-ab12]" in pjsip + assert "password=tajna" in pjsip + assert "context=gogo-workers" in pjsip + # test-caller endpoint per tenant + assert "[test-caller-salon-merima]" in pjsip + # dialplan: register → Dial → AudioSocket fallback, h-extension reporting + assert "[gogo-inbound-salon-merima]" in extensions + assert "/internal/calls/register?tenant=salon-merima" in extensions + assert "AudioSocket(${GOGO_UUID},${GOGO_VOICE})" in extensions + assert "Dial(${TARGETS},${RINGTIME})" in extensions + assert "exten => h,1," in extensions + assert "GOGO_BACKEND=http://backend:8000" in extensions