P0-R2 HF Bucket flag-OFF deployment

huggingface/hf_bucket_store.py

@@ -0,0 +1,380 @@
+# -*- coding: utf-8 -*-
+"""Best-effort private Hugging Face Bucket storage for completed ZIP delivery.
+
+This module is deliberately outside the paper-generation core. Bucket storage is
+an optional recovery layer only: a Bucket error must never turn an already valid
+local paper package into a failed generation or alter quota/accounting state.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import re
+import tempfile
+import zipfile
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+
+_TRUE = {"1", "true", "yes", "on", "enabled"}
+_DEFAULT_TTL_HOURS = 168
+_DEFAULT_WARNING_GIB = 70.0
+_DEFAULT_HARD_STOP_GIB = 80.0
+_MAX_HARD_STOP_GIB = 80.0  # Free-only safety ceiling approved for Article Agent.
+_GIB = 1024 ** 3
+_TASK_RE = re.compile(r"^[A-Za-z0-9_.-]{1,160}$")
+_BUCKET_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
+
+
+@dataclass(frozen=True)
+class BucketConfig:
+    enabled: bool
+    bucket_id: str
+    token: str
+    ttl_hours: int
+    warning_bytes: int
+    hard_stop_bytes: int
+
+
+def _truthy(value: Any) -> bool:
+    return str(value or "").strip().casefold() in _TRUE
+
+
+def _bounded_float(name: str, default: float, low: float, high: float) -> float:
+    try:
+        value = float(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        value = default
+    return max(low, min(high, value))
+
+
+def _bounded_int(name: str, default: int, low: int, high: int) -> int:
+    try:
+        value = int(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        value = default
+    return max(low, min(high, value))
+
+
+def config() -> BucketConfig:
+    enabled = _truthy(os.environ.get("SMARTPAPER_HF_ARTIFACTS_ENABLED"))
+    bucket_id = str(os.environ.get("SMARTPAPER_HF_ARTIFACT_BUCKET") or "").strip()
+    token = str(os.environ.get("SMARTPAPER_HF_BUCKET_TOKEN") or "").strip()
+    ttl_hours = _bounded_int("SMARTPAPER_HF_ARTIFACT_TTL_HOURS", _DEFAULT_TTL_HOURS, 1, _DEFAULT_TTL_HOURS)
+    requested_hard = _bounded_float(
+        "SMARTPAPER_HF_ARTIFACT_HARD_STOP_GIB", _DEFAULT_HARD_STOP_GIB, 1.0, _MAX_HARD_STOP_GIB
+    )
+    requested_warning = _bounded_float(
+        "SMARTPAPER_HF_ARTIFACT_WARNING_GIB", _DEFAULT_WARNING_GIB, 1.0, requested_hard
+    )
+    return BucketConfig(
+        enabled=enabled,
+        bucket_id=bucket_id,
+        token=token,
+        ttl_hours=ttl_hours,
+        warning_bytes=int(requested_warning * _GIB),
+        hard_stop_bytes=int(requested_hard * _GIB),
+    )
+
+
+def _new_api(token: str):
+    # huggingface_hub reads these at import time. Keep recovery bounded without
+    # importing the optional dependency when the feature is disabled.
+    os.environ.setdefault("HF_HUB_ETAG_TIMEOUT", "8")
+    os.environ.setdefault("HF_HUB_DOWNLOAD_TIMEOUT", "8")
+    from huggingface_hub import HfApi  # lazy: absence must not break normal generation
+
+    return HfApi(token=token)
+
+
+def _validate_config(cfg: BucketConfig) -> None:
+    if not cfg.enabled:
+        raise RuntimeError("hf_bucket_disabled")
+    if not cfg.bucket_id or not _BUCKET_RE.fullmatch(cfg.bucket_id):
+        raise RuntimeError("hf_bucket_id_invalid")
+    if not cfg.token:
+        raise RuntimeError("hf_bucket_token_missing")
+    if cfg.hard_stop_bytes > int(_MAX_HARD_STOP_GIB * _GIB):
+        raise RuntimeError("hf_bucket_hard_stop_exceeds_free_safety_ceiling")
+
+
+def _trusted_key_parts(user_id: int | str, task_id: str) -> tuple[str, str]:
+    try:
+        trusted_user_id = str(int(user_id))
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError("hf_bucket_user_id_invalid") from exc
+    trusted_task_id = str(task_id or "").strip()
+    if not _TASK_RE.fullmatch(trusted_task_id):
+        raise RuntimeError("hf_bucket_task_id_invalid")
+    return trusted_user_id, trusted_task_id
+
+
+def object_paths(user_id: int | str, task_id: str) -> tuple[str, str]:
+    trusted_user_id, trusted_task_id = _trusted_key_parts(user_id, task_id)
+    prefix = f"v1/users/{trusted_user_id}/tasks/{trusted_task_id}"
+    return f"{prefix}/final.zip", f"{prefix}/final.json"
+
+
+def _safe_output_filename(value: str) -> str:
+    text = str(value or "").strip()
+    name = os.path.basename(text)
+    if not name or name != text or not name.lower().endswith(".zip") or len(name) > 240:
+        raise RuntimeError("hf_bucket_output_filename_invalid")
+    return name
+
+
+def _sha256_file(path: Path) -> tuple[str, int]:
+    digest = hashlib.sha256()
+    size = 0
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            size += len(block)
+            digest.update(block)
+    return digest.hexdigest(), size
+
+
+def _private_bucket_info(api, cfg: BucketConfig):
+    info = api.bucket_info(cfg.bucket_id)
+    if getattr(info, "private", None) is not True:
+        raise RuntimeError("hf_bucket_not_private")
+    return info
+
+
+def _file_map(api, bucket_id: str, paths: list[str]) -> dict[str, Any]:
+    return {str(item.path): item for item in api.get_bucket_paths_info(bucket_id, paths)}
+
+
+def _metadata_bytes(*, user_id: str, task_id: str, filename: str, sha256: str, size_bytes: int, uploaded_at: datetime) -> bytes:
+    payload = {
+        "schema_version": "articleagent-hf-bucket-v1",
+        "user_id": user_id,
+        "task_id": task_id,
+        "output_filename": filename,
+        "sha256": sha256,
+        "size_bytes": int(size_bytes),
+        "uploaded_at": uploaded_at.astimezone(timezone.utc).isoformat(),
+    }
+    return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def _parse_uploaded_at(value: Any) -> datetime:
+    text = str(value or "").strip()
+    if not text:
+        raise RuntimeError("hf_bucket_metadata_time_missing")
+    try:
+        observed = datetime.fromisoformat(text.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise RuntimeError("hf_bucket_metadata_time_invalid") from exc
+    if observed.tzinfo is None:
+        observed = observed.replace(tzinfo=timezone.utc)
+    return observed.astimezone(timezone.utc)
+
+
+def _download_metadata(api, bucket_id: str, remote_path: str) -> dict[str, Any]:
+    with tempfile.TemporaryDirectory(prefix="sp_hf_meta_") as temp_dir:
+        target = Path(temp_dir) / "final.json"
+        api.download_bucket_files(bucket_id, [(remote_path, target)], raise_on_missing_files=True)
+        raw = target.read_bytes()
+    if len(raw) > 64 * 1024:
+        raise RuntimeError("hf_bucket_metadata_too_large")
+    value = json.loads(raw.decode("utf-8"))
+    if not isinstance(value, dict):
+        raise RuntimeError("hf_bucket_metadata_invalid")
+    return value
+
+
+def cleanup_expired(*, api, cfg: BucketConfig, now: datetime | None = None) -> dict[str, Any]:
+    """Delete only files older than the approved 7-day retention window.
+
+    Cleanup itself is best-effort at the call site. This function raises on API
+    failure so callers can log it, while never guessing that a younger object is
+    safe to delete.
+    """
+    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+    cutoff = current - timedelta(hours=cfg.ttl_hours)
+    expired: list[str] = []
+    for item in api.list_bucket_tree(cfg.bucket_id, prefix="v1/users/", recursive=True):
+        if str(getattr(item, "type", "")) != "file":
+            continue
+        mtime = getattr(item, "mtime", None)
+        if not isinstance(mtime, datetime):
+            continue
+        if mtime.tzinfo is None:
+            mtime = mtime.replace(tzinfo=timezone.utc)
+        if mtime.astimezone(timezone.utc) < cutoff:
+            expired.append(str(item.path))
+    if expired:
+        api.batch_bucket_files(cfg.bucket_id, delete=expired)
+    return {"deleted": len(expired), "paths": expired}
+
+
+def archive_completed_zip(
+    *,
+    user_id: int | str,
+    task_id: str,
+    output_filename: str,
+    output_path: str | Path,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Best-effort archive primitive. Caller must catch all errors.
+
+    This function never modifies task/quota state and should only be called after
+    the authoritative Worker has accepted the completed task transition.
+    """
+    cfg = config()
+    if not cfg.enabled:
+        return {"ok": True, "stored": False, "reason": "disabled"}
+    _validate_config(cfg)
+    trusted_user_id, trusted_task_id = _trusted_key_parts(user_id, task_id)
+    filename = _safe_output_filename(output_filename)
+    local = Path(output_path).resolve()
+    if not local.is_file() or local.name != filename:
+        raise RuntimeError("hf_bucket_local_zip_missing")
+    sha256, size_bytes = _sha256_file(local)
+    if size_bytes <= 0:
+        raise RuntimeError("hf_bucket_local_zip_empty")
+
+    api = _new_api(cfg.token)
+    _private_bucket_info(api, cfg)
+
+    cleanup_warning = ""
+    try:
+        cleanup_expired(api=api, cfg=cfg, now=now)
+    except Exception as exc:  # cleanup must never destroy local delivery
+        cleanup_warning = type(exc).__name__
+        logging.warning("[SP_BUCKET] cleanup_failed task_id=%s error=%s", trusted_task_id, cleanup_warning)
+
+    zip_path, metadata_path = object_paths(trusted_user_id, trusted_task_id)
+    remote = _file_map(api, cfg.bucket_id, [zip_path, metadata_path])
+    if zip_path in remote or metadata_path in remote:
+        if zip_path in remote and metadata_path in remote:
+            existing = _download_metadata(api, cfg.bucket_id, metadata_path)
+            same = (
+                str(existing.get("user_id") or "") == trusted_user_id
+                and str(existing.get("task_id") or "") == trusted_task_id
+                and str(existing.get("output_filename") or "") == filename
+                and str(existing.get("sha256") or "") == sha256
+                and int(existing.get("size_bytes") or -1) == size_bytes
+                and int(getattr(remote[zip_path], "size", -1) or -1) == size_bytes
+            )
+            if same:
+                return {
+                    "ok": True,
+                    "stored": True,
+                    "idempotent": True,
+                    "sha256": sha256,
+                    "size_bytes": size_bytes,
+                    "cleanup_warning": cleanup_warning,
+                }
+        raise RuntimeError("hf_bucket_artifact_conflict")
+
+    info = _private_bucket_info(api, cfg)
+    current_size = int(getattr(info, "size", 0) or 0)
+    if current_size >= cfg.warning_bytes:
+        logging.warning("[SP_BUCKET] capacity_warning bytes=%s", current_size)
+    uploaded_at = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+    metadata = _metadata_bytes(
+        user_id=trusted_user_id,
+        task_id=trusted_task_id,
+        filename=filename,
+        sha256=sha256,
+        size_bytes=size_bytes,
+        uploaded_at=uploaded_at,
+    )
+    projected = current_size + size_bytes + len(metadata)
+    if projected >= cfg.hard_stop_bytes:
+        raise RuntimeError("hf_bucket_hard_stop_reached")
+
+    api.batch_bucket_files(
+        cfg.bucket_id,
+        add=[(local, zip_path), (metadata, metadata_path)],
+    )
+    verified = _file_map(api, cfg.bucket_id, [zip_path, metadata_path])
+    if zip_path not in verified or metadata_path not in verified:
+        raise RuntimeError("hf_bucket_upload_incomplete")
+    if int(getattr(verified[zip_path], "size", -1) or -1) != size_bytes:
+        raise RuntimeError("hf_bucket_upload_size_mismatch")
+    existing = _download_metadata(api, cfg.bucket_id, metadata_path)
+    if (
+        str(existing.get("sha256") or "") != sha256
+        or int(existing.get("size_bytes") or -1) != size_bytes
+        or str(existing.get("task_id") or "") != trusted_task_id
+        or str(existing.get("user_id") or "") != trusted_user_id
+    ):
+        raise RuntimeError("hf_bucket_upload_metadata_mismatch")
+    return {
+        "ok": True,
+        "stored": True,
+        "idempotent": False,
+        "sha256": sha256,
+        "size_bytes": size_bytes,
+        "cleanup_warning": cleanup_warning,
+    }
+
+
+def restore_completed_zip(
+    *,
+    user_id: int | str,
+    task_id: str,
+    output_filename: str,
+    output_path: str | Path,
+    now: datetime | None = None,
+) -> bool:
+    """Restore a completed ZIP from a private bucket, with ownership/integrity checks."""
+    cfg = config()
+    if not cfg.enabled:
+        return False
+    _validate_config(cfg)
+    trusted_user_id, trusted_task_id = _trusted_key_parts(user_id, task_id)
+    filename = _safe_output_filename(output_filename)
+    target = Path(output_path).resolve()
+    if target.is_file():
+        return True
+    if target.name != filename:
+        raise RuntimeError("hf_bucket_restore_target_invalid")
+
+    api = _new_api(cfg.token)
+    _private_bucket_info(api, cfg)
+    zip_path, metadata_path = object_paths(trusted_user_id, trusted_task_id)
+    remote = _file_map(api, cfg.bucket_id, [zip_path, metadata_path])
+    if zip_path not in remote or metadata_path not in remote:
+        return False
+
+    metadata = _download_metadata(api, cfg.bucket_id, metadata_path)
+    uploaded_at = _parse_uploaded_at(metadata.get("uploaded_at"))
+    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
+    if uploaded_at > current + timedelta(minutes=5):
+        raise RuntimeError("hf_bucket_metadata_time_future")
+    if current - uploaded_at > timedelta(hours=cfg.ttl_hours):
+        return False
+    expected_size = int(metadata.get("size_bytes") or -1)
+    expected_sha = str(metadata.get("sha256") or "")
+    if (
+        str(metadata.get("schema_version") or "") != "articleagent-hf-bucket-v1"
+        or str(metadata.get("user_id") or "") != trusted_user_id
+        or str(metadata.get("task_id") or "") != trusted_task_id
+        or str(metadata.get("output_filename") or "") != filename
+        or expected_size <= 0
+        or not re.fullmatch(r"[0-9a-f]{64}", expected_sha)
+        or int(getattr(remote[zip_path], "size", -1) or -1) != expected_size
+    ):
+        raise RuntimeError("hf_bucket_restore_metadata_mismatch")
+
+    target.parent.mkdir(parents=True, exist_ok=True)
+    temporary = target.with_suffix(target.suffix + ".bucket.tmp")
+    try:
+        temporary.unlink(missing_ok=True)
+        api.download_bucket_files(cfg.bucket_id, [(zip_path, temporary)], raise_on_missing_files=True)
+        observed_sha, observed_size = _sha256_file(temporary)
+        if observed_size != expected_size or observed_sha != expected_sha:
+            raise RuntimeError("hf_bucket_restore_integrity_mismatch")
+        with zipfile.ZipFile(temporary, "r") as archive:
+            if archive.testzip() is not None:
+                raise RuntimeError("hf_bucket_restore_zip_invalid")
+        os.replace(temporary, target)
+    finally:
+        temporary.unlink(missing_ok=True)
+    return True

huggingface/managed_auth.py

@@ -0,0 +1,867 @@
+# -*- 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",
+    ):
+        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",
+    ):
+        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:
+    bucket = int(time.time() // 30)
+    revision = int(st.session_state.get("sp_account_revision") or 0)
+    command_id = f"account:{revision}:{bucket}"
+    result = _edge_call("account", command_id, {}, key_prefix="sp_account")
+    if result is None:
+        if block:
+            st.stop()
+        return 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"])
+    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"):
+        _restore_completed_artifact_if_needed(task)
+        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"] = str(task.get("status") or "")
+    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,
+) -> 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
+    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 = ""
+    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_order = st.session_state.get("sp_payment_order")
+        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"] = ""
+    else:
+        st.warning(tr(language, "payment_notice"))
+
+    if st.button(tr(language, "payment_create"), key="sp_create_payment_order"):
+        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
+
+    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 {}
+    qr_code = str(order.get("qr_code") or "").strip()
+    if qr_code:
+        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)
+        payment_provider = str(order.get("payment_provider") or "").strip().upper()
+        if payment_provider == "MANUAL_ALIPAY":
+            st.caption(tr(language, "payment_scan_manual"))
+        else:
+            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 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"] = ""
+                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 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.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:
+        free_remaining = user.get("daily_free_remaining", user.get("daily_free_limit", 3))
+        bonus_papers = user.get("bonus_papers", 0)
+
+    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: rgba(30, 41, 59, 0.58);
+            color: var(--text-color, #f3f4f6);
+            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: transparent;
+            color: var(--text-color, #f3f4f6) !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: rgba(255, 75, 75, 0.78);
+            background: rgba(255, 255, 255, 0.035);
+            color: var(--text-color, #f3f4f6) !important;
+        }
+
+        .sp-logout-link:focus-visible {
+            outline: 2px solid rgba(255, 75, 75, 0.85);
+            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:
+        if product_mode:
+            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>"
+            )
+        else:
+            balance_markup = (
+                f"<span>{html.escape(tr(language, 'account_chip_free', value=free_remaining))}</span>"
+                f'<span class="sp-account-separator">•</span>'
+                f"<span>{html.escape(tr(language, 'account_chip_bonus', value=bonus_papers))}</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,
+        )

huggingface/requirements.txt

@@ -0,0 +1,16 @@
+streamlit==1.38.0
+requests>=2.31,<3
+python-docx>=1.1,<2
+pandas>=2.1,<3
+numpy>=1.26,<3
+scikit-learn>=1.4,<2
+tiktoken>=0.7,<1
+langdetect==1.0.9
+mistune>=3,<4
+beautifulsoup4>=4.12,<5
+PyMuPDF>=1.24,<2
+openpyxl>=3.1,<4
+tabulate>=0.9,<1
+qrcode[pil]>=8,<9
+
+huggingface_hub>=1.16,<2

huggingface/task_executor.py

@@ -29,6 +29,7 @@ import requests
 
 from config import APP_CONFIG
 from core.checkpoint_store import TaskCheckpointStore, checkpoint_authority_lock
+from core.advanced_method_registry import METHOD_FAMILIES, method_capability
 
 _log_handlers = [logging.StreamHandler(sys.stdout)]
 _executor_log_path = str(os.environ.get("SMARTPAPER_EXECUTOR_LOG_PATH") or "").strip()
@@ -125,34 +126,69 @@ def _is_local_checkpoint_authority_failure(exc: Exception) -> bool:
     return isinstance(exc, RuntimeError) and str(exc or "").startswith("checkpoint_authority_")
 
 
-def _is_recoverable_generation_failure(exc: Exception) -> bool:
-    """Classify only post-claim execution failures.
-
-    This deliberately avoids importing the generation pipeline at module import
-    time. Provider/model exception class names are inspected only after
-    claim/bootstrap has succeeded, so future provider adapters do not require
-    executor imports or vendor-specific branching.
-    """
+def _generation_failure_family(exc: Exception) -> str:
+    """Classify post-claim execution failures for the P0 delivery-first contract."""
     if isinstance(exc, ExecutorApiError):
-        return exc.status_code == 429 or exc.status_code >= 500
+        return "TRANSPORT" if exc.status_code == 429 or exc.status_code >= 500 else "AUTH_POLICY_SECURITY"
     class_name = exc.__class__.__name__
+    text = str(exc or "").casefold()
+
     if class_name in {
-        "DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError",
-        "ModelAuthenticationError", "ModelBalanceError", "ModelRequestError",
-        "ModelProviderUnavailableError", "ModelGatewayError",
+        "DeepSeekAuthenticationError", "DeepSeekBalanceError",
+        "ModelAuthenticationError", "ModelBalanceError",
     }:
-        return False
+        return "AUTH_POLICY_SECURITY"
+    stable_policy_markers = (
+        "trusted_product_policy", "trusted_product_route_", "request_hash_mismatch",
+        "policy_snapshot", "entitlement", "authentication", "balance is insufficient",
+        "rejected the request format (400)", "not found (404)", "parameters are invalid (422)",
+        "unknown or unconfigured deepseek model role", "no deepseek model configured",
+    )
+    if any(marker in text for marker in stable_policy_markers):
+        return "AUTH_POLICY_SECURITY"
     if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
-        return True
-    if class_name == "DeepSeekAPIError":
-        return True
-    text = str(exc or "").lower()
+        return "TRANSPORT"
+    if isinstance(exc, requests.exceptions.HTTPError):
+        response = getattr(exc, "response", None)
+        status = int(getattr(response, "status_code", 0) or 0)
+        if status in {408, 409, 425, 429, 500, 502, 503, 504}:
+            return "TRANSPORT"
+    if class_name in {"ModelProviderUnavailableError", "DeepSeekAPIError"}:
+        return "TRANSPORT"
+    if class_name == "DeepSeekRequestError":
+        content_validation_markers = (
+            "final abstract", "front matter", "translation", "citation", "hypothesis",
+            "manuscript", "qualitative", "glossary", "marker mismatch", "semantic",
+            "must not contain", "empty after validation", "structured validation",
+        )
+        return "CONTENT_VALIDATION" if any(marker in text for marker in content_validation_markers) else "AUTH_POLICY_SECURITY"
+    if class_name == "ModelRequestError":
+        # Gateway request errors are trusted route/request-shape failures; provider
+        # availability has its own recoverable exception class.
+        return "AUTH_POLICY_SECURITY"
+    if class_name == "ModelGatewayError":
+        transient_markers = ("unavailable", "timeout", "timed out", "connection", "provider")
+        return "TRANSPORT" if any(marker in text for marker in transient_markers) else "CONTENT_VALIDATION"
     transient_markers = (
         "connection", "timed out", "timeout", "temporarily unavailable",
         "streaming failed after", "api failed after", "stream was interrupted",
         "insufficient eligible references",
     )
-    return any(marker in text for marker in transient_markers)
+    if any(marker in text for marker in transient_markers):
+        return "TRANSPORT"
+    return "TERMINAL_INTEGRITY"
+
+
+def _is_recoverable_generation_failure(exc: Exception) -> bool:
+    """Return True only for failures that should resume instead of refund/fail.
+
+    CONTENT_VALIDATION reaches this layer only after the runner's local repair/
+    fallback paths were unable to complete the current attempt.  A new executor
+    attempt is still preferable to destroying the paper because model output is
+    nondeterministic; stable auth/policy/request-shape faults remain terminal.
+    """
+    return _generation_failure_family(exc) in {"TRANSPORT", "CONTENT_VALIDATION"}
+
 
 
 class ExecutorApi:
@@ -415,6 +451,23 @@ def _snapshot_nullable_version(value):
     return text or None
 
 
+def _validate_request_method_family(request_data: dict, task: dict) -> None:
+    """Validate new U1-R2 method identity while preserving legacy requests."""
+    raw_family = str((request_data or {}).get("method_family") or "").strip().casefold()
+    if not raw_family:
+        return
+    if raw_family not in METHOD_FAMILIES:
+        raise RuntimeError("request_method_family_invalid")
+    capability = method_capability(raw_family)
+    expected_lane = str(capability.get("legacy_policy_lane") or "").strip().casefold()
+    request_lane = str((request_data or {}).get("derived_policy_lane") or "").strip().casefold()
+    task_lane = str((task or {}).get("mode") or "").strip().casefold()
+    if expected_lane not in {"amj", "asq"} or request_lane != expected_lane or task_lane != expected_lane:
+        raise RuntimeError("request_method_family_lane_mismatch")
+    if str(capability.get("capability_status") or "").strip().casefold() != "production_supported":
+        raise RuntimeError("request_method_family_capability_blocked")
+
+
 def _trusted_task_policy_snapshot_envelope(snapshot: dict, resolved: dict) -> dict:
     return {
         "task_id": str(snapshot.get("task_id") or ""),
@@ -486,6 +539,7 @@ def _validate_trusted_task_policy_snapshot(claim: dict, task: dict, request_data
         if _snapshot_nullable_version(resolved.get(field)) != _snapshot_nullable_version(snapshot.get(field)):
             raise RuntimeError("task_policy_snapshot_policy_version_mismatch")
 
+    _validate_request_method_family(request_data, task)
     if str(resolved.get("manuscript_mode") or "").strip().lower() != str(task.get("mode") or "").strip().lower():
         raise RuntimeError("task_policy_snapshot_mode_mismatch")
     target_words = int(task.get("target_words") or 0)
@@ -675,6 +729,7 @@ def main() -> int:
             state.stop_heartbeat()
         return 2
 
+    runner = None
     try:
         # Heavy paper modules are deliberately imported only after D1 claim and
         # the local claim handshake. Any import/init failure is now observable
@@ -700,6 +755,7 @@ def main() -> int:
             logging.warning("[SP_EXECUTOR] final_heartbeat_error task_id=%s error=%s", args.task_id, _safe(exc, 180))
         if state.cancelled():
             raise RuntimeError("task_cancel_requested")
+        output_filename = str(result.get("output_filename") or "")[:240]
         api.finish(
             {
                 "task_id": args.task_id,
@@ -708,7 +764,7 @@ def main() -> int:
                 "executor_instance_id": instance_id,
                 "status": "completed",
                 "current_node": "package_validated",
-                "output_filename": str(result.get("output_filename") or "")[:240],
+                "output_filename": output_filename,
                 # R14-TQ2-PQ1-R3-R1: completed tasks may carry a nonfatal,
                 # backend-trusted quality warning. Worker/D1 already persists
                 # safe_error_code/detail for completed terminal transitions;
@@ -717,6 +773,28 @@ def main() -> int:
                 "safe_error_detail": str(result.get("safe_error_detail") or "")[:800],
             }
         )
+        # P0-R2: private HF Bucket storage is a best-effort recovery layer only.
+        # The authoritative completed/quota transition above must succeed first;
+        # any storage failure below is nonfatal and cannot alter task accounting.
+        try:
+            from hf_bucket_store import archive_completed_zip
+
+            archive_result = archive_completed_zip(
+                user_id=args.user_id,
+                task_id=args.task_id,
+                output_filename=output_filename,
+                output_path=str(result.get("output_path") or ""),
+            )
+            if archive_result.get("stored"):
+                logging.info(
+                    "[SP_EXECUTOR] bucket_backup_ok task_id=%s bytes=%s idempotent=%s",
+                    args.task_id, archive_result.get("size_bytes"), bool(archive_result.get("idempotent")),
+                )
+        except Exception as exc:
+            logging.warning(
+                "[SP_EXECUTOR] bucket_backup_nonfatal task_id=%s error=%s",
+                args.task_id, type(exc).__name__,
+            )
         logging.info("[SP_EXECUTOR] completed task_id=%s epoch=%s", args.task_id, epoch)
         return 0
     except Exception as exc:
@@ -742,6 +820,15 @@ def main() -> int:
             logging.exception("[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
             return 75
         logging.exception("[SP_EXECUTOR] failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+        recovery = {}
+        if runner is not None:
+            try:
+                recovery = dict(runner.build_recovery_artifact(exc) or {})
+            except Exception as recovery_exc:
+                logging.warning(
+                    "[SP_EXECUTOR] recovery_package_failed task_id=%s error=%s",
+                    args.task_id, _safe(recovery_exc, 220),
+                )
         try:
             api.finish(
                 {
@@ -753,6 +840,7 @@ def main() -> int:
                     "safe_error_code": "E_GENERATION_FAILED",
                     "safe_error_detail": _safe(exc, 800),
                     "current_node": "generation_failed",
+                    "output_filename": str(recovery.get("output_filename") or "")[:240],
                 }
             )
         except Exception as finish_exc: