Send Leo checkout QR photo in Telegram

telegram/bot.py

@@ -58,6 +58,7 @@ from http_chat_proxy import (
     classify_smart_research_auto_resume_response,
     extract_auto_smart_research_followup_goal,
     extract_auto_smart_research_goal,
+    extract_checkout_qr_url,
     extract_paid_work_order_id,
     extract_smart_research_goal,
     post_chat_proxy,
@@ -193,6 +194,22 @@ async def _reply_text_native(msg, text: str, *, do_quote: bool = True):
         first = False
 
 
+async def _reply_checkout_qr_photo(msg, proxy_body: dict | None) -> bool:
+    qr_url = extract_checkout_qr_url(proxy_body)
+    if not qr_url:
+        return False
+    try:
+        await msg.reply_photo(
+            photo=qr_url,
+            caption="QR checkout for Leo paid research",
+            do_quote=False,
+        )
+        return True
+    except Exception as e:
+        logger.warning("Telegram checkout QR photo reply failed: %s", e)
+        return False
+
+
 async def _typing_keepalive(chat, stop_event: asyncio.Event, interval_seconds: float = 4.0) -> None:
     while not stop_event.is_set():
         try:
@@ -1419,6 +1436,7 @@ async def handle_tagged(update: Update, context: ContextTypes.DEFAULT_TYPE):
             return
 
         await _reply_text_native(msg, proxy_reply, do_quote=True)
+        await _reply_checkout_qr_photo(msg, proxy_body)
 
         if _should_start_smart_research_auto_resume_poll(
             paid_work_order_id=paid_work_order_id,

telegram/http_chat_proxy.py

@@ -174,6 +174,24 @@ def extract_paid_work_order_id(message: str) -> str | None:
     return match.group(1)
 
 
+def extract_checkout_qr_url(payload: dict[str, Any] | None) -> str | None:
+    """Return a Telegram-safe checkout QR image URL from a Leo research response."""
+    if not isinstance(payload, dict):
+        return None
+    checkout = payload.get("checkout")
+    if not isinstance(checkout, dict):
+        return None
+    raw_url = checkout.get("checkoutQrUrl") or checkout.get("checkout_qr_url")
+    if not isinstance(raw_url, str):
+        return None
+    url = raw_url.strip()
+    if len(url) > 2000:
+        return None
+    if not re.match(r"^https://leo\.livingip\.xyz/api/agents/leo/research/checkout-qr\?", url):
+        return None
+    return url
+
+
 def should_attach_structured_market_context(message: str) -> bool:
     """Return true only for explicit market-data questions, not social narrative research."""
     text = message.strip()

tests/test_telegram_leo_x402_bridge.py

@@ -16,6 +16,7 @@ from http_chat_proxy import (  # noqa: E402
     classify_smart_research_auto_resume_response,
     extract_auto_smart_research_followup_goal,
     extract_auto_smart_research_goal,
+    extract_checkout_qr_url,
     extract_paid_work_order_id,
     extract_chat_proxy_reply,
     extract_smart_research_goal,
@@ -238,6 +239,55 @@ def test_extract_paid_work_order_id(message, expected):
     assert extract_paid_work_order_id(message) == expected
 
 
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        (
+            {
+                "checkout": {
+                    "checkoutQrUrl": (
+                        "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?"
+                        "url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest"
+                    )
+                }
+            },
+            (
+                "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?"
+                "url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest"
+            ),
+        ),
+        (
+            {
+                "checkout": {
+                    "checkout_qr_url": (
+                        "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x"
+                    )
+                }
+            },
+            "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x",
+        ),
+        ({"checkout": {"checkoutQrUrl": "https://example.com/qr.png"}}, None),
+        ({"checkout": {"checkoutQrUrl": "http://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x"}}, None),
+        ({"checkout": {"checkoutQrUrl": "https://leo.livingip.xyz/not-qr?url=x"}}, None),
+        (
+            {
+                "checkout": {
+                    "checkoutQrUrl": (
+                        "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?"
+                        + ("x" * 2100)
+                    )
+                }
+            },
+            None,
+        ),
+        ({}, None),
+        (None, None),
+    ],
+)
+def test_extract_checkout_qr_url(payload, expected):
+    assert extract_checkout_qr_url(payload) == expected
+
+
 @pytest.mark.parametrize(
     ("message", "expected"),
     [