"""Inbound voice-call capture (S12) — call a number, the agent forges an estimate. Flow (all reuse; no new business logic): 1. Twilio routes an inbound call to ``POST /api/voice/incoming``. We answer with TwiML: a short greeting then ````, which posts the ``RecordingUrl`` to ``POST /api/voice/recording`` when the caller hangs up. 2. The recording webhook downloads the ``.wav``/``.mp3``, transcribes it (Audio role — Nemotron Omni on the Best Stack, Cohere Transcribe on the Private Stack, same ``transcribe_audio`` resolution as the mic button), forges an estimate, and saves it as a **DRAFT** under ``account_id="demo"`` (ADR-0013). It then ````s the spoken total and ````s the caller's reply (Tier A — a conversation). 3. Each reply hits ``POST /api/voice/refine`` with Twilio's own ``SpeechResult`` transcript. "Done/no" → recalc, persist, text the PDF, end the call. Otherwise the spoken edit runs through the SAME ``chat_about_estimate`` ops (add / remove / change) the desktop chat uses, the new total is read back, and we ```` again. Per-call state is held under the Twilio ``CallSid`` (in-process, demo-scoped). Honesty (ADR-0004, ADR-0013): the estimate's numbers all come from the catalog + ``recalc`` (Facts-from-Tools); the call produces a draft a human approves later. On a call the agent runs to completion without the interactive Agent Pause (``forge_estimate`` is the non-streaming path — a missing price is auto-flagged in the trace, never blocks). The public base URL is read from ``FF_PUBLIC_BASE_URL`` (the ngrok/cloudflared tunnel), so Twilio can fetch the recording-action URL and the PDF media URL. SMS reuses the S10 ``send_estimate`` SMS provider and the tokenized PDF registry — nothing new is sent. """ import os from collections.abc import Callable from xml.sax.saxutils import escape # Spoken voice for every . Amazon Polly Neural voices (rendered by Twilio at no extra # cost beyond standard call rates) sound far more natural than the default. Override with # FF_VOICE if you prefer another (e.g. Polly.Joanna-Neural, Polly.Stephen-Neural). VOICE = os.environ.get("FF_VOICE", "Polly.Matthew-Neural") def _say(message: str) -> str: """A with the configured natural voice.""" return f'{escape(message)}' def public_base_url() -> str: """The tunnel's public base URL (FF_PUBLIC_BASE_URL), trailing slash stripped, or ''.""" return os.environ.get("FF_PUBLIC_BASE_URL", "").rstrip("/") def _action_url(path: str, base_url: str | None = None) -> str: """An absolute URL on the public base when known, else a relative path (Twilio resolves a relative against the request host).""" base = (base_url if base_url is not None else public_base_url()).rstrip("/") return f"{base}{path}" if base else path def greeting_twiml(base_url: str | None = None) -> str: """Answer an inbound call: greet, then record the caller's job description. ```` posts the RecordingUrl to /api/voice/recording on hang-up (or after the silence timeout). ``playBeep`` cues the caller; ``maxLength`` caps a runaway call. """ action = _action_url("/api/voice/recording", base_url) return ( '\n' "" + _say( "Welcome to Quillwright. After the beep, describe the job — the parts you " "used and the labor — then stop talking. I'll forge an estimate, read it back, " "and you can tell me what to change." ) + f'' + _say("I didn't catch a recording. Goodbye.") + "" ) def _say_response(message: str) -> str: """A bare spoken TwiML response (no recording).""" return f'\n{_say(message)}' # --- Conversational refine loop (Tier A): after forging, the agent reads the total and # keeps the call open, asking "anything else?" via . Each reply # is a new /api/voice/refine turn that runs the SAME chat_about_estimate ops (add / # remove / change), so Facts-from-Tools holds — the agent only READS totals that recalc # produced. State is held per Twilio CallSid (in-process, demo-scoped, like pairing). --- # call_sid -> {"rows": list[dict], "job_title": str, "tax_rate": float, "from_number": str} _CALLS: dict[str, dict] = {} # Phrases that end the conversation (caller says they're done). _DONE_WORDS = ( "no", "nope", "nothing", "that's it", "thats it", "done", "all set", "good", "send it", ) def _call_state(call_sid: str) -> dict | None: return _CALLS.get(call_sid) def reset_calls() -> None: """Drop all in-flight call + job state (tests).""" _CALLS.clear() _JOBS.clear() def _is_done(speech: str) -> bool: s = (speech or "").strip().lower() if not s: return False return any(w in s for w in _DONE_WORDS) def _ask_twiml(message: str, base_url: str | None = None) -> str: """Speak `message`, then the caller's spoken reply to /api/voice/refine. If they stay silent, end politely (the Gather falls through to the closing Say).""" action = _action_url("/api/voice/refine", base_url) return ( '\n' "" + f'' + _say(message) + "" + _say("I didn't catch that — I'll text you what I have. Goodbye.") + "" ) def _download_recording(url: str) -> str: """Fetch a Twilio RecordingUrl to a local temp file. Twilio serves the media at ``.wav`` (a safer container for Omni than the browser's webm). Auth with the standard Twilio creds when present (recordings on a real account are protected).""" import tempfile import time import requests # already a core dep media_url = url if url.endswith((".wav", ".mp3")) else f"{url}.wav" auth = None sid, token = os.environ.get("TWILIO_ACCOUNT_SID"), os.environ.get("TWILIO_AUTH_TOKEN") if sid and token: auth = (sid, token) # Twilio posts the recording webhook the instant recording ends, but the media file # is often not encoded/available for a beat — an immediate GET 404s/403s. Retry a few # times with a short backoff so the not-ready race doesn't fail the call. last = None for attempt in range(5): resp = requests.get(media_url, auth=auth, timeout=20) if resp.status_code == 200 and resp.content: with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: tmp.write(resp.content) return tmp.name last = resp if resp.status_code not in (404, 403, 401): break # a different error won't fix itself by waiting time.sleep(1.5) if last is not None: last.raise_for_status() raise RuntimeError(f"could not fetch recording at {media_url}") def _send_sms(*, to: str, body: str, media_url: str) -> dict: """Text the caller via Twilio (lazy import — same optional [send] dep as S10).""" from twilio.rest import Client # noqa: PLC0415 — lazy: optional dep, not in the Space client = Client(os.environ["TWILIO_ACCOUNT_SID"], os.environ["TWILIO_AUTH_TOKEN"]) msg = client.messages.create( to=to, from_=os.environ["FF_SEND_FROM"], body=body, media_url=[media_url] if media_url else None, ) return {"sid": msg.sid} def _rows_from_est(est: dict) -> list[dict]: """The editable rows (no subtotal/source) the chat ops + PDF renderer expect.""" return [ { "description": li["description"], "quantity": li["quantity"], "unit": li["unit"], "rate": li["rate"], } for li in est["line_items"] ] def _summary(est: dict) -> str: n = len(est["line_items"]) items = "item" if n == 1 else "items" return f"{n} {items}, totaling {est['total']:.2f} dollars" def _text_pdf(*, rows, job_title, tax_rate, total, n, from_number, base, sms) -> bool: """Render + register the PDF and text it to the caller. Best-effort: returns True on a send, False on any failure (the draft is already saved either way).""" from quillwright.api.pdf_links import public_pdf_url, register_pdf from quillwright.api.send import _render_pdf_bytes if not from_number: return False try: pdf_bytes = _render_pdf_bytes(rows, job_title=job_title, tax_rate=tax_rate) token = register_pdf(pdf_bytes) media_url = public_pdf_url(token, base_url=base or "") items = "item" if n == 1 else "items" sms( to=from_number, body=( f"Your Quillwright estimate: {n} {items}, total ${total:.2f}. " "AI-generated draft — review before accepting." ), media_url=media_url, ) return True except Exception: # noqa: BLE001 — texting is best-effort; the draft is saved return False def handle_recording( *, recording_url: str, from_number: str, call_sid: str = "default", download: Callable[[str], str] | None = None, transcribe: Callable[[str], dict] | None = None, sms: Callable | None = None, base_url: str | None = None, ) -> dict: """Transcribe the recording, forge + save a draft estimate, then ASK the caller if they want to change anything (the conversational refine loop — Tier A). Returns ``{"estimate": , "twiml": , "transcript": str}``. The PDF is NOT texted here — it goes out when the caller says they're done (see ``handle_refine``). Per-call state is held under ``call_sid``. Side-effects (download / SMS) are injectable so tests need no network or twilio. """ from quillwright.api.estimate import estimate_store, forge_estimate from quillwright.api.transcribe import transcribe_audio download = download or _download_recording transcribe = transcribe or (lambda path: transcribe_audio(path)) base = (base_url if base_url is not None else public_base_url()).rstrip("/") path = download(recording_url) transcript = (transcribe(path) or {}).get("transcript", "").strip() if not transcript: return { "estimate": None, "transcript": "", "twiml": _say_response( "Sorry, I couldn't make out the job from that recording. " "Please call back and describe the parts and labor after the beep." ), } forged = forge_estimate(transcript, trade="hvac") est = forged.get("estimate") if est is None or not est.get("line_items"): return { "estimate": None, "transcript": transcript, "twiml": _say_response( "I heard the job but couldn't build an estimate from it. " "I've made a note — please call back with the parts and labor." ), } # Hold the rows for this call so refine turns edit the same estimate (forge_estimate # already auto-saved a DRAFT — ADR-0013; this is the same store the desktop reads). _CALLS[call_sid] = { "rows": _rows_from_est(est), "job_title": est["job_title"], "tax_rate": est["tax_rate"], "from_number": from_number, } estimate_store() # touch so a misconfigured store surfaces in logs spoken = ( f"Done. I forged an estimate with {_summary(est)}. " "Want to add or change anything, or should I text it to you?" ) return {"estimate": est, "transcript": transcript, "twiml": _ask_twiml(spoken, base)} # --- Async job pattern: forge+transcribe take ~tens of seconds (model load + brain), # far over Twilio's ~15s webhook timeout. So the recording webhook kicks the work off # on a background thread and returns a holding response immediately; Twilio is parked on # a + to /api/voice/status, which polls until the job finishes. Each # webhook response stays well under the timeout. --- # call_sid -> {"status": "working"|"done"|"error", "twiml": } _JOBS: dict[str, dict] = {} def _hold_twiml(message: str, base_url: str | None = None) -> str: """Speak a short status line, pause, then redirect to /api/voice/status to poll again.""" action = _action_url("/api/voice/status", base_url) return ( '\n' "" + _say(message) + '' + f'{escape(action)}' + "" ) def start_recording_job( *, recording_url: str, from_number: str, call_sid: str, base_url: str | None = None ) -> str: """Kick the forge off on a background thread; return holding TwiML immediately. The webhook never blocks on the slow work (model load + brain).""" import threading base = (base_url if base_url is not None else public_base_url()).rstrip("/") _JOBS[call_sid] = {"status": "working", "twiml": None} def _work(): try: result = handle_recording( recording_url=recording_url, from_number=from_number, call_sid=call_sid, base_url=base, ) _JOBS[call_sid] = {"status": "done", "twiml": result["twiml"]} except Exception as exc: # noqa: BLE001 — surface as an error status, not a crash print(f"[quillwright] voice forge job failed: {exc}", flush=True) _JOBS[call_sid] = { "status": "error", "twiml": _say_response( "Sorry, I couldn't build that estimate. Please call back and try again." ), } threading.Thread(target=_work, daemon=True).start() return _hold_twiml("Got it. I'm forging your estimate now — this takes a moment.", base) def handle_status(*, call_sid: str, base_url: str | None = None) -> str: """Poll the background forge: still working → hold + redirect again; done → the ask (or error) TwiML the job produced. Unknown call → polite fallback.""" base = (base_url if base_url is not None else public_base_url()).rstrip("/") job = _JOBS.get(call_sid) if job is None: return _say_response( "Sorry, I lost track of that estimate. Please call back to start again." ) if job["status"] == "working": return _hold_twiml("Still working on it — just a few more seconds.", base) # done or error: hand back the prepared TwiML and clear the job marker. _JOBS.pop(call_sid, None) return job["twiml"] def handle_refine( *, call_sid: str, speech_result: str, base_url: str | None = None, sms: Callable | None = None, ) -> dict: """One caller turn in the refine loop. If they're done, text the PDF and end; else apply the spoken edit through the SAME chat_about_estimate ops (Facts-from-Tools — the catalog owns every price) and ask again. Returns ``{"estimate": , "twiml": str}``. ``sms`` is injectable for tests. """ from quillwright.api.chat import chat_about_estimate from quillwright.api.estimate import save_estimate_record sms = sms or _send_sms base = (base_url if base_url is not None else public_base_url()).rstrip("/") state = _CALLS.get(call_sid) if state is None: # Lost the thread (server restart / stale call) — fail politely, don't crash. return { "estimate": None, "twiml": _say_response( "Sorry, I lost track of that estimate. Please call back to start again." ), } rows = state["rows"] job_title, tax_rate = state["job_title"], state["tax_rate"] # Caller signalled they're finished → recalc to authoritative numbers, persist the # final draft, text the PDF, and end the call. if _is_done(speech_result): from quillwright.api.recalc import recalc_estimate est = recalc_estimate(rows, job_title=job_title, tax_rate=tax_rate) save_estimate_record(rows, job_title, tax_rate, thread=[]) # update the saved draft n = len(est["line_items"]) sent = _text_pdf( rows=rows, job_title=job_title, tax_rate=tax_rate, total=est["total"], n=n, from_number=state.get("from_number", ""), base=base, sms=sms, ) _CALLS.pop(call_sid, None) tail = ( "I've texted you the PDF. It's a draft — review before sending it on. Goodbye." if sent else "It's saved as a draft on your dashboard. Goodbye." ) return {"estimate": est, "twiml": _say_response(f"Got it. {tail}")} # Otherwise it's an edit: run it through the shared chat ops (catalog owns the price). out = chat_about_estimate(speech_result, rows, tax_rate=tax_rate) est = out["estimate"] state["rows"] = _rows_from_est(est) # carry the edit forward to the next turn save_estimate_record(state["rows"], job_title, tax_rate, thread=[]) # keep the draft current spoken = f"{out['reply']} That's now {est['total']:.2f} dollars. Anything else?" return {"estimate": est, "twiml": _ask_twiml(spoken, base)}