diff --git a/adapters/telnyx/webhook.py b/adapters/telnyx/webhook.py index b3f683f..0d11164 100644 --- a/adapters/telnyx/webhook.py +++ b/adapters/telnyx/webhook.py @@ -129,9 +129,13 @@ def validate_telnyx_signature( else None ) if capture_on and not provisioner: - logger.error("TELNYX_AUTO_SIPREC/STREAM is on but TELNYX_API_KEY is unset; capturing nothing") + logger.error( + "TELNYX_AUTO_SIPREC/STREAM is on but TELNYX_API_KEY is unset; capturing nothing" + ) if config.auto_stream and not config.stream_url: - logger.error("TELNYX_AUTO_STREAM is on but TELNYX_STREAM_URL is unset; not streaming any calls") + logger.error( + "TELNYX_AUTO_STREAM is on but TELNYX_STREAM_URL is unset; not streaming any calls" + ) if config.auto_siprec and config.auto_stream: logger.warning( "Both TELNYX_AUTO_SIPREC and TELNYX_AUTO_STREAM are on; Telnyx allows one " @@ -215,6 +219,10 @@ async def recording_event( logger.error(f"Failed to parse JSON body: {e}") raise HTTPException(status_code=400, detail="Invalid JSON") from None + return await _process_recording_event(event_data) + + async def _process_recording_event(event_data: dict) -> str: + """Shared by the Call Control JSON webhook and the TeXML form callback.""" # Check event type data = event_data.get("data", event_data) event_type = data.get("event_type", "") @@ -295,6 +303,53 @@ async def recording_event( return "OK" + @app.post("/webhook/texml-recording", response_class=PlainTextResponse) + async def texml_recording_event(request: Request): + """TeXML `recordingStatusCallback` (form-encoded, path B of the smart trunk). + + Reshaped into the Call Control `call.recording.saved` payload so one + builder serves both. Measured fields, 2026-09-13: RecordingSid, + CallSessionId, RecordingUrl (pre-signed S3 mp3, 600 s), RecordingChannels + "1"|"2", RecordingStartTime/EndTime, From, To, Direction, ConnectionId. + Unlike Call Control webhooks these carry From and To, so no lookup is needed. + + ponytail: no signature check here. TeXML callbacks are not ed25519 signed the + way Call Control webhooks are; gate this path with a per-tenant token in the + URL when the multi-tenant hook lands (CON-846). + """ + form = await request.form() + f = {k: str(v) for k, v in form.items()} + if f.get("RecordingStatus", "completed") != "completed" or not f.get("RecordingUrl"): + return "OK" + url = f["RecordingUrl"] + ext = url.split("?")[0].rsplit(".", 1)[-1].lower() + direction = {"inbound": "incoming", "outbound": "outgoing"}.get( + f.get("Direction", "").lower(), "" + ) + payload = { + "recording_id": f.get("RecordingSid", ""), + "call_session_id": f.get("CallSessionId", ""), + "call_control_id": f.get("CallSid", ""), + "connection_id": f.get("ConnectionId", ""), + "recording_urls": {ext if ext in ("mp3", "wav") else "mp3": url}, + "channels": "dual" if f.get("RecordingChannels") == "2" else "single", + "recording_started_at": f.get("RecordingStartTime"), + "recording_ended_at": f.get("RecordingEndTime"), + "from": f.get("From", ""), + "to": f.get("To", ""), + "direction": direction, + "recording_source": f.get("RecordingSource", "texml"), + } + event_data = { + "data": { + "event_type": "call.recording.saved", + "record_type": "event", + "occurred_at": f.get("RecordingEndTime"), + "payload": payload, + } + } + return await _process_recording_event(event_data) + @app.get("/status/{recording_id}") async def get_recording_status(recording_id: str): """Get processing status for a recording. diff --git a/scripts/telnyx_stream_spike.py b/scripts/telnyx_stream_spike.py index 21bae1e..512ab14 100644 --- a/scripts/telnyx_stream_spike.py +++ b/scripts/telnyx_stream_spike.py @@ -153,7 +153,7 @@ def build_app() -> FastAPI: "recording_saved": False, "stream_started_ok": None, "media_format": None, - "frames": {}, # track -> bytearray + "frames": {}, # track -> bytearray "start_seen": False, } @@ -185,21 +185,29 @@ async def hook(request: Request): # WS upgrade Telnyx opens during that window never gets accepted in # time -> Telnyx times out with 90046 while the socket lands late. if not NO_RECORD: - await asyncio.to_thread(command, ccid, "record_start", - {"format": "wav", "channels": "dual", "max_length": MAX_SECONDS}) - sc, _ = await asyncio.to_thread(command, ccid, "streaming_start", streaming_start_payload()) + await asyncio.to_thread( + command, + ccid, + "record_start", + {"format": "wav", "channels": "dual", "max_length": MAX_SECONDS}, + ) + sc, _ = await asyncio.to_thread( + command, ccid, "streaming_start", streaming_start_payload() + ) state["stream_started_ok"] = sc < 400 if sc >= 400: - logger.error("streaming_start rejected; adjust TELNYX_STREAM_* / TELNYX_STREAM_EXTRA") + logger.error( + "streaming_start rejected; adjust TELNYX_STREAM_* / TELNYX_STREAM_EXTRA" + ) if TRY_SIPREC and SIPREC_CONNECTOR: - threading.Timer( - SIPREC_DELAY, _start_siprec, args=(ccid,) - ).start() + threading.Timer(SIPREC_DELAY, _start_siprec, args=(ccid,)).start() elif event_type in ("call.recording.saved", "recording.saved"): state["recording_saved"] = True - logger.info(" RECORDING SAVED while streaming=%s <- question 1 answered", - state["stream_started_ok"]) + logger.info( + " RECORDING SAVED while streaming=%s <- question 1 answered", + state["stream_started_ok"], + ) elif event_type == "call.hangup": logger.info(" caller hung up (%s)", payload.get("hangup_cause", "?")) @@ -244,9 +252,13 @@ def _on_text(state: dict, text: str) -> None: start = m.get("start", m) state["media_format"] = start.get("media_format", {}) save_event("ws_start", m) - logger.info(" START media_format=%s call_control_id=%s from=%s to=%s", - state["media_format"], start.get("call_control_id"), - start.get("from"), start.get("to")) + logger.info( + " START media_format=%s call_control_id=%s from=%s to=%s", + state["media_format"], + start.get("call_control_id"), + start.get("from"), + start.get("to"), + ) elif event == "media": media = m.get("media", {}) track = media.get("track", "inbound") @@ -309,10 +321,12 @@ def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--port", type=int, default=8080) ap.add_argument("--max-seconds", type=int, default=MAX_SECONDS) - ap.add_argument("--try-siprec", action="store_true", - help="start a SIPREC fork mid-stream to test the one-slot rule") - ap.add_argument("--no-record", action="store_true", - help="isolate streaming; skip record_start") + ap.add_argument( + "--try-siprec", + action="store_true", + help="start a SIPREC fork mid-stream to test the one-slot rule", + ) + ap.add_argument("--no-record", action="store_true", help="isolate streaming; skip record_start") ap.add_argument("--selfcheck", action="store_true", help="run the offline self-check and exit") args = ap.parse_args() if args.selfcheck: @@ -327,8 +341,13 @@ def main() -> int: TRY_SIPREC = args.try_siprec NO_RECORD = args.no_record - logger.info("listening on :%d media sink at %s record=%s try_siprec=%s", - args.port, PUBLIC_WSS, not NO_RECORD, TRY_SIPREC) + logger.info( + "listening on :%d media sink at %s record=%s try_siprec=%s", + args.port, + PUBLIC_WSS, + not NO_RECORD, + TRY_SIPREC, + ) logger.info("streaming_start payload: %s", json.dumps(streaming_start_payload())) logger.info("writing to %s", OUT) uvicorn.run(build_app(), host="0.0.0.0", port=args.port, log_level="warning") diff --git a/tests/adapters/telnyx/test_texml_recording.py b/tests/adapters/telnyx/test_texml_recording.py new file mode 100644 index 0000000..abb07f2 --- /dev/null +++ b/tests/adapters/telnyx/test_texml_recording.py @@ -0,0 +1,84 @@ +"""TeXML `recordingStatusCallback` (form-encoded) reaches the same builder as Call Control JSON. + +Payload fields are the ones a real Telnyx TeXML callback carried on 2026-09-13 (smart trunk +Phase 0, session abde437e). Values are synthetic. +""" + +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient + +from adapters.telnyx.config import TelnyxConfig +from adapters.telnyx.webhook import create_app + +TEXML_FORM = { + "AccountSid": "acct-1", + "CallSid": "v3:CALLSID", + "CallSessionId": "sess-texml-1", + "CallStatus": "completed", + "ConnectionId": "3047649853975824124", + "Direction": "inbound", + "From": "+15085550100", + "To": "+14015550100", + "RecordingChannels": "2", + "RecordingDuration": "23", + "RecordingEndTime": "2026-09-13T02:35:25.068185Z", + "RecordingSid": "13a61448-d5cd-4f16-b231-308b4c61c953", + "RecordingSource": "DialVerb", + "RecordingStartTime": "2026-09-13T02:35:02.522170Z", + "RecordingStatus": "completed", + "RecordingUrl": "https://s3.amazonaws.com/bucket/x.mp3?X-Amz-Expires=600&X-Amz-Signature=abc", +} + + +@pytest.fixture +def config(monkeypatch, tmp_path): + monkeypatch.setenv("CONSERVER_URL", "https://conserver.example.com/vcon") + monkeypatch.setenv("STATE_FILE", str(tmp_path / "state.json")) + monkeypatch.setenv("VALIDATE_TELNYX_WEBHOOK", "false") + return TelnyxConfig() + + +def test_texml_callback_is_reshaped_and_built(config): + with ( + patch("adapters.telnyx.webhook.HttpPoster") as poster_cls, + patch("adapters.telnyx.webhook.TelnyxVconBuilder") as builder_cls, + ): + vcon = MagicMock(uuid="vcon-1") + builder = MagicMock() + builder.build.return_value = vcon + builder_cls.return_value = builder + poster = MagicMock() + poster.post.return_value = True + poster_cls.return_value = poster + + client = TestClient(create_app(config)) + r = client.post("/webhook/texml-recording", data=TEXML_FORM) + + assert r.status_code == 200 and r.text == "OK" + (data,), _ = builder.build.call_args + assert data.recording_id == TEXML_FORM["RecordingSid"] + assert data.call_session_id == "sess-texml-1" + assert data.from_number == "+15085550100" + assert data.to_number == "+14015550100" + assert data.direction == "inbound" + assert data.recording_url == TEXML_FORM["RecordingUrl"] + assert data.recording_urls == {"mp3": TEXML_FORM["RecordingUrl"]} + assert data.duration_seconds == pytest.approx(22.5, abs=0.1) + assert data.platform_tags["recording_channels"] == "dual" + poster.post.assert_called_once_with(vcon) + + # second delivery of the same RecordingSid is a no-op + r2 = client.post("/webhook/texml-recording", data=TEXML_FORM) + assert r2.text == "OK" and poster.post.call_count == 1 + + +def test_texml_callback_ignores_non_completed(config): + with patch("adapters.telnyx.webhook.TelnyxVconBuilder") as builder_cls: + client = TestClient(create_app(config)) + r = client.post( + "/webhook/texml-recording", data={**TEXML_FORM, "RecordingStatus": "in-progress"} + ) + assert r.text == "OK" + builder_cls.return_value.build.assert_not_called()