91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Prove the Telegram Leo bridge can consume the public smart-research route."""
|
|
|
|
# ruff: noqa: E402, I001
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
TELEGRAM_DIR = REPO_ROOT / "telegram"
|
|
sys.path.insert(0, str(TELEGRAM_DIR))
|
|
|
|
from http_chat_proxy import build_smart_research_proxy_payload, post_chat_proxy
|
|
|
|
|
|
DEFAULT_URL = "https://leo.livingip.xyz/api/agents/leo/research"
|
|
DEFAULT_OUTPUT = "docs/reports/telegram-leo-x402-smart-research-bridge-proof.json"
|
|
|
|
|
|
async def run_check(url: str, research_goal: str) -> dict:
|
|
payload = build_smart_research_proxy_payload(
|
|
research_goal=research_goal,
|
|
source="telegram-proof",
|
|
agent="leo",
|
|
chat_id=0,
|
|
message_id=0,
|
|
username="codex-proof",
|
|
allow_paid_execution=False,
|
|
max_amount_usd=0.01,
|
|
include_synthesis=True,
|
|
)
|
|
status, body, reply = await post_chat_proxy(url=url, payload=payload, timeout_seconds=90)
|
|
funds_moved = bool(body.get("fundsMoved")) if isinstance(body, dict) else False
|
|
selected_provider = body.get("selectedProvider") if isinstance(body, dict) else None
|
|
exact_blocker = body.get("exactBlocker") if isinstance(body, dict) else None
|
|
return {
|
|
"schema": "livingip.telegramLeoX402SmartResearchBridgeProof.v1",
|
|
"generatedAt": datetime.now(timezone.utc).isoformat(),
|
|
"ok": bool(reply) and status in {200, 402} and not funds_moved,
|
|
"requiredTier": "T3_live_readonly",
|
|
"currentTier": body.get("currentTier", "T2_runtime") if isinstance(body, dict) else "T2_runtime",
|
|
"url": url,
|
|
"httpStatus": status,
|
|
"routeSchema": body.get("schema") if isinstance(body, dict) else None,
|
|
"selectedProvider": selected_provider,
|
|
"exactBlocker": exact_blocker,
|
|
"reply": reply,
|
|
"paidPostAttempted": bool(body.get("paidPostAttempted")) if isinstance(body, dict) else False,
|
|
"fundsMoved": funds_moved,
|
|
"secretValuesIncluded": False,
|
|
"strongestClaimAllowed": (
|
|
"Telegram bridge helper can POST a no-secret smart-research payload to the public Leo "
|
|
"research route and extract a usable fail-closed reply. This proves route shape and "
|
|
"readback only; it does not prove a Telegram bot service is deployed or a paid Telegram "
|
|
"message executed."
|
|
),
|
|
"notProven": [
|
|
"teleo-agent@leo-wallet-test.service active",
|
|
"Telegram message delivery",
|
|
"Telegram reply delivery",
|
|
"Telegram-triggered paid execution",
|
|
],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--url", default=DEFAULT_URL)
|
|
parser.add_argument(
|
|
"--research-goal",
|
|
default="Find current public evidence on x402 agent payments and recommend what Living IP Leo should test next.",
|
|
)
|
|
parser.add_argument("--output", default=DEFAULT_OUTPUT)
|
|
args = parser.parse_args()
|
|
|
|
proof = asyncio.run(run_check(args.url, args.research_goal))
|
|
output_path = REPO_ROOT / args.output
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
output_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
|
|
print(json.dumps(proof, indent=2, sort_keys=True))
|
|
return 0 if proof["ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|