From fe21726a825b65574d44b94e19a71ddc4acb4127 Mon Sep 17 00:00:00 2001 From: Thomas Howe Date: Fri, 25 Sep 2026 15:29:32 -0400 Subject: [PATCH] feat(telnyx): keep TeXML call timing and relayed annotations as tags The TeXML recording callback carries CallInitiatedAt and RecordingStartTime. The gap between them is the ringback, where a pre-answer recording notice plays, so a downstream consent check needs both. They now become the call_initiated_at and recording_started_at tags. A service that relays the callback can add Annotation- form fields, for example which notice it played and that file's hash. Each becomes an annotation_ tag. Names are limited to [A-Za-z0-9_]{1,64}; others are dropped. They carry the same trust as the rest of the unsigned TeXML form. Co-Authored-By: Claude Opus 5.5 --- adapters/telnyx/builder.py | 9 ++++++ adapters/telnyx/webhook.py | 15 ++++++++++ tests/adapters/telnyx/test_texml_recording.py | 28 +++++++++++++++++++ 3 files changed, 52 insertions(+) diff --git a/adapters/telnyx/builder.py b/adapters/telnyx/builder.py index 9116c61..66f9562 100644 --- a/adapters/telnyx/builder.py +++ b/adapters/telnyx/builder.py @@ -227,6 +227,15 @@ def platform_tags(self) -> dict[str, str]: if self._payload.get("channels"): tags["recording_channels"] = self._payload["channels"] + # TeXML only: when the call began and when recording began. The gap between + # them is the ringback, where a pre-answer notice plays. + for key in ("call_initiated_at", "recording_started_at"): + if self._payload.get(key): + tags[key] = self._payload[key] + + for name, value in (self._payload.get("annotations") or {}).items(): + tags[f"annotation_{name}"] = value + return tags diff --git a/adapters/telnyx/webhook.py b/adapters/telnyx/webhook.py index 0d11164..fbb91e6 100644 --- a/adapters/telnyx/webhook.py +++ b/adapters/telnyx/webhook.py @@ -3,6 +3,7 @@ import asyncio import base64 import logging +import re from fastapi import FastAPI, Header, HTTPException, Request from fastapi.responses import PlainTextResponse @@ -15,6 +16,9 @@ from .config import TelnyxConfig from .provision import TelnyxProvisioner, handle_call_event +# Relayed TeXML form fields with this prefix become annotation_ tags. +ANNOTATION_PREFIX = "Annotation-" + logger = logging.getLogger(__name__) @@ -313,6 +317,10 @@ async def texml_recording_event(request: Request): "1"|"2", RecordingStartTime/EndTime, From, To, Direction, ConnectionId. Unlike Call Control webhooks these carry From and To, so no lookup is needed. + A service that relays this callback can add `Annotation-` fields, for + example which recording notice it played. Each becomes an `annotation_` + tag on the vCon. They are exactly as trusted as the rest of this unsigned form. + 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). @@ -339,6 +347,13 @@ async def texml_recording_event(request: Request): "to": f.get("To", ""), "direction": direction, "recording_source": f.get("RecordingSource", "texml"), + "call_initiated_at": f.get("CallInitiatedAt"), + "annotations": { + k[len(ANNOTATION_PREFIX) :]: v + for k, v in f.items() + if k.startswith(ANNOTATION_PREFIX) + and re.fullmatch(r"[A-Za-z0-9_]{1,64}", k[len(ANNOTATION_PREFIX) :]) + }, } event_data = { "data": { diff --git a/tests/adapters/telnyx/test_texml_recording.py b/tests/adapters/telnyx/test_texml_recording.py index abb07f2..13bbe5d 100644 --- a/tests/adapters/telnyx/test_texml_recording.py +++ b/tests/adapters/telnyx/test_texml_recording.py @@ -16,6 +16,7 @@ "AccountSid": "acct-1", "CallSid": "v3:CALLSID", "CallSessionId": "sess-texml-1", + "CallInitiatedAt": "2026-09-13T02:34:48.101000Z", "CallStatus": "completed", "ConnectionId": "3047649853975824124", "Direction": "inbound", @@ -67,6 +68,9 @@ def test_texml_callback_is_reshaped_and_built(config): 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" + assert data.platform_tags["call_initiated_at"] == "2026-09-13T02:34:48.101000Z" + assert data.platform_tags["recording_started_at"] == TEXML_FORM["RecordingStartTime"] + assert not any(k.startswith("annotation_") for k in data.platform_tags) poster.post.assert_called_once_with(vcon) # second delivery of the same RecordingSid is a no-op @@ -74,6 +78,30 @@ def test_texml_callback_is_reshaped_and_built(config): assert r2.text == "OK" and poster.post.call_count == 1 +def test_texml_annotations_become_tags(config): + form = { + **TEXML_FORM, + "Annotation-notice": "notice-v1.wav", + "Annotation-notice_sha256": "ab" * 32, + "Annotation-bad name": "dropped", + "Annotation-": "dropped", + } + with ( + patch("adapters.telnyx.webhook.HttpPoster") as poster_cls, + patch("adapters.telnyx.webhook.TelnyxVconBuilder") as builder_cls, + ): + builder_cls.return_value.build.return_value = MagicMock(uuid="vcon-2") + poster_cls.return_value.post.return_value = True + client = TestClient(create_app(config)) + assert client.post("/webhook/texml-recording", data=form).text == "OK" + (data,), _ = builder_cls.return_value.build.call_args + annotations = {k: v for k, v in data.platform_tags.items() if k.startswith("annotation_")} + assert annotations == { + "annotation_notice": "notice-v1.wav", + "annotation_notice_sha256": "ab" * 32, + } + + def test_texml_callback_ignores_non_completed(config): with patch("adapters.telnyx.webhook.TelnyxVconBuilder") as builder_cls: client = TestClient(create_app(config))