server.py

"""Cross-language harness adapter for the Python PayKit umbrella surface.

One TCP server, two settle paths (x402:exact and mpp:charge), picked per
scenario by which env namespace the harness orchestrator sets (or by the
explicit ``PAY_KIT_HARNESS_PROTOCOL`` hint). Mirrors ``harness/php-server/
server.php`` and the Ruby/Lua pay-kit-server pattern.

This adapter routes every request through the unified ``solana_pay_kit`` surface:

  * x402 exact  -> ``solana_pay_kit.protocols.x402.X402Adapter`` (the umbrella adapter)
  * MPP charge  -> ``solana_pay_kit.protocols.mpp.server.charge.Mpp`` (the lower-level wire)

This split mirrors the canonical PHP adapter (``harness/php-server/
server.php``): x402 routes through the umbrella adapter, while MPP charge
routes through the lower-level ``solana_pay_kit.protocols.mpp`` handler. The umbrella's
ticker-based currency model (``Stablecoin`` enum -> ``Mints.resolve``) is the
right surface for x402, where the offer's ``asset`` is the resolved on-chain
mint; but the harness MPP charge matrix runs in *pubkey mode* (the harness
deploys the scenario mint at an arbitrary ``MPP_HARNESS_MINT`` pubkey, not the
canonical USDC mint), so the MPP challenge must advertise that literal mint as
its ``currency``. The lower-level ``solana_pay_kit.protocols.mpp`` handler takes the raw mint
directly, exactly as the PHP ``SolanaChargeHandler`` path does.

Cross-route replay protection on the MPP path is enforced by
``Mpp.verify_credential_with_expected`` (pins amount/currency/recipient per
route); the x402 path pins via ``X402Adapter``'s offer-equality gate.

Stdout discipline: ONLY the ``ready`` JSON line is written to stdout. All
diagnostics go to stderr.
"""

from __future__ import annotations

import asyncio
import json
import os
import socket
import sys
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import Any


def _find_repo_root(start: Path) -> Path:
    for candidate in [start, *start.parents]:
        if (candidate / ".git").exists() or (candidate / "python" / "pyproject.toml").is_file():
            return candidate
    return start.parents[-1]


_repo_root = _find_repo_root(Path(__file__).resolve())
_python_src = _repo_root / "python" / "src"
if _python_src.is_dir():
    sys.path.insert(0, str(_python_src))

from solana_pay_kit import (  # noqa: E402
    Config,
    Gate,
    Network,
    Operator,
    Price,
    Protocol,
    Signer,
    Stablecoin,
)
from solana_pay_kit._paycore.errors import (  # noqa: E402
    CODE_SETTLEMENT_IN_PROGRESS,
    PaymentError,
    canonical_code,
    settlement_in_progress_response,
)
from solana_pay_kit._paycore.rpc import SolanaRpc  # noqa: E402
from solana_pay_kit._paycore.store import MemoryStore  # noqa: E402
from solana_pay_kit.errors import InvalidProofError, SettlementInProgressError  # noqa: E402
from solana_pay_kit.protocols.mpp.core.headers import format_www_authenticate, parse_authorization, parse_receipt  # noqa: E402
from solana_pay_kit.protocols.mpp.intents.charge import ChargeRequest  # noqa: E402
from solana_pay_kit.protocols.mpp._paymentchannels import PROGRAM_ID as PAYMENT_CHANNELS_PROGRAM_ID  # noqa: E402
from solana_pay_kit.protocols.mpp.server import (  # noqa: E402
    MemoryChannelStore,
    Session,
    SessionChallengeOptions,
    SessionConfig,
    SessionServer,
    session_routes,
)
from solana_pay_kit.protocols.mpp.server.charge import (  # noqa: E402
    ChargeOptions,
    Mpp,
)
from solana_pay_kit.protocols.mpp.server.charge import Config as MppServerConfig  # noqa: E402
from solana_pay_kit.protocols.x402 import X402Adapter  # noqa: E402
from solana_pay_kit.protocols.x402.upto import X402Upto  # noqa: E402
from solana_pay_kit.usage import Charge, finalize_usage  # noqa: E402


def require_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        print(f"Missing required env: {name}", file=sys.stderr)
        sys.exit(2)
    return value


def optional_env(name: str, default: str) -> str:
    value = os.environ.get(name)
    return value if value else default


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


def _resolve_network(raw: str) -> Network:
    """Map the harness network string to a solana_pay_kit Network enum.

    Charge scenarios send the short slug ``localnet``; x402 scenarios send a
    CAIP-2 string (``solana:<genesis>``). Mirrors PHP ``resolve_network``.
    """
    if raw.startswith("solana:"):
        if raw == "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp":
            return Network.SOLANA_MAINNET
        # Devnet genesis and any other CAIP-2 fall to devnet (the localnet
        # surfpool fixtures are funded under the devnet genesis hash).
        return Network.SOLANA_DEVNET
    return {
        "mainnet": Network.SOLANA_MAINNET,
        "devnet": Network.SOLANA_DEVNET,
    }.get(raw, Network.SOLANA_LOCALNET)


def _base_units_to_human(base_units: str, decimals: int) -> str:
    """Convert a base-units string (e.g. ``"1000"``) into a decimal string."""
    if decimals <= 0:
        return str(int(base_units))
    units = int(base_units)
    sign = "-" if units < 0 else ""
    units = abs(units)
    quotient, remainder = divmod(units, 10**decimals)
    fraction = f"{remainder:0{decimals}d}".rstrip("0")
    if not fraction:
        return f"{sign}{quotient}"
    return f"{sign}{quotient}.{fraction}"


def _fetch_recent_state_sync(rpc_url: str) -> tuple[str | None, int | None]:
    """Fetch a recent blockhash + slot via a blocking JSON-RPC call (no asyncio).

    The x402 upto challenge requires ``extra.recentBlockhash`` (so the client can
    build the channel-open transaction) and ``extra.recentSlot`` (the channel's
    ``openSlot``, a PDA seed). Both come from the one ``getLatestBlockhash``
    response — the slot rides in its ``context``. ``accepts_entry`` runs both at
    challenge time and inside ``verify_open`` (itself under ``asyncio.run``), so
    the provider must be synchronous to avoid nesting event loops.
    """
    body = json.dumps(
        {"jsonrpc": "2.0", "id": 1, "method": "getLatestBlockhash", "params": [{"commitment": "confirmed"}]}
    ).encode("utf-8")
    request = urllib.request.Request(rpc_url, data=body, headers={"content-type": "application/json"})  # noqa: S310
    try:
        with urllib.request.urlopen(request, timeout=10) as resp:  # noqa: S310
            data = json.loads(resp.read())
        blockhash = data["result"]["value"]["blockhash"]
        slot = data["result"]["context"]["slot"]
        return (
            blockhash if isinstance(blockhash, str) and blockhash else None,
            slot if isinstance(slot, int) else None,
        )
    except Exception:  # noqa: BLE001 - recent-state fetch is best-effort at challenge time
        return (None, None)


def _coin_for_mint(mint: str) -> Stablecoin:
    """Pick the settlement Stablecoin for the scenario mint.

    The harness sends an on-chain mint pubkey (pubkey mode) or a ticker
    (symbol mode). The harness matrix's stablecoin is USDC; map any ticker we
    recognise, else default to USDC. The on-chain mint is asserted by the
    harness from the SDK's own resolver, so the ticker only selects which
    6-decimal coin the offer advertises.
    """
    try:
        return Stablecoin(mint)
    except ValueError:
        return Stablecoin.USDC


def _detect_protocol() -> str:
    """Decide which protocol this run exercises (mirror PHP detection)."""
    explicit = optional_env("PAY_KIT_HARNESS_PROTOCOL", "").lower()
    if explicit in ("x402-upto", "upto"):
        return "upto"
    if explicit in ("x402", "mpp", "charge", "session"):
        return "mpp" if explicit == "charge" else explicit
    x402_set = bool(os.environ.get("X402_HARNESS_RPC_URL"))
    mpp_set = bool(os.environ.get("MPP_HARNESS_RPC_URL"))
    if x402_set == mpp_set:
        print(
            "set exactly one of X402_HARNESS_RPC_URL / MPP_HARNESS_RPC_URL, or set PAY_KIT_HARNESS_PROTOCOL",
            file=sys.stderr,
        )
        sys.exit(2)
    return "x402" if x402_set else "mpp"


class _Adapter:
    """Holds the built solana_pay_kit adapter plus per-route gate amounts."""

    def __init__(self) -> None:
        self.protocol = _detect_protocol()
        self.x402 = self.protocol == "x402"
        if self.protocol == "x402":
            self._build_x402()
        elif self.protocol == "upto":
            self._build_upto()
        elif self.protocol == "session":
            self._build_session()
        else:
            self._build_mpp()

    # -- x402 -----------------------------------------------------------------

    def _build_x402(self) -> None:
        rpc_url = require_env("X402_HARNESS_RPC_URL")
        pay_to = require_env("X402_HARNESS_PAY_TO")
        facilitator_json = require_env("X402_HARNESS_FACILITATOR_SECRET_KEY")
        amount_units = optional_env("X402_HARNESS_AMOUNT", "1000")
        mint = optional_env("X402_HARNESS_MINT", "USDC")
        network_raw = optional_env("X402_HARNESS_NETWORK", "solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1")
        self.resource_path = optional_env("X402_HARNESS_RESOURCE_PATH", "/protected")
        self.settlement_header = optional_env("X402_HARNESS_SETTLEMENT_HEADER", "x-fixture-settlement").lower()
        self.coin = _coin_for_mint(mint)

        signer = Signer.json(facilitator_json)
        config = Config(
            network=_resolve_network(network_raw),
            accept=(Protocol.X402,),
            stablecoins=(self.coin,),
            rpc_url=rpc_url,
            operator=Operator(recipient=pay_to, signer=signer, fee_payer=True),
            preflight=False,
        ).model_copy()
        self.config = config
        self.adapter = X402Adapter(config)
        self.pay_to = pay_to
        decimals = int(optional_env("X402_HARNESS_DECIMALS", "6"))
        self.routes = {self.resource_path: _base_units_to_human(amount_units, decimals)}
        self.replay_path = ""

    # -- x402 upto ------------------------------------------------------------

    def _build_upto(self) -> None:
        rpc_url = require_env("X402_HARNESS_RPC_URL")
        pay_to = require_env("X402_HARNESS_PAY_TO")
        fee_payer_json = os.environ.get("X402_HARNESS_FEE_PAYER_SECRET_KEY") or require_env(
            "X402_HARNESS_FACILITATOR_SECRET_KEY"
        )
        mint = optional_env("X402_HARNESS_MINT", "USDC")
        network_raw = optional_env("X402_HARNESS_NETWORK", "solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1")
        self.resource_path = optional_env("X402_HARNESS_RESOURCE_PATH", "/usage")
        self.settlement_header = optional_env(
            "X402_HARNESS_SETTLEMENT_HEADER", "x-payment-settlement-signature"
        ).lower()
        self.price = optional_env("X402_HARNESS_PRICE", "0.10")
        # The metered amount the handler "charges" after serving (base units).
        self.actual_amount = int(optional_env("X402_HARNESS_ACTUAL_AMOUNT", "0"))
        self.coin = _coin_for_mint(mint)
        program_id = os.environ.get("PAYMENT_CHANNELS_PROGRAM_ID") or None

        signer = Signer.json(fee_payer_json)
        config = Config(
            network=_resolve_network(network_raw),
            accept=(Protocol.X402,),
            stablecoins=(self.coin,),
            rpc_url=rpc_url,
            operator=Operator(recipient=pay_to, signer=signer, fee_payer=True),
            preflight=False,
        ).model_copy()
        self.config = config
        self.pay_to = pay_to
        self.rpc_url = rpc_url
        self.upto_engine = X402Upto(
            config,
            channel_program=program_id,
            recent_state_provider=lambda: _fetch_recent_state_sync(rpc_url),
        )
        self.routes = {self.resource_path: self.price}
        self.replay_path = ""

    # -- mpp ------------------------------------------------------------------

    def _build_mpp(self) -> None:
        self.rpc_url = require_env("MPP_HARNESS_RPC_URL")
        pay_to = require_env("MPP_HARNESS_PAY_TO")
        # Pubkey mode: the literal scenario mint pubkey is the MPP currency.
        self.mint = require_env("MPP_HARNESS_MINT")
        amount_units = require_env("MPP_HARNESS_AMOUNT")
        secret = optional_env("MPP_HARNESS_SECRET_KEY", "mpp-harness-secret-key-with-32b-pad")
        network_raw = optional_env("MPP_HARNESS_NETWORK", "localnet")
        self.resource_path = optional_env("MPP_HARNESS_RESOURCE_PATH", "/paid")
        self.settlement_header = optional_env("MPP_HARNESS_SETTLEMENT_HEADER", "x-payment-settlement-signature").lower()
        realm = optional_env("MPP_HARNESS_REALM", "MPP Harness")
        self.splits = json.loads(optional_env("MPP_HARNESS_SPLITS", "[]"))
        if not isinstance(self.splits, list):
            print("MPP_HARNESS_SPLITS must decode to a JSON array", file=sys.stderr)
            sys.exit(2)

        fee_payer = None
        fee_payer_raw = os.environ.get("MPP_HARNESS_FEE_PAYER_SECRET_KEY")
        if fee_payer_raw:
            fee_payer = Signer.json(fee_payer_raw)
        self.fee_payer = fee_payer

        # Build the lower-level solana_pay_kit.protocols.mpp handler with the raw mint. The
        # ``Mpp`` server boots with ``rpc=None``; a request-lifetime
        # ``SolanaRpc`` is scoped via ``using_rpc`` in the request path.
        config = MppServerConfig(
            recipient=pay_to,
            currency=self.mint,
            decimals=int(optional_env("MPP_HARNESS_DECIMALS", "6")),
            network=network_raw,
            rpc_url=self.rpc_url,
            secret_key=secret,
            realm=realm,
            fee_payer_signer=fee_payer,
            store=MemoryStore(),
            rpc=None,
        )
        self.handler = Mpp(config)
        self.pay_to = pay_to

        decimals = int(optional_env("MPP_HARNESS_DECIMALS", "6"))
        self.routes = {self.resource_path: _base_units_to_human(amount_units, decimals)}
        replay_path = os.environ.get("MPP_HARNESS_REPLAY_SOURCE_PATH") or ""
        if replay_path:
            replay_amount = os.environ.get("MPP_HARNESS_REPLAY_SOURCE_AMOUNT") or amount_units
            self.routes[replay_path] = _base_units_to_human(replay_amount, decimals)
        self.replay_path = replay_path

    def _build_session(self) -> None:
        self.rpc_url = require_env("MPP_HARNESS_RPC_URL")
        pay_to = require_env("MPP_HARNESS_PAY_TO")
        amount_units = require_env("MPP_HARNESS_AMOUNT")
        secret = optional_env("MPP_HARNESS_SECRET_KEY", "mpp-harness-secret-key-with-32b-pad")
        network_raw = optional_env("MPP_HARNESS_NETWORK", "localnet")
        self.resource_path = optional_env("MPP_HARNESS_RESOURCE_PATH", "/session")
        self.settlement_header = optional_env("MPP_HARNESS_SETTLEMENT_HEADER", "x-session-settlement-signature").lower()
        fee_payer_raw = require_env("MPP_HARNESS_FEE_PAYER_SECRET_KEY")
        signer = Signer.json(fee_payer_raw)
        # The final session API (mpp-specs e702dd8) drops the draft
        # cap/modes/pullVoucherStrategy/openTxSubmitter options: the per-action
        # price is ``amount``, the pull + clientVoucher pair collapsed into
        # voucher_signer="client", and the challenge's recentBlockhash /
        # recentSlot come from a shared blockhash cache instead of a
        # recent_state_provider hook. ``new_session`` also now fails closed
        # without an RPC client, but this scenario's surfnet runs no
        # payment-channels program, so the wire-level trust model must stay
        # intact: the harness composes the lower-level SessionServer directly
        # with accept-all tx verifiers (open verification / broadcast /
        # settle-at-close stay off, exactly like the draft-era rpc=None mode)
        # and stamps challenge fields from one blocking getLatestBlockhash
        # call (the slot rides in its context).
        config = SessionConfig(
            operator=str(signer.pubkey),
            recipient=pay_to,
            amount=int(amount_units),
            currency=optional_env("MPP_HARNESS_SESSION_CURRENCY", "USDC"),
            decimals=int(optional_env("MPP_HARNESS_DECIMALS", "6")),
            network=network_raw,
            channel_program=os.environ.get("PAYMENT_CHANNELS_PROGRAM_ID") or str(PAYMENT_CHANNELS_PROGRAM_ID),
            # The final client derives its channel deposit from the challenge
            # (suggestedDeposit, falling back to minimumDeposit) and refuses a
            # challenge advertising neither.
            suggested_deposit=int(optional_env("MPP_HARNESS_SESSION_DEPOSIT", str(int(amount_units) * 10))),
            grace_period_seconds=900,
            voucher_signer="client",
            idle_timeout_seconds=300,
        )

        async def _accept_wire_only(*_: object) -> None:
            return None

        config.verify_open_tx = _accept_wire_only
        config.verify_top_up_tx = _accept_wire_only

        def _blockhash_cache() -> tuple[str, int] | None:
            blockhash, slot = _fetch_recent_state_sync(self.rpc_url)
            if blockhash is None or slot is None:
                return None
            return (blockhash, slot)

        def _current_slot() -> int:
            _, slot = _fetch_recent_state_sync(self.rpc_url)
            if slot is None:
                raise ValueError("session harness RPC did not return a current slot")
            return slot

        config.current_slot_provider = _current_slot

        core = SessionServer(config, MemoryChannelStore()).with_blockhash_cache(_blockhash_cache)
        self.session_method = Session(
            core=core,
            secret_key=secret,
            realm=optional_env("MPP_HARNESS_REALM", "MPP Harness"),
            amount=int(amount_units),
            currency=config.currency,
            recipient=pay_to,
            network=network_raw,
            rpc=None,
            lifecycle=None,
            signer=None,
        )
        # No touch hook: the harness Session runs without an idle-close
        # lifecycle (lifecycle=None), and session_routes expects a *sync*
        # TouchFn — passing the async Session._touch would leave un-awaited
        # coroutines behind.
        self.session_routes = session_routes(core)
        self.session_challenge = SessionChallengeOptions(amount=amount_units, description="Harness session")
        decimals = int(optional_env("MPP_HARNESS_DECIMALS", "6"))
        self.routes = {self.resource_path: _base_units_to_human(amount_units, decimals)}
        self.replay_path = ""

    def charge_options(self) -> ChargeOptions:
        options = ChargeOptions(
            description="PayKit Python harness protected content",
            splits=self.splits or [],
        )
        if self.fee_payer is not None:
            options.fee_payer = True
        return options

    # -- x402 gate ------------------------------------------------------------

    def gate_for(self, path: str) -> Gate:
        amount = self.routes[path]
        return Gate.build(
            name=path.lstrip("/") or "root",
            amount=Price.usd(amount, self.coin),
            default_pay_to=self.pay_to,
            accept=(Protocol.X402,),
            description="PayKit Python harness protected content",
        )


class HarnessHandler(BaseHTTPRequestHandler):
    server_version = "python-harness/1.0"

    def log_message(self, format: str, *args: Any) -> None:  # noqa: A002
        return

    @property
    def adapter(self) -> _Adapter:
        return self.server.adapter  # type: ignore[attr-defined]

    def _send_json(self, status: int, body: dict, extra_headers: dict | None = None) -> None:
        payload = json.dumps(body).encode("utf-8")
        self.send_response(status)
        headers = {"content-type": "application/json"}
        if extra_headers:
            for name, value in extra_headers.items():
                headers[name.lower()] = value
        headers["content-length"] = str(len(payload))
        headers["connection"] = "close"
        for name, value in headers.items():
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(payload)

    def _request_bag(self) -> dict[str, Any]:
        # Build the framework-agnostic request bag both adapters accept
        # (``.headers``-style getter and ``path``). Header names are
        # lower-cased so the adapters' case-tolerant lookups hit.
        headers = {name.lower(): value for name, value in self.headers.items()}
        return {"headers": headers, "path": self.path}

    def do_GET(self) -> None:  # noqa: N802
        if self.path == "/health":
            self._send_json(200, {"ok": True})
            return

        adapter = self.adapter
        if adapter.protocol == "session" and self.path == adapter.resource_path:
            self._handle_session(adapter)
            return
        if self.path not in adapter.routes:
            self._send_json(404, {"error": "not_found"})
            return

        request = self._request_bag()

        if adapter.protocol == "upto":
            self._handle_upto(adapter, adapter.gate_for(self.path), request)
        elif adapter.x402:
            self._handle_x402(adapter, adapter.gate_for(self.path), request)
        else:
            self._handle_mpp(adapter, request)

    def do_POST(self) -> None:  # noqa: N802
        adapter = self.adapter
        if adapter.protocol == "session":
            if self.path == adapter.resource_path:
                self._handle_session(adapter)
                return
            raw = self.rfile.read(int(self.headers.get("content-length", "0") or "0"))
            if self.path == "/__402/session/deliveries":
                response = asyncio.run(adapter.session_routes.deliveries("POST", raw or b"{}"))
                self._send_json(response.status, response.body)
                return
            if self.path == "/__402/session/commit":
                response = asyncio.run(adapter.session_routes.commit("POST", raw or b"{}"))
                self._send_json(response.status, response.body)
                return
            if self.path == "/__402/session/close":
                self._handle_session(adapter)
                return
        self._send_json(404, {"error": "not_found"})

    def _handle_x402(self, adapter: _Adapter, gate: Gate, request: dict[str, Any]) -> None:
        if not request["headers"].get("payment-signature"):
            challenge_headers = adapter.adapter.challenge_headers(gate, request)
            accepts = adapter.adapter.accepts_entry(gate, request)
            self._send_json(
                402,
                {"error": "payment_required", "resource": self.path, "accepts": [accepts]},
                extra_headers=challenge_headers,
            )
            return
        try:
            payment = asyncio.run(adapter.adapter.verify_and_settle(gate, request))
        except SettlementInProgressError as err:
            self._send_settlement_in_progress(str(err))
            return
        except InvalidProofError as err:
            self._send_json(
                402,
                {"error": err.code or "invalid_proof", "code": err.code, "message": str(err)},
                extra_headers=adapter.adapter.challenge_headers(gate, request),
            )
            return
        headers = dict(payment.settlement_headers)
        headers[adapter.settlement_header] = payment.transaction
        self._send_json(
            200,
            {"ok": True, "paid": True, "protocol": "x402", "transaction": payment.transaction},
            extra_headers=headers,
        )

    def _handle_upto(self, adapter: _Adapter, gate: Gate, request: dict[str, Any]) -> None:
        engine = adapter.upto_engine
        if not request["headers"].get("payment-signature"):
            challenge_headers = engine.challenge_headers(gate, request)
            accepts = engine.accepts_entry(gate, request)
            self._send_json(
                402,
                {"error": "payment_required", "resource": self.path, "accepts": [accepts]},
                extra_headers=challenge_headers,
            )
            return
        try:
            verified = asyncio.run(engine.verify_open(gate, request))
        except InvalidProofError as err:
            self._send_json(
                402,
                {"error": err.code or "invalid_proof", "code": err.code, "message": str(err)},
                extra_headers=engine.challenge_headers(gate, request),
            )
            return
        # Serve the resource: meter the configured actual amount, then settle
        # after the handler (fail-closed on a zero/absent charge, like Go PayKit).
        charge = Charge(verified.max_amount)
        if adapter.actual_amount > 0:
            charge.charge(adapter.actual_amount)
        outcome = asyncio.run(finalize_usage(engine, verified, charge))
        if not outcome.ok:
            self._send_json(
                402,
                {"error": "payment_required", "code": outcome.code, "message": outcome.detail or ""},
                extra_headers=engine.challenge_headers(gate, request),
            )
            return
        headers = dict(outcome.settlement_headers)
        headers[adapter.settlement_header] = outcome.transaction
        self._send_json(
            200,
            {"ok": True, "paid": True, "protocol": "x402-upto", "transaction": outcome.transaction},
            extra_headers=headers,
        )

    def _handle_session(self, adapter: _Adapter) -> None:
        auth = self.headers.get("authorization", "")
        result = asyncio.run(adapter.session_method.handle(auth or None, adapter.session_challenge))
        if not result.ok:
            self._send_json(result.status, result.body or {"error": "payment_required"}, extra_headers=result.headers)
            return
        receipt_header = result.headers.get("payment-receipt", "")
        reference = parse_receipt(receipt_header).reference if receipt_header else ""
        body = {"ok": True, "paid": True, "protocol": "session", "reference": reference}
        if reference:
            body["settledSignature"] = reference
        self._send_json(200, body, extra_headers={**result.headers, adapter.settlement_header: reference})

    def _handle_mpp(self, adapter: _Adapter, request: dict[str, Any]) -> None:
        amount = adapter.routes[self.path]
        options = adapter.charge_options()
        auth = request["headers"].get("authorization", "")

        if not auth:
            self._issue_mpp_challenge(adapter, amount, options, message="missing authorization")
            return

        try:
            credential = parse_authorization(auth)
        except Exception as exc:  # noqa: BLE001 - parse errors map to 402
            self._issue_mpp_challenge(
                adapter,
                amount,
                options,
                message=f"could not parse Authorization: {exc}",
                code="payment_invalid",
            )
            return

        try:
            challenge = adapter.handler.charge_with_options(amount, options)
            expected = ChargeRequest.from_dict(challenge.decode_request())

            async def _verify_with_fresh_rpc():
                fresh_rpc = SolanaRpc(adapter.rpc_url)
                try:
                    async with adapter.handler.using_rpc(fresh_rpc):
                        return await adapter.handler.verify_credential_with_expected(credential, expected)
                finally:
                    await fresh_rpc.aclose()

            receipt = asyncio.run(_verify_with_fresh_rpc())
        except PaymentError as err:
            if canonical_code(err.code) == CODE_SETTLEMENT_IN_PROGRESS:
                self._send_settlement_in_progress(str(err))
                return
            self._issue_mpp_challenge(
                adapter, amount, options, message=str(err) or "verification failed", code=err.code
            )
            return
        except Exception as err:  # noqa: BLE001 framework guard
            print(f"harness python server error: {err}", file=sys.stderr)
            self._issue_mpp_challenge(adapter, amount, options, message=str(err))
            return

        self._send_json(
            200,
            {"ok": True, "paid": True},
            extra_headers={
                "payment-receipt": receipt.reference,
                adapter.settlement_header: receipt.reference,
            },
        )

    def _send_settlement_in_progress(self, message: str) -> None:
        response = settlement_in_progress_response(message)
        self._send_json(response["status_code"], response["body"], extra_headers=response["headers"])

    def _issue_mpp_challenge(
        self,
        adapter: _Adapter,
        amount: str,
        options: ChargeOptions,
        *,
        message: str = "Payment required",
        code: str = "payment_invalid",
    ) -> None:
        try:
            challenge = adapter.handler.charge_with_options(amount, options)
        except PaymentError as exc:
            # Audit #21 promoted too-many-splits to a refuse-to-issue. The
            # conformance harness expects the 402-class outcome (no challenge to
            # advertise), not a 500. Re-raise anything else.
            if "too many splits" not in str(exc):
                raise
            invalid = canonical_code("payment_invalid")
            self._send_json(
                402,
                {
                    "type": f"https://paymentauth.org/problems/{invalid}",
                    "title": "Payment Required",
                    "status": 402,
                    "code": invalid,
                    "error": invalid,
                    "message": str(exc),
                },
                extra_headers={
                    "content-type": "application/problem+json",
                    "cache-control": "no-store",
                },
            )
            return
        canonical = canonical_code(code) if code else "payment_invalid"
        body = {
            "type": f"https://paymentauth.org/problems/{canonical}",
            "title": "Payment Required",
            "status": 402,
            "code": canonical,
            "error": canonical,
            "message": message,
        }
        self._send_json(
            402,
            body,
            extra_headers={
                "content-type": "application/problem+json",
                "www-authenticate": format_www_authenticate(challenge),
                "cache-control": "no-store",
            },
        )


def main() -> None:
    adapter = _Adapter()
    port = _free_port()
    server = HTTPServer(("127.0.0.1", port), HarnessHandler)
    server.adapter = adapter  # type: ignore[attr-defined]

    ready = {
        "type": "ready",
        "implementation": "python",
        "role": "server",
        "port": port,
        "capabilities": [adapter.protocol],
    }
    sys.stdout.write(json.dumps(ready) + "\n")
    sys.stdout.flush()

    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        thread.join()
    except KeyboardInterrupt:
        server.shutdown()


if __name__ == "__main__":
    main()