feat(telnyx): keep TeXML call timing and relayed annotations as tags

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
 
 

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_<name> 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-<name>` fields, for
+        example which recording notice it played. Each becomes an `annotation_<name>`
+        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": {

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,13 +68,40 @@ 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
         r2 = client.post("/webhook/texml-recording", data=TEXML_FORM)
         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))