Render Leo paid research as Telegram payment card

telegram/bot.py

@@ -58,7 +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_checkout_qr_photo_message,
     extract_paid_work_order_id,
     extract_smart_research_goal,
     post_chat_proxy,
@@ -194,15 +194,16 @@ 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:
+async def _reply_checkout_qr_photo(msg, proxy_body: dict | None, *, do_quote: bool = True) -> bool:
-    qr_url = extract_checkout_qr_url(proxy_body)
+    card = extract_checkout_qr_photo_message(proxy_body)
-    if not qr_url:
+    if not card:
         return False
     try:
         await msg.reply_photo(
-            photo=qr_url,
+            photo=card["photo_url"],
-            caption="QR checkout for Leo paid research",
+            caption=card["caption_html"],
-            do_quote=False,
+            parse_mode=card.get("parse_mode") or "HTML",
+            do_quote=do_quote,
         )
         return True
     except Exception as e:
@@ -1435,8 +1436,8 @@ async def handle_tagged(update: Update, context: ContextTypes.DEFAULT_TYPE):
             )
             return
 
-        await _reply_text_native(msg, proxy_reply, do_quote=True)
+        if not await _reply_checkout_qr_photo(msg, proxy_body, do_quote=True):
-        await _reply_checkout_qr_photo(msg, proxy_body)
+            await _reply_text_native(msg, proxy_reply, do_quote=True)
 
         if _should_start_smart_research_auto_resume_poll(
             paid_work_order_id=paid_work_order_id,

telegram/http_chat_proxy.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import re
+from html import escape
 from typing import Any
 
 DEFAULT_SMART_RESEARCH_COMMAND_PREFIXES = ("/smart_research", "/paid_research")
@@ -192,6 +193,53 @@ def extract_checkout_qr_url(payload: dict[str, Any] | None) -> str | None:
     return url
 
 
+def extract_checkout_qr_photo_message(payload: dict[str, Any] | None) -> dict[str, str] | None:
+    """Return a compact Telegram photo+caption payload for a Leo paid-research quote."""
+    qr_url = extract_checkout_qr_url(payload)
+    if not qr_url or not isinstance(payload, dict):
+        return None
+
+    checkout = payload.get("checkout")
+    if not isinstance(checkout, dict):
+        return None
+
+    telegram = checkout.get("telegram")
+    caption_html = None
+    if isinstance(telegram, dict):
+        raw_caption = telegram.get("captionHtml") or telegram.get("caption_html")
+        if isinstance(raw_caption, str) and 1 <= len(raw_caption) <= 1024:
+            caption_html = raw_caption.strip()
+
+    if not caption_html:
+        price = str(checkout.get("priceUsd") or checkout.get("price_usd") or "0.07").strip()
+        network = str(checkout.get("networkLabel") or checkout.get("network") or "Solana mainnet").strip()
+        payment_address = checkout.get("paymentAddress") or checkout.get("payment_address")
+        address_line = (
+            f"Recipient: <code>{escape(str(payment_address), quote=False)}</code>"
+            if isinstance(payment_address, str) and payment_address.strip()
+            else None
+        )
+        caption_html = "\n".join(
+            line
+            for line in [
+                "<b>Paid Leo research</b>",
+                f"{escape(price, quote=False)} USDC on {escape(network, quote=False)}.",
+                "Scan the QR to pay. I will continue here after payment.",
+                address_line,
+            ]
+            if line
+        )
+
+    if len(caption_html) > 1024:
+        caption_html = caption_html[:1000].rstrip() + "..."
+
+    return {
+        "photo_url": qr_url,
+        "caption_html": caption_html,
+        "parse_mode": "HTML",
+    }
+
+
 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_photo_message,
     extract_checkout_qr_url,
     extract_paid_work_order_id,
     extract_chat_proxy_reply,
@@ -288,6 +289,69 @@ def test_extract_checkout_qr_url(payload, expected):
     assert extract_checkout_qr_url(payload) == expected
 
 
+def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card():
+    payload = {
+        "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"
+            ),
+            "paymentAddress": "8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4",
+            "telegram": {
+                "captionHtml": (
+                    "<b>Paid Leo research</b>\n"
+                    "0.07 USDC on Solana mainnet.\n"
+                    '<a href="https://leo.livingip.xyz/agents/leo/research/checkout?q=test">'
+                    "Pay with x402</a>\n"
+                    "Recipient: <code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>"
+                )
+            },
+        }
+    }
+
+    card = extract_checkout_qr_photo_message(payload)
+
+    assert card == {
+        "photo_url": payload["checkout"]["checkoutQrUrl"],
+        "caption_html": payload["checkout"]["telegram"]["captionHtml"],
+        "parse_mode": "HTML",
+    }
+    assert "npx agentcash" not in card["caption_html"]
+    assert "Human checkout:" not in card["caption_html"]
+    assert "QR checkout:" not in card["caption_html"]
+
+
+def test_extract_checkout_qr_photo_message_builds_address_fallback_caption():
+    payload = {
+        "checkout": {
+            "checkoutQrUrl": "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x",
+            "priceUsd": "0.07",
+            "networkLabel": "Solana mainnet",
+            "paymentAddress": "8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4",
+        }
+    }
+
+    card = extract_checkout_qr_photo_message(payload)
+
+    assert card["photo_url"] == payload["checkout"]["checkoutQrUrl"]
+    assert "0.07 USDC on Solana mainnet" in card["caption_html"]
+    assert "<code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>" in card["caption_html"]
+
+
+def test_extract_checkout_qr_photo_message_rejects_non_leo_qr_url():
+    assert (
+        extract_checkout_qr_photo_message(
+            {
+                "checkout": {
+                    "checkoutQrUrl": "https://example.com/qr.png",
+                    "telegram": {"captionHtml": "<b>Paid Leo research</b>"},
+                }
+            }
+        )
+        is None
+    )
+
+
 @pytest.mark.parametrize(
     ("message", "expected"),
     [