diff --git a/CHANGELOG.md b/CHANGELOG.md index 47930b0..7e4068d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,43 @@ All notable changes to blockrun-llm will be documented in this file. +## Unreleased + +### Changed +- **Solana media polls prove job ownership instead of signing a payment.** + `SolanaLLMClient` / `AsyncSolanaLLMClient` poll `video()`, slow `image()` / + `image_edit()` and slow `music()` jobs with an ed25519 signature by the + submitting wallet (`x-poll-wallet` / `x-poll-timestamp` / `x-poll-signature` + over `blockrun-poll:v1:::`), signed fresh per + poll. A video signs exactly one payment, from the 402 its finished-but-unpaid + poll returns; images and music never pay on a poll. A refused proof falls back + once, per job, to the per-poll payment signature, with a warning on the + `blockrun_llm.solana_client` logger. Wallet rail on sol.blockrun.ai only; the + account rail and Base are unchanged. + +### Fixed +- **Slow `music()` tracks on the Solana wallet rail are polled to completion.** + The paid POST's `202 + poll_url` stub was returned as the result and failed + `MusicResponse` validation after the track had been paid for. +- **Solana `image()`, `image_edit()`, `video()` and `video_from_content()` + enforce `max_cost_per_call` and `max_session_cost`.** They are checked before + the submit payment is signed, and again before a video's completion payment + is signed. Before this, these calls ignored both limits. +- **A video's completion payment that may have settled is never treated as + free.** If the paid poll is lost in transit, polling resumes and reads the + job's settlement record instead of abandoning the clip. A Retry-After on a + paid poll the gateway could not settle yet is honoured. An error or timeout + after the payment went out says it may still settle and is tagged as paid, + so no fallback chain buys a second job. +- **A completion challenge is not a delivered clip.** Under ownership polling, + only a 200 delivers. A 5xx on the paid poll is no longer returned as the + video and booked to the session. A clip that an earlier poll or call already + paid for is logged at zero cost. +- **A raw paid POST's challenge is kept per thread and per asyncio task.** + Concurrent music calls on one shared client can no longer poll, or fall back + to paying, against another call's challenge, and a cached answer no longer + inherits a previous call's challenge. + ## 1.18.1 — 2026-10-02 ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index cce2c3b..33d75c8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,6 +23,7 @@ blockrun_llm/ ├── wallet.py # EVM wallet management ├── solana_wallet.py # Solana wallet management ├── x402.py # x402 payment protocol +├── poll_auth.py # Solana media-poll ownership proofs (x-poll-* headers) ├── router_core/ # Port of @blockrun/router-core (shared with the TS SDK + gateway) ├── router_adapter.py # Host glue: catalog ids, payment floors, free profile ├── router.py # Back-compat shim over router_core diff --git a/README.md b/README.md index 960a1c3..177b5b0 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,25 @@ export SOLANA_WALLET_KEY="your-bs58-solana-key" > what to switch to instead of failing with a cryptic "must be 66 characters" > error. +**Polling slow media jobs.** `video()`, slow `image()` / `image_edit()` models +and slow `music()` tracks answer `202 + poll_url`, and the client polls until the +job finishes. On Solana a payment dies with its blockhash in ~60-90s, so instead +of signing a payment on every poll the client proves it owns the job: each poll +carries `x-poll-wallet`, `x-poll-timestamp` and `x-poll-signature`, an ed25519 +signature by your wallet over `blockrun-poll:v1:::`. No transaction, nothing a facilitator can settle. + +- **Video** is charged once, when it is ready: the finished-but-unpaid poll + answers 402, the client signs one payment from that challenge and re-sends + the poll, which settles the clip. A poll budget that runs out costs nothing. +- **Images and music** were paid at submit, so their polls never sign a payment. +- If the gateway refuses the proof, the client falls back once, for that job, + to signing a payment per poll (logged as a warning on + `blockrun_llm.solana_client`). + +The account rail (API key) and Base are unchanged: an API key is its own proof, +and a Base authorization lives for hours. + ## Arc Support The same `LLMClient` pays on [Circle's Arc](https://www.arc.network) via [arc.blockrun.ai](https://arc.blockrun.ai) — point `api_url` at it and hold USDC on Arc in the same EVM wallet: diff --git a/blockrun_llm/poll_auth.py b/blockrun_llm/poll_auth.py new file mode 100644 index 0000000..a9bb8f0 --- /dev/null +++ b/blockrun_llm/poll_auth.py @@ -0,0 +1,124 @@ +"""Proof of job ownership for Solana media polls, without a payment. + +A Solana x402 payment is a transaction pinned to a recent blockhash that dies +in ~60-90s, so polling a multi-minute video by re-signing a payment meant a +fresh signature on almost every poll. sol.blockrun.ai also accepts a short +ed25519 signature over the job id from the wallet that submitted it: + + message = "blockrun-poll:v1:::" + headers x-poll-wallet: base58 public key of the submitting wallet + x-poll-timestamp: the in the message + x-poll-signature: base58 ed25519 signature over the UTF-8 message + +```` is ``video``, ``audio`` (music) or ``image``; ```` is the +last path segment of the ``poll_url`` the POST returned, URL-decoded, which is +the value the gateway verifies against. The gateway accepts a timestamp within +300s of its clock, so the headers are signed fresh for every poll. + +Server contract: blockrun-sol ``src/lib/poll-ownership.ts``. Base never sees +these headers: an EIP-3009 authorization lives for hours, so the Base clients +keep polling with their payment signature. +""" + +from __future__ import annotations + +import time +from collections.abc import Mapping +from typing import Any, Literal +from urllib.parse import unquote, urlsplit + +PollKind = Literal["video", "audio", "image"] + +POLL_WALLET_HEADER = "x-poll-wallet" +POLL_TIMESTAMP_HEADER = "x-poll-timestamp" +POLL_SIGNATURE_HEADER = "x-poll-signature" + +# The gateway's poll routes, by the its signature must name. Matched on +# the path the POST returned, so an unknown route never gets a proof. +_KIND_BY_ROUTE: dict[str, PollKind] = { + "videos": "video", + "audio": "audio", + "images": "image", +} + +# What each poll route answers a refused proof with: the same bytes as a job id +# that does not exist (wrong wallet, bad signature, stale timestamp, a job bound +# to no wallet). +_REFUSAL_BY_KIND: dict[PollKind, tuple[int, str]] = { + "video": (400, "Invalid job id"), + "audio": (404, "Job not found"), + "image": (404, "Job not found"), +} + + +def poll_kind_for_url(poll_url: str) -> PollKind | None: + """The ```` for a ``.../v1//generations/`` poll URL, or + ``None`` for any other path (no proof is sent there).""" + segments = [s for s in urlsplit(poll_url).path.split("/") if s] + if len(segments) < 4 or segments[-2] != "generations" or segments[-4] != "v1": + return None + return _KIND_BY_ROUTE.get(segments[-3]) + + +def poll_job_id(poll_url: str) -> str: + """The job id the gateway verifies: the poll URL's last path segment, + URL-decoded (the route reads ``decodeURIComponent(params.id)``).""" + path = urlsplit(poll_url).path.rstrip("/") + return unquote(path.rsplit("/", 1)[-1]) + + +def poll_auth_message(kind: PollKind, job_id: str, timestamp_s: int) -> str: + """The exact text the wallet signs.""" + return f"blockrun-poll:v1:{kind}:{job_id}:{timestamp_s}" + + +def poll_auth_headers( + keypair: Any, kind: PollKind, job_id: str, *, now_s: int | None = None +) -> dict[str, str]: + """Sign a fresh ownership proof for one poll. + + ``keypair`` is the wallet's ``solders.keypair.Keypair`` — the same key that + signed the submit payment, which is the payer the gateway bound the job to. + """ + timestamp_s = int(time.time()) if now_s is None else int(now_s) + message = poll_auth_message(kind, job_id, timestamp_s).encode("utf-8") + return { + POLL_WALLET_HEADER: str(keypair.pubkey()), + POLL_TIMESTAMP_HEADER: str(timestamp_s), + POLL_SIGNATURE_HEADER: str(keypair.sign_message(message)), + } + + +def is_poll_ownership_refusal(kind: PollKind, status_code: int, body: Any) -> bool: + """True when a poll carrying x-poll-* got the route's "no such job" answer.""" + expected_status, expected_error = _REFUSAL_BY_KIND[kind] + return ( + status_code == expected_status + and isinstance(body, Mapping) + and body.get("error") == expected_error + ) + + +def poll_keypair_of(signer: Any) -> Any: + """The ``solders`` Keypair behind an x402 ``KeypairSigner``, or ``None`` + when the signer does not expose one (then polls keep the payment path).""" + try: + from solders.keypair import Keypair + except ImportError: # pragma: no cover - solders ships with x402[svm] + return None + keypair = getattr(signer, "keypair", None) + return keypair if isinstance(keypair, Keypair) else None + + +__all__ = [ + "POLL_SIGNATURE_HEADER", + "POLL_TIMESTAMP_HEADER", + "POLL_WALLET_HEADER", + "PollKind", + "is_poll_ownership_refusal", + "poll_auth_headers", + "poll_auth_message", + "poll_job_id", + "poll_keypair_of", + "poll_kind_for_url", +] diff --git a/blockrun_llm/solana_client.py b/blockrun_llm/solana_client.py index c0fbef8..0d57f4e 100644 --- a/blockrun_llm/solana_client.py +++ b/blockrun_llm/solana_client.py @@ -19,11 +19,14 @@ import asyncio import json as _json +import logging import os import re import sys import threading from collections.abc import Iterator +from contextvars import ContextVar +from types import SimpleNamespace from typing import Any import httpx @@ -46,6 +49,13 @@ # no cycle. from .client import _SETTLED_ATTR, _enforce_spend_limits, _mark_settled from .jobs import absolute_poll_url +from .poll_auth import ( + is_poll_ownership_refusal, + poll_auth_headers, + poll_job_id, + poll_keypair_of, + poll_kind_for_url, +) from .price import Category, Market, Resolution, Session from .realface import _GROUP_ID_RE from .router_adapter import ( @@ -65,6 +75,7 @@ ) from .types import ( APIError, + BlockrunError, ChatCompletionChunk, ChatResponse, ImageResponse, @@ -117,22 +128,25 @@ SOLANA_API_URL = "https://sol.blockrun.ai/api" +logger = logging.getLogger(__name__) + def _create_signer(private_key: str) -> KeypairSigner: """Create a KeypairSigner, handling both full keypair and seed-only formats.""" try: return KeypairSigner.from_base58(private_key) except (ValueError, Exception): - # Fallback: might be a 32-byte seed (agentcash, etc.) - import base58 as b58 + # Fallback: might be a 32-byte seed (agentcash, etc.). Decoded with + # solders (a 32-byte base58 string parses as a Pubkey) rather than the + # `base58` package, which is not a declared dependency: a clean install + # refused every seed-only key as invalid. from solders.keypair import Keypair + from solders.pubkey import Pubkey - decoded = b58.b58decode(private_key) - if len(decoded) == 32: - kp = Keypair.from_seed(decoded) - full_key = b58.b58encode(bytes(kp)).decode() - return KeypairSigner.from_base58(full_key) - raise + # Anything that is not 32 bytes of base58 raises; the caller turns it + # into "Invalid Solana private key". + seed = bytes(Pubkey.from_string(private_key.strip())) + return KeypairSigner.from_base58(str(Keypair.from_seed(seed))) DEFAULT_MAX_TOKENS = 1024 @@ -524,6 +538,48 @@ def _receipt_from_headers(headers: Any) -> str | None: return headers.get("x-payment-receipt") or headers.get("X-Payment-Receipt") +def _is_pending_media_job(data: Any) -> bool: + """A paid POST answered with an accepted job (``202 + poll_url``) rather + than the finished media.""" + return ( + isinstance(data, dict) + and bool(data.get("poll_url")) + and data.get("status") in ("queued", "in_progress") + ) + + +def _quoted_cost_usd(payment_required: Any) -> float | None: + """The USD a 402 challenge can cost this wallet, read BEFORE signing. + + The highest amount among the challenge's Solana ``exact`` accepts (the only + scheme this client registers), so a spend limit checked against it can + never be exceeded by the accept the signer then picks. ``None`` when the + challenge has no such accept to read; the caller then checks the signed + payload's amount before anything is sent. + """ + amounts: list[int] = [] + for req in getattr(payment_required, "accepts", None) or (): + network = str(getattr(req, "network", "") or "") + if getattr(req, "scheme", None) != "exact" or not network.startswith("solana"): + continue + raw = getattr(req, "amount", None) + if raw is None: # x402 v1 names it max_amount_required + raw = getattr(req, "max_amount_required", None) + try: + amounts.append(int(str(raw))) + except (TypeError, ValueError): + return None + return max(amounts) / 1e6 if amounts else None + + +def _retry_after_seconds(response: Any, cap: float = 120.0) -> float | None: + """A response's ``Retry-After`` in seconds (delta-seconds only), capped.""" + raw = retry_after_of(response) + if raw is None or not str(raw).strip().isdigit(): + return None + return min(float(str(raw).strip()), cap) + + def _assert_same_payment_terms(signed_payload: Any, orig_amount: Any, orig_pay_to: Any) -> None: """Guard a mid-poll re-sign: the fresh 402 challenge must charge the same amount to the same recipient as the payment originally authorized for this @@ -706,6 +762,17 @@ def __init__( # metadata the shared JSON-only helper would otherwise drop. Read it # immediately after the helper returns (no intervening await). self._last_raw_headers: httpx.Headers | None = None + # (challenge, signed payload) of the most recent raw paid POST: music + # polls a slow track against the terms it paid. Held per thread / per + # asyncio task (see the property), never on the shared client. + self._raw_payment_var: ContextVar[tuple[Any, Any] | None] = ContextVar( + f"blockrun_raw_payment_{id(self)}", default=None + ) + + # The wallet's ed25519 key, for signing poll ownership proofs + # (poll_auth.py). None on the account rail and for a signer that does + # not expose one; those polls keep the payment-signature path. + self._poll_keypair: Any = None # Account calls need neither an x402 client nor a local signer. Keep # all shared request/receipt state above this branch initialized. @@ -726,12 +793,24 @@ def __init__( "Invalid Solana private key (expected a base58-encoded keypair " "or 32-byte seed)." ) from e _register_svm_with_headers(self._x402_client, signer, resolved_url, resolved_headers) + self._poll_keypair = poll_keypair_of(signer) # x402ClientSync is NOT thread-safe: concurrent payment signing on one # shared client races on nonce/authorization state. This lock serializes # just the (fast) signing step so a single client can be shared across # threads — see _sign_payment. self._payment_lock = threading.Lock() + @property + def _last_raw_payment(self) -> tuple[Any, Any] | None: + """(challenge, signed payload) of the raw paid POST made by THIS thread + or asyncio task. A shared client's concurrent music calls must never + poll (or fall back to paying) against another call's challenge.""" + return self._raw_payment_var.get() + + @_last_raw_payment.setter + def _last_raw_payment(self, value: tuple[Any, Any] | None) -> None: + self._raw_payment_var.set(value) + def _sign_payment(self, payment_required: Any) -> Any: """Thread-safe wrapper around ``x402_client.create_payment_payload``. @@ -1679,6 +1758,11 @@ def _request_with_payment_raw_once( """Make a request with Solana x402 payment, returning raw JSON.""" from .cache import get_cached, save_to_cache + # Before the cache check: a cached answer must not inherit a prior + # call's paid challenge (music would poll, or fall back to paying, + # against it). + self._last_raw_payment = None + # Check cache first — don't pay twice for same data cached = get_cached(endpoint, body) if cached is not None: @@ -1763,6 +1847,7 @@ def _handle_payment_and_retry_raw( # means nothing settles. _enforce_spend_limits(self, float(payment_payload.accepted.amount) / 1e6, body.get("model")) encoded_payment = encode_payment_signature_header(payment_payload) + self._last_raw_payment = (payment_required, payment_payload) payment_headers = { "Content-Type": "application/json", @@ -2099,9 +2184,17 @@ def _request_image_with_payment( raise PaymentError("402 response but no payment requirements found") payment_required = decode_payment_required_header(payment_header_str) + # Spend limits, before signing: image settles at this POST, and a + # video's completion payment is pinned to these terms. + quote_usd = _quoted_cost_usd(payment_required) + if quote_usd is not None: + _enforce_spend_limits(self, quote_usd, body.get("model")) payment_payload_obj = self._sign_payment(payment_required) - encoded_payment = encode_payment_signature_header(payment_payload_obj) cost_usd = float(payment_payload_obj.accepted.amount) / 1e6 + if quote_usd is None: + # No readable quote: check the signed amount before it is sent. + _enforce_spend_limits(self, cost_usd, body.get("model")) + encoded_payment = encode_payment_signature_header(payment_payload_obj) # Terms this job is authorized to pay — any mid-poll re-sign must match. orig_amount = payment_payload_obj.accepted.amount orig_pay_to = payment_payload_obj.accepted.pay_to @@ -2157,6 +2250,89 @@ def _request_image_with_payment( except Exception: submit_data = {} + return self._poll_media_job( + endpoint, + body, + submit_data, + eff_timeout=eff_timeout, + poll_budget_seconds=poll_budget_seconds, + poll_interval_seconds=poll_interval_seconds, + max_resigns=max_resigns, + label=label, + settled_at_submit=settled_at_submit, + account_job=account_job, + payment_required=payment_required, + encoded_payment=encoded_payment, + cost_usd=cost_usd, + orig_amount=orig_amount, + orig_pay_to=orig_pay_to, + ) + + def _poll_media_job( + self, endpoint: str, body: dict[str, Any], submit_data: dict[str, Any], **kwargs: Any + ) -> dict[str, Any]: + """Poll an accepted media job; see :meth:`_poll_media_job_impl`. + + Anything that escapes once this job's money moved, or may have moved, + is tagged settled (:func:`blockrun_llm.client._mark_settled`) so no + fallback chain buys a second job: an image or music job paid at its + POST, and a video once its completion payment went out (a paid poll + lost in transit may still have settled). + """ + state = SimpleNamespace(completion_sent=False) + try: + return self._poll_media_job_impl(endpoint, body, submit_data, state=state, **kwargs) + except (httpx.HTTPError, BlockrunError) as exc: + if state.completion_sent or ( + kwargs.get("settled_at_submit") and not kwargs.get("account_job") + ): + _mark_settled(exc) + raise + + def _poll_media_job_impl( + self, + endpoint: str, + body: dict[str, Any], + submit_data: dict[str, Any], + *, + state: Any, + eff_timeout: float, + poll_budget_seconds: float | None, + poll_interval_seconds: float | None, + max_resigns: int, + label: str, + settled_at_submit: bool, + account_job: bool, + payment_required: Any, + encoded_payment: str | None, + cost_usd: float, + orig_amount: Any, + orig_pay_to: Any, + book_completion: bool = True, + ) -> dict[str, Any]: + """Poll an accepted media job (``202 + poll_url``) to completion. + + Wallet rail, Solana poll routes: each poll carries a fresh x-poll-* + ownership proof (see :mod:`blockrun_llm.poll_auth`) instead of a payment + signature, so a pending job costs no signatures at all. Images and music + were paid at POST, so their polls never pay. A finished, unpaid video + answers 402 with the x402 challenge and ``status: "completed"``: ONE + payment is signed from that challenge and the same GET is re-sent with + it, which settles the clip. A poll that refuses the proof as "no such + job" falls back, once, to the payment-signature polling below; so does + any other 402 (a batch-billed image job). + + Account rail, or a wallet whose signer exposes no keypair: the + payment-signature polling, unchanged. + + ``book_completion=False`` is for a caller that already booked and logged + the paid POST (music on the wallet rail): the completed poll is returned + without being counted, cached or logged a second time. + """ + import time as _time + + from .cache import save_to_cache + poll_url_rel = submit_data.get("poll_url") job_id = submit_data.get("id") if not poll_url_rel: @@ -2170,7 +2346,7 @@ def _request_image_with_payment( # wallet route booked it above, and the account rail has no x402 charge # (it reserves credit at accept and settles it when the job completes). charged_at_submit = settled_at_submit or account_job - if settled_at_submit and not account_job: + if settled_at_submit and not account_job and book_completion: # Solana image routes settle at POST (a signed transaction dies with # its ~60-90s blockhash, so the gateway cannot wait for a long # render). The charge has already happened: book it now, or a job @@ -2182,6 +2358,25 @@ def _request_image_with_payment( if encoded_payment is not None: poll_headers["PAYMENT-SIGNATURE"] = encoded_payment + # Ownership polling: wallet rail only (an API key is its own proof), on + # a route the gateway knows a for, with a key that can sign. + poll_kind = None if account_job else poll_kind_for_url(poll_url) + poll_keypair = self._poll_keypair if poll_kind is not None else None + ownership = poll_keypair is not None + poll_job = poll_job_id(poll_url) + # Payment signatures a finished video may cost under ownership polling: + # the one that settles it, plus the same re-sign allowance the payment + # path has for a settle that failed (stale blockhash, a concurrent claim). + completion_payments_left = 1 + max_resigns + completion_paid = False + # Set when a completed video's 402 was just signed: the next request is + # that paid GET, sent at once rather than after the poll interval. + paid_completion_header: str | None = None + # The gateway's Retry-After on a paid completion poll it could not + # settle yet (a claim held while an earlier settle's outcome is + # resolved): honoured by the next ownership poll instead of the interval. + next_wait: float | None = None + budget = ( poll_budget_seconds if poll_budget_seconds is not None @@ -2197,39 +2392,174 @@ def _request_image_with_payment( resigns_left = max_resigns last_resign_at = _time.monotonic() - while _time.monotonic() < deadline: - _time.sleep(interval) - - # Keep the settlement blockhash fresh (poll-based media path only, - # gated on max_resigns). Re-sign the ORIGINAL challenge — same amount/pay_to, - # only a freshly-fetched blockhash — so that whenever upstream flips to - # "completed" the signature is 0 - and _time.monotonic() - last_resign_at >= self.MEDIA_RESIGN_FRESH_SECONDS - ): + def _fall_back_to_payment_polls(reason: str) -> None: + nonlocal ownership, last_resign_at + ownership = False + logger.warning( + "%s job %s: %s; falling back to payment-signature polling for this job.", + label, + job_id, + reason, + ) + if payment_required is not None: try: - fresh_payload = self._sign_payment(payment_required) poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( - fresh_payload + self._sign_payment(payment_required) ) last_resign_at = _time.monotonic() except Exception: - # Best-effort only: a failed proactive re-sign (RPC hiccup, - # SolanaRpcException, etc.) must never abort the poll loop — - # we simply keep the prior signature (pre-fix behaviour). + # Same tolerance as the proactive re-sign: keep the submit + # signature, and the payment path's 402 re-sign takes over. pass - poll_resp = self._client.get(poll_url, headers=poll_headers, timeout=eff_timeout) + # A completion payment signed just before the deadline is still sent: + # leaving it unsent would strand a finished clip behind a spent signature. + while paid_completion_header is not None or _time.monotonic() < deadline: + sending_paid_completion = paid_completion_header is not None + # Whether this request is on the ownership flow (a proof poll, or + # the paid completion GET). There, success is a 200: the + # completion 402's body already says status "completed", so the + # carried-over status alone must never turn an error into a result. + ownership_request = ownership or sending_paid_completion + request_headers: dict[str, str] + if paid_completion_header is not None: + request_headers = { + "User-Agent": _get_user_agent(), + "PAYMENT-SIGNATURE": paid_completion_header, + } + paid_completion_header = None + state.completion_sent = True + elif ownership and poll_kind is not None: + # A Retry-After never sleeps past the poll budget. + _time.sleep( + interval + if next_wait is None + else min(next_wait, max(0.0, deadline - _time.monotonic())) + ) + next_wait = None + request_headers = { + "User-Agent": _get_user_agent(), + **poll_auth_headers(poll_keypair, poll_kind, poll_job), + } + else: + _time.sleep(interval) + + # Keep the settlement blockhash fresh (poll-based media path only, + # gated on max_resigns). Re-sign the ORIGINAL challenge — same + # amount/pay_to, only a freshly-fetched blockhash — so that whenever + # upstream flips to "completed" the signature is + # 0 + and payment_required is not None + and _time.monotonic() - last_resign_at >= self.MEDIA_RESIGN_FRESH_SECONDS + ): + try: + fresh_payload = self._sign_payment(payment_required) + poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( + fresh_payload + ) + last_resign_at = _time.monotonic() + except Exception: + # Best-effort only: a failed proactive re-sign (RPC hiccup, + # SolanaRpcException, etc.) must never abort the poll loop — + # we simply keep the prior signature (pre-fix behaviour). + pass + request_headers = poll_headers + + try: + poll_resp = self._client.get(poll_url, headers=request_headers, timeout=eff_timeout) + except httpx.TransportError as exc: + if not sending_paid_completion: + raise + # The payment may have reached the gateway and settled, with + # only the answer lost. The next ownership poll reads the job's + # settlement record: "already settled" (no new charge) or, if + # it never settled, a fresh challenge. + if completion_payments_left > 0: + continue + raise PaymentError( + f"{label}: the paid poll for the finished job failed in transit " + f"({type(exc).__name__}). The payment may have settled: re-poll " + f"{poll_url} with the same wallet before submitting again." + ) from exc try: poll_data = poll_resp.json() except Exception: poll_data = {} + if not isinstance(poll_data, dict): + poll_data = {} last_status = poll_data.get("status", last_status) + if ownership and sending_paid_completion and poll_resp.status_code == 402: + # The one payment did not settle on this poll: another poll holds + # the job's settlement claim, the blockhash went stale, or an + # earlier settle's outcome is still unknown. The next ownership + # poll answers either "already settled" (no new charge) or a + # fresh challenge, so go back to polling rather than re-signing + # blind. + if completion_payments_left > 0: + next_wait = _retry_after_seconds(poll_resp) + continue + refusal = build_payment_rejected_error(poll_resp) + raise PaymentError( + f"{refusal} The payment for the finished job may still settle: " + f"re-poll {poll_url} with the same wallet before submitting again." + ) from refusal + + if ownership and not sending_paid_completion: + if poll_kind is not None and is_poll_ownership_refusal( + poll_kind, poll_resp.status_code, poll_data + ): + _fall_back_to_payment_polls( + f"the gateway refused the poll ownership proof " + f"(HTTP {poll_resp.status_code} {poll_data.get('error')!r})" + ) + continue + if poll_resp.status_code == 402: + challenge_header = self._extract_payment_header(poll_resp) + if ( + poll_kind == "video" + and poll_data.get("status") == "completed" + and challenge_header + ): + # The clip is ready and unpaid: this 402 is the challenge + # for the one payment that settles it. + if completion_payments_left <= 0: + raise build_payment_rejected_error(poll_resp) + if _time.monotonic() >= deadline: + # Out of budget: sign nothing new (the payment a + # signature is already out for is the timeout's + # concern); the 504 below says how to resume. + break + completion_payments_left -= 1 + completion_required = decode_payment_required_header(challenge_header) + # Spend limits, before signing the one payment that + # settles the clip. The terms guard below pins it to the + # submit's amount, so that is the quote when the + # challenge has none to read. + completion_quote = _quoted_cost_usd(completion_required) + _enforce_spend_limits( + self, + completion_quote if completion_quote is not None else cost_usd, + body.get("model"), + ) + completion_payload = self._sign_payment(completion_required) + if orig_amount is not None: + # Same guard as a mid-poll re-sign: the completion + # challenge must charge what the submit authorized. + _assert_same_payment_terms(completion_payload, orig_amount, orig_pay_to) + paid_completion_header = encode_payment_signature_header(completion_payload) + completion_paid = True + continue + # Any other 402 on an ownership poll (a batch-billed image + # job asks for its batch payment): the payment path. + _fall_back_to_payment_polls("the poll answered 402") + continue + if poll_resp.status_code == 402: # Account rail: a 402 is the account being out of credit, not a # challenge to sign. Nothing here can sign, so say so plainly. @@ -2285,7 +2615,7 @@ def _request_image_with_payment( else " (payment was settled at submit)" if settled_at_submit else "" ), poll_resp.status_code, - sanitize_error_response(poll_data if isinstance(poll_data, dict) else {}), + sanitize_error_response(poll_data), retry_after=retry_after_of(poll_resp), ) @@ -2293,22 +2623,40 @@ def _request_image_with_payment( # gateway settles the moment a poll reports completed, so a # completed-but-non-200 poll (which the caller was already charged # for) must still be treated as success. - if last_status == "completed": + if last_status == "completed" and ( + not ownership_request or poll_resp.status_code == 200 + ): tx_hash = poll_resp.headers.get("x-payment-receipt") or poll_resp.headers.get( "X-Payment-Receipt" ) - if tx_hash and isinstance(poll_data, dict) and not poll_data.get("txHash"): + if tx_hash and not poll_data.get("txHash"): poll_data["txHash"] = tx_hash - if not charged_at_submit: + if not book_completion: + return poll_data + # An ownership poll can be handed a clip some earlier poll paid + # for ("already_settled"); only a payment this call signed is + # booked to this session. + paid_by_this_call = completion_paid or not ownership + if not charged_at_submit and paid_by_this_call: self._session_calls += 1 self._session_total_usd += cost_usd self._last_call_cost = cost_usd + # The cost log and transaction log record what this call paid: + # nothing, for a clip an earlier poll or call already settled. + logged_cost = cost_usd if charged_at_submit or paid_by_this_call else 0.0 self._capture_settlement(poll_resp) - save_to_cache(endpoint, body, poll_data, cost_usd=cost_usd, **self._billing_meta()) - self._log_transaction(endpoint, body, poll_data, cost_usd) + save_to_cache( + endpoint, body, poll_data, cost_usd=logged_cost, **self._billing_meta() + ) + self._log_transaction(endpoint, body, poll_data, logged_cost) return poll_data - if poll_resp.status_code in (202, 504): + if poll_resp.status_code in (202, 504) or ( + # A paid or proof poll answered 503 (the gateway's verifier + # is briefly down): nothing settled, so it is retriable. + ownership_request + and poll_resp.status_code == 503 + ): # 202 = still queued/in_progress; 504 = transient upstream # hiccup. Both are retriable inside the budget. continue @@ -2344,11 +2692,20 @@ def _request_image_with_payment( ) if settled_at_submit else ( - f"{label} did not complete within {budget:.0f}s " - f"(last status: {last_status}). Settlement only happens on " - "completion, so no payment was taken. The job stays claimable " - "for ~48h — re-poll poll_url with a fresh signature from the " - "same wallet to fetch (and settle) the finished result." + ( + f"{label} did not complete within {budget:.0f}s " + f"(last status: {last_status}). A payment for the finished job " + "was sent and may still settle: re-poll poll_url with the same " + "wallet before submitting again (no new charge if it settled)." + ) + if state.completion_sent + else ( + f"{label} did not complete within {budget:.0f}s " + f"(last status: {last_status}). Settlement only happens on " + "completion, so no payment was taken. The job stays claimable " + "for ~48h — re-poll poll_url with a fresh signature from the " + "same wallet to fetch (and settle) the finished result." + ) ) ) ), @@ -2589,14 +2946,48 @@ def _request_audio_with_payment( ) -> dict[str, Any]: """POST an audio request (music, speech, sound effect) on either rail. - Wallet rail: the raw x402 helper, unchanged. Account rail: the first - POST carries the key and is the billed submit, so it goes through the - media helper instead, which never replays it on a 502/503 (that could - bill twice) and polls a 202 job to completion unsigned rather than - returning the ``{id, poll_url}`` stub as the result. + Wallet rail: the raw x402 helper. A slow track answers ``202 + + poll_url`` after its payment settled at POST; it is polled to + completion with ownership proofs (never another payment) rather than + returning the ``{id, poll_url}`` stub as the result. Account rail: the + first POST carries the key and is the billed submit, so it goes through + the media helper instead, which never replays it on a 502/503 (that + could bill twice) and polls a 202 job to completion unsigned. """ if not self.api_key: - return self._request_with_payment_raw(endpoint, body, timeout=timeout) + data = self._request_with_payment_raw(endpoint, body, timeout=timeout) + if not _is_pending_media_job(data): + return data + # The receipt callers attach is the paid POST's, not a poll's. + paid_headers = self._last_raw_headers + required, payload = self._last_raw_payment or (None, None) + try: + return self._poll_media_job( + endpoint, + body, + data, + eff_timeout=timeout if timeout is not None else self._timeout, + poll_budget_seconds=self.AUDIO_POLL_BUDGET_SECONDS, + poll_interval_seconds=self.AUDIO_POLL_INTERVAL_SECONDS, + max_resigns=self.MEDIA_POLL_MAX_RESIGNS, + label=label, + settled_at_submit=True, + account_job=False, + payment_required=required, + encoded_payment=( + encode_payment_signature_header(payload) if payload is not None else None + ), + cost_usd=self._last_call_cost, + orig_amount=payload.accepted.amount if payload is not None else None, + orig_pay_to=payload.accepted.pay_to if payload is not None else None, + book_completion=False, + ) + except (httpx.HTTPError, APIError) as exc: + # The track was paid for at POST: nothing may buy a retry. + _mark_settled(exc) + raise + finally: + self._last_raw_headers = paid_headers self._last_raw_headers = None return self._request_image_with_payment( endpoint, @@ -3583,7 +3974,15 @@ def __init__( # metadata the shared JSON-only helper would otherwise drop. Read it # immediately after the helper returns (no intervening await). self._last_raw_headers: httpx.Headers | None = None + # (challenge, signed payload) of the most recent raw paid POST: music + # polls a slow track against the terms it paid. Held per thread / per + # asyncio task (see the property), never on the shared client. + self._raw_payment_var: ContextVar[tuple[Any, Any] | None] = ContextVar( + f"blockrun_raw_payment_{id(self)}", default=None + ) + # The wallet's ed25519 key for poll ownership proofs; see the sync class. + self._poll_keypair: Any = None if api_key: self._x402_client = None self._payment_lock = None @@ -3603,11 +4002,23 @@ def __init__( "Invalid Solana private key (expected a base58-encoded keypair " "or 32-byte seed)." ) from e _register_svm_with_headers(self._x402_client, signer, resolved_url, resolved_headers) + self._poll_keypair = poll_keypair_of(signer) # Lazily created on first sign (avoids binding asyncio.Lock to a loop at # construction time). Serializes the async signing critical section so a # shared client is safe across concurrent coroutines — see _sign_payment. self._payment_lock: asyncio.Lock | None = None + @property + def _last_raw_payment(self) -> tuple[Any, Any] | None: + """(challenge, signed payload) of the raw paid POST made by THIS thread + or asyncio task. A shared client's concurrent music calls must never + poll (or fall back to paying) against another call's challenge.""" + return self._raw_payment_var.get() + + @_last_raw_payment.setter + def _last_raw_payment(self, value: tuple[Any, Any] | None) -> None: + self._raw_payment_var.set(value) + async def _sign_payment(self, payment_required: Any) -> Any: """Task-safe async wrapper around ``x402_client.create_payment_payload``. @@ -4232,6 +4643,7 @@ async def _sign_payment_from_response( # See the sync path: refusing here means nothing is ever sent. _enforce_spend_limits(self, float(payment_payload.accepted.amount) / 1e6) encoded_payment = encode_payment_signature_header(payment_payload) + self._last_raw_payment = (payment_required, payment_payload) cost_usd = float(payment_payload.accepted.amount) / 1e6 return ( { @@ -4398,6 +4810,9 @@ async def _request_with_payment_raw_once( the sync :class:`SolanaLLMClient` helper).""" from .cache import get_cached, save_to_cache + # Before the cache check: see the sync helper. + self._last_raw_payment = None + cached = get_cached(endpoint, body) if cached is not None: return cached @@ -4845,10 +5260,43 @@ async def _request_audio_with_payment( self, endpoint: str, body: dict[str, Any], timeout: float | None, label: str ) -> dict[str, Any]: """Async mirror of :meth:`SolanaLLMClient._request_audio_with_payment`: - wallet rail unchanged; on the account rail the keyed first POST is the + on the wallet rail a slow track's ``202 + poll_url`` is polled with + ownership proofs; on the account rail the keyed first POST is the billed submit, so no 5xx replay and a 202 job is polled unsigned.""" if not self.api_key: - return await self._request_with_payment_raw(endpoint, body, timeout=timeout) + data = await self._request_with_payment_raw(endpoint, body, timeout=timeout) + if not _is_pending_media_job(data): + return data + # The receipt callers attach is the paid POST's, not a poll's. + paid_headers = self._last_raw_headers + required, payload = self._last_raw_payment or (None, None) + try: + return await self._poll_media_job( + endpoint, + body, + data, + eff_timeout=timeout if timeout is not None else self._timeout, + poll_budget_seconds=SolanaLLMClient.AUDIO_POLL_BUDGET_SECONDS, + poll_interval_seconds=SolanaLLMClient.AUDIO_POLL_INTERVAL_SECONDS, + max_resigns=SolanaLLMClient.MEDIA_POLL_MAX_RESIGNS, + label=label, + settled_at_submit=True, + account_job=False, + payment_required=required, + encoded_payment=( + encode_payment_signature_header(payload) if payload is not None else None + ), + cost_usd=self._last_call_cost, + orig_amount=payload.accepted.amount if payload is not None else None, + orig_pay_to=payload.accepted.pay_to if payload is not None else None, + book_completion=False, + ) + except (httpx.HTTPError, APIError) as exc: + # The track was paid for at POST: nothing may buy a retry. + _mark_settled(exc) + raise + finally: + self._last_raw_headers = paid_headers self._last_raw_headers = None return await self._request_image_with_payment( endpoint, @@ -5232,7 +5680,6 @@ async def _request_image_with_payment( :meth:`image` and :meth:`video` (``max_resigns`` re-signs to survive the 600s x402 authorization window on long video polls). """ - import time as _time from .cache import get_cached, save_to_cache @@ -5300,9 +5747,15 @@ async def _request_image_with_payment( if not probe_payment_header: raise PaymentError("402 response but no payment requirements found") payment_required = decode_payment_required_header(probe_payment_header) + # Spend limits, before signing (see the sync helper). + quote_usd = _quoted_cost_usd(payment_required) + if quote_usd is not None: + _enforce_spend_limits(self, quote_usd, body.get("model")) payment_payload_obj = await self._sign_payment(payment_required) - encoded_payment = encode_payment_signature_header(payment_payload_obj) cost_usd = float(payment_payload_obj.accepted.amount) / 1e6 + if quote_usd is None: + _enforce_spend_limits(self, cost_usd, body.get("model")) + encoded_payment = encode_payment_signature_header(payment_payload_obj) # Terms this job is authorized to pay — any mid-poll re-sign must match. orig_amount = payment_payload_obj.accepted.amount orig_pay_to = payment_payload_obj.accepted.pay_to @@ -5357,16 +5810,106 @@ async def _request_image_with_payment( except Exception: submit_data = {} + return await self._poll_media_job( + endpoint, + body, + submit_data, + eff_timeout=eff_timeout, + poll_budget_seconds=poll_budget_seconds, + poll_interval_seconds=poll_interval_seconds, + max_resigns=max_resigns, + label=label, + settled_at_submit=settled_at_submit, + account_job=account_job, + payment_required=payment_required, + encoded_payment=encoded_payment, + cost_usd=cost_usd, + orig_amount=orig_amount, + orig_pay_to=orig_pay_to, + ) + + async def _poll_media_job( + self, endpoint: str, body: dict[str, Any], submit_data: dict[str, Any], **kwargs: Any + ) -> dict[str, Any]: + """Poll an accepted media job; see :meth:`_poll_media_job_impl`. + + Anything that escapes once this job's money moved, or may have moved, + is tagged settled (:func:`blockrun_llm.client._mark_settled`) so no + fallback chain buys a second job: an image or music job paid at its + POST, and a video once its completion payment went out (a paid poll + lost in transit may still have settled). + """ + state = SimpleNamespace(completion_sent=False) + try: + return await self._poll_media_job_impl( + endpoint, body, submit_data, state=state, **kwargs + ) + except (httpx.HTTPError, BlockrunError) as exc: + if state.completion_sent or ( + kwargs.get("settled_at_submit") and not kwargs.get("account_job") + ): + _mark_settled(exc) + raise + + async def _poll_media_job_impl( + self, + endpoint: str, + body: dict[str, Any], + submit_data: dict[str, Any], + *, + state: Any, + eff_timeout: float, + poll_budget_seconds: float | None, + poll_interval_seconds: float | None, + max_resigns: int, + label: str, + settled_at_submit: bool, + account_job: bool, + payment_required: Any, + encoded_payment: str | None, + cost_usd: float, + orig_amount: Any, + orig_pay_to: Any, + book_completion: bool = True, + ) -> dict[str, Any]: + """Async mirror of :meth:`SolanaLLMClient._poll_media_job`: poll an + accepted media job (``202 + poll_url``) to completion. + + Wallet rail, Solana poll routes: each poll carries a fresh x-poll-* + ownership proof (see :mod:`blockrun_llm.poll_auth`) instead of a payment + signature, so a pending job costs no signatures at all. Images and music + were paid at POST, so their polls never pay. A finished, unpaid video + answers 402 with the x402 challenge and ``status: "completed"``: ONE + payment is signed from that challenge and the same GET is re-sent with + it, which settles the clip. A poll that refuses the proof as "no such + job" falls back, once, to the payment-signature polling below; so does + any other 402 (a batch-billed image job). + + Account rail, or a wallet whose signer exposes no keypair: the + payment-signature polling, unchanged. + + ``book_completion=False`` is for a caller that already booked and logged + the paid POST (music on the wallet rail): the completed poll is returned + without being counted, cached or logged a second time. + """ + import time as _time + + from .cache import save_to_cache + poll_url_rel = submit_data.get("poll_url") job_id = submit_data.get("id") if not poll_url_rel: - raise APIError("Slow-path 202 missing poll_url", 202, {"response": submit_data}) + raise APIError( + "Slow-path 202 missing poll_url", + 202, + {"response": submit_data}, + ) poll_url = self._absolute_url(poll_url_rel, job_id) # Neither rail books spend on completion here: a settled-at-submit # wallet route booked it above, and the account rail has no x402 charge # (it reserves credit at accept and settles it when the job completes). charged_at_submit = settled_at_submit or account_job - if settled_at_submit and not account_job: + if settled_at_submit and not account_job and book_completion: # Solana image routes settle at POST (a signed transaction dies with # its ~60-90s blockhash, so the gateway cannot wait for a long # render). The charge has already happened: book it now, or a job @@ -5378,6 +5921,25 @@ async def _request_image_with_payment( if encoded_payment is not None: poll_headers["PAYMENT-SIGNATURE"] = encoded_payment + # Ownership polling: wallet rail only (an API key is its own proof), on + # a route the gateway knows a for, with a key that can sign. + poll_kind = None if account_job else poll_kind_for_url(poll_url) + poll_keypair = self._poll_keypair if poll_kind is not None else None + ownership = poll_keypair is not None + poll_job = poll_job_id(poll_url) + # Payment signatures a finished video may cost under ownership polling: + # the one that settles it, plus the same re-sign allowance the payment + # path has for a settle that failed (stale blockhash, a concurrent claim). + completion_payments_left = 1 + max_resigns + completion_paid = False + # Set when a completed video's 402 was just signed: the next request is + # that paid GET, sent at once rather than after the poll interval. + paid_completion_header: str | None = None + # The gateway's Retry-After on a paid completion poll it could not + # settle yet (a claim held while an earlier settle's outcome is + # resolved): honoured by the next ownership poll instead of the interval. + next_wait: float | None = None + budget = ( poll_budget_seconds if poll_budget_seconds is not None @@ -5393,47 +5955,189 @@ async def _request_image_with_payment( resigns_left = max_resigns last_resign_at = _time.monotonic() - while _time.monotonic() < deadline: - await asyncio.sleep(interval) - - # Keep the settlement blockhash fresh (poll-based media path only, - # gated on max_resigns) — mirror of the sync helper. Re-sign the - # ORIGINAL challenge (same amount/ - # pay_to, fresh blockhash) every MEDIA_RESIGN_FRESH_SECONDS so a slow / - # flaky-status model (1080p Seedance) can't age the signature out - # before the settling "completed" poll lands. Only completed settles. - if ( - max_resigns > 0 - and _time.monotonic() - last_resign_at >= SolanaLLMClient.MEDIA_RESIGN_FRESH_SECONDS - ): + async def _fall_back_to_payment_polls(reason: str) -> None: + nonlocal ownership, last_resign_at + ownership = False + logger.warning( + "%s job %s: %s; falling back to payment-signature polling for this job.", + label, + job_id, + reason, + ) + if payment_required is not None: try: - fresh_payload = await self._sign_payment(payment_required) poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( - fresh_payload + await self._sign_payment(payment_required) ) last_resign_at = _time.monotonic() except Exception: + # Same tolerance as the proactive re-sign: keep the submit + # signature, and the payment path's 402 re-sign takes over. pass - poll_resp = await self._client.get(poll_url, headers=poll_headers, timeout=eff_timeout) + # A completion payment signed just before the deadline is still sent: + # leaving it unsent would strand a finished clip behind a spent signature. + while paid_completion_header is not None or _time.monotonic() < deadline: + sending_paid_completion = paid_completion_header is not None + # Whether this request is on the ownership flow (a proof poll, or + # the paid completion GET). There, success is a 200: the + # completion 402's body already says status "completed", so the + # carried-over status alone must never turn an error into a result. + ownership_request = ownership or sending_paid_completion + request_headers: dict[str, str] + if paid_completion_header is not None: + request_headers = { + "User-Agent": _get_user_agent(), + "PAYMENT-SIGNATURE": paid_completion_header, + } + paid_completion_header = None + state.completion_sent = True + elif ownership and poll_kind is not None: + # A Retry-After never sleeps past the poll budget. + await asyncio.sleep( + interval + if next_wait is None + else min(next_wait, max(0.0, deadline - _time.monotonic())) + ) + next_wait = None + request_headers = { + "User-Agent": _get_user_agent(), + **poll_auth_headers(poll_keypair, poll_kind, poll_job), + } + else: + await asyncio.sleep(interval) + + # Keep the settlement blockhash fresh (poll-based media path only, + # gated on max_resigns). Re-sign the ORIGINAL challenge — same + # amount/pay_to, only a freshly-fetched blockhash — so that whenever + # upstream flips to "completed" the signature is + # 0 + and payment_required is not None + and _time.monotonic() - last_resign_at + >= SolanaLLMClient.MEDIA_RESIGN_FRESH_SECONDS + ): + try: + fresh_payload = await self._sign_payment(payment_required) + poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( + fresh_payload + ) + last_resign_at = _time.monotonic() + except Exception: + # Best-effort only: a failed proactive re-sign (RPC hiccup, + # SolanaRpcException, etc.) must never abort the poll loop — + # we simply keep the prior signature (pre-fix behaviour). + pass + request_headers = poll_headers + + try: + poll_resp = await self._client.get( + poll_url, headers=request_headers, timeout=eff_timeout + ) + except httpx.TransportError as exc: + if not sending_paid_completion: + raise + # See the sync helper: the payment may have settled; the next + # ownership poll reads the job's settlement record. + if completion_payments_left > 0: + continue + raise PaymentError( + f"{label}: the paid poll for the finished job failed in transit " + f"({type(exc).__name__}). The payment may have settled: re-poll " + f"{poll_url} with the same wallet before submitting again." + ) from exc try: poll_data = poll_resp.json() except Exception: poll_data = {} + if not isinstance(poll_data, dict): + poll_data = {} last_status = poll_data.get("status", last_status) + if ownership and sending_paid_completion and poll_resp.status_code == 402: + # The one payment did not settle on this poll: another poll holds + # the job's settlement claim, the blockhash went stale, or an + # earlier settle's outcome is still unknown. The next ownership + # poll answers either "already settled" (no new charge) or a + # fresh challenge, so go back to polling rather than re-signing + # blind. + if completion_payments_left > 0: + next_wait = _retry_after_seconds(poll_resp) + continue + refusal = build_payment_rejected_error(poll_resp) + raise PaymentError( + f"{refusal} The payment for the finished job may still settle: " + f"re-poll {poll_url} with the same wallet before submitting again." + ) from refusal + + if ownership and not sending_paid_completion: + if poll_kind is not None and is_poll_ownership_refusal( + poll_kind, poll_resp.status_code, poll_data + ): + await _fall_back_to_payment_polls( + f"the gateway refused the poll ownership proof " + f"(HTTP {poll_resp.status_code} {poll_data.get('error')!r})" + ) + continue + if poll_resp.status_code == 402: + challenge_header = SolanaLLMClient._extract_payment_header(poll_resp) + if ( + poll_kind == "video" + and poll_data.get("status") == "completed" + and challenge_header + ): + # The clip is ready and unpaid: this 402 is the challenge + # for the one payment that settles it. + if completion_payments_left <= 0: + raise build_payment_rejected_error(poll_resp) + if _time.monotonic() >= deadline: + # Out of budget: sign nothing new (the payment a + # signature is already out for is the timeout's + # concern); the 504 below says how to resume. + break + completion_payments_left -= 1 + completion_required = decode_payment_required_header(challenge_header) + # Spend limits, before signing the one payment that + # settles the clip. The terms guard below pins it to the + # submit's amount, so that is the quote when the + # challenge has none to read. + completion_quote = _quoted_cost_usd(completion_required) + _enforce_spend_limits( + self, + completion_quote if completion_quote is not None else cost_usd, + body.get("model"), + ) + completion_payload = await self._sign_payment(completion_required) + if orig_amount is not None: + # Same guard as a mid-poll re-sign: the completion + # challenge must charge what the submit authorized. + _assert_same_payment_terms(completion_payload, orig_amount, orig_pay_to) + paid_completion_header = encode_payment_signature_header(completion_payload) + completion_paid = True + continue + # Any other 402 on an ownership poll (a batch-billed image + # job asks for its batch payment): the payment path. + await _fall_back_to_payment_polls("the poll answered 402") + continue + if poll_resp.status_code == 402: # Account rail: a 402 is the account being out of credit, not a # challenge to sign. Nothing here can sign, so say so plainly. raise_for_api_key_402(poll_resp, self.api_key) - # Mid-poll 402 = settlement failed, almost always a stale - # blockhash (the payment was signed at submit time but only - # settles when the job completes; by then the signed tx's - # recent-blockhash can be expired -> transaction_simulation_failed). - # The failing poll carries NO fresh challenge, so re-GET poll_url - # WITHOUT the stale signature to solicit a fresh 402 (new - # blockhash), re-sign, and keep polling. Mirrors the sync helper / - # Base VideoClient. + # Mid-poll 402 = settlement of the signed payment failed. For + # long jobs this is almost always a stale blockhash: the payment + # was signed at submit time, but the on-chain settlement only + # runs once the job completes, and by then the signed + # transaction's recent-blockhash can be expired — the facilitator + # reports ``transaction_simulation_failed``. The failing poll + # response carries NO fresh challenge, so re-GET poll_url WITHOUT + # the stale signature to solicit a fresh 402 (new blockhash), + # re-sign, and keep polling. Mirrors the Base VideoClient. A + # fresh signature that 402s again is a genuine payment problem. if resigns_left > 0: resigns_left -= 1 resign_payload = None @@ -5448,14 +6152,16 @@ async def _request_image_with_payment( resign_required = decode_payment_required_header(resign_header) resign_payload = await self._sign_payment(resign_required) except (PaymentError, httpx.HTTPError): - # Challenge GET or re-sign failed — surface the gateway's - # real 402 reason, not a network/signing error. + # Challenge GET failed, or signing was rejected — fall + # through to surface the gateway's real 402 reason rather + # than masking it with a network/signing error. Nothing + # settled here. resign_payload = None if resign_payload is not None: # Refuse a re-challenge that reprices or redirects the # payment vs. what this job originally authorized. This # PaymentError must propagate (NOT fall through to the - # generic 402); the guard also pins the amount, so the + # generic 402). The guard also pins the amount, so the # submit-time cost_usd stays correct for the ledger. _assert_same_payment_terms(resign_payload, orig_amount, orig_pay_to) poll_headers["PAYMENT-SIGNATURE"] = encode_payment_signature_header( @@ -5473,28 +6179,50 @@ async def _request_image_with_payment( else " (payment was settled at submit)" if settled_at_submit else "" ), poll_resp.status_code, - sanitize_error_response(poll_data if isinstance(poll_data, dict) else {}), + sanitize_error_response(poll_data), retry_after=retry_after_of(poll_resp), ) - # Terminal success is keyed on status, NOT the HTTP code (see the - # sync helper) — a completed-but-non-200 poll is still success. - if last_status == "completed": + # Terminal success is keyed on status, NOT the HTTP code — the + # gateway settles the moment a poll reports completed, so a + # completed-but-non-200 poll (which the caller was already charged + # for) must still be treated as success. + if last_status == "completed" and ( + not ownership_request or poll_resp.status_code == 200 + ): tx_hash = poll_resp.headers.get("x-payment-receipt") or poll_resp.headers.get( "X-Payment-Receipt" ) - if tx_hash and isinstance(poll_data, dict) and not poll_data.get("txHash"): + if tx_hash and not poll_data.get("txHash"): poll_data["txHash"] = tx_hash - if not charged_at_submit: + if not book_completion: + return poll_data + # An ownership poll can be handed a clip some earlier poll paid + # for ("already_settled"); only a payment this call signed is + # booked to this session. + paid_by_this_call = completion_paid or not ownership + if not charged_at_submit and paid_by_this_call: self._session_calls += 1 self._session_total_usd += cost_usd self._last_call_cost = cost_usd + # The cost log and transaction log record what this call paid: + # nothing, for a clip an earlier poll or call already settled. + logged_cost = cost_usd if charged_at_submit or paid_by_this_call else 0.0 self._capture_settlement(poll_resp) - save_to_cache(endpoint, body, poll_data, cost_usd=cost_usd, **self._billing_meta()) - self._log_transaction(endpoint, body, poll_data, cost_usd) + save_to_cache( + endpoint, body, poll_data, cost_usd=logged_cost, **self._billing_meta() + ) + self._log_transaction(endpoint, body, poll_data, logged_cost) return poll_data - if poll_resp.status_code in (202, 504): + if poll_resp.status_code in (202, 504) or ( + # A paid or proof poll answered 503 (the gateway's verifier + # is briefly down): nothing settled, so it is retriable. + ownership_request + and poll_resp.status_code == 503 + ): + # 202 = still queued/in_progress; 504 = transient upstream + # hiccup. Both are retriable inside the budget. continue if poll_resp.status_code != 200: @@ -5528,11 +6256,20 @@ async def _request_image_with_payment( ) if settled_at_submit else ( - f"{label} did not complete within {budget:.0f}s " - f"(last status: {last_status}). Settlement only happens on " - "completion, so no payment was taken. The job stays claimable " - "for ~48h — re-poll poll_url with a fresh signature from the " - "same wallet to fetch (and settle) the finished result." + ( + f"{label} did not complete within {budget:.0f}s " + f"(last status: {last_status}). A payment for the finished job " + "was sent and may still settle: re-poll poll_url with the same " + "wallet before submitting again (no new charge if it settled)." + ) + if state.completion_sent + else ( + f"{label} did not complete within {budget:.0f}s " + f"(last status: {last_status}). Settlement only happens on " + "completion, so no payment was taken. The job stays claimable " + "for ~48h — re-poll poll_url with a fresh signature from the " + "same wallet to fetch (and settle) the finished result." + ) ) ) ), diff --git a/tests/unit/test_solana_poll_ownership.py b/tests/unit/test_solana_poll_ownership.py new file mode 100644 index 0000000..c2d63fa --- /dev/null +++ b/tests/unit/test_solana_poll_ownership.py @@ -0,0 +1,1379 @@ +"""Solana media polls prove job ownership with an ed25519 signature, not a payment. + +Server contract: blockrun-sol ``src/lib/poll-ownership.ts`` and the poll routes +under ``src/app/api/v1/{videos,audio,images}/generations/[id]``. The fake gateway +below enforces it the way those routes do: an x-poll-* proof must verify (with +an ed25519 implementation independent of the SDK's) over +``blockrun-poll:v1:::`` for the wallet that paid the +submit, or the poll gets the route's "no such job" answer. + +The x402 codec and payment signer are stubbed as in test_solana_media.py; the +poll keypair is a real ``solders`` Keypair. +""" + +from __future__ import annotations + +import hashlib +import logging +import time +from types import SimpleNamespace +from typing import Any +from unittest import mock + +import httpx +import pytest + +pytest.importorskip("x402") +pytest.importorskip("solders") + +from solders.keypair import Keypair + +from blockrun_llm.poll_auth import ( + is_poll_ownership_refusal, + poll_auth_headers, + poll_auth_message, + poll_job_id, + poll_kind_for_url, +) +from blockrun_llm.solana_client import AsyncSolanaLLMClient, SolanaLLMClient +from blockrun_llm.types import APIError, MusicResponse, PaymentError + +# --------------------------------------------------------------------------- +# An ed25519 verifier and base58 decoder that share no code with the SDK +# (RFC 8032 section 5.1.7, affine coordinates; slow but exact). +# --------------------------------------------------------------------------- + +_P = 2**255 - 19 +_L = 2**252 + 27742317777372353535851937790883648493 +_D = -121665 * pow(121666, _P - 2, _P) % _P +_SQRT_M1 = pow(2, (_P - 1) // 4, _P) + + +def _inv(x: int) -> int: + return pow(x, _P - 2, _P) + + +def _recover_x(y: int, sign: int) -> int | None: + xx = (y * y - 1) * _inv(_D * y * y + 1) % _P + x = pow(xx, (_P + 3) // 8, _P) + if (x * x - xx) % _P != 0: + x = x * _SQRT_M1 % _P + if (x * x - xx) % _P != 0: + return None + if x == 0 and sign: + return None + if x & 1 != sign: + x = _P - x + return x + + +def _add(p: tuple[int, int], q: tuple[int, int]) -> tuple[int, int]: + (x1, y1), (x2, y2) = p, q + t = _D * x1 * x2 * y1 * y2 % _P + return ( + (x1 * y2 + x2 * y1) * _inv(1 + t) % _P, + (y1 * y2 + x1 * x2) * _inv(1 - t) % _P, + ) + + +def _mul(p: tuple[int, int], e: int) -> tuple[int, int]: + q = (0, 1) + while e: + if e & 1: + q = _add(q, p) + p = _add(p, p) + e >>= 1 + return q + + +_BY = 4 * _inv(5) % _P +_BASE = (_recover_x(_BY, 0) or 0, _BY) + + +def _decode_point(s: bytes) -> tuple[int, int] | None: + y = int.from_bytes(s, "little") + sign = y >> 255 + y &= (1 << 255) - 1 + if y >= _P: + return None + x = _recover_x(y, sign) + return None if x is None else (x, y) + + +def ed25519_verify(public_key: bytes, message: bytes, signature: bytes) -> bool: + if len(public_key) != 32 or len(signature) != 64: + return False + a = _decode_point(public_key) + r = _decode_point(signature[:32]) + if a is None or r is None: + return False + s = int.from_bytes(signature[32:], "little") + if s >= _L: + return False + h = int.from_bytes(hashlib.sha512(signature[:32] + public_key + message).digest(), "little") + return _mul(_BASE, s) == _add(r, _mul(a, h % _L)) + + +_B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" + + +def b58decode(value: str) -> bytes: + n = 0 + for ch in value: + n = n * 58 + _B58.index(ch) + body = n.to_bytes((n.bit_length() + 7) // 8, "big") + return b"\0" * (len(value) - len(value.lstrip("1"))) + body + + +def proof_is_valid( + headers: httpx.Headers, kind: str, job_id: str, wallet: str, *, skew_s: int = 300 +) -> bool: + """What the gateway checks: all three headers, a fresh timestamp, the + submitting wallet, and an ed25519 signature over the exact message.""" + w = headers.get("x-poll-wallet") + ts = headers.get("x-poll-timestamp") + sig = headers.get("x-poll-signature") + if not (w and ts and sig) or not ts.isdigit() or w != wallet: + return False + if abs(time.time() - int(ts)) > skew_s: + return False + message = f"blockrun-poll:v1:{kind}:{job_id}:{ts}".encode() + return ed25519_verify(b58decode(w), message, b58decode(sig)) + + +def has_proof(request: httpx.Request) -> bool: + return any(h in request.headers for h in ("x-poll-wallet", "x-poll-signature")) + + +# --------------------------------------------------------------------------- +# The proof itself +# --------------------------------------------------------------------------- + + +class TestProof: + def test_signature_verifies_over_the_exact_message(self) -> None: + kp = Keypair() + headers = httpx.Headers(poll_auth_headers(kp, "video", "job:abc/1", now_s=1_800_000_000)) + + assert headers["x-poll-wallet"] == str(kp.pubkey()) + assert headers["x-poll-timestamp"] == "1800000000" + pub, sig = b58decode(headers["x-poll-wallet"]), b58decode(headers["x-poll-signature"]) + assert len(pub) == 32 and len(sig) == 64 + assert ed25519_verify(pub, b"blockrun-poll:v1:video:job:abc/1:1800000000", sig) + # The verifier is not vacuous: the same signature fails for any other + # kind, job, timestamp, or version string. + for other in ( + b"blockrun-poll:v1:image:job:abc/1:1800000000", + b"blockrun-poll:v1:video:job:abc/2:1800000000", + b"blockrun-poll:v1:video:job:abc/1:1800000001", + b"blockrun-poll:v2:video:job:abc/1:1800000000", + ): + assert not ed25519_verify(pub, other, sig) + + def test_signature_matches_solders_for_a_fixed_key(self) -> None: + # Deterministic ed25519: a fixed seed and message give fixed bytes. + kp = Keypair.from_seed(bytes(range(32))) + headers = poll_auth_headers(kp, "audio", "mus_1", now_s=1_700_000_000) + assert poll_auth_message("audio", "mus_1", 1_700_000_000) == ( + "blockrun-poll:v1:audio:mus_1:1700000000" + ) + assert ed25519_verify( + bytes(kp.pubkey()), + b"blockrun-poll:v1:audio:mus_1:1700000000", + b58decode(headers["x-poll-signature"]), + ) + + def test_timestamp_defaults_to_now(self) -> None: + headers = poll_auth_headers(Keypair(), "image", "img_1") + assert abs(int(headers["x-poll-timestamp"]) - time.time()) <= 2 + + @pytest.mark.parametrize( + ("poll_url", "kind", "job_id"), + [ + ("/api/v1/videos/generations/vid%2F1?duration=8&sig=abc", "video", "vid/1"), + ("https://sol.blockrun.ai/api/v1/audio/generations/mus_9", "audio", "mus_9"), + ("/api/v1/images/generations/img%3A7", "image", "img:7"), + ("/v1/images/generations/img_2", "image", "img_2"), + ], + ) + def test_kind_and_job_id_from_poll_url(self, poll_url: str, kind: str, job_id: str) -> None: + assert poll_kind_for_url(poll_url) == kind + assert poll_job_id(poll_url) == job_id + + @pytest.mark.parametrize( + "poll_url", + ["/api/v1/audio/speech/x", "/api/v1/videos/x", "/api/v2/videos/generations/x", "/x"], + ) + def test_other_paths_get_no_kind(self, poll_url: str) -> None: + assert poll_kind_for_url(poll_url) is None + + def test_refusal_shapes(self) -> None: + assert is_poll_ownership_refusal("video", 400, {"error": "Invalid job id"}) + assert is_poll_ownership_refusal("audio", 404, {"error": "Job not found"}) + assert is_poll_ownership_refusal("image", 404, {"error": "Job not found"}) + # A different error, or the right error on another route's status, is not one. + assert not is_poll_ownership_refusal("video", 404, {"error": "Job not found"}) + assert not is_poll_ownership_refusal("video", 400, {"error": "Invalid duration"}) + assert not is_poll_ownership_refusal("image", 404, "Job not found") + + def test_real_wallet_key_gives_the_payer_keypair(self) -> None: + kp = Keypair() + with mock.patch("blockrun_llm.solana_client.register_exact_svm_client"): + full = SolanaLLMClient(private_key=str(kp), rpc_url="http://test") + seed_only = SolanaLLMClient( + private_key=_b58encode(bytes(kp)[:32]), rpc_url="http://test" + ) + assert full._poll_keypair.pubkey() == kp.pubkey() + assert seed_only._poll_keypair.pubkey() == kp.pubkey() + + def test_account_rail_has_no_poll_keypair(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BLOCKRUN_API_KEY", "brk_live_testkey") + assert SolanaLLMClient()._poll_keypair is None + + +def _b58encode(raw: bytes) -> str: + n = int.from_bytes(raw, "big") + out = "" + while n: + n, r = divmod(n, 58) + out = _B58[r] + out + return "1" * (len(raw) - len(raw.lstrip(b"\0"))) + out + + +# --------------------------------------------------------------------------- +# Fake gateway +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _stub_x402_codec(monkeypatch: pytest.MonkeyPatch) -> None: + # The challenge a payment is signed from travels through to the header the + # client sends, so a test can tell WHICH 402 a payment answered. + monkeypatch.setattr( + "blockrun_llm.solana_client.decode_payment_required_header", + lambda header: {"challenge": header}, + ) + monkeypatch.setattr( + "blockrun_llm.solana_client.encode_payment_signature_header", + lambda payload: f"paid-from:{payload.challenge['challenge']}", + ) + + +@pytest.fixture(autouse=True) +def _no_disk_cache(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("blockrun_llm.cache.get_cached", lambda *a, **k: None) + monkeypatch.setattr("blockrun_llm.cache.save_to_cache", lambda *a, **k: None) + + +def _payload_for(challenge: dict[str, str]) -> SimpleNamespace: + return SimpleNamespace( + accepted=SimpleNamespace(amount="1000000", pay_to="TREASURY"), challenge=challenge + ) + + +def _make_client(handler: Any, *, keypair: Keypair | None) -> SolanaLLMClient: + with ( + mock.patch("blockrun_llm.solana_client.register_exact_svm_client"), + mock.patch("blockrun_llm.solana_client._create_signer"), + ): + client = SolanaLLMClient( + private_key="bogus_signer_is_patched", + api_url="https://sol.blockrun.ai/api", + rpc_url="http://test", + ) + client._x402_client = mock.MagicMock() + client._x402_client.create_payment_payload.side_effect = _payload_for + client._client = httpx.Client(transport=httpx.MockTransport(handler)) + client._address = "11111111111111111111111111111111" + client._poll_keypair = keypair + return client + + +def _make_async_client(handler: Any, *, keypair: Keypair | None) -> AsyncSolanaLLMClient: + with ( + mock.patch("blockrun_llm.solana_client.register_exact_svm_client"), + mock.patch("blockrun_llm.solana_client._create_signer"), + ): + client = AsyncSolanaLLMClient( + private_key="bogus_signer_is_patched", + api_url="https://sol.blockrun.ai/api", + rpc_url="http://test", + ) + client._x402_client = mock.MagicMock() + client._x402_client.create_payment_payload = mock.AsyncMock(side_effect=_payload_for) + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + client._address = "11111111111111111111111111111111" + client._poll_keypair = keypair + return client + + +def _signed_challenges(client: Any) -> list[str]: + """Every challenge the wallet signed a payment from, in order.""" + return [ + c.args[0]["challenge"] for c in client._x402_client.create_payment_payload.call_args_list + ] + + +def _json(status: int, body: dict[str, Any], **headers: str) -> httpx.Response: + return httpx.Response( + status, json=body, headers={"content-type": "application/json", **headers} + ) + + +_VIDEO_DONE = { + "status": "completed", + "created": 1, + "model": "xai/grok-imagine-video", + "data": [{"url": "https://cdn/v.mp4"}], +} + + +class FakeVideoGateway: + """The sol video poll route: x-poll polls report status and never charge; + a finished unpaid job answers 402 + challenge + ``status: completed``; a + paid poll settles once; later polls answer ``already_settled``.""" + + JOB = "vid/1" + POLL_URL = "/api/v1/videos/generations/vid%2F1?duration=8&sig=abc" + + def __init__( + self, + wallet: str, + *, + pending: int = 2, + refuse_proofs: bool = False, + first_paid_poll: int = 200, + fail_after_pending: bool = False, + refuse_payments: bool = False, + ) -> None: + self.wallet = wallet + self.pending = pending + self.refuse_proofs = refuse_proofs + self.first_paid_poll = first_paid_poll + self.proof_polls: list[httpx.Request] = [] + self.paid_polls: list[httpx.Request] = [] + self.bare_polls: list[httpx.Request] = [] + self.settled = False + self.claim_in_flight_once = first_paid_poll == 402 + self.fail_after_pending = fail_after_pending + self.refuse_payments = refuse_payments + + def __call__(self, request: httpx.Request) -> httpx.Response: + paid = "PAYMENT-SIGNATURE" in request.headers + if request.method == "POST": + if not paid: + return _json( + 402, {"error": "Payment Required"}, **{"payment-required": "submit-challenge"} + ) + return _json(202, {"id": self.JOB, "poll_url": self.POLL_URL, "status": "queued"}) + + # The id is sent percent-encoded, exactly as the POST minted it. + assert request.url.raw_path.split(b"?")[0] == b"/api/v1/videos/generations/vid%2F1" + if paid: + assert not has_proof(request), "a paid poll carried an ownership proof too" + self.paid_polls.append(request) + if self.refuse_payments: + return _json( + 402, + {"error": "Payment verification failed", "details": "insufficient_funds"}, + ) + if self.claim_in_flight_once: + # A concurrent poll holds the claim and then settles THIS + # payment; the paid poll itself only hears "in progress". + self.claim_in_flight_once = False + self.settled = True + return _json(402, {"error": "Payment settlement in progress"}) + # Legacy payment polls report progress until the job is done. + if len(self.proof_polls) + len(self.paid_polls) <= self.pending: + return _json(202, {"status": "in_progress", "payment_status": "verified"}) + self.settled = True + return _json(200, _VIDEO_DONE, **{"x-payment-receipt": "SETTLE_TX"}) + + if has_proof(request): + self.proof_polls.append(request) + if self.refuse_proofs or not proof_is_valid( + request.headers, "video", self.JOB, self.wallet + ): + return _json( + 400, + { + "error": "Invalid job id", + "details": "Pass the job id from POST back verbatim", + }, + ) + if len(self.proof_polls) <= self.pending: + return _json(202, {"status": "in_progress", "payment_status": "not_charged"}) + if self.fail_after_pending: + return _json( + 200, + {"status": "failed", "error": "moderation", "payment_status": "not_charged"}, + ) + if self.settled: + return _json( + 200, + {**_VIDEO_DONE, "payment": {"status": "already_settled", "tx_hash": "TX"}}, + ) + return _json( + 402, + {"status": "completed", "error": "Payment Required", "job_id": self.JOB}, + **{"payment-required": f"completion-challenge-{len(self.proof_polls)}"}, + ) + + self.bare_polls.append(request) + return _json(402, {"error": "Payment Required"}, **{"payment-required": "bare-challenge"}) + + +_VIDEO_KW: dict[str, Any] = { + "poll_budget_seconds": 5.0, + "poll_interval_seconds": 0.001, + "max_resigns": 2, + "label": "Video generation", +} +_VIDEO_BODY = {"model": "xai/grok-imagine-video", "prompt": "a cat"} + + +# --------------------------------------------------------------------------- +# Video: status by proof, ONE payment at completion +# --------------------------------------------------------------------------- + + +class TestVideo: + def test_pending_polls_carry_a_proof_and_completion_pays_once(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=3) + client = _make_client(gw, keypair=kp) + + data = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert data["txHash"] == "SETTLE_TX" + # 3 pending proofs + the one that found the clip finished and unpaid. + assert len(gw.proof_polls) == 4 + for req in gw.proof_polls: + assert "PAYMENT-SIGNATURE" not in req.headers + assert proof_is_valid(req.headers, "video", "vid/1", str(kp.pubkey())) + # Two payments in all: the submit, and one from the completion 402. + assert _signed_challenges(client) == ["submit-challenge", "completion-challenge-4"] + assert [r.headers["PAYMENT-SIGNATURE"] for r in gw.paid_polls] == [ + "paid-from:completion-challenge-4" + ] + assert gw.bare_polls == [] + # Booked once, for the clip. + assert client._session_calls == 1 + assert client._session_total_usd == pytest.approx(1.0) + + def test_pending_polls_never_call_the_payment_signer(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=6, fail_after_pending=True) + client = _make_client(gw, keypair=kp) + + with pytest.raises(APIError, match="failed upstream: moderation"): + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert len(gw.proof_polls) == 7 + assert _signed_challenges(client) == ["submit-challenge"] + assert gw.paid_polls == [] and gw.bare_polls == [] + assert client._session_calls == 0 + + def test_settlement_in_flight_is_not_paid_twice(self) -> None: + # The paid poll hears "settlement in progress"; the next proof poll + # finds the job settled and delivers it without another payment. + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=1, first_paid_poll=402) + client = _make_client(gw, keypair=kp) + + data = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert data["payment"]["status"] == "already_settled" + assert _signed_challenges(client) == ["submit-challenge", "completion-challenge-2"] + assert len(gw.paid_polls) == 1 + assert client._session_calls == 1 + + def test_completion_payments_are_bounded_and_surface_the_settle_reason(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0, refuse_payments=True) + client = _make_client(gw, keypair=kp) + + with pytest.raises(PaymentError, match="insufficient_funds"): + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + # 1 + max_resigns (2) completion payments, then stop. + assert len(gw.paid_polls) == 3 + assert client._session_calls == 0 + + def test_refused_proof_falls_back_once_to_payment_polls( + self, caplog: pytest.LogCaptureFixture + ) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=2, refuse_proofs=True) + client = _make_client(gw, keypair=kp) + + with caplog.at_level(logging.WARNING, logger="blockrun_llm.solana_client"): + data = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + # One refused proof, then payment-signature polls only. + assert len(gw.proof_polls) == 1 + assert len(gw.paid_polls) >= 1 + assert all(not has_proof(r) for r in gw.paid_polls) + assert any("falling back to payment-signature polling" in m for m in caplog.messages) + assert client._session_calls == 1 + + def test_a_proof_from_another_wallet_is_refused_then_falls_back(self) -> None: + # The gateway binds the job to the submit's payer; a proof by any other + # key is "no such job", and the client recovers through payments. + gw = FakeVideoGateway(str(Keypair().pubkey()), pending=0) + client = _make_client(gw, keypair=Keypair()) + + data = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert len(gw.proof_polls) == 1 and len(gw.paid_polls) == 1 + + def test_without_a_poll_keypair_polls_pay_as_before(self) -> None: + gw = FakeVideoGateway("unused", pending=1) + client = _make_client(gw, keypair=None) + + client._request_image_with_payment("/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW) + + assert gw.proof_polls == [] + assert len(gw.paid_polls) == 2 + + def test_completion_challenge_that_reprices_is_refused(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + client = _make_client(gw, keypair=kp) + + def reprice(challenge: dict[str, str]) -> SimpleNamespace: + payload = _payload_for(challenge) + if challenge["challenge"].startswith("completion"): + payload.accepted.amount = "99000000" + return payload + + client._x402_client.create_payment_payload.side_effect = reprice + with pytest.raises(Exception, match="changed the payment terms"): + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + assert gw.paid_polls == [] + + async def test_async_pending_proofs_then_one_payment(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=3) + client = _make_async_client(gw, keypair=kp) + try: + data = await client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + finally: + await client._client.aclose() + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert len(gw.proof_polls) == 4 + assert all( + proof_is_valid(r.headers, "video", "vid/1", str(kp.pubkey())) for r in gw.proof_polls + ) + assert all("PAYMENT-SIGNATURE" not in r.headers for r in gw.proof_polls) + assert _signed_challenges(client) == ["submit-challenge", "completion-challenge-4"] + assert len(gw.paid_polls) == 1 + + async def test_async_refused_proof_falls_back_once(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=1, refuse_proofs=True) + client = _make_async_client(gw, keypair=kp) + try: + data = await client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + finally: + await client._client.aclose() + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert len(gw.proof_polls) == 1 + + +# --------------------------------------------------------------------------- +# Image and music: paid at POST, polls never pay +# --------------------------------------------------------------------------- + + +class FakePaidAtPostGateway: + """The sol image / music poll routes: the job was paid at POST, a proof + poll returns status and results, a refused proof is 404 "Job not found".""" + + def __init__( + self, + kind: str, + route: str, + wallet: str, + *, + pending: int = 2, + done: dict[str, Any], + refuse_proofs: bool = False, + batch_billed: bool = False, + ) -> None: + self.kind = kind + self.route = route + self.wallet = wallet + self.pending = pending + self.done = done + self.refuse_proofs = refuse_proofs + self.batch_billed = batch_billed + self.proof_polls: list[httpx.Request] = [] + self.paid_polls: list[httpx.Request] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + paid = "PAYMENT-SIGNATURE" in request.headers + if request.method == "POST": + if not paid: + return _json( + 402, {"error": "Payment Required"}, **{"payment-required": "submit-challenge"} + ) + return _json( + 202, + { + "id": "job_1", + "poll_url": f"/api/v1/{self.route}/generations/job_1", + "status": "queued", + }, + **{"x-payment-receipt": "SUBMIT_TX"}, + ) + assert request.url.path == f"/api/v1/{self.route}/generations/job_1" + if paid: + self.paid_polls.append(request) + return _json(200, self.done) + if has_proof(request): + self.proof_polls.append(request) + if self.refuse_proofs or not proof_is_valid( + request.headers, self.kind, "job_1", self.wallet + ): + return _json(404, {"error": "Job not found"}) + if self.batch_billed: + return _json(402, {"error": "Payment Required", "details": "batch payment"}) + if len(self.proof_polls) <= self.pending: + return _json(202, {"status": "in_progress", "payment_status": "settled_optimistic"}) + return _json(200, self.done) + return _json(402, {"error": "Payment Required"}, **{"payment-required": "bare-challenge"}) + + +_IMAGE_DONE = { + "status": "completed", + "created": 1, + "data": [{"url": "https://cdn/i.png"}], +} +_MUSIC_DONE = { + "id": "job_1", + "status": "completed", + "created": 1, + "model": "minimax/music-2.5+", + "data": [{"url": "https://cdn/x.mp3", "duration_seconds": 30}], + "payment": {"status": "settled_at_post"}, +} +_IMAGE_KW: dict[str, Any] = { + "poll_budget_seconds": 5.0, + "poll_interval_seconds": 0.001, + "max_resigns": 2, + "label": "Image", + "settled_at_submit": True, +} +_IMAGE_BODY = {"model": "openai/gpt-image-2", "prompt": "a cat"} + + +class TestImage: + def test_proof_polls_never_pay(self) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway("image", "images", str(kp.pubkey()), done=_IMAGE_DONE) + client = _make_client(gw, keypair=kp) + + data = client._request_image_with_payment( + "/v1/images/generations", dict(_IMAGE_BODY), **_IMAGE_KW + ) + + assert data["data"][0]["url"] == "https://cdn/i.png" + assert len(gw.proof_polls) == 3 and gw.paid_polls == [] + assert all( + proof_is_valid(r.headers, "image", "job_1", str(kp.pubkey())) for r in gw.proof_polls + ) + assert _signed_challenges(client) == ["submit-challenge"] + assert client._session_calls == 1 # the POST, booked once + + def test_batch_billed_402_falls_back_to_the_payment_path(self) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway( + "image", "images", str(kp.pubkey()), done=_IMAGE_DONE, batch_billed=True + ) + client = _make_client(gw, keypair=kp) + + data = client._request_image_with_payment( + "/v1/images/generations", dict(_IMAGE_BODY), **_IMAGE_KW + ) + + assert data["data"][0]["url"] == "https://cdn/i.png" + assert len(gw.proof_polls) == 1 and len(gw.paid_polls) == 1 + + def test_refused_proof_falls_back_once(self) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway( + "image", "images", str(kp.pubkey()), done=_IMAGE_DONE, refuse_proofs=True + ) + client = _make_client(gw, keypair=kp) + + client._request_image_with_payment("/v1/images/generations", dict(_IMAGE_BODY), **_IMAGE_KW) + + assert len(gw.proof_polls) == 1 and len(gw.paid_polls) == 1 + + async def test_async_proof_polls_never_pay(self) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway("image", "images", str(kp.pubkey()), done=_IMAGE_DONE) + client = _make_async_client(gw, keypair=kp) + try: + data = await client._request_image_with_payment( + "/v1/images/generations", dict(_IMAGE_BODY), **_IMAGE_KW + ) + finally: + await client._client.aclose() + + assert data["data"][0]["url"] == "https://cdn/i.png" + assert len(gw.proof_polls) == 3 and gw.paid_polls == [] + assert _signed_challenges(client) == ["submit-challenge"] + + +@pytest.fixture +def _fast_audio_polls(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(SolanaLLMClient, "AUDIO_POLL_INTERVAL_SECONDS", 0.001) + monkeypatch.setattr(SolanaLLMClient, "AUDIO_POLL_BUDGET_SECONDS", 5.0) + + +@pytest.mark.usefixtures("_fast_audio_polls") +class TestMusic: + def test_slow_track_is_polled_with_proofs_and_never_pays(self) -> None: + # Before this change the wallet rail returned the {id, poll_url} stub + # and MusicResponse failed validation after the track was paid for. + kp = Keypair() + gw = FakePaidAtPostGateway("audio", "audio", str(kp.pubkey()), done=_MUSIC_DONE) + client = _make_client(gw, keypair=kp) + + resp = client.music("lo-fi beats") + + assert isinstance(resp, MusicResponse) + assert resp.data[0].url == "https://cdn/x.mp3" + assert resp.txHash == "SUBMIT_TX" # the POST's receipt, not a poll's + assert len(gw.proof_polls) == 3 and gw.paid_polls == [] + assert all( + proof_is_valid(r.headers, "audio", "job_1", str(kp.pubkey())) for r in gw.proof_polls + ) + assert _signed_challenges(client) == ["submit-challenge"] + assert client._session_calls == 1 + + def test_refused_proof_falls_back_once(self, caplog: pytest.LogCaptureFixture) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway( + "audio", "audio", str(kp.pubkey()), done=_MUSIC_DONE, refuse_proofs=True + ) + client = _make_client(gw, keypair=kp) + + with caplog.at_level(logging.WARNING, logger="blockrun_llm.solana_client"): + resp = client.music("lo-fi beats") + + assert resp.data[0].url == "https://cdn/x.mp3" + assert len(gw.proof_polls) == 1 and len(gw.paid_polls) == 1 + # The fallback poll carries a payment signed from the submit's own + # challenge (verify-only on this route: never settled again). + assert gw.paid_polls[0].headers["PAYMENT-SIGNATURE"] == "paid-from:submit-challenge" + assert any("falling back" in m for m in caplog.messages) + assert client._session_calls == 1 + + async def test_async_slow_track_is_polled_with_proofs(self) -> None: + kp = Keypair() + gw = FakePaidAtPostGateway("audio", "audio", str(kp.pubkey()), done=_MUSIC_DONE) + client = _make_async_client(gw, keypair=kp) + try: + resp = await client.music("lo-fi beats") + finally: + await client._client.aclose() + + assert resp.data[0].url == "https://cdn/x.mp3" + assert resp.txHash == "SUBMIT_TX" + assert len(gw.proof_polls) == 3 and gw.paid_polls == [] + assert _signed_challenges(client) == ["submit-challenge"] + + def test_inline_track_is_unchanged(self) -> None: + kp = Keypair() + + def handler(request: httpx.Request) -> httpx.Response: + if "PAYMENT-SIGNATURE" not in request.headers: + return _json(402, {"error": "x"}, **{"payment-required": "submit-challenge"}) + return _json(200, _MUSIC_DONE) + + resp = _make_client(handler, keypair=kp).music("lo-fi beats") + assert resp.data[0].url == "https://cdn/x.mp3" + + +# --------------------------------------------------------------------------- +# Rails that never send a proof +# --------------------------------------------------------------------------- + + +class TestOtherRails: + def test_base_music_polls_keep_the_payment_signature( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from blockrun_llm import MusicClient + + from ..helpers import TEST_PRIVATE_KEY, build_payment_required_response + + monkeypatch.setattr(MusicClient, "MUSIC_POLL_INTERVAL_SECONDS", 0.0) + polls: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + if "PAYMENT-SIGNATURE" not in request.headers: + return _json( + 402, + {"error": "Payment Required"}, + **{"payment-required": build_payment_required_response()}, + ) + return _json( + 202, + {"id": "m1", "status": "queued", "poll_url": "/api/v1/audio/generations/m1"}, + ) + polls.append(request) + if len(polls) == 1: + return _json(202, {"id": "m1", "status": "in_progress"}) + return _json(200, _MUSIC_DONE) + + client = MusicClient(private_key=TEST_PRIVATE_KEY) + client._client = httpx.Client(transport=httpx.MockTransport(handler)) + client.generate("chill") + + assert len(polls) == 2 + assert all("PAYMENT-SIGNATURE" in r.headers and not has_proof(r) for r in polls) + + def test_account_rail_polls_carry_no_proof(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("BLOCKRUN_API_KEY", "brk_live_testkey") + client = SolanaLLMClient() + polls: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + return _json( + 202, + { + "id": "job_1", + "status": "queued", + "poll_url": "/api/v1/images/generations/job_1", + }, + ) + polls.append(request) + return _json(200, _IMAGE_DONE) + + client._client = httpx.Client( + transport=httpx.MockTransport(handler), headers=client._client.headers + ) + client._request_image_with_payment("/v1/images/generations", dict(_IMAGE_BODY), **_IMAGE_KW) + assert len(polls) == 1 and not has_proof(polls[0]) + assert "PAYMENT-SIGNATURE" not in polls[0].headers + + +# --------------------------------------------------------------------------- +# Money already moved (or may have): a lost paid poll, logging, timeouts +# --------------------------------------------------------------------------- + + +class LostPaidPollGateway(FakeVideoGateway): + """The paid completion poll reaches the gateway, which settles it, but the + answer is lost in transit (``settles=True``); or it never arrives and + nothing settles (``settles=False``).""" + + def __init__(self, wallet: str, *, settles: bool, **kw: Any) -> None: + super().__init__(wallet, **kw) + self.settles = settles + + def __call__(self, request: httpx.Request) -> httpx.Response: + if request.method == "GET" and "PAYMENT-SIGNATURE" in request.headers: + self.paid_polls.append(request) + if self.settles: + self.settled = True + raise httpx.ReadTimeout("answer lost", request=request) + return super().__call__(request) + + +def _is_tagged_settled(exc: BaseException) -> bool: + return bool(getattr(exc, "blockrun_payment_settled", False)) + + +class TestPaidCompletionInDoubt: + def test_lost_paid_poll_resumes_and_delivers_without_paying_again(self) -> None: + kp = Keypair() + gw = LostPaidPollGateway(str(kp.pubkey()), settles=True, pending=1) + client = _make_client(gw, keypair=kp) + + data = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert data["payment"]["status"] == "already_settled" + # The submit and ONE completion payment; the next proof poll found it settled. + assert _signed_challenges(client) == ["submit-challenge", "completion-challenge-2"] + assert len(gw.paid_polls) == 1 + assert client._session_calls == 1 + assert client._session_total_usd == pytest.approx(1.0) + + async def test_async_lost_paid_poll_resumes_and_delivers(self) -> None: + kp = Keypair() + gw = LostPaidPollGateway(str(kp.pubkey()), settles=True, pending=1) + client = _make_async_client(gw, keypair=kp) + try: + data = await client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + finally: + await client._client.aclose() + + assert data["payment"]["status"] == "already_settled" + assert _signed_challenges(client) == ["submit-challenge", "completion-challenge-2"] + assert client._session_calls == 1 + + def test_lost_paid_polls_exhausted_raise_tagged_possibly_charged(self) -> None: + kp = Keypair() + gw = LostPaidPollGateway(str(kp.pubkey()), settles=False, pending=0) + client = _make_client(gw, keypair=kp) + + with pytest.raises(PaymentError, match="may have settled") as info: + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + + assert _is_tagged_settled(info.value) + assert len(gw.paid_polls) == 3 # 1 + max_resigns, then stop + + async def test_async_lost_paid_polls_exhausted_raise_tagged(self) -> None: + kp = Keypair() + gw = LostPaidPollGateway(str(kp.pubkey()), settles=False, pending=0) + client = _make_async_client(gw, keypair=kp) + try: + with pytest.raises(PaymentError, match="may have settled") as info: + await client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + finally: + await client._client.aclose() + + assert _is_tagged_settled(info.value) + assert len(gw.paid_polls) == 3 + + def test_refused_completion_payments_are_tagged(self) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0, refuse_payments=True) + client = _make_client(gw, keypair=kp) + + with pytest.raises(PaymentError) as info: + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + # A payment for the finished job went out: no fallback chain may buy + # another job on top of it. + assert _is_tagged_settled(info.value) + + @pytest.mark.parametrize("use_async", [False, True]) + async def test_budget_runs_out_after_the_completion_payment(self, use_async: bool) -> None: + # Every paid poll hears "settlement in progress" with a Retry-After + # longer than the budget: the call ends, but a payment is out there. + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "GET" and "PAYMENT-SIGNATURE" in request.headers: + gw.paid_polls.append(request) + return _json( + 402, {"error": "Payment settlement in progress"}, **{"retry-after": "60"} + ) + return gw(request) + + kw = {**_VIDEO_KW, "poll_budget_seconds": 1.5} + if use_async: + aclient = _make_async_client(handler, keypair=kp) + try: + with pytest.raises(APIError) as info: + await aclient._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **kw + ) + finally: + await aclient._client.aclose() + else: + client = _make_client(handler, keypair=kp) + with pytest.raises(APIError) as info: + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **kw + ) + + assert info.value.status_code == 504 + assert "may still settle" in str(info.value) + assert "no payment was taken" not in str(info.value) + assert _is_tagged_settled(info.value) + # The Retry-After was honoured: one paid poll, not one per interval. + assert len(gw.paid_polls) == 1 + + @pytest.mark.parametrize("use_async", [False, True]) + async def test_clip_settled_elsewhere_is_logged_at_zero_cost(self, use_async: bool) -> None: + # The first proof poll finds the job already paid (an earlier call or + # poll settled it): this call paid nothing and must not log a charge. + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + gw.settled = True + logged: list[float] = [] + cached: list[float] = [] + + def fake_save(*a: Any, cost_usd: float = -1.0, **k: Any) -> None: + cached.append(cost_usd) + + with mock.patch("blockrun_llm.cache.save_to_cache", fake_save): + if use_async: + aclient = _make_async_client(gw, keypair=kp) + aclient._log_transaction = lambda e, b, r, cost: logged.append(cost) # type: ignore[method-assign] + try: + await aclient._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + finally: + await aclient._client.aclose() + assert aclient._session_calls == 0 + else: + client = _make_client(gw, keypair=kp) + client._log_transaction = lambda e, b, r, cost: logged.append(cost) # type: ignore[method-assign] + client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + assert client._session_calls == 0 + + assert logged == [0.0] + assert cached == [0.0] + assert gw.paid_polls == [] + + +# --------------------------------------------------------------------------- +# Spend limits: image and video refuse BEFORE signing +# --------------------------------------------------------------------------- + + +class _Challenge(dict): + """A decoded 402 the stubbed codec can carry, with readable x402 accepts.""" + + def __init__(self, header: str, amount: str = "1000000") -> None: + super().__init__(challenge=header) + self.accepts = [ + SimpleNamespace(scheme="exact", network="solana:mainnet", amount=amount), + ] + + +@pytest.fixture +def _readable_quotes(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "blockrun_llm.solana_client.decode_payment_required_header", + lambda header: _Challenge(header), + ) + + +def _limit(client: Any, *, per_call: float | None = None, session: float | None = None) -> None: + client._max_cost_per_call = per_call + client._max_session_cost = session + + +class TestSpendLimits: + @pytest.mark.usefixtures("_readable_quotes") + @pytest.mark.parametrize("use_async", [False, True]) + @pytest.mark.parametrize("method", ["image", "video"]) + async def test_per_call_limit_refuses_before_signing( + self, use_async: bool, method: str + ) -> None: + from blockrun_llm.types import SpendLimitError + + kp = Keypair() + if method == "image": + gw: Any = FakePaidAtPostGateway("image", "images", str(kp.pubkey()), done=_IMAGE_DONE) + else: + gw = FakeVideoGateway(str(kp.pubkey())) + sent: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + sent.append(request) + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + _limit(client, per_call=0.5) + try: + with pytest.raises(SpendLimitError): + call = getattr(client, method)("a cat") + if use_async: + await call + finally: + if use_async: + await client._client.aclose() + + assert _signed_challenges(client) == [] # refused before signing + assert all("PAYMENT-SIGNATURE" not in r.headers for r in sent) + assert client._session_total_usd == 0 + + @pytest.mark.parametrize("use_async", [False, True]) + async def test_unreadable_quote_is_checked_before_sending(self, use_async: bool) -> None: + # The stub codec's challenge has no accepts: the signed amount is + # checked instead, still before the paid POST goes out. + from blockrun_llm.types import SpendLimitError + + kp = Keypair() + gw = FakePaidAtPostGateway("image", "images", str(kp.pubkey()), done=_IMAGE_DONE) + sent: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + sent.append(request) + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + _limit(client, session=1.5) + client._session_total_usd = 1.0 + try: + with pytest.raises(SpendLimitError): + call = client.image("a cat") + if use_async: + await call + finally: + if use_async: + await client._client.aclose() + + assert all("PAYMENT-SIGNATURE" not in r.headers for r in sent) + assert client._session_total_usd == pytest.approx(1.0) + + @pytest.mark.usefixtures("_readable_quotes") + @pytest.mark.parametrize("use_async", [False, True]) + async def test_session_limit_refuses_the_completion_payment(self, use_async: bool) -> None: + # The submit fits the session budget; other calls spend while the clip + # renders, so the completion payment would breach it. Refused before + # the completion challenge is signed; the job stays claimable. + from blockrun_llm.types import SpendLimitError + + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=1) + holder: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "GET" and has_proof(request): + holder["client"]._session_total_usd = 1.0 # spent elsewhere meanwhile + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + holder["client"] = client + _limit(client, session=1.5) + try: + with pytest.raises(SpendLimitError) as info: + call = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + if use_async: + await call + finally: + if use_async: + await client._client.aclose() + + assert _signed_challenges(client) == ["submit-challenge"] + assert gw.paid_polls == [] + assert not _is_tagged_settled(info.value) + + +# --------------------------------------------------------------------------- +# The raw paid POST's challenge is per thread / task, and never cached over +# --------------------------------------------------------------------------- + + +class TestRawPaymentIsolation: + def test_another_thread_never_sees_this_threads_payment(self) -> None: + import threading + + client = _make_client(lambda r: _json(200, {}), keypair=Keypair()) + client._last_raw_payment = ("challenge-A", "payload-A") + seen: list[Any] = [] + t = threading.Thread(target=lambda: seen.append(client._last_raw_payment)) + t.start() + t.join() + + assert seen == [None] + assert client._last_raw_payment == ("challenge-A", "payload-A") + + async def test_another_task_never_sees_this_tasks_payment(self) -> None: + import asyncio + + client = _make_async_client(lambda r: _json(200, {}), keypair=Keypair()) + try: + gate = asyncio.Event() + + async def payer() -> None: + client._last_raw_payment = ("challenge-A", "payload-A") + gate.set() + + async def other() -> Any: + await gate.wait() + return client._last_raw_payment + + _, seen = await asyncio.gather(payer(), other()) + finally: + await client._client.aclose() + assert seen is None + + def test_a_cache_hit_does_not_inherit_a_prior_payment(self) -> None: + client = _make_client(lambda r: _json(200, {}), keypair=Keypair()) + client._last_raw_payment = ("old-challenge", "old-payload") + with mock.patch("blockrun_llm.cache.get_cached", lambda *a, **k: {"cached": True}): + client._request_with_payment_raw_once("/v1/audio/generations", {"prompt": "x"}) + assert client._last_raw_payment is None + + +# --------------------------------------------------------------------------- +# The completion 402 says status "completed": that is not a delivered clip +# --------------------------------------------------------------------------- + + +class TestCompletedIsNotDelivered: + @pytest.mark.parametrize("use_async", [False, True]) + async def test_verifier_outage_on_the_paid_poll_is_retried_not_returned( + self, use_async: bool + ) -> None: + # The paid GET answers 503 (no status field) once; the clip is then + # paid and delivered. The 503 body must never be returned as the video. + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + outage = {"left": 1} + + def handler(request: httpx.Request) -> httpx.Response: + if ( + request.method == "GET" + and "PAYMENT-SIGNATURE" in request.headers + and outage["left"] + ): + outage["left"] -= 1 + return _json(503, {"error": "Payment verification temporarily unavailable"}) + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + try: + call = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + data = await call if use_async else call + finally: + if use_async: + await client._client.aclose() + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert client._session_calls == 1 + assert client._session_total_usd == pytest.approx(1.0) + + @pytest.mark.parametrize("use_async", [False, True]) + @pytest.mark.parametrize("status", [500, 524]) + async def test_an_error_after_the_paid_poll_raises_tagged( + self, use_async: bool, status: int + ) -> None: + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "GET" and "PAYMENT-SIGNATURE" in request.headers: + gw.paid_polls.append(request) + return httpx.Response(status, text="edge error") + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + try: + with pytest.raises(APIError) as info: + call = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + if use_async: + await call + finally: + if use_async: + await client._client.aclose() + + assert info.value.status_code == status + assert _is_tagged_settled(info.value) + assert client._session_calls == 0 and client._session_total_usd == 0 + + +@pytest.mark.usefixtures("_fast_audio_polls") +class TestMusicFallbackRefusal: + @pytest.mark.parametrize("use_async", [False, True]) + async def test_a_refused_fallback_payment_is_tagged_paid(self, use_async: bool) -> None: + # The track was paid at POST; the proof is refused and the fallback's + # verify-only payment poll is rejected (the wallet just spent its USDC). + kp = Keypair() + gw = FakePaidAtPostGateway( + "audio", "audio", str(kp.pubkey()), done=_MUSIC_DONE, refuse_proofs=True + ) + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "GET" and "PAYMENT-SIGNATURE" in request.headers: + gw.paid_polls.append(request) + return _json( + 402, + {"error": "Payment verification failed", "details": "insufficient_funds"}, + ) + return gw(request) + + client: Any = (_make_async_client if use_async else _make_client)(handler, keypair=kp) + try: + with pytest.raises(PaymentError) as info: + call = client.music("lo-fi beats") + if use_async: + await call + finally: + if use_async: + await client._client.aclose() + + assert _is_tagged_settled(info.value) + + +@pytest.mark.parametrize("use_async", [False, True]) +async def test_a_completion_payment_signed_at_the_deadline_is_still_sent( + use_async: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + # The budget expires while the completion challenge is being signed: the + # signed payment must still go out and deliver the clip. + kp = Keypair() + gw = FakeVideoGateway(str(kp.pubkey()), pending=0) + real_monotonic = time.monotonic + clock = {"skew": 0.0} + # Only ever jumps forward, so the event loop's timers stay sane. + monkeypatch.setattr(time, "monotonic", lambda: real_monotonic() + clock["skew"]) + + def slow_sign(challenge: dict[str, str]) -> SimpleNamespace: + if challenge["challenge"].startswith("completion"): + clock["skew"] += 3600.0 # well past any budget + return _payload_for(challenge) + + client: Any = (_make_async_client if use_async else _make_client)(gw, keypair=kp) + if use_async: + client._x402_client.create_payment_payload = mock.AsyncMock(side_effect=slow_sign) + else: + client._x402_client.create_payment_payload.side_effect = slow_sign + try: + call = client._request_image_with_payment( + "/v1/videos/generations", dict(_VIDEO_BODY), **_VIDEO_KW + ) + data = await call if use_async else call + finally: + if use_async: + await client._client.aclose() + + assert data["data"][0]["url"] == "https://cdn/v.mp4" + assert len(gw.paid_polls) == 1 + assert client._session_calls == 1