Merge pull request #27 from living-ip/codex/leo-telegram-payment-card-20260701
Some checks are pending
CI / lint-and-test (push) Waiting to run

Render Leo paid research as Telegram payment card
This commit is contained in:
twentyOne2x 2026-07-01 00:27:40 +02:00 committed by GitHub
commit a27b9b3440
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 122 additions and 9 deletions

View file

@ -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:
qr_url = extract_checkout_qr_url(proxy_body)
if not qr_url:
async def _reply_checkout_qr_photo(msg, proxy_body: dict | None, *, do_quote: bool = True) -> bool:
card = extract_checkout_qr_photo_message(proxy_body)
if not card:
return False
try:
await msg.reply_photo(
photo=qr_url,
caption="QR checkout for Leo paid research",
do_quote=False,
photo=card["photo_url"],
caption=card["caption_html"],
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)
await _reply_checkout_qr_photo(msg, proxy_body)
if not await _reply_checkout_qr_photo(msg, proxy_body, do_quote=True):
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,

View file

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

View file

@ -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"),
[