feat(telnyx): accept TeXML recordingStatusCallback form posts

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.

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")

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()