Blame for managed_auth.py

# -*- coding: utf-8 -*-
"""SmartPaper edge-first authentication and browser task/payment bridge.

Cloudflare owns the durable browser session, quota, tasks, and payments. The
Hugging Face process verifies the signed edge assertion but does not call the
Worker over the server-to-server network. Authenticated business operations are
executed by the user's browser through the same-origin invisible component.
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import html
import io
import json
import logging
import os
import re
import secrets
import time
from pathlib import Path
from typing import Any, Callable
from urllib.parse import unquote, urlsplit

import qrcode
import streamlit as st

from config import APP_CONFIG
from i18n import tr
from smartpaper_restore_bridge import edge_bridge

EDGE_ASSERTION_COOKIE_NAME = "smartpaper_edge_assertion"
EDGE_ASSERTION_AUDIENCE = "smartpaper-streamlit"
ASSERTION_CLOCK_SKEW_SECONDS = 30
def _runtime_value(name: str) -> str:
    value = os.environ.get(name, "")
    if not value:
        try:
            value = st.secrets.get(name, "")
        except Exception:
            value = ""
    return str(value or "").strip()


def _api_base() -> str:
    value = str(st.session_state.get("sp_auth_api_origin") or "").strip().rstrip("/")
    if not value:
        value = _runtime_value("SMARTPAPER_AUTH_API").rstrip("/")
    parsed = urlsplit(value)
    if (
        parsed.scheme != "https"
        or not parsed.netloc
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in ("", "/")
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("SMARTPAPER_AUTH_API must be an HTTPS origin.")
    return f"https://{parsed.netloc}"


def _service_key() -> str:
    value = _runtime_value("SMARTPAPER_SERVICE_KEY")
    if not value:
        raise RuntimeError("SMARTPAPER_SERVICE_KEY is not configured.")
    return value


def _context_cookie(name: str) -> str:
    try:
        cookies = st.context.cookies
        value = cookies.get(name, "")
        if value:
            return unquote(str(value).strip())
        target = name.lower()
        for key, item in cookies.items():
            if str(key).lower() == target:
                return unquote(str(item or "").strip())
    except Exception:
        return ""
    return ""


def _decode_b64url(value: str) -> bytes:
    return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def _verify_edge_assertion(token: str) -> dict[str, Any]:
    try:
        encoded, supplied_signature = token.split(".", 1)
    except ValueError as exc:
        raise RuntimeError("invalid_edge_assertion") from exc
    expected_signature = hmac.new(
        _service_key().encode("utf-8"),
        encoded.encode("ascii"),
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected_signature, supplied_signature):
        raise RuntimeError("invalid_edge_assertion_signature")
    try:
        payload = json.loads(_decode_b64url(encoded).decode("utf-8"))
    except Exception as exc:
        raise RuntimeError("invalid_edge_assertion_payload") from exc
    now = int(time.time())
    if payload.get("ver") != 1 or payload.get("aud") != EDGE_ASSERTION_AUDIENCE:
        raise RuntimeError("invalid_edge_assertion_claims")
    api_origin = str(payload.get("api_origin") or "").strip().rstrip("/")
    parsed = urlsplit(api_origin)
    if (
        parsed.scheme != "https"
        or not parsed.netloc
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in ("", "/")
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("invalid_edge_assertion_origin")
    issued_at = int(payload.get("iat") or 0)
    expires_at = int(payload.get("exp") or 0)
    if issued_at > now + ASSERTION_CLOCK_SKEW_SECONDS or expires_at <= now - ASSERTION_CLOCK_SKEW_SECONDS:
        raise RuntimeError("expired_edge_assertion")
    if expires_at - issued_at > 9 * 60 * 60:
        raise RuntimeError("invalid_edge_assertion_lifetime")
    user = payload.get("user") or {}
    if not isinstance(user, dict) or not user.get("id") or user.get("status") != "active":
        raise RuntimeError("invalid_edge_assertion_identity")
    payload["api_origin"] = f"https://{parsed.netloc}"
    return payload


def _clear_generation_session_state() -> None:
    for key in (
        "sp_generation_task", "sp_tracked_task_id", "sp_current_node",
        "sp_task_recovery_initialized", "sp_last_ui_task_status",
        "sp_generation_idempotency_key", "zip_buffer", "zip_filename",
        "is_running", "start_pipeline", "resume_pipeline", "run_mode",
        "rag_engine", "up_amj_ref", "up_amj_data", "up_asq_ref",
        "up_asq_data", "sp_payment_required", "sp_payment_order",
        "sp_payment_request_id", "sp_payment_package_sku",
        "sp_payment_channel", "sp_payment_order_channel",
        "sp_payment_dialog_pending",
    ):
        st.session_state.pop(key, None)


def _clear_session() -> None:
    for key in (
        "sp_session_expires_at", "sp_user",
        "sp_last_verified_at", "sp_edge_assertion", "sp_auth_api_origin",
        "sp_product_ui_projection", "sp_product_tier",
        "sp_account_pending_command_id",
    ):
        st.session_state.pop(key, None)


def require_alipay_login(language: str = "zh") -> dict[str, Any]:
    """Return the user authenticated by the Cloudflare edge gateway."""
    token = _context_cookie(EDGE_ASSERTION_COOKIE_NAME)
    if token and token == st.session_state.get("sp_edge_assertion") and st.session_state.get("sp_user"):
        return dict(st.session_state["sp_user"])
    if not token:
        _clear_session()
        st.stop()
    try:
        payload = _verify_edge_assertion(token)
    except Exception:
        _clear_session()
        st.stop()
    user = dict(payload["user"])
    st.session_state["sp_edge_assertion"] = token
    st.session_state["sp_session_expires_at"] = payload["exp"]
    st.session_state["sp_user"] = user
    st.session_state["sp_auth_api_origin"] = str(payload["api_origin"]).rstrip("/")
    st.session_state["sp_last_verified_at"] = time.time()
    return user


def _bridge_key(prefix: str, value: str) -> str:
    digest = hashlib.sha256(str(value).encode("utf-8")).hexdigest()[:20]
    return f"{prefix}_{digest}"


def _edge_call(
    action: str,
    command_id: str,
    payload: dict[str, Any] | None = None,
    *,
    key_prefix: str = "sp_edge",
) -> dict[str, Any] | None:
    return edge_bridge(
        action=action,
        command_id=command_id,
        api_origin=_api_base(),
        payload=payload or {},
        key=_bridge_key(key_prefix, command_id),
    )


def _raise_edge_error(result: dict[str, Any], fallback: str) -> None:
    message = (
        result.get("message")
        or result.get("reason")
        or result.get("error")
        or fallback
    )
    raise RuntimeError(str(message))


def _merge_quota_into_session(result: dict[str, Any]) -> None:
    """Refresh the approved account card immediately after a quota mutation."""
    quota = result.get("quota")
    if not isinstance(quota, dict):
        return
    user = dict(st.session_state.get("sp_user") or {})
    for key in (
        "quota_date",
        "daily_free_limit",
        "daily_free_reserved",
        "daily_free_used",
        "daily_free_remaining",
        "bonus_papers",
    ):
        if key in quota:
            user[key] = quota[key]
    if user:
        st.session_state["sp_user"] = user
        st.session_state["sp_account_revision"] = int(
            st.session_state.get("sp_account_revision") or 0
        ) + 1


def _legacy_product_ui_projection() -> dict[str, Any]:
    return {
        "schema_version": "a5-product-ui-v1",
        "mode": "legacy",
        "product_tier_selection_required": False,
        "products": {},
    }


def _normalize_product_ui_projection(value: Any) -> dict[str, Any]:
    """Accept only the Worker-owned A5 UI projection; never reinterpret legacy quota fields."""
    if not isinstance(value, dict):
        return _legacy_product_ui_projection()
    mode = str(value.get("mode") or "legacy").strip().lower()
    if mode == "legacy":
        return _legacy_product_ui_projection()
    if mode != "a5_product":
        raise RuntimeError("product_ui_projection_invalid_mode")

    raw_products = value.get("products")
    raw_entitlement = value.get("entitlement")
    if not isinstance(raw_products, dict) or not isinstance(raw_entitlement, dict):
        raise RuntimeError("product_ui_projection_incomplete")

    products: dict[str, dict[str, Any]] = {}
    for tier in ("STANDARD", "FLAGSHIP"):
        raw = raw_products.get(tier)
        if not isinstance(raw, dict):
            raise RuntimeError("product_ui_projection_missing_tier")
        try:
            minimum = int(raw.get("min_body_words"))
            maximum = int(raw.get("max_body_words"))
        except (TypeError, ValueError) as exc:
            raise RuntimeError("product_ui_projection_invalid_word_policy") from exc
        if minimum <= 0 or maximum < minimum:
            raise RuntimeError("product_ui_projection_invalid_word_policy")
        try:
            amount_fen = int(raw.get("amount_fen"))
            credits_granted = int(raw.get("credits_granted"))
        except (TypeError, ValueError) as exc:
            raise RuntimeError("product_ui_projection_invalid_billing_policy") from exc
        products[tier] = {
            "product_tier": tier,
            "product_policy_version": str(raw.get("product_policy_version") or ""),
            "word_count_policy_version": str(raw.get("word_count_policy_version") or ""),
            "body_metric": str(raw.get("body_metric") or ""),
            "min_body_words": minimum,
            "max_body_words": maximum,
            "counting_contract_version": str(raw.get("counting_contract_version") or ""),
            "package_sku": str(raw.get("package_sku") or ""),
            "billing_policy_version": str(raw.get("billing_policy_version") or ""),
            "amount_fen": amount_fen,
            "currency": str(raw.get("currency") or ""),
            "credits_granted": credits_granted,
        }
        if not all((
            products[tier]["product_policy_version"],
            products[tier]["word_count_policy_version"],
            products[tier]["body_metric"],
            products[tier]["counting_contract_version"],
            products[tier]["package_sku"],
            products[tier]["billing_policy_version"],
            products[tier]["currency"],
        )) or amount_fen <= 0 or credits_granted <= 0:
            raise RuntimeError("product_ui_projection_incomplete")

    try:
        standard_remaining = int(raw_entitlement.get("standard_remaining"))
        flagship_remaining = int(raw_entitlement.get("flagship_remaining"))
    except (TypeError, ValueError) as exc:
        raise RuntimeError("product_ui_projection_invalid_entitlement") from exc
    if standard_remaining < 0 or flagship_remaining < 0:
        raise RuntimeError("product_ui_projection_invalid_entitlement")

    return {
        "schema_version": str(value.get("schema_version") or "a5-product-ui-v1"),
        "mode": "a5_product",
        "product_tier_selection_required": True,
        "products": products,
        "entitlement": {
            "standard_remaining": standard_remaining,
            "flagship_remaining": flagship_remaining,
        },
    }


def get_product_ui_projection() -> dict[str, Any]:
    value = st.session_state.get("sp_product_ui_projection")
    return _normalize_product_ui_projection(value)


def get_account_snapshot(*, block: bool = False) -> dict[str, Any] | None:
    # External entitlement mutations (administrator adjustment/payment credit/
    # another device) cannot increment this browser's local revision. Keep one
    # command id stable only for the browser-bridge handshake, then retire it
    # after a completed response so the next Streamlit rerun performs a fresh
    # Worker/D1 account read instead of reusing a time-bucketed cached result.
    revision = int(st.session_state.get("sp_account_revision") or 0)
    command_id = str(st.session_state.get("sp_account_pending_command_id") or "").strip()
    if not command_id:
        command_id = f"account:{revision}:{secrets.token_urlsafe(10)}"
        st.session_state["sp_account_pending_command_id"] = command_id
    result = _edge_call("account", command_id, {}, key_prefix="sp_account")
    if result is None:
        if block:
            st.stop()
        return None
    st.session_state.pop("sp_account_pending_command_id", None)
    if not result.get("ok"):
        if block:
            _raise_edge_error(result, "account_query_failed")
        return None
    user = result.get("user")
    if isinstance(user, dict):
        st.session_state["sp_user"] = dict(user)
    projection = _normalize_product_ui_projection(result.get("product_ui"))
    st.session_state["sp_product_ui_projection"] = projection
    if projection["mode"] != "a5_product":
        st.session_state.pop("sp_product_tier", None)
    # R3.1 single-writer rule: account refresh owns identity/quota only.
    # Task UI state is initialized by recover_generation_task() and thereafter
    # updated only by generation_task_status()/explicit task actions.
    return result


def start_generation_task(
    *,
    idempotency_key: str,
    mode: str,
    interface_language: str,
    target_words: int,
    product_tier: str | None = None,
) -> dict[str, Any]:
    normalized = str(idempotency_key or "").strip()
    if not normalized:
        raise RuntimeError("idempotency_key_required")
    projection = get_product_ui_projection()
    normalized_tier = str(product_tier or "").strip().upper()
    product_mode = projection.get("mode") == "a5_product"
    if product_mode:
        if normalized_tier not in projection.get("products", {}):
            raise RuntimeError("product_tier_required")
    elif normalized_tier:
        raise RuntimeError("product_tier_not_available_in_legacy_mode")
    payload = {
        "idempotency_key": normalized,
        "mode": str(mode or "").strip().lower(),
        "interface_language": str(interface_language or "zh").strip(),
        "target_words": int(target_words),
    }
    if product_mode:
        # Product tier is user intent only. Worker/D1 resolves entitlement and
        # freezes the trusted TaskPolicySnapshot; browser state is never authority.
        payload["product_tier"] = normalized_tier
    result = _edge_call(
        "task_start",
        normalized,
        payload,
        key_prefix="sp_task_start",
    )
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "task_start_failed")
    _merge_quota_into_session(result)
    started_task = result.get("task")
    if isinstance(started_task, dict) and started_task.get("task_id"):
        st.session_state["sp_tracked_task_id"] = str(started_task["task_id"])
        # A user-started task must never inherit the prior-login completed-task UI mode.
        st.session_state["sp_recovered_completed_task_id"] = ""
    return dict(result)


def _completed_output_path(task: dict[str, Any]) -> Path | None:
    user = dict(st.session_state.get("sp_user") or {})
    user_key = re.sub(r"[^A-Za-z0-9_.-]", "_", str(user.get("id") or "").strip())[:100]
    task_key = re.sub(r"[^A-Za-z0-9_.-]", "_", str((task or {}).get("task_id") or "").strip())[:100]
    filename = os.path.basename(str((task or {}).get("output_filename") or "").strip())
    if not user_key or not task_key or not filename or not filename.lower().endswith(".zip"):
        return None
    workspace = Path(str(APP_CONFIG.get("workspace_dir", ".workspace")))
    return workspace / "users" / user_key / "tasks" / task_key / "outputs" / filename


def _restore_completed_artifact_if_needed(task: dict[str, Any]) -> None:
    """Best-effort restore for a completed task when this runtime lost its local ZIP."""
    if str((task or {}).get("status") or "").strip().lower() != "completed":
        return
    output_path = _completed_output_path(task)
    if output_path is None or output_path.is_file():
        return
    user = dict(st.session_state.get("sp_user") or {})
    numeric_user_id = user.get("numeric_id")
    if numeric_user_id in (None, ""):
        logging.warning("[SP_AUTH] bucket_restore_skipped task_id=%s reason=numeric_user_missing", str(task.get("task_id") or ""))
        return
    try:
        from hf_bucket_store import restore_completed_zip

        restored = restore_completed_zip(
            user_id=numeric_user_id,
            task_id=str(task.get("task_id") or ""),
            output_filename=str(task.get("output_filename") or ""),
            output_path=output_path,
        )
        if restored:
            logging.info("[SP_AUTH] bucket_restore_ok task_id=%s", str(task.get("task_id") or ""))
    except Exception as exc:
        # Recovery storage is never authoritative for task state or quota.
        logging.warning(
            "[SP_AUTH] bucket_restore_nonfatal task_id=%s error=%s",
            str((task or {}).get("task_id") or ""), type(exc).__name__,
        )


def recover_generation_task() -> dict[str, Any]:
    """Initialize current task state once from the authoritative Worker.

    Account polling is intentionally forbidden from writing task state. This
    recovery call is the only fresh-session initializer; later changes come
    from generation_task_status() or explicit start/cancel operations.
    """
    command_id = f"recover:{int(time.time() // 10)}"
    result = _edge_call("task_recover", command_id, {}, key_prefix="sp_task_recover")
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "task_recover_failed")
    _merge_quota_into_session(result)
    task = result.get("task")
    if isinstance(task, dict) and task.get("task_id"):
        recovered_status = str(task.get("status") or "").strip().lower()
        if recovered_status == "completed":
            # Historical completed tasks remain authoritative in D1/Bucket, but a
            # fresh login must open a clean workspace. Do not restore the artifact,
            # attach the completed task to Streamlit session state, or render any
            # historical generation/download UI.
            st.session_state["sp_generation_task"] = None
            st.session_state["sp_last_ui_task_status"] = ""
            st.session_state.pop("sp_tracked_task_id", None)
            logging.info(
                "[SP_AUTH] completed_task_hidden_on_fresh_login task_id=%s",
                str(task.get("task_id") or ""),
            )
        else:
            st.session_state["sp_generation_task"] = dict(task)
            st.session_state["sp_tracked_task_id"] = str(task["task_id"])
            st.session_state["sp_last_ui_task_status"] = recovered_status
    else:
        st.session_state["sp_generation_task"] = None
        st.session_state["sp_last_ui_task_status"] = ""
        st.session_state.pop("sp_tracked_task_id", None)
    return dict(result)


def generation_task_status(task_id: str) -> dict[str, Any]:
    normalized = str(task_id or "").strip()
    command_id = f"status:{normalized}:{int(time.time() // 5)}"
    result = _edge_call(
        "task_status", command_id, {"task_id": normalized}, key_prefix="sp_task_status"
    )
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "task_status_failed")
    _merge_quota_into_session(result)
    task = result.get("task")
    if isinstance(task, dict):
        st.session_state["sp_generation_task"] = dict(task)
        if task.get("task_id"):
            st.session_state["sp_tracked_task_id"] = str(task["task_id"])
    return dict(result)


def cancel_generation_task(task_id: str | None = None) -> dict[str, Any]:
    """Cancel synchronously so quota release is confirmed before local cleanup."""
    normalized = str(task_id or "").strip()
    command_id = f"cancel:{normalized or 'latest'}"
    result = _edge_call(
        "task_cancel",
        command_id,
        {"task_id": normalized} if normalized else {},
        key_prefix="sp_task_cancel",
    )
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "task_cancel_failed")
    reservation = result.get("reservation")
    if result.get("cancelled") and reservation and str(reservation.get("status") or "").lower() != "released":
        raise RuntimeError("task_cancel_quota_not_released")
    _merge_quota_into_session(result)
    return dict(result)


def logout_alipay_session(on_logout: Callable[[str, str], None] | None = None) -> None:
    user_id = str((st.session_state.get("sp_user") or {}).get("id") or "")
    task = st.session_state.get("sp_generation_task") or {}
    task_id = str(task.get("task_id") or "")
    if on_logout is not None:
        try:
            on_logout(user_id, task_id)
        except Exception:
            pass
    _clear_session()
    _clear_generation_session_state()


def create_payment_order(
    idempotency_key: str,
    package_sku: str | None = None,
    payment_channel: str | None = None,
) -> dict[str, Any]:
    normalized = str(idempotency_key or "").strip()
    payload: dict[str, Any] = {"idempotency_key": normalized}
    normalized_sku = str(package_sku or "").strip()
    if normalized_sku:
        payload["package_sku"] = normalized_sku
    normalized_channel = str(payment_channel or "").strip().upper()
    if normalized_channel:
        if normalized_channel not in {"ALIPAY", "WECHAT"}:
            raise RuntimeError("payment_channel_invalid")
        payload["payment_channel"] = normalized_channel
    result = _edge_call(
        "pay_create",
        normalized,
        payload,
        key_prefix="sp_pay_create",
    )
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "payment_create_failed")
    return dict(result)


def payment_order_status(merchant_order_no: str) -> dict[str, Any]:
    order_no = str(merchant_order_no or "").strip()
    command_id = f"pay-status:{order_no}:{int(time.time() // 3)}"
    result = _edge_call(
        "pay_status",
        command_id,
        {"merchant_order_no": order_no},
        key_prefix="sp_pay_status",
    )
    if result is None:
        st.stop()
    if not result.get("ok"):
        _raise_edge_error(result, "payment_status_failed")
    return dict(result)


def _qr_png_bytes(value: str) -> bytes:
    image = qrcode.make(str(value))
    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    return buffer.getvalue()


def _render_payment_purchase_body(language: str) -> None:
    projection = get_product_ui_projection()
    product_mode = projection.get("mode") == "a5_product"
    selected_product: dict[str, Any] | None = None
    package_sku = ""
    existing_order = st.session_state.get("sp_payment_order")
    if product_mode:
        selected_tier = str(st.session_state.get("sp_product_tier") or "STANDARD").strip().upper()
        raw_product = (projection.get("products") or {}).get(selected_tier)
        if not isinstance(raw_product, dict):
            raise RuntimeError("product_payment_projection_missing_tier")
        selected_product = dict(raw_product)
        package_sku = str(selected_product.get("package_sku") or "").strip()
        amount_fen = int(selected_product.get("amount_fen") or 0)
        credits = int(selected_product.get("credits_granted") or 0)
        if not package_sku or amount_fen <= 0 or credits <= 0:
            raise RuntimeError("product_payment_projection_invalid")
        amount_text = f"{amount_fen / 100:.2f}".rstrip("0").rstrip(".")
        currency = str(selected_product.get("currency") or "").strip()
        if not currency:
            raise RuntimeError("product_payment_projection_invalid_currency")
        tier_name = tr(language, f"product_tier_{selected_tier.lower()}")
        st.warning(
            tr(
                language,
                "payment_notice_product",
                tier=tier_name,
                amount=amount_text,
                currency=currency,
                credits=credits,
            )
        )
        existing_sku = str(st.session_state.get("sp_payment_package_sku") or "").strip()
        order_sku = str((existing_order or {}).get("package_sku") or "").strip() if isinstance(existing_order, dict) else ""
        if (existing_sku and existing_sku != package_sku) or (existing_order and order_sku != package_sku):
            st.session_state["sp_payment_order"] = None
            st.session_state["sp_payment_request_id"] = ""
            existing_order = None
    else:
        st.warning(tr(language, "payment_notice"))

    # R4-1D: opening the purchase dialog is the explicit purchase action.
    # Create one trusted order for the selected server-authoritative SKU;
    # users no longer preselect a payment channel or click a second create button.
    if not existing_order and not str(st.session_state.get("sp_payment_request_id") or "").strip():
        st.session_state["sp_payment_request_id"] = (
            f"pay_{int(time.time() * 1000)}_{secrets.token_urlsafe(12)}"
        )
        st.session_state["sp_payment_package_sku"] = package_sku
        st.session_state["sp_payment_order_channel"] = ""

    request_id = str(st.session_state.get("sp_payment_request_id") or "").strip()
    order = st.session_state.get("sp_payment_order")
    if request_id and not order:
        st.info(tr(language, "payment_creating"))
        order_result = create_payment_order(
            request_id,
            package_sku or None,
        )
        order = dict(order_result.get("order") or {})
        if not order.get("merchant_order_no") or not order.get("qr_code"):
            raise RuntimeError("payment_order_missing_qr_code")
        if product_mode and str(order.get("package_sku") or "") != package_sku:
            raise RuntimeError("payment_order_package_mismatch")
        st.session_state["sp_payment_order"] = order
        projected = order_result.get("product_ui")
        if isinstance(projected, dict):
            st.session_state["sp_product_ui_projection"] = _normalize_product_ui_projection(projected)

    order = st.session_state.get("sp_payment_order") or {}
    if str(order.get("merchant_order_no") or "").strip():
        # The app-level pending flag exists only to survive the initial
        # browser-bridge handshake. Once the trusted order exists, clear it
        # inside the dialog fragment so manually closing the dialog does not
        # cause an unrelated later rerun to reopen it.
        st.session_state["sp_payment_dialog_pending"] = False
    qr_code = str(order.get("qr_code") or "").strip()
    if qr_code:
        payment_provider = str(order.get("payment_provider") or "").strip().upper()
        if payment_provider == "MANUAL_ALIPAY":
            alipay_bytes = _qr_png_bytes(qr_code)
            wechat_path = Path(__file__).resolve().parent / "static" / "payment_wechat_qr.png"
            if not wechat_path.is_file():
                raise RuntimeError("payment_wechat_qr_missing")
            wechat_bytes = wechat_path.read_bytes()
            alipay_uri = "data:image/png;base64," + base64.b64encode(alipay_bytes).decode("ascii")
            wechat_uri = "data:image/png;base64," + base64.b64encode(wechat_bytes).decode("ascii")
            st.markdown(
                f"""
                <style media="all">
                .sp-payment-grid {{
                    display:grid;
                    grid-template-columns:repeat(2,minmax(0,1fr));
                    gap:1rem;
                    align-items:start;
                    margin:.35rem 0 .6rem;
                }}
                .sp-payment-card {{text-align:center; min-width:0;}}
                .sp-payment-card strong {{display:block; margin-bottom:.4rem;}}
                .sp-payment-qr-frame {{
                    width:min(100%,300px);
                    aspect-ratio:1/1;
                    margin:0 auto;
                    display:flex;
                    align-items:center;
                    justify-content:center;
                    overflow:hidden;
                }}
                .sp-payment-qr-frame img {{
                    width:100%;
                    height:100%;
                    object-fit:contain;
                    display:block;
                }}
                @media (max-width:640px) {{
                    .sp-payment-grid {{grid-template-columns:1fr;}}
                    .sp-payment-qr-frame {{width:min(86vw,300px);}}
                }}
                </style>
                <div class="sp-payment-grid">
                  <div class="sp-payment-card">
                    <strong>{html.escape(tr(language, 'payment_channel_alipay'))}</strong>
                    <div class="sp-payment-qr-frame"><img src="{alipay_uri}" alt="Alipay QR"></div>
                  </div>
                  <div class="sp-payment-card">
                    <strong>{html.escape(tr(language, 'payment_channel_wechat'))}</strong>
                    <div class="sp-payment-qr-frame"><img src="{wechat_uri}" alt="WeChat QR"></div>
                  </div>
                </div>
                """,
                unsafe_allow_html=True,
            )
            st.caption(tr(language, "payment_scan_dual"))
        else:
            st.image(_qr_png_bytes(qr_code), width=280)
        order_no = str(order.get("merchant_order_no") or "").strip()
        if order_no:
            st.caption(tr(language, "payment_order_no", order_no=order_no))
            st.code(order_no)
        if payment_provider != "MANUAL_ALIPAY":
            st.caption(tr(language, "payment_scan"))
        support = order.get("payment_support") if isinstance(order.get("payment_support"), dict) else {}
        support_wechat = str((support or {}).get("wechat") or "").strip()
        support_whatsapp = str((support or {}).get("whatsapp") or "").strip()
        if support_wechat or support_whatsapp:
            st.info(tr(language, "payment_support_prompt"))
            if support_wechat:
                st.caption(tr(language, "payment_support_wechat", value=support_wechat))
            if support_whatsapp:
                st.caption(tr(language, "payment_support_whatsapp", value=support_whatsapp))
        if payment_provider != "MANUAL_ALIPAY" and st.button(
            tr(language, "payment_check"), key="sp_check_payment_status"
        ):
            status_result = payment_order_status(order.get("merchant_order_no", ""))
            updated = dict(status_result.get("order") or {})
            st.session_state["sp_payment_order"] = updated
            projected = status_result.get("product_ui")
            if isinstance(projected, dict):
                st.session_state["sp_product_ui_projection"] = _normalize_product_ui_projection(projected)
            if updated.get("status") == "paid" and updated.get("credit_applied"):
                user = status_result.get("user")
                if isinstance(user, dict):
                    st.session_state["sp_user"] = dict(user)
                    st.session_state["sp_account_revision"] = int(
                        st.session_state.get("sp_account_revision") or 0
                    ) + 1
                st.session_state["sp_payment_required"] = False
                st.session_state["sp_payment_order"] = None
                st.session_state["sp_payment_request_id"] = ""
                st.session_state["sp_payment_package_sku"] = ""
                st.session_state["sp_payment_order_channel"] = ""
                if product_mode and selected_product is not None:
                    tier_name = tr(language, f"product_tier_{str(selected_product.get('product_tier') or '').lower()}")
                    st.success(
                        tr(
                            language,
                            "payment_paid_product",
                            tier=tier_name,
                            credits=int(selected_product.get("credits_granted") or 0),
                        )
                    )
                else:
                    st.success(tr(language, "payment_paid"))
            elif str(updated.get("manual_review_status") or "").upper() == "REJECTED":
                st.session_state["sp_payment_order"] = None
                st.session_state["sp_payment_request_id"] = ""
                st.session_state["sp_payment_package_sku"] = ""
                st.session_state["sp_payment_order_channel"] = ""
                st.warning(tr(language, "payment_rejected"))
            elif updated.get("status") in {"closed", "failed"}:
                st.session_state["sp_payment_order"] = None
                st.session_state["sp_payment_request_id"] = ""
                st.session_state["sp_payment_package_sku"] = ""
                st.session_state["sp_payment_order_channel"] = ""
                st.warning(tr(language, "payment_expired"))


def render_payment_purchase(language: str) -> None:
    """Show payment in a modal so the approved normal-page layout is unchanged."""
    dialog_factory = getattr(st, "dialog", None)
    if callable(dialog_factory):
        @dialog_factory(tr(language, "payment_dialog_title"))
        def _payment_dialog():
            _render_payment_purchase_body(language)
        _payment_dialog()
        return
    # Compatibility fallback for older Streamlit builds.
    _render_payment_purchase_body(language)


def render_logged_in_account(
    language: str,
    user: dict[str, Any] | None = None,
    on_logout: Callable[[str, str], None] | None = None,
) -> None:
    """Render the approved responsive account summary with live quota values."""
    live = get_account_snapshot(block=False)
    if live and isinstance(live.get("user"), dict):
        user = dict(live["user"])
    else:
        user = user or st.session_state.get("sp_user", {})
    user_id = user.get("id", "—")
    projection = get_product_ui_projection()
    product_mode = projection.get("mode") == "a5_product"
    if product_mode:
        entitlement = projection["entitlement"]
        standard_remaining = entitlement["standard_remaining"]
        flagship_remaining = entitlement["flagship_remaining"]
    else:
        standard_remaining = "—"
        flagship_remaining = "—"

    st.markdown(
        """
        <style>
        section[data-testid="stSidebar"] {
            display: none !important;
        }

        [data-testid="collapsedControl"] {
            display: none !important;
        }

        .sp-account-card {
            display: flex;
            align-items: center;
            justify-content: center;
            gap: clamp(0.3rem, 0.55vw, 0.58rem);
            width: 100%;
            min-width: 0;
            min-height: 2.85rem;
            padding: 0.48rem clamp(0.55rem, 0.85vw, 0.9rem);
            border: 1px solid rgba(148, 163, 184, 0.30);
            border-radius: 13px;
            background: #dbeafe;
            color: #1e3a5f;
            font-size: clamp(0.76rem, 0.84vw, 0.92rem);
            font-weight: 500;
            line-height: 1.25;
            white-space: nowrap;
            overflow: visible;
            box-sizing: border-box;
        }

        .sp-account-card span {
            display: inline-block;
            white-space: nowrap;
        }

        .sp-account-separator {
            flex: 0 0 auto;
            opacity: 0.48;
        }

        div[data-testid="stMarkdownContainer"]:has(.sp-logout-link) {
            width: 100%;
        }

        div[data-testid="stMarkdownContainer"]:has(.sp-logout-link) p {
            width: 100%;
            margin: 0;
        }

        .sp-logout-link {
            display: flex;
            align-items: center;
            justify-content: center;
            width: 100%;
            min-width: 6.8rem;
            min-height: 2.85rem;
            padding: 0.45rem 0.72rem;
            border: 1px solid rgba(148, 163, 184, 0.30);
            border-radius: 0.5rem;
            background: #dbeafe;
            color: #1e3a5f !important;
            text-decoration: none !important;
            white-space: nowrap !important;
            overflow-wrap: normal !important;
            word-break: keep-all !important;
            line-height: 1.15;
            font-size: clamp(0.78rem, 0.82vw, 0.92rem);
            font-weight: 400;
            box-sizing: border-box;
            cursor: pointer;
            transition: border-color 120ms ease, background-color 120ms ease;
        }

        .sp-logout-link:hover {
            border-color: #93c5fd;
            background: #bfdbfe;
            color: #1e3a5f !important;
        }

        .sp-logout-link:focus-visible {
            outline: 2px solid #60a5fa;
            outline-offset: 2px;
        }

        @media (max-width: 1150px) {
            .sp-account-card {
                font-size: 0.75rem;
                gap: 0.3rem;
                padding-left: 0.48rem;
                padding-right: 0.48rem;
            }

            .sp-logout-link {
                min-width: 5.8rem;
                font-size: 0.78rem;
                padding-left: 0.46rem;
                padding-right: 0.46rem;
            }
        }

        @media (max-width: 760px) {
            .sp-account-card {
                justify-content: flex-start;
                overflow-x: auto;
                scrollbar-width: none;
            }

            .sp-account-card::-webkit-scrollbar {
                display: none;
            }
        }
        </style>
        """,
        unsafe_allow_html=True,
    )

    info_col, logout_col = st.columns([5.9, 1.65])
    with info_col:
        balance_markup = (
            f"<span>{html.escape(tr(language, 'account_chip_standard', value=standard_remaining))}</span>"
            f'<span class="sp-account-separator">•</span>'
            f"<span>{html.escape(tr(language, 'account_chip_flagship', value=flagship_remaining))}</span>"
        )
        st.markdown(
            f"""
            <div class="sp-account-card">
              <span>{html.escape(tr(language, 'account_chip_user', value=user_id))}</span>
              <span class="sp-account-separator">•</span>
              {balance_markup}
            </div>
            """,
            unsafe_allow_html=True,
        )
    with logout_col:
        logout_label = html.escape(tr(language, "logout_short"))
        logout_help = html.escape(tr(language, "logout"), quote=True)
        st.markdown(
            f'<a class="sp-logout-link" href="/edge/logout" target="_self" '
            f'rel="nofollow" role="button" aria-label="{logout_help}" '
            f'title="{logout_help}">{logout_label}</a>',
            unsafe_allow_html=True,
        )