fix(ad): detach late impressions from abandoned coupons

fix(ad): detach late impressions from failed coupon traces
feat(huawei-review): 华为审核开关(admin 可切 + 客户端下发 + 审计) (#147 )
2026-07-21 11:33:21 +08:00 · 2026-07-21 11:33:21 +08:00 · 2026-07-21 10:43:58 +08:00 · 2026-07-21 10:11:39 +08:00 · 2026-07-20 18:55:38 +08:00
33 changed files with 2743 additions and 16 deletions
@@ -137,3 +137,19 @@ PANGLE_REPORT_SECURITY_KEY=
 # GroMore AppId(报表 site_id 维度)→ 应用环境;默认取现网两个应用,按需覆盖。
 PANGLE_REPORT_SITE_ID_PROD=5830519
 PANGLE_REPORT_SITE_ID_TEST=5832303
 # ===== 可观测(OpenObserve 接口指标)=====
 # 采集每个接口 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker,见 deploy/openobserve/)。
 # 默认关;开启需 ENABLED=true 且填 USER/PASSWORD(与 docker-compose 里 root 账号一致)。
 # 未开/缺凭证 → 中间件透传、worker 不启动,整套 no-op,不影响业务。
 OBSERVE_ENABLED=false
 OBSERVE_ENDPOINT=http://localhost:5080
 OBSERVE_ORG=default
 OBSERVE_STREAM=app_requests
 OBSERVE_USER=admin@shaguabijia.local
 OBSERVE_PASSWORD=Complexpass#123
 # 进阶(一般不用改):攒批间隔秒 / 单批最大条数 / 有界队列上限(满则丢) / 上报超时秒
 OBSERVE_FLUSH_INTERVAL_SEC=5
 OBSERVE_BATCH_MAX=200
 OBSERVE_QUEUE_MAX=10000
 OBSERVE_TIMEOUT_SEC=5
@@ -30,6 +30,7 @@ from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
 from app.admin.routers.huawei_review import router as huawei_review_router
 from app.admin.routers.onboarding import router as onboarding_router
 from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
 from app.admin.routers.price_report import router as price_report_router
@@ -109,4 +110,5 @@ admin_app.include_router(cps_router)
 admin_app.include_router(coupon_data_router)
 admin_app.include_router(ad_audit_router)
 admin_app.include_router(ad_config_router)
 admin_app.include_router(huawei_review_router)
 admin_app.include_router(ad_revenue_router)
@@ -31,6 +31,7 @@ PERMISSION_CATALOG: list[dict] = [
    {"group": "数据配置", "pages": [
        {"key": "config", "label": "系统配置"},
        {"key": "ad-revenue", "label": "广告配置"},
        {"key": "huawei-review", "label": "华为审核开关"},
        {"key": "users", "label": "用户管理"},
    ]},
    {"group": "其他", "pages": [
@@ -51,13 +52,14 @@ BUILTIN_ROLES: list[dict] = [
    {"name": SUPER_ADMIN_ROLE, "label": "管理员", "pages": []},
    {"name": "operator", "label": "运营", "pages": [
        "dashboard", "coupon-data", "ad-revenue-report", "comparison-records",
-        "cps", "device-liveness", "price-reports", "feedbacks",
+        "cps", "device-liveness", "price-reports", "feedbacks", "huawei-review",
    ]},
    {"name": "finance", "label": "财务", "pages": [
        "dashboard", "ad-revenue-report", "cps", "withdraws",
    ]},
    {"name": "tech", "label": "技术", "pages": [
-        "dashboard", "device-liveness", "config", "ad-revenue", "event-logs", "audit-logs",
+        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
        "event-logs", "audit-logs",
    ]},
 ]
@@ -7,7 +7,7 @@ from __future__ import annotations
 from collections import Counter
 from datetime import date, datetime, time, timedelta, timezone
-from decimal import Decimal, InvalidOperation
+from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from sqlalchemy import case, func, select
 from sqlalchemy.orm import Session
@@ -109,6 +109,23 @@ def _date_range(date_from: date, date_to: date) -> list[date]:
    return [date_from + timedelta(days=i) for i in range(days + 1)]
 def _duration_percentile(sorted_values: list[int], q: float) -> int | None:
    """Linear-interpolated percentile with the same half-up rounding as Math.round."""
    if not sorted_values:
        return None
    if len(sorted_values) == 1:
        return sorted_values[0]
    index = (len(sorted_values) - 1) * q
    lower = int(index)
    upper = min(lower + 1, len(sorted_values) - 1)
    fraction = Decimal(str(index - lower))
    value = (
        Decimal(sorted_values[lower]) * (Decimal(1) - fraction)
        + Decimal(sorted_values[upper]) * fraction
    )
    return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
 def _id_set(db: Session, stmt) -> set[int]:
    return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}
@@ -242,16 +259,46 @@ def dashboard_overview(
        ComparisonRecord.created_at >= start_local,
        ComparisonRecord.created_at < end_local,
    )
-    period_comparison_total = _count(ComparisonRecord, *period_comparison_conds)
+    period_comparison_stats = db.execute(
-    period_comparison_success = _count(
+        select(
-        ComparisonRecord,
+            func.count(ComparisonRecord.id),
-        *period_comparison_conds,
+            func.coalesce(
-        ComparisonRecord.status == "success",
+                func.sum(
                    case(
                        (ComparisonRecord.status.in_(("success", "failed")), 1),
                        else_=0,
                    )
                ),
                0,
            ),
            func.coalesce(
                func.sum(
                    case((ComparisonRecord.status == "cancelled", 1), else_=0)
                ),
                0,
            ),
            func.coalesce(
                func.sum(case((ComparisonRecord.status == "success", 1), else_=0)),
                0,
            ),
            func.coalesce(func.sum(ComparisonRecord.llm_cost_yuan), 0.0),
        ).where(*period_comparison_conds)
    ).one()
    period_comparison_total = int(period_comparison_stats[0])
    period_comparison_completed = int(period_comparison_stats[1])
    period_comparison_cancelled = int(period_comparison_stats[2])
    period_comparison_success = int(period_comparison_stats[3])
    period_comparison_token_cost_yuan = float(period_comparison_stats[4])
    period_comparison_success_denominator = (
        period_comparison_total - period_comparison_cancelled
    )
    period_comparison_success_rate = (
-        round(period_comparison_success / period_comparison_total, 4)
+        round(
-        if period_comparison_total
+            period_comparison_success / period_comparison_success_denominator,
-        else 0.0
+            4,
        )
        if period_comparison_success_denominator > 0
        else None
    )
    period_saved_positive_count = _count(
        ComparisonRecord,
@@ -282,6 +329,47 @@ def dashboard_overview(
        if period_avg_duration_ms is not None
        else None
    )
    completed_duration_conds = (
        *period_comparison_conds,
        ComparisonRecord.status.in_(("success", "failed")),
        ComparisonRecord.total_ms.is_not(None),
    )
    if db.bind is not None and db.bind.dialect.name == "postgresql":
        period_median_duration_ms, period_p95_duration_ms = db.execute(
            select(
                func.percentile_cont(0.5).within_group(ComparisonRecord.total_ms),
                func.percentile_cont(0.95).within_group(ComparisonRecord.total_ms),
            ).where(*completed_duration_conds)
        ).one()
        period_median_duration_ms = (
            int(
                Decimal(str(period_median_duration_ms)).quantize(
                    Decimal("1"), rounding=ROUND_HALF_UP
                )
            )
            if period_median_duration_ms is not None
            else None
        )
        period_p95_duration_ms = (
            int(
                Decimal(str(period_p95_duration_ms)).quantize(
                    Decimal("1"), rounding=ROUND_HALF_UP
                )
            )
            if period_p95_duration_ms is not None
            else None
        )
    else:
        # SQLite 测试环境没有 percentile_cont；仅回退读取耗时单列，不加载完整记录。
        completed_durations = list(
            db.execute(
                select(ComparisonRecord.total_ms)
                .where(*completed_duration_conds)
                .order_by(ComparisonRecord.total_ms)
            ).scalars()
        )
        period_median_duration_ms = _duration_percentile(completed_durations, 0.5)
        period_p95_duration_ms = _duration_percentile(completed_durations, 0.95)
    ordered_exists = (
        select(SavingsRecord.id)
@@ -622,11 +710,16 @@ def dashboard_overview(
            },
            "comparison": {
                "total": period_comparison_total,
                "completed": period_comparison_completed,
                "cancelled": period_comparison_cancelled,
                "success": period_comparison_success,
                "success_rate": period_comparison_success_rate,
                "ordered": period_ordered_count,
                "average_duration_ms": period_avg_duration_ms,
                "median_duration_ms": period_median_duration_ms,
                "p95_duration_ms": period_p95_duration_ms,
                "average_saved_cents": period_avg_saved_cents,
                "token_cost_total_yuan": period_comparison_token_cost_yuan,
            },
            "coupon": {
                "started": coupon_started,
@@ -0,0 +1,58 @@
 """admin 华为审核开关:控制新手引导页(快速设置)在华为 ROM 客户端能否被用户关闭。
 存在 app_config 表的 huawei_review dict(见 repositories/app_config.get_huawei_review/set_huawei_review)。
 客户端经 /api/v1/platform/huawei-review 拉取(且只有华为 ROM 机型会去拉)。权限 operator/tech + 审计。
 """
 from __future__ import annotations
 from typing import Annotated
 from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy.orm import Session
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.schemas.huawei_review import HuaweiReviewOut, HuaweiReviewUpdate
 from app.models.admin import AdminUser
 from app.models.app_config import AppConfig
 from app.repositories import app_config
 router = APIRouter(
    prefix="/admin/api/huawei-review",
    tags=["admin-huawei-review"],
    dependencies=[Depends(get_current_admin)],
 )
 def _out(db: Session) -> HuaweiReviewOut:
    row = db.get(AppConfig, app_config.HUAWEI_REVIEW_KEY)
    return HuaweiReviewOut(
        mode=app_config.get_huawei_review(db)["mode"],
        updated_at=row.updated_at.isoformat() if row is not None else None,
    )
@router.get("", response_model=HuaweiReviewOut, summary="华为审核开关当前状态")
 def get_huawei_review(db: AdminDb) -> HuaweiReviewOut:
    return _out(db)
@router.patch("", response_model=HuaweiReviewOut, summary="切换华为审核开关(带审计)")
 def update_huawei_review(
    body: HuaweiReviewUpdate,
    request: Request,
    admin: Annotated[AdminUser, Depends(require_role("operator", "tech"))],
    db: AdminDb,
 ) -> HuaweiReviewOut:
    before = app_config.get_huawei_review(db)["mode"]
    try:
        app_config.set_huawei_review(db, body.mode, admin_id=admin.id, commit=False)
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    write_audit(
        db, admin, action="huawei_review.set", target_type="huawei_review",
        target_id=app_config.HUAWEI_REVIEW_KEY,
        detail={"before": before, "after": body.mode}, ip=get_client_ip(request), commit=False,
    )
    db.commit()
    return _out(db)
@@ -53,11 +53,16 @@ class DashboardPeriodUsers(BaseModel):
 class DashboardPeriodComparison(BaseModel):
    total: int
    completed: int
    cancelled: int
    success: int
-    success_rate: float
+    success_rate: float | None = None
    ordered: int
    average_duration_ms: int | None = None
    median_duration_ms: int | None = None
    p95_duration_ms: int | None = None
    average_saved_cents: int | None = None
    token_cost_total_yuan: float = 0.0
 class DashboardPeriodCoupon(BaseModel):
@@ -0,0 +1,19 @@
 """admin 华为审核开关 schemas(两态:default / review)。"""
 from __future__ import annotations
 from typing import Literal
 from pydantic import BaseModel
 class HuaweiReviewOut(BaseModel):
    """当前开关状态。updated_at 给后台展示「谁什么时候切的」提供时间锚点。"""
    mode: Literal["default", "review"]
    updated_at: str | None = None  # ISO 字符串;从未切过为 None
 class HuaweiReviewUpdate(BaseModel):
    """切换开关。整值覆盖,不做部分更新(就一个字段)。"""
    mode: Literal["default", "review"]
@@ -280,13 +280,21 @@ def ecpm_report(payload: EcpmReportIn, user: CurrentUser, db: DbSession) -> Ecpm
    Bearer 鉴权,user_id 取自 JWT(不信 body)。best-effort:落库即 ok,客户端 fire-and-forget,
    丢一两条不影响业务(穿山甲后台报表是结算权威)。eCPM 与发奖(S2S)是两条独立流,不逐条关联。
    """
    attributed_trace_id = crud_ecpm.attributable_trace_id(
        db, feed_scene=payload.feed_scene, trace_id=payload.trace_id
    )
    if payload.trace_id and attributed_trace_id is None:
        logger.info(
            "detach late coupon ad impression from failed trace user_id=%d trace=%s session=%s",
            user.id, payload.trace_id, payload.ad_session_id,
        )
    crud_ecpm.create_ecpm_record(
        db, user.id,
        ad_type=payload.ad_type, ecpm_raw=payload.ecpm,
        ad_session_id=payload.ad_session_id,
        adn=payload.adn, slot_id=payload.slot_id,
        feed_scene=payload.feed_scene,
-        trace_id=payload.trace_id,
+        trace_id=attributed_trace_id,
        app_env=payload.app_env, our_code_id=payload.our_code_id,
    )
    logger.info(
@@ -20,6 +20,7 @@ from app.schemas.platform import (
    AdConfigPublicOut,
    AppFlagsOut,
    AppVersionOut,
    HuaweiReviewOut,
    PlatformStatsOut,
    SavingsFeedItem,
    SavingsFeedOut,
@@ -72,6 +73,15 @@ def ad_config(db: DbSession) -> AdConfigPublicOut:
    )
@router.get("/huawei-review", response_model=HuaweiReviewOut, summary="华为审核开关(不鉴权)")
 def huawei_review(db: DbSession) -> HuaweiReviewOut:
    """客户端进新手引导前拉一次,决定「快速设置」权限步左上角要不要给退出按钮。
    不鉴权:引导页在登录之前就展示,此时必然没有 token。空库回退 default(=不给退出按钮,维持现状)。
    只有华为 ROM 客户端会来拉(荣耀 MagicOS 不拉),故这里不做机型判断,由客户端自己 gate。"""
    mode = app_config.get_huawei_review(db)["mode"]
    return HuaweiReviewOut(mode=mode, onboarding_closable=(mode == "review"))
@router.get("/app-version", response_model=AppVersionOut, summary="最新 App 版本(OTA 检查更新,不鉴权)")
 def app_version(db: DbSession) -> AppVersionOut:
    """客户端启动 / 手动检查更新时拉取。不鉴权:版本信息非敏感,且检查更新可能在登录前。
@@ -347,6 +347,31 @@ class Settings(BaseSettings):
            return []
        return [o.strip() for o in self.CORS_ALLOW_ORIGINS.split(",") if o.strip()]
    # ===== 可观测(OpenObserve 接口指标)=====
    # 采集每个接口的 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker)。
    # 默认关(prod 安全):未开启 → 中间件透传、worker 不启动,整套 no-op。
    # 开启需 ENABLED=true 且 ENDPOINT/USER/PASSWORD 齐全(见 observe_configured)。
    OBSERVE_ENABLED: bool = False
    OBSERVE_ENDPOINT: str = "http://localhost:5080"   # OpenObserve base URL
    OBSERVE_ORG: str = "default"                        # 组织名
    OBSERVE_STREAM: str = "app_requests"               # stream 名(首次上报自动建)
    OBSERVE_USER: str = ""                             # Basic auth 邮箱
    OBSERVE_PASSWORD: str = ""                         # Basic auth 密码/token
    OBSERVE_FLUSH_INTERVAL_SEC: float = 5.0            # worker 最长攒批间隔
    OBSERVE_BATCH_MAX: int = 200                       # 单批最大事件数
    OBSERVE_QUEUE_MAX: int = 10000                     # 有界队列上限,满则丢
    OBSERVE_TIMEOUT_SEC: float = 5.0                   # 上报 HTTP 超时
    @property
    def observe_configured(self) -> bool:
        """观测上报可用 = 总开关开 且 endpoint/账号/密码齐全(缺则整套 no-op)。"""
        return bool(
            self.OBSERVE_ENABLED
            and self.OBSERVE_ENDPOINT
            and self.OBSERVE_USER
            and self.OBSERVE_PASSWORD
        )
    @property
    def is_prod(self) -> bool:
        return self.APP_ENV == "prod"
@@ -0,0 +1,110 @@
 """接口指标埋点:有界事件队列 + 纯 ASGI 中间件。
 每个 HTTP 请求测总耗时、抓路由模板 + 状态码,非阻塞塞进有界队列;由 observe_worker
 后台批量上报到 OpenObserve。请求路径上无任何 I/O。未配置观测时中间件直接透传。
 """
 from __future__ import annotations
 import asyncio
 import os
 import time
 from starlette.routing import Match
 from app.core.config import settings
 # 不采集的路径(纯噪音):健康检查。
 _SKIP_PATHS = frozenset({"/health"})
 # 未匹配路由(404/扫描器)归一到此,防维度爆炸。
 _UNMATCHED = "__unmatched__"
 # service 字段:与 logging.py 同源(LOG_SERVICE_NAME),默认 app-server。
 _SERVICE = os.getenv("LOG_SERVICE_NAME", "app-server")
 # 有界事件队列(懒创建,见 get_queue):首次取用时在运行中的 loop 里建,避免 import 期
 # 无 loop 的边角问题;put_nowait/get_nowait 不需运行中的 loop → 可在无 loop 下测试。
 _queue: asyncio.Queue[dict] | None = None
 # 队列满时的丢弃计数,worker 定期取出打日志。
 _dropped = 0
 def get_queue() -> asyncio.Queue[dict]:
    """返回全局有界事件队列(懒创建)。测试可 monkeypatch 模块级 _queue 换成小队列。"""
    global _queue
    if _queue is None:
        _queue = asyncio.Queue(maxsize=settings.OBSERVE_QUEUE_MAX)
    return _queue
 def take_dropped() -> int:
    """取出并清零累计丢弃数(供 worker 打点)。"""
    global _dropped
    n, _dropped = _dropped, 0
    return n
 def record_event(event: dict) -> None:
    """非阻塞入队;队列满则丢弃当前事件并计数。永不抛异常、永不阻塞请求。"""
    global _dropped
    try:
        get_queue().put_nowait(event)
    except asyncio.QueueFull:
        _dropped += 1
 def _resolve_route(scope) -> str:
    """从 scope 取路由模板(如 /things/{tid})。优先 scope['route'](现代 Starlette
    路由后写入);取不到则手动匹配一次(老版本兜底);仍无 → __unmatched__(404/扫描器)。"""
    route = scope.get("route")
    path = getattr(route, "path", None)
    if path:
        return path
    app_ = scope.get("app")
    router = getattr(app_, "router", None)
    for candidate in getattr(router, "routes", []):
        try:
            match, _ = candidate.matches(scope)
        except Exception:  # noqa: BLE001 - 匹配兜底,任一路由异常不影响整体
            continue
        if match == Match.FULL and getattr(candidate, "path", None):
            return candidate.path
    return _UNMATCHED
 class RequestMetricsMiddleware:
    """纯 ASGI 中间件:测每个 http 请求耗时,记 method/route/status/duration。
    放在最外层(main.py 里 CORS 之后 add),测到含 CORS 的完整耗时。未配置观测 → 透传。
    """
    def __init__(self, app) -> None:
        self.app = app
    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http" or not settings.observe_configured:
            await self.app(scope, receive, send)
            return
        if scope.get("path") in _SKIP_PATHS:
            await self.app(scope, receive, send)
            return
        start = time.perf_counter()
        status_holder = {"status": 500}  # 下游异常未产出 response 时兜底 500
        async def send_wrapper(message) -> None:
            if message["type"] == "http.response.start":
                status_holder["status"] = message["status"]
            await send(message)
        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            duration_ms = (time.perf_counter() - start) * 1000.0
            record_event({
                "_timestamp": int(time.time() * 1_000_000),  # µs,OpenObserve 时间列
                "service": _SERVICE,
                "env": settings.APP_ENV,
                "method": scope.get("method", ""),
                "route": _resolve_route(scope),
                "status": status_holder["status"],
                "duration_ms": round(duration_ms, 3),
            })
@@ -0,0 +1,128 @@
 """接口指标后台上报 worker:批量 drain 事件队列 → POST 到 OpenObserve。
 对齐 heartbeat_monitor_worker 等的 start_*/stop_* 形态。best-effort 遥测:catch 全部
 异常,上报失败直接丢批不重试。未配置观测 → start 返回 None(不启动),整套 no-op。
 """
 from __future__ import annotations
 import asyncio
 import contextlib
 import logging
 import httpx
 from app.core.config import settings
 from app.core.observe import get_queue, take_dropped
 logger = logging.getLogger("shagua.observe")
 # 上报用的 httpx client,start 时建、stop 时关。
 _client: httpx.AsyncClient | None = None
 async def _collect_batch() -> list[dict]:
    """等到 ≥1 条(或到 flush 间隔)后,连抽到 BATCH_MAX 条或抽空。超时且空 → 返回 []。"""
    queue = get_queue()
    batch: list[dict] = []
    try:
        first = await asyncio.wait_for(
            queue.get(), timeout=settings.OBSERVE_FLUSH_INTERVAL_SEC
        )
    except asyncio.TimeoutError:  # noqa: UP041 - 3.10 兼容:该版 wait_for 抛的 asyncio.TimeoutError ≠ 内置 TimeoutError
        return batch
    batch.append(first)
    while len(batch) < settings.OBSERVE_BATCH_MAX:
        try:
            batch.append(queue.get_nowait())
        except asyncio.QueueEmpty:
            break
    return batch
 async def _post_batch(client: httpx.AsyncClient, batch: list[dict]) -> None:
    """POST 一批事件到 OpenObserve 的 _json ingest 端点。非 2xx 仅告警。"""
    url = f"/api/{settings.OBSERVE_ORG}/{settings.OBSERVE_STREAM}/_json"
    resp = await client.post(url, json=batch)
    if resp.status_code >= 300:
        logger.warning(
            "observe ingest failed status=%s body=%s",
            resp.status_code,
            resp.text[:200],
        )
 async def _run_loop(client: httpx.AsyncClient) -> None:
    try:
        while True:
            batch = await _collect_batch()
            dropped = take_dropped()
            if dropped:
                logger.warning("observe dropped %d events (queue full)", dropped)
            if not batch:
                continue
            try:
                await _post_batch(client, batch)
            except Exception:  # noqa: BLE001 - best-effort 遥测,失败丢批不重试、不退出
                logger.warning(
                    "observe post batch failed, dropped %d events",
                    len(batch),
                    exc_info=True,
                )
    except asyncio.CancelledError:
        logger.info("observe worker stopped")
        raise
 def start_observe_worker() -> asyncio.Task | None:
    """启动上报 worker。未配置观测 → 返回 None(no-op)。约定每进程只调一次(lifespan)。"""
    global _client
    if not settings.observe_configured:
        return None
    if _client is not None:
        # 约定 start 每进程只调一次;已启动则不重复建 client(避免泄漏旧连接池)。
        logger.warning("observe worker already started; ignoring duplicate start")
        return None
    _client = httpx.AsyncClient(
        base_url=settings.OBSERVE_ENDPOINT,
        auth=(settings.OBSERVE_USER, settings.OBSERVE_PASSWORD),
        timeout=settings.OBSERVE_TIMEOUT_SEC,
    )
    logger.info(
        "observe worker started endpoint=%s org=%s stream=%s",
        settings.OBSERVE_ENDPOINT,
        settings.OBSERVE_ORG,
        settings.OBSERVE_STREAM,
    )
    return asyncio.create_task(_run_loop(_client), name="observe-worker")
 async def stop_observe_worker(task: asyncio.Task | None) -> None:
    """收尾:cancel worker → best-effort 发最后一批 → 关 client。"""
    global _client
    if task is None:
        return
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task
    dropped = take_dropped()  # 收口:补记最后一个 flush 窗口累计的丢弃数,不让账丢在关停期
    if dropped:
        logger.warning("observe dropped %d events (queue full) before shutdown", dropped)
    if _client is not None:
        # worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停);
        # 超过一批(BATCH_MAX)的剩余直接丢,不做多轮 flush(best-effort,关停从速)。
        try:
            queue = get_queue()
            final: list[dict] = []
            while len(final) < settings.OBSERVE_BATCH_MAX:
                try:
                    final.append(queue.get_nowait())
                except asyncio.QueueEmpty:
                    break
            if final:
                await asyncio.wait_for(
                    _post_batch(_client, final), timeout=settings.OBSERVE_TIMEOUT_SEC
                )
        except Exception:  # noqa: BLE001 - 关停期尽力而为,失败忽略
            pass
        await _client.aclose()
        _client = None
@@ -54,6 +54,11 @@ from app.core.inactivity_reset_worker import (
    stop_inactivity_reset_worker,
 )
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
    start_observe_worker,
    stop_observe_worker,
 )
 from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
 from app.core.withdraw_reconcile_worker import (
    start_withdraw_reconcile_worker,
@@ -84,6 +89,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    reconcile_task = start_withdraw_reconcile_worker()
    heartbeat_task = start_heartbeat_monitor()
    daily_exchange_task = start_daily_exchange_worker()
    observe_task = start_observe_worker()
    inactivity_task = start_inactivity_reset_worker()
    try:
        yield
@@ -91,6 +97,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        await stop_heartbeat_monitor(heartbeat_task)
        await stop_withdraw_reconcile_worker(reconcile_task)
        await stop_daily_exchange_worker(daily_exchange_task)
        await stop_observe_worker(observe_task)
        await stop_inactivity_reset_worker(inactivity_task)
        await aclose_pricebot_client()
        logger.info("shutting down")
@@ -113,6 +120,9 @@ if settings.cors_origins_list:
        allow_headers=["*"],
    )
 # 接口指标埋点(放在 CORS 之后 = 最外层:测到含 CORS 的完整耗时)。未配置观测时中间件自 no-op。
 app.add_middleware(RequestMetricsMiddleware)
@app.get("/health", tags=["meta"])
 def health() -> dict[str, str]:
@@ -13,6 +13,25 @@ from sqlalchemy.orm import Session
 from app.core import rewards
 from app.core.rewards import cn_today
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.coupon_state import CouponSession
 def attributable_trace_id(
    db: Session, *, feed_scene: str | None, trace_id: str | None
 ) -> str | None:
    """返回广告展示允许归属的业务 trace。
    领券任务可能在 Draw 广告异步加载完成前已经失败或被用户放弃。非完成终态先落库、
    广告回调后到时，收益仍需保留在总广告报表中，但不能再挂到该死亡领券明细，
    因此清空关联 trace。其它场景、找不到 session、进行中或已完成状态保持原值，
    由客户端生命周期修复负责主防线。
    """
    if feed_scene != "coupon" or not trace_id:
        return trace_id
    session_status = db.execute(
        select(CouponSession.status).where(CouponSession.trace_id == trace_id)
    ).scalar_one_or_none()
    return None if session_status in {"failed", "abandoned"} else trace_id
 def create_ecpm_record(
@@ -142,3 +142,47 @@ def set_ad_config(db: Session, data: dict, *, admin_id: int, commit: bool = True
    else:
        db.flush()
    return row
 # ── 华为审核开关(admin 可切,仅华为 ROM 客户端拉)────────────────────────────────
 # 同 ad_config:复用 AppConfig 表但不进 CONFIG_DEFS——它由「华为审核开关」专用页管理,
 # 有自己的两态语义,混进通用系统配置页只会显示成一个没头没尾的 on/off。
 #   default → 新手引导页(快速设置)强制展示,用户无法关闭(= 上线至今的现状)
 #   review  → 快速设置的权限步左上角出现退出按钮,可直接进首页(过华为应用市场审核用)
 # 空库 = default = 行为完全不变。客户端经 /api/v1/platform/huawei-review 拉取。
 HUAWEI_REVIEW_KEY = "huawei_review"
 HUAWEI_REVIEW_MODES = ("default", "review")
 _HUAWEI_REVIEW_DEFAULTS: dict[str, Any] = {
    "mode": "default",
 }
 def get_huawei_review(db: Session) -> dict:
    """读华为审核开关。DB 无 / 脏值一律回退 default(宁可不给退出按钮,也不误放开)。"""
    row = db.get(AppConfig, HUAWEI_REVIEW_KEY)
    merged = dict(_HUAWEI_REVIEW_DEFAULTS)
    if row is not None and isinstance(row.value, dict):
        merged.update(row.value)
    if merged.get("mode") not in HUAWEI_REVIEW_MODES:
        merged["mode"] = _HUAWEI_REVIEW_DEFAULTS["mode"]
    return merged
 def set_huawei_review(db: Session, mode: str, *, admin_id: int, commit: bool = True) -> AppConfig:
    """admin 切换华为审核开关。非法 mode 抛 ValueError(路由转 400)。"""
    if mode not in HUAWEI_REVIEW_MODES:
        raise ValueError(f"invalid mode: {mode} (expected one of {list(HUAWEI_REVIEW_MODES)})")
    row = db.get(AppConfig, HUAWEI_REVIEW_KEY)
    value = {"mode": mode}
    if row is None:
        row = AppConfig(key=HUAWEI_REVIEW_KEY, value=value, updated_by_admin_id=admin_id)
        db.add(row)
    else:
        row.value = value
        row.updated_by_admin_id = admin_id
    if commit:
        db.commit()
        db.refresh(row)
    else:
        db.flush()
    return row
@@ -46,6 +46,17 @@ class AdConfigPublicOut(BaseModel):
    withdrawal_ad_enabled: bool  # 提现激励视频开关(关=客户端直接放行提现)
 class HuaweiReviewOut(BaseModel):
    """华为审核开关下发给客户端(不鉴权,引导页在登录前就要展示)。
    客户端只需读 onboarding_closable 决策;mode 仅供排查问题时看后台切成了哪态。
    只有华为 ROM(HarmonyOS/EMUI,不含荣耀 MagicOS)的客户端才会来拉这个端点。
    """
    mode: str = "default"  # default | review
    onboarding_closable: bool = False  # 快速设置权限步是否允许用户退出(mode == review)
 class AppVersionOut(BaseModel):
    """最新 App 版本信息(OTA 检查更新,不鉴权)。
@@ -19,7 +19,11 @@ server {
    ssl_ciphers         HIGH:!aNULL:!MD5;
    ssl_session_cache   shared:SSL:10m;
-    client_max_body_size 4m;
+    # 上传接口(反馈/上报截图、头像)业务上限 = 最多 6 张 × 每张 5MB
    # (见 app _MAX_IMAGES / AVATAR_MAX_BYTES)≈ 30MB,留余量设 32m。
    # 低于此值时带截图的反馈会在到达 uvicorn 前就被 nginx 413,表现为「提交经常失败」
    # (纯文字反馈体积小、不受影响 → 呈现为「时好时坏」)。根治仍需客户端上传前压缩。
    client_max_body_size 32m;
    location / {
        proxy_pass http://127.0.0.1:8770;
@@ -0,0 +1,57 @@
 # OpenObserve 监控台反代(observe.shaguabijia.com)。证书走 Certbot/Let's Encrypt,与 admin-web 一致。
 #
 # 前置(一次性):
 #   1) DNS:  observe.shaguabijia.com  A 记录 → 本服务器公网 IP
 #   2) 证书: sudo certbot certonly --nginx -d observe.shaguabijia.com
 #            (options-ssl-nginx.conf / ssl-dhparams.pem 首次跑 certbot 时已生成,admin-web 在用即已存在)
 #   3) OpenObserve 只绑 127.0.0.1:5080(见 docker-compose.prod.yml),本文件把它反代出公网
 #   4) nginx -t 通过后 systemctl reload nginx
 #
 # 安全:OO 有自身登录。监控台不必对全网裸开——本机办公网无固定出口 IP,故在 nginx 层加 Basic Auth 兜底;
 #      将来有固定 IP 可改用【IP 白名单】块(更省事,可去掉 Basic Auth)。
 server {
    server_name observe.shaguabijia.com;
    client_max_body_size 10m;
    # —— IP 白名单:办公网无固定出口 IP,暂不用;将来有固定 IP 可改用这块(比 Basic Auth 省事)——
    # allow 1.2.3.4;      # ← 换成你的真实出口 IP,可多行
    # deny  all;
    # —— Basic Auth:无固定 IP 的兜底密码(生成 .htpasswd_observe 的命令见 README/下方)——
    auth_basic           "OpenObserve";
    auth_basic_user_file /etc/nginx/conf.d/.htpasswd_observe;
    location / {
        proxy_pass http://127.0.0.1:5080;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        # OpenObserve 有实时/流式面板,需透传 WebSocket
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_read_timeout 300s;
    }
    # IPv6 这行不带 ipv6only=on:该选项对 [::]:443 全局只能设一次,admin-web 那个 server 块已设(否则 nginx 报 duplicate listen options)
    listen [::]:443 ssl; # managed by Certbot
    listen 443 ssl; # managed by Certbot
    ssl_certificate /etc/letsencrypt/live/observe.shaguabijia.com/fullchain.pem; # managed by Certbot
    ssl_certificate_key /etc/letsencrypt/live/observe.shaguabijia.com/privkey.pem; # managed by Certbot
    include /etc/letsencrypt/options-ssl-nginx.conf; # managed by Certbot
    ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; # managed by Certbot
 }
 server {
    if ($host = observe.shaguabijia.com) {
        return 301 https://$host$request_uri;
    } # managed by Certbot
    listen 80;
    listen [::]:80;
    server_name observe.shaguabijia.com;
    return 404; # managed by Certbot
 }
@@ -0,0 +1,4 @@
 # OpenObserve 落盘数据(parquet/索引/元数据),运行时产生,不入库。
 data/
 # 生产 compose 的密码文件(OO_ROOT_PASSWORD),含机密,不入库。
 .env
@@ -0,0 +1,126 @@
 # OpenObserve 本地部署（接口 QPS / 耗时可观测）
 app-server 通过中间件采集每个接口的 QPS + 耗时 + 错误率，批量上报到这里。
 设计见 [../../docs/superpowers/specs/2026-07-06-openobserve-api-metrics-design.md](../../docs/superpowers/specs/2026-07-06-openobserve-api-metrics-design.md)。
 ## 启动
 ```bash
 cd deploy/openobserve
 docker compose up -d
 ```
 - Web UI：http://localhost:5080
 - 登录：`admin@shaguabijia.local` / `Complexpass#123`（见 `docker-compose.yml`）
 - 数据落 `deploy/openobserve/data/`（已挂卷持久化；该目录已 gitignore）
 ## 让 app-server 上报
 在项目根的 `.env` 打开观测（`OBSERVE_*`，账号密码与 compose 里 root 一致）：
 ```dotenv
 OBSERVE_ENABLED=true
 OBSERVE_ENDPOINT=http://localhost:5080
 OBSERVE_ORG=default
 OBSERVE_STREAM=app_requests
 OBSERVE_USER=admin@shaguabijia.local
 OBSERVE_PASSWORD=Complexpass#123
 ```
 重启 app-server，随便打几个接口。stream `app_requests` **首次上报自动创建**，
 在 UI 的 Logs → 选 `app_requests` 就能看到逐条请求事件（字段：`method` / `route` /
 `status` / `duration_ms` / `service` / `env`）。
 > 未开 `OBSERVE_ENABLED` 或缺账号密码时，中间件透传、worker 不启动，整套 no-op，不影响业务。
 ## 查询（Logs 页 SQL，或建 Dashboard 面板）
 各接口 QPS（1 分钟分桶，面板里再除 60 得每秒）：
 ```sql
 SELECT route, histogram(_timestamp, '1 minute') AS ts, count(*) AS cnt
 FROM app_requests GROUP BY route, ts ORDER BY ts
 ```
 各接口 P95 耗时（毫秒）：
 ```sql
 SELECT route, approx_percentile_cont(duration_ms, 0.95) AS p95_ms
 FROM app_requests GROUP BY route ORDER BY p95_ms DESC
 ```
 各接口错误率（5xx 占比）：
 ```sql
 SELECT route,
       count(*) FILTER (WHERE status >= 500) * 100.0 / count(*) AS err_pct
 FROM app_requests GROUP BY route ORDER BY err_pct DESC
 ```
 ## 一键导入现成仪表盘（QPS / P95 / 分位 / 错误率）
 备好了 [dashboard-api-metrics.json](dashboard-api-metrics.json)，4 个面板：各接口每分钟请求数(QPS 源)、
 P95 耗时折线、P50/P95/P99 分位表、5xx 错误率表。
 - **UI 导入**：Dashboards → 右上 **Import** → 选该 JSON 文件 → Import（每次导入新建，不覆盖）。
 - **或 API 导入**：
  ```bash
  curl -u admin@shaguabijia.local:Complexpass#123 -H 'Content-Type: application/json' \
    -X POST 'http://localhost:5080/api/default/dashboards?folder=default' \
    --data-binary @deploy/openobserve/dashboard-api-metrics.json
  ```
 导入后进仪表盘，右上角时间调到「最近 15 分钟 / 1 小时」、开自动刷新即可。低流量下 QPS 面板看「每分钟请求数」比「每秒」直观。
 ## 停止 / 清数据
 ```bash
 docker compose down            # 停止(保留数据)
 docker compose down -v && rm -rf data   # 停止并清空数据
 ```
 ## 生产部署（单机）+ UI 访问
 前提:app-server 与 OpenObserve **同机**,app→OO 走 localhost(`127.0.0.1:5080`)、不出网、无需 TLS。
 唯一要防的是**别把 :5080 裸暴露公网**。硬化版编排见 [docker-compose.prod.yml](docker-compose.prod.yml)。
 ### 部署步骤
 ```bash
 # 1) 密码文件(本目录,已 gitignore)
 echo "OO_ROOT_PASSWORD=$(python -c 'import secrets;print(secrets.token_urlsafe(24))')" > deploy/openobserve/.env
 # 2) 起 OpenObserve(只绑 127.0.0.1、命名卷持久化、mem 1g)
 cd deploy/openobserve && docker compose -f docker-compose.prod.yml up -d
 sudo systemctl enable docker            # 开机自起
 ```
 3) app-server 的 `.env` 打开观测并**重启**(用非 root 的专用 ingest 账号):
 ```dotenv
 OBSERVE_ENABLED=true
 OBSERVE_ENDPOINT=http://127.0.0.1:5080
 OBSERVE_ORG=default
 OBSERVE_STREAM=app_requests
 OBSERVE_USER=ingest@shaguabijia.com     # UI → Users 建的非 root 账号
 OBSERVE_PASSWORD=<该账号密码>
 ```
 ```bash
 sudo systemctl restart shaguabijia-app-server   # 日志出现 "observe worker started" 即生效
 ```
 4) 两个必做收口(磁盘/安全):
 - **保留期**:UI → Streams → `app_requests` → Data Retention 设 14/30 天(一请求一行,不封顶迟早撑爆盘)。
 - **专用账号**:UI → Users 建非 root 账号给 app 上报,root 只留人工登 UI。
 ### UI 访问(二选一)
 **A. SSH 隧道(推荐,零暴露、不用域名/证书):**
 ```bash
 ssh -L 5080:127.0.0.1:5080  用户@服务器IP
 # 然后本机浏览器开 http://localhost:5080
 ```
 **B. nginx 子域名反代(要固定 URL / 团队常看):** 见 [../nginx/observe.shaguabijia.com.conf](../nginx/observe.shaguabijia.com.conf)。
 需 DNS `observe.shaguabijia.com` → 本机 + 证书放 `/etc/nginx/ssl/`;含 IP 白名单 + TLS + WebSocket 透传。
 > ⚠️ prod compose 必须保持 `127.0.0.1:5080:5080`;写成 `5080:5080`(绑 0.0.0.0)= 裸暴露公网,这是唯一真正的坑。
@@ -0,0 +1,302 @@
 {
  "version": 8,
  "dashboardId": "api-metrics",
  "title": "接口监控 (QPS / 耗时 / 错误率)",
  "description": "app-server 接口 QPS、P50/P95/P99 耗时、5xx 错误率。数据流 app_requests。",
  "role": "",
  "tabs": [
    {
      "tabId": "default",
      "name": "Default",
      "panels": [
        {
          "id": "panel_qps",
          "type": "line",
          "title": "各接口 每分钟请求数 (QPS 源)",
          "description": "",
          "config": {
            "show_legends": true,
            "legends_position": null,
            "decimals": 2.0,
            "axis_border_show": false,
            "base_map": null,
            "map_view": null
          },
          "queryType": "sql",
          "queries": [
            {
              "query": "SELECT histogram(_timestamp, '1 minute') as ts, route, count(*) as reqs FROM app_requests GROUP BY ts, route ORDER BY ts",
              "vrlFunctionQuery": "",
              "customQuery": true,
              "fields": {
                "stream": "app_requests",
                "stream_type": "logs",
                "x": [
                  {
                    "label": "ts",
                    "alias": "ts",
                    "column": "ts",
                    "color": null,
                    "sortBy": "ASC"
                  }
                ],
                "y": [
                  {
                    "label": "reqs",
                    "alias": "reqs",
                    "column": "reqs",
                    "color": null
                  }
                ],
                "z": [],
                "breakdown": [
                  {
                    "label": "route",
                    "alias": "route",
                    "column": "route",
                    "color": null
                  }
                ],
                "filter": {
                  "filterType": "group",
                  "logicalOperator": "AND",
                  "conditions": []
                }
              },
              "config": {
                "promql_legend": "",
                "layer_type": "scatter",
                "weight_fixed": 1.0
              }
            }
          ],
          "layout": {
            "x": 0,
            "y": 0,
            "w": 24,
            "h": 9,
            "i": 1
          }
        },
        {
          "id": "panel_p95",
          "type": "line",
          "title": "各接口 P95 耗时 (ms)",
          "description": "",
          "config": {
            "show_legends": true,
            "legends_position": null,
            "decimals": 2.0,
            "axis_border_show": false,
            "base_map": null,
            "map_view": null
          },
          "queryType": "sql",
          "queries": [
            {
              "query": "SELECT histogram(_timestamp, '1 minute') as ts, route, approx_percentile_cont(duration_ms, 0.95) as p95_ms FROM app_requests GROUP BY ts, route ORDER BY ts",
              "vrlFunctionQuery": "",
              "customQuery": true,
              "fields": {
                "stream": "app_requests",
                "stream_type": "logs",
                "x": [
                  {
                    "label": "ts",
                    "alias": "ts",
                    "column": "ts",
                    "color": null,
                    "sortBy": "ASC"
                  }
                ],
                "y": [
                  {
                    "label": "p95_ms",
                    "alias": "p95_ms",
                    "column": "p95_ms",
                    "color": null
                  }
                ],
                "z": [],
                "breakdown": [
                  {
                    "label": "route",
                    "alias": "route",
                    "column": "route",
                    "color": null
                  }
                ],
                "filter": {
                  "filterType": "group",
                  "logicalOperator": "AND",
                  "conditions": []
                }
              },
              "config": {
                "promql_legend": "",
                "layer_type": "scatter",
                "weight_fixed": 1.0
              }
            }
          ],
          "layout": {
            "x": 24,
            "y": 0,
            "w": 24,
            "h": 9,
            "i": 2
          }
        },
        {
          "id": "panel_pctl",
          "type": "table",
          "title": "各接口 耗时分位 P50/P95/P99 (ms)",
          "description": "",
          "config": {
            "show_legends": true,
            "legends_position": null,
            "decimals": 2.0,
            "axis_border_show": false,
            "base_map": null,
            "map_view": null
          },
          "queryType": "sql",
          "queries": [
            {
              "query": "SELECT route, approx_percentile_cont(duration_ms,0.5) as p50, approx_percentile_cont(duration_ms,0.95) as p95, approx_percentile_cont(duration_ms,0.99) as p99, count(*) as cnt FROM app_requests GROUP BY route ORDER BY p95 DESC",
              "vrlFunctionQuery": "",
              "customQuery": true,
              "fields": {
                "stream": "app_requests",
                "stream_type": "logs",
                "x": [
                  {
                    "label": "route",
                    "alias": "route",
                    "column": "route",
                    "color": null
                  }
                ],
                "y": [
                  {
                    "label": "p50",
                    "alias": "p50",
                    "column": "p50",
                    "color": null
                  },
                  {
                    "label": "p95",
                    "alias": "p95",
                    "column": "p95",
                    "color": null
                  },
                  {
                    "label": "p99",
                    "alias": "p99",
                    "column": "p99",
                    "color": null
                  },
                  {
                    "label": "cnt",
                    "alias": "cnt",
                    "column": "cnt",
                    "color": null
                  }
                ],
                "z": [],
                "breakdown": [],
                "filter": {
                  "filterType": "group",
                  "logicalOperator": "AND",
                  "conditions": []
                }
              },
              "config": {
                "promql_legend": "",
                "layer_type": "scatter",
                "weight_fixed": 1.0
              }
            }
          ],
          "layout": {
            "x": 0,
            "y": 9,
            "w": 24,
            "h": 9,
            "i": 3
          }
        },
        {
          "id": "panel_err",
          "type": "table",
          "title": "各接口 错误率 (5xx %)",
          "description": "",
          "config": {
            "show_legends": true,
            "legends_position": null,
            "decimals": 2.0,
            "axis_border_show": false,
            "base_map": null,
            "map_view": null
          },
          "queryType": "sql",
          "queries": [
            {
              "query": "SELECT route, count(*) FILTER (WHERE status >= 500) * 100.0 / count(*) as err_pct, count(*) as cnt FROM app_requests GROUP BY route ORDER BY err_pct DESC",
              "vrlFunctionQuery": "",
              "customQuery": true,
              "fields": {
                "stream": "app_requests",
                "stream_type": "logs",
                "x": [
                  {
                    "label": "route",
                    "alias": "route",
                    "column": "route",
                    "color": null
                  }
                ],
                "y": [
                  {
                    "label": "err_pct",
                    "alias": "err_pct",
                    "column": "err_pct",
                    "color": null
                  },
                  {
                    "label": "cnt",
                    "alias": "cnt",
                    "column": "cnt",
                    "color": null
                  }
                ],
                "z": [],
                "breakdown": [],
                "filter": {
                  "filterType": "group",
                  "logicalOperator": "AND",
                  "conditions": []
                }
              },
              "config": {
                "promql_legend": "",
                "layer_type": "scatter",
                "weight_fixed": 1.0
              }
            }
          ],
          "layout": {
            "x": 24,
            "y": 9,
            "w": 24,
            "h": 9,
            "i": 4
          }
        }
      ]
    }
  ],
  "variables": {
    "list": []
  }
 }
@@ -0,0 +1,33 @@
 # 生产用 OpenObserve(单机)。相对本地版 docker-compose.yml 的区别:
 #   - 端口只绑 127.0.0.1 → 公网/外网都到不了(UI 访问走 SSH 隧道或 nginx 反代,见 README)
 #   - root 密码走环境变量(放同目录 .env,已 gitignore,勿提交)
 #   - 数据 bind-mount 到宿主 /data 分区(需预建目录 + 确认容器可写)+ CPU/内存上限(与 app/PG 共存防抢内存)
 #
 # 用法:
 #   1) 本目录建 .env(已 gitignore):
 #        OO_ROOT_PASSWORD=<强随机串>     # 生成: python -c "import secrets;print(secrets.token_urlsafe(24))"
 #   2) docker compose -f docker-compose.prod.yml up -d
 #   3) 开机自起: sudo systemctl enable docker
 services:
  openobserve:
    image: public.ecr.aws/zinclabs/openobserve:v0.91.2
    container_name: openobserve
    ports:
      - "127.0.0.1:5080:5080"          # 只绑本机，安全
    environment:
      ZO_ROOT_USER_EMAIL: "admin@shaguabijia.com"
      ZO_ROOT_USER_PASSWORD: "${OO_ROOT_PASSWORD:?请先在 deploy/openobserve/.env 里设 OO_ROOT_PASSWORD}"
      ZO_DATA_DIR: "/data"
      ZO_COMPACT_DATA_RETENTION_DAYS: "30"                # 超 30 天自动删,防爆盘(默认 3650 天=10年)
      ZO_TELEMETRY: "false"                               # 关匿名遥测(内网自用);变量名是 ZO_TELEMETRY,不是 *_ENABLED
    volumes:
      - /data/openobserve/data:/data   # 绑定挂载到宿主机的 /data/openobserve/data 目录（建议该目录所在分区有 20G+ 空间）
    restart: unless-stopped
    deploy:
      resources:
        limits:                        # 硬上限:防 OO 查询/ingest 抢爆 CPU/内存,拖垮同机 PG+app
          cpus: '2.0'                  
          memory: 3G                   
    logging:                                              # 容器 stdout 日志上限,防爆盘
      driver: json-file
      options: { max-size: "10m", max-file: "3" }
@@ -0,0 +1,16 @@
 # 本地开发用 OpenObserve(单容器 = local 模式)。用于接收 app-server 的接口指标(QPS/耗时/错误率)。
 # 启动: cd deploy/openobserve && docker compose up -d
 # Web UI: http://localhost:5080  (账号见下方 env)
 services:
  openobserve:
    image: public.ecr.aws/zinclabs/openobserve:latest
    container_name: openobserve
    ports:
      - "5080:5080"
    environment:
      ZO_ROOT_USER_EMAIL: "admin@shaguabijia.local"
      ZO_ROOT_USER_PASSWORD: "Complexpass#123"
      ZO_DATA_DIR: "/data"
    volumes:
      - ./data:/data
    restart: unless-stopped
@@ -115,6 +115,7 @@
 | 40a | `GET /api/v1/platform/flags` | 无 | [详情](./platform/platform-flags.md)（客户端运营 feature flag,比价/领券期广告开关等,拉取后缓存） |
 | 40b | `GET /api/v1/platform/ad-config` | 无 | [详情](./platform/platform-ad-config.md)（客户端拉广告配置:穿山甲 app_id+各位ID+各场景开关;不含验签密钥） |
 | 40c | `GET /api/v1/platform/app-version` | 无 | [详情](./platform/platform-app-version.md)（最新 App 版本,OTA 检查更新;与本机 versionCode 比） |
 | 40d | `GET /api/v1/platform/huawei-review` | 无 | [详情](./platform/platform-huawei-review.md)（华为审核开关:快速设置权限步能否被用户关闭;仅华为 ROM 客户端拉） |
 | **微信支付回调**（前缀 `/api/v1/wxpay`） |||
 | W1 | `POST /api/v1/wxpay/transfer-auth-notify` | 无 | 免确认收款授权结果通知（一期 stub:仅应答 200 不验签不改账,授权状态靠主动查询兜底)(无单独文档) |
 | **CPS 群发短链落地**（**无前缀**,挂域名根;公网不鉴权） |||
@@ -153,6 +154,7 @@
 | A12 | `GET /admin/api/ad-revenue-report` | admin | [详情](./admin/ad/admin-ad-revenue-report.md)(广告收益报表:分页/场景/`app_env` 筛 + **DAU/ARPU** #120;真实收益侧接穿山甲日表 #92) |
 | A13 | `GET / PATCH /admin/api/ad-config` | operator/finance | 广告配置(穿山甲 ID/验签密钥/各场景开关;C 端只读版见 40b)(无单独文档,见 `app/admin/routers/ad_config.py`) |
 | A14 | `GET /admin/api/config`、`PATCH /config/{key}` | operator/finance | 运营可配置项([app_config](../database/app_config.md):奖励常量/提现地板价等;#117 修系统配置下发)(无单独文档,见 `app/admin/routers/config.py`) |
 | A16 | `GET / PATCH /admin/api/huawei-review` | operator/tech | 华为审核开关(快速设置权限步能否被用户关闭,落 `app_config.huawei_review`;C 端只读版见 40d)(无单独文档,见 `app/admin/routers/huawei_review.py`) |
 | **A·管理员与角色**(super_admin):`GET`/`POST` `/admins`、`PATCH`/`DELETE` `/admins/{id}`(#126 删除+`pages_override`)、`GET`/`POST` `/roles`、`GET /roles/catalog`、`PATCH`/`DELETE` `/roles/{id}`(#117/#126 自定义角色) ||| [列表](./admin/admins/admin-admins-list.md) / [建](./admin/admins/admin-admin-create.md) / [改+删](./admin/admins/admin-admin-update.md) / [角色](./admin/admin-roles.md) |
 | A15 | `GET /admin/api/audit-logs` | admin | [详情](./admin/admin-audit-logs.md) |
 | **A·CPS 运营台**:群/活动 CRUD、`POST /referral-links`、`POST /orders/reconcile`(美团+京东 #90)、`GET /orders`、`/stats`、群 `timeseries`/`daily`/`wx-users`/`day-users`(#79) ||| [详情](./admin/admin-cps.md) |
@@ -0,0 +1,36 @@
 # GET /api/v1/platform/huawei-review — 华为审核开关
 > 所属:Platform 组（前缀 `/api/v1/platform`） | 鉴权:无 | [← 返回 API 索引](../README.md)
 华为应用市场审核要求:新手引导的**「快速设置」权限步必须可被用户关闭**（引导视频页不在要求内）。本端点把运营后台配的开关下发给客户端,决定该步左上角是否出现退出按钮。
 **不鉴权**:引导页在登录之前就展示,此时客户端必然没有 token。
 值来自 `app_config` 表的 `huawei_review` 行（admin 页 `GET / PATCH /admin/api/huawei-review` 可改），空库回退 `default`。
 ## 入参
 无。
 ## 出参
 响应 `200`:`HuaweiReviewOut`
 | 字段 | 类型 | 说明 |
 |---|---|---|
 | `mode` | string | `default`（强制展示,不可关闭 = 上线至今的现状）/ `review`（可关闭,过审用）。仅供排查时看后台切成了哪态 |
 | `onboarding_closable` | bool | 快速设置权限步是否允许用户退出（= `mode == "review"`）。**客户端只读这一个字段决策** |
 Mock 出参:
 ```json
 {
  "mode": "review",
  "onboarding_closable": true
 }
 ```
 ## 说明
 - **只有华为 ROM 客户端会来拉**（HarmonyOS / EMUI；荣耀 MagicOS 不拉）。机型判断在客户端做（`OemDetector`），服务端不看 UA，也就不用维护机型名单。
 - 客户端在进新手引导前拉一次并本地缓存；请求失败/超时用上次缓存值，从未拉到过则按 `onboarding_closable=false`（宁可不给退出按钮，也不误放开）。
 - 脏值兜底:DB 里 `mode` 不在枚举内时服务端一律回退 `default`。
 - 切回 `default` 即可一键收回退出按钮（审核通过后无需发版）。
@@ -29,3 +29,12 @@
 ## 注意
 - 不缓存:配置读频率低(每次福利操作读一次,主键查极快),admin 改了立即生效、跨进程一致(多 worker 也对)。
 - 新增可配项 = 在 `CONFIG_DEFS` 加一条 + 业务处改用 `app_config.get_value(db, key)` 读;不需要建迁移(行是动态插的,表结构不变)。
 ## 专用 key(借表不进 CONFIG_DEFS)
 有自己的语义与专用管理页的配置,复用本表但**不注册进 `CONFIG_DEFS`**——混进通用「系统配置」页只会显示成一个没头没尾的 on/off。它们各有一对 `get_*` / `set_*` 函数(仍在 `repositories/app_config.py`),`value` 存 dict,空行回退各自的模块内默认值。
 | key | 管理页 / admin 端点 | C 端读取 | 说明 |
 |---|---|---|---|
 | `ad_config` | `GET/PATCH /admin/api/ad-config` | `GET /api/v1/platform/ad-config`(去密钥) | 穿山甲 app_id / 各代码位 / 各场景开关 |
 | `app_version` | 内部写入(`X-Internal-Secret`) | `GET /api/v1/platform/app-version` | OTA 最新版本信息 |
 | `huawei_review` | `GET/PATCH /admin/api/huawei-review` | `GET /api/v1/platform/huawei-review` | 华为审核开关:`{"mode": "default"｜"review"}`,决定新手引导「快速设置」权限步能否被用户关闭。脏值/空行一律回退 `default`(不给退出按钮) |
@@ -0,0 +1,897 @@
 # 接口 QPS + 耗时可观测（OpenObserve）实现计划
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
 **Goal:** 给 app-server 每个接口采集 QPS + 耗时 + 错误率，经轻量 ASGI 中间件 + 后台 worker 批量直采到本地 Docker 的 OpenObserve。
 **Architecture:** 纯 ASGI 中间件测每请求耗时/抓路由模板+状态码 → 非阻塞入有界队列（满则丢、绝不阻塞）→ 后台 asyncio worker 批量 POST 到 OpenObserve `_json` ingest 端点。请求路径零 I/O；未配置观测则整套 no-op；上报失败丢批不重试。
 **Tech Stack:** FastAPI / Starlette ASGI 中间件、`asyncio.Queue`、`httpx.AsyncClient`（已有依赖）、pydantic-settings、OpenObserve（Docker）。
 参考 spec：[docs/superpowers/specs/2026-07-06-openobserve-api-metrics-design.md](2026-07-06-openobserve-api-metrics-design.md)
 ---
 ## 文件结构
 | 文件 | 职责 |
 |---|---|
 | `app/core/config.py`（改） | 新增 `OBSERVE_*` 配置 + `observe_configured` 门槛属性 |
 | `app/core/observe.py`（新） | 有界事件队列 + `record_event` + 路由模板解析 + `RequestMetricsMiddleware` |
 | `app/core/observe_worker.py`（新） | 后台批量上报 worker：`_collect_batch` / `_post_batch` / `start_*` / `stop_*` |
 | `app/main.py`（改） | 挂中间件（最外层）+ lifespan 启停 worker |
 | `.env.example`（改） | 新增 `OBSERVE_*` 注释段 |
 | `deploy/openobserve/docker-compose.yml`（新） | 本地 OpenObserve 容器 |
 | `deploy/openobserve/README.md`（新） | 部署步骤 + 查询/仪表盘 SQL |
 | `tests/test_observe.py`（新） | 配置门槛 / 队列 / 中间件 / worker 单测 |
 **关键接口契约（跨任务一致，勿改名）：**
 - `app.core.observe.get_queue() -> asyncio.Queue[dict]`
 - `app.core.observe.record_event(event: dict) -> None`
 - `app.core.observe.take_dropped() -> int`
 - `app.core.observe.RequestMetricsMiddleware`（ASGI class，`__init__(self, app)`）
 - 事件字段：`_timestamp`(µs int) / `service` / `env` / `method` / `route` / `status` / `duration_ms`(float)
 - `app.core.observe_worker.start_observe_worker() -> asyncio.Task | None`
 - `app.core.observe_worker.stop_observe_worker(task) -> None`
 - `settings.observe_configured -> bool`
 ---
 ## Task 1: 配置项 `OBSERVE_*` + `observe_configured`
 **Files:**
 - Modify: `app/core/config.py`（在 `cors_origins_list` property 之后、`is_prod` property 之前插入）
 - Test: `tests/test_observe.py`（新建）
 - [ ] **Step 1: 写失败测试**
 新建 `tests/test_observe.py`：
 ```python
 """接口指标可观测(observe)单测:配置门槛 / 队列 / 中间件 / worker。
 沿用仓库约定:TestClient + monkeypatch,绝不打真网络。observe 默认关(conftest 未设
 OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。
 """
 from __future__ import annotations
 from app.core.config import settings
 def test_observe_configured_requires_switch_and_creds(monkeypatch):
    # 开关开 + endpoint(默认 localhost)+ user + password 齐全 → True
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    assert settings.observe_configured is True
    # 缺密码 → False
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "")
    assert settings.observe_configured is False
    # 开关关 → False(即便凭证齐全)
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert settings.observe_configured is False
 ```
 - [ ] **Step 2: 运行,确认失败**
 Run: `pytest tests/test_observe.py::test_observe_configured_requires_switch_and_creds -q`
 Expected: FAIL —— `AttributeError`（`settings` 无 `OBSERVE_ENABLED` / 无 `observe_configured`）
 - [ ] **Step 3: 实现配置**
 在 `app/core/config.py` 的 `cors_origins_list` property 之后、`is_prod` property 之前插入：
 ```python
    # ===== 可观测(OpenObserve 接口指标)=====
    # 采集每个接口的 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker)。
    # 默认关(prod 安全):未开启 → 中间件透传、worker 不启动,整套 no-op。
    # 开启需 ENABLED=true 且 ENDPOINT/USER/PASSWORD 齐全(见 observe_configured)。
    OBSERVE_ENABLED: bool = False
    OBSERVE_ENDPOINT: str = "http://localhost:5080"   # OpenObserve base URL
    OBSERVE_ORG: str = "default"                        # 组织名
    OBSERVE_STREAM: str = "app_requests"               # stream 名(首次上报自动建)
    OBSERVE_USER: str = ""                             # Basic auth 邮箱
    OBSERVE_PASSWORD: str = ""                         # Basic auth 密码/token
    OBSERVE_FLUSH_INTERVAL_SEC: float = 5.0            # worker 最长攒批间隔
    OBSERVE_BATCH_MAX: int = 200                       # 单批最大事件数
    OBSERVE_QUEUE_MAX: int = 10000                     # 有界队列上限,满则丢
    OBSERVE_TIMEOUT_SEC: float = 5.0                   # 上报 HTTP 超时
    @property
    def observe_configured(self) -> bool:
        """观测上报可用 = 总开关开 且 endpoint/账号/密码齐全(缺则整套 no-op)。"""
        return bool(
            self.OBSERVE_ENABLED
            and self.OBSERVE_ENDPOINT
            and self.OBSERVE_USER
            and self.OBSERVE_PASSWORD
        )
 ```
 - [ ] **Step 4: 运行,确认通过**
 Run: `pytest tests/test_observe.py::test_observe_configured_requires_switch_and_creds -q`
 Expected: PASS
 - [ ] **Step 5: 提交**
 ```bash
 git add app/core/config.py tests/test_observe.py
 git commit -m "feat(observe): 加 OBSERVE_* 配置与 observe_configured 门槛"
 ```
 ---
 ## Task 2: 事件队列 + `record_event` + `take_dropped`
 **Files:**
 - Create: `app/core/observe.py`
 - Test: `tests/test_observe.py`（追加）
 - [ ] **Step 1: 写失败测试**
 在 `tests/test_observe.py` 顶部 import 区补 `import asyncio` 和 `from app.core import observe`，并追加：
 ```python
 def test_record_event_enqueues(monkeypatch):
    q = asyncio.Queue(maxsize=10)
    monkeypatch.setattr(observe, "_queue", q)
    observe.record_event({"route": "/x"})
    assert q.get_nowait() == {"route": "/x"}
 def test_record_event_drops_when_full(monkeypatch):
    q = asyncio.Queue(maxsize=1)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(observe, "_dropped", 0)
    observe.record_event({"n": 1})   # 占满
    observe.record_event({"n": 2})   # 满 → 丢弃当前,不抛异常
    assert observe.take_dropped() == 1
    assert observe.take_dropped() == 0   # 取出后清零
    assert q.get_nowait() == {"n": 1}    # 保留的是先到的
 ```
 - [ ] **Step 2: 运行,确认失败**
 Run: `pytest tests/test_observe.py -q -k record_event`
 Expected: FAIL —— `ModuleNotFoundError: app.core.observe` 或无 `record_event`
 - [ ] **Step 3: 实现 `app/core/observe.py`（先只放队列部分）**
 > 注意:本步只放队列相关代码。中间件用到的 `os`/`time`/`Match` 及 `_SKIP_PATHS`/`_UNMATCHED`/`_SERVICE` 常量放到 Task 3 一并加入——否则本步提交时 ruff 会报 F401 未用导入。
 新建 `app/core/observe.py`：
 ```python
 """接口指标埋点:有界事件队列 + 纯 ASGI 中间件。
 每个 HTTP 请求测总耗时、抓路由模板 + 状态码,非阻塞塞进有界队列;由 observe_worker
 后台批量上报到 OpenObserve。请求路径上无任何 I/O。未配置观测时中间件直接透传。
 """
 from __future__ import annotations
 import asyncio
 from app.core.config import settings
 # 有界事件队列(懒创建,见 get_queue):首次取用时在运行中的 loop 里建,避免 import 期
 # 无 loop 的边角问题;put_nowait/get_nowait 不需运行中的 loop → 可在无 loop 下测试。
 _queue: asyncio.Queue[dict] | None = None
 # 队列满时的丢弃计数,worker 定期取出打日志。
 _dropped = 0
 def get_queue() -> asyncio.Queue[dict]:
    """返回全局有界事件队列(懒创建)。测试可 monkeypatch 模块级 _queue 换成小队列。"""
    global _queue
    if _queue is None:
        _queue = asyncio.Queue(maxsize=settings.OBSERVE_QUEUE_MAX)
    return _queue
 def take_dropped() -> int:
    """取出并清零累计丢弃数(供 worker 打点)。"""
    global _dropped
    n, _dropped = _dropped, 0
    return n
 def record_event(event: dict) -> None:
    """非阻塞入队;队列满则丢弃当前事件并计数。永不抛异常、永不阻塞请求。"""
    global _dropped
    try:
        get_queue().put_nowait(event)
    except asyncio.QueueFull:
        _dropped += 1
 ```
 - [ ] **Step 4: 运行,确认通过**
 Run: `pytest tests/test_observe.py -q -k record_event`
 Expected: PASS
 - [ ] **Step 5: 提交**
 ```bash
 git add app/core/observe.py tests/test_observe.py
 git commit -m "feat(observe): 加有界事件队列与 record_event(满则丢)"
 ```
 ---
 ## Task 3: `RequestMetricsMiddleware`（路由模板 + 状态码 + 耗时）
 **Files:**
 - Modify: `app/core/observe.py`（追加 `_resolve_route` 和中间件 class）
 - Test: `tests/test_observe.py`（追加）
 - [ ] **Step 1: 写失败测试**
 在 `tests/test_observe.py` 顶部 import 区补：
 ```python
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 ```
 并追加：
 ```python
 def _make_probe_app() -> FastAPI:
    """独立最小 app:只挂中间件 + 两个无鉴权路由,不碰真业务 DB/auth。"""
    app = FastAPI()
    app.add_middleware(observe.RequestMetricsMiddleware)
    @app.get("/things/{tid}")
    def get_thing(tid: str):
        return {"tid": tid}
    @app.get("/health")
    def health():
        return {"ok": True}
    return app
@pytest.fixture
 def observe_on(monkeypatch):
    """开启观测 + 换一个干净小队列,返回该队列供断言。"""
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    return q
 def test_middleware_records_route_template(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/things/42")
    assert r.status_code == 200
    evt = observe_on.get_nowait()
    assert evt["route"] == "/things/{tid}"   # 模板,不是 /things/42
    assert evt["method"] == "GET"
    assert evt["status"] == 200
    assert evt["duration_ms"] >= 0
    assert evt["service"] and "env" in evt and isinstance(evt["_timestamp"], int)
 def test_middleware_skips_health(observe_on):
    client = TestClient(_make_probe_app())
    client.get("/health")
    assert observe_on.empty()
 def test_middleware_unmatched_route_is_normalized(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/definitely-not-a-route")
    assert r.status_code == 404
    evt = observe_on.get_nowait()
    assert evt["route"] == "__unmatched__"
    assert evt["status"] == 404
 def test_middleware_noop_when_disabled(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    client = TestClient(_make_probe_app())
    client.get("/things/1")
    assert q.empty()   # 未配置观测 → 零入队
 ```
 - [ ] **Step 2: 运行,确认失败**
 Run: `pytest tests/test_observe.py -q -k middleware`
 Expected: FAIL —— `AttributeError: module 'app.core.observe' has no attribute 'RequestMetricsMiddleware'`
 - [ ] **Step 3a: 给 `app/core/observe.py` 补中间件用的导入与常量**
 把顶部 import 段从
 ```python
 from __future__ import annotations
 import asyncio
 from app.core.config import settings
 ```
 改成
 ```python
 from __future__ import annotations
 import asyncio
 import os
 import time
 from starlette.routing import Match
 from app.core.config import settings
 # 不采集的路径(纯噪音):健康检查。
 _SKIP_PATHS = frozenset({"/health"})
 # 未匹配路由(404/扫描器)归一到此,防维度爆炸。
 _UNMATCHED = "__unmatched__"
 # service 字段:与 logging.py 同源(LOG_SERVICE_NAME),默认 app-server。
 _SERVICE = os.getenv("LOG_SERVICE_NAME", "app-server")
 ```
 （`_queue` / `_dropped` / `get_queue` / `take_dropped` / `record_event` 保持不动。）
 - [ ] **Step 3b: 实现中间件（追加到 `app/core/observe.py` 末尾）**
 ```python
 def _resolve_route(scope) -> str:
    """从 scope 取路由模板(如 /things/{tid})。优先 scope['route'](现代 Starlette
    路由后写入);取不到则手动匹配一次(老版本兜底);仍无 → __unmatched__(404/扫描器)。"""
    route = scope.get("route")
    path = getattr(route, "path", None)
    if path:
        return path
    app_ = scope.get("app")
    router = getattr(app_, "router", None)
    for candidate in getattr(router, "routes", []):
        try:
            match, _ = candidate.matches(scope)
        except Exception:  # noqa: BLE001 - 匹配兜底,任一路由异常不影响整体
            continue
        if match == Match.FULL and getattr(candidate, "path", None):
            return candidate.path
    return _UNMATCHED
 class RequestMetricsMiddleware:
    """纯 ASGI 中间件:测每个 http 请求耗时,记 method/route/status/duration。
    放在最外层(main.py 里 CORS 之后 add),测到含 CORS 的完整耗时。未配置观测 → 透传。
    """
    def __init__(self, app) -> None:
        self.app = app
    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http" or not settings.observe_configured:
            await self.app(scope, receive, send)
            return
        if scope.get("path") in _SKIP_PATHS:
            await self.app(scope, receive, send)
            return
        start = time.perf_counter()
        status_holder = {"status": 500}  # 下游异常未产出 response 时兜底 500
        async def send_wrapper(message) -> None:
            if message["type"] == "http.response.start":
                status_holder["status"] = message["status"]
            await send(message)
        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            duration_ms = (time.perf_counter() - start) * 1000.0
            record_event({
                "_timestamp": int(time.time() * 1_000_000),  # µs,OpenObserve 时间列
                "service": _SERVICE,
                "env": settings.APP_ENV,
                "method": scope.get("method", ""),
                "route": _resolve_route(scope),
                "status": status_holder["status"],
                "duration_ms": round(duration_ms, 3),
            })
 ```
 - [ ] **Step 4: 运行,确认通过**
 Run: `pytest tests/test_observe.py -q -k middleware`
 Expected: PASS（4 个中间件用例全过）
 > 若 `test_middleware_records_route_template` 拿到的是 `/things/42` 而非模板，说明该 Starlette 版本未在 `scope["route"]` 写模板——此时 `_resolve_route` 的手动匹配兜底应已生效并返回模板；若仍不对，检查兜底分支是否被 import 顺序影响。
 - [ ] **Step 5: 提交**
 ```bash
 git add app/core/observe.py tests/test_observe.py
 git commit -m "feat(observe): 加 RequestMetricsMiddleware(路由模板+状态码+耗时)"
 ```
 ---
 ## Task 4: 后台上报 worker
 **Files:**
 - Create: `app/core/observe_worker.py`
 - Test: `tests/test_observe.py`（追加）
 - [ ] **Step 1: 写失败测试**
 在 `tests/test_observe.py` 顶部 import 区补：
 ```python
 import httpx
 from app.core import observe_worker
 ```
 并追加：
 ```python
 async def test_collect_batch_drains_up_to_batch_max(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.1)
    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
    for i in range(3):
        q.put_nowait({"n": i})
    batch = await observe_worker._collect_batch()
    assert [e["n"] for e in batch] == [0, 1, 2]
 async def test_collect_batch_timeout_returns_empty(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.05)
    batch = await observe_worker._collect_batch()
    assert batch == []
 async def test_post_batch_hits_json_ingest_url(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ORG", "default")
    monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests")
    captured = {}
    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        captured["json"] = request.content
        return httpx.Response(200, json={"code": 200})
    client = httpx.AsyncClient(
        base_url="http://oo", transport=httpx.MockTransport(handler)
    )
    await observe_worker._post_batch(client, [{"route": "/x", "status": 200}])
    await client.aclose()
    assert captured["url"] == "http://oo/api/default/app_requests/_json"
    assert b"/x" in captured["json"]
 def test_start_observe_worker_noop_when_not_configured(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert observe_worker.start_observe_worker() is None
 ```
 - [ ] **Step 2: 运行,确认失败**
 Run: `pytest tests/test_observe.py -q -k "collect_batch or post_batch or start_observe"`
 Expected: FAIL —— `ModuleNotFoundError: app.core.observe_worker`
 - [ ] **Step 3: 实现 `app/core/observe_worker.py`**
 新建 `app/core/observe_worker.py`：
 ```python
 """接口指标后台上报 worker:批量 drain 事件队列 → POST 到 OpenObserve。
 对齐 heartbeat_monitor_worker 等的 start_*/stop_* 形态。best-effort 遥测:catch 全部
 异常,上报失败直接丢批不重试。未配置观测 → start 返回 None(不启动),整套 no-op。
 """
 from __future__ import annotations
 import asyncio
 import contextlib
 import logging
 import httpx
 from app.core.config import settings
 from app.core.observe import get_queue, take_dropped
 logger = logging.getLogger("shagua.observe")
 # 上报用的 httpx client,start 时建、stop 时关。
 _client: httpx.AsyncClient | None = None
 async def _collect_batch() -> list[dict]:
    """等到 ≥1 条(或到 flush 间隔)后,连抽到 BATCH_MAX 条或抽空。超时且空 → 返回 []。"""
    queue = get_queue()
    batch: list[dict] = []
    try:
        first = await asyncio.wait_for(
            queue.get(), timeout=settings.OBSERVE_FLUSH_INTERVAL_SEC
        )
    except asyncio.TimeoutError:
        return batch
    batch.append(first)
    while len(batch) < settings.OBSERVE_BATCH_MAX:
        try:
            batch.append(queue.get_nowait())
        except asyncio.QueueEmpty:
            break
    return batch
 async def _post_batch(client: httpx.AsyncClient, batch: list[dict]) -> None:
    """POST 一批事件到 OpenObserve 的 _json ingest 端点。非 2xx 仅告警。"""
    url = f"/api/{settings.OBSERVE_ORG}/{settings.OBSERVE_STREAM}/_json"
    resp = await client.post(url, json=batch)
    if resp.status_code >= 300:
        logger.warning(
            "observe ingest failed status=%s body=%s",
            resp.status_code,
            resp.text[:200],
        )
 async def _run_loop(client: httpx.AsyncClient) -> None:
    try:
        while True:
            batch = await _collect_batch()
            dropped = take_dropped()
            if dropped:
                logger.warning("observe dropped %d events (queue full)", dropped)
            if not batch:
                continue
            try:
                await _post_batch(client, batch)
            except Exception:  # noqa: BLE001 - best-effort 遥测,失败丢批不重试、不退出
                logger.warning(
                    "observe post batch failed, dropped %d events",
                    len(batch),
                    exc_info=True,
                )
    except asyncio.CancelledError:
        logger.info("observe worker stopped")
        raise
 def start_observe_worker() -> asyncio.Task | None:
    """启动上报 worker。未配置观测 → 返回 None(no-op)。"""
    global _client
    if not settings.observe_configured:
        return None
    _client = httpx.AsyncClient(
        base_url=settings.OBSERVE_ENDPOINT,
        auth=(settings.OBSERVE_USER, settings.OBSERVE_PASSWORD),
        timeout=settings.OBSERVE_TIMEOUT_SEC,
    )
    logger.info(
        "observe worker started endpoint=%s org=%s stream=%s",
        settings.OBSERVE_ENDPOINT,
        settings.OBSERVE_ORG,
        settings.OBSERVE_STREAM,
    )
    return asyncio.create_task(_run_loop(_client), name="observe-worker")
 async def stop_observe_worker(task: asyncio.Task | None) -> None:
    """收尾:cancel worker → best-effort 发最后一批 → 关 client。"""
    global _client
    if task is None:
        return
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task
    if _client is not None:
        # worker 已停,安全 drain 剩余并 best-effort 发最后一批(短超时,不卡关停)。
        try:
            queue = get_queue()
            final: list[dict] = []
            while len(final) < settings.OBSERVE_BATCH_MAX:
                try:
                    final.append(queue.get_nowait())
                except asyncio.QueueEmpty:
                    break
            if final:
                await asyncio.wait_for(
                    _post_batch(_client, final), timeout=settings.OBSERVE_TIMEOUT_SEC
                )
        except Exception:  # noqa: BLE001 - 关停期尽力而为,失败忽略
            pass
        await _client.aclose()
        _client = None
 ```
 - [ ] **Step 4: 运行,确认通过**
 Run: `pytest tests/test_observe.py -q -k "collect_batch or post_batch or start_observe"`
 Expected: PASS
 - [ ] **Step 5: 提交**
 ```bash
 git add app/core/observe_worker.py tests/test_observe.py
 git commit -m "feat(observe): 加后台批量上报 worker(失败丢批不重试)"
 ```
 ---
 ## Task 5: 接线到 `app/main.py`（挂中间件 + lifespan 启停）
 **Files:**
 - Modify: `app/main.py`（import 区、lifespan、CORS 之后）
 - Test: `tests/test_observe.py`（追加）
 - [ ] **Step 1: 写失败测试**
 在 `tests/test_observe.py` 追加：
 ```python
 def test_app_has_metrics_middleware():
    from app.main import app
    names = [m.cls.__name__ for m in app.user_middleware]
    assert "RequestMetricsMiddleware" in names
 ```
 - [ ] **Step 2: 运行,确认失败**
 Run: `pytest tests/test_observe.py::test_app_has_metrics_middleware -q`
 Expected: FAIL —— 断言失败（中间件尚未挂载）
 - [ ] **Step 3: 实现接线**
 3a. 在 `app/main.py` import 区（`withdraw_reconcile_worker` import 块之后）加：
 ```python
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
    start_observe_worker,
    stop_observe_worker,
 )
 ```
 3b. lifespan 里加启停（现有 `daily_exchange_task = start_daily_exchange_worker()` 之后、`try:` 之前加一行；`finally` 里在 `stop_daily_exchange_worker` 之后加一行）：
 ```python
    daily_exchange_task = start_daily_exchange_worker()
    observe_task = start_observe_worker()
    try:
        yield
    finally:
        await stop_heartbeat_monitor(heartbeat_task)
        await stop_withdraw_reconcile_worker(reconcile_task)
        await stop_daily_exchange_worker(daily_exchange_task)
        await stop_observe_worker(observe_task)
        await aclose_pricebot_client()
        logger.info("shutting down")
 ```
 3c. 挂中间件——在 CORS 的 `if settings.cors_origins_list:` 整块之后加（使其成为最外层，测到含 CORS 的完整耗时）：
 ```python
 # 接口指标埋点(最外层:测含 CORS 的完整耗时)。未配置观测时中间件自 no-op。
 app.add_middleware(RequestMetricsMiddleware)
 ```
 - [ ] **Step 4: 运行,确认通过**
 Run: `pytest tests/test_observe.py::test_app_has_metrics_middleware -q`
 Expected: PASS
 - [ ] **Step 5: 跑整套 observe 测试 + 全量回归,确认无破坏**
 Run: `pytest tests/test_observe.py -q && pytest -q`
 Expected: 全 PASS（现有用例不受影响：conftest 未设 `OBSERVE_*` → 观测关 → worker no-op、中间件透传）
 - [ ] **Step 6: 提交**
 ```bash
 git add app/main.py tests/test_observe.py
 git commit -m "feat(observe): main.py 挂中间件 + lifespan 启停上报 worker"
 ```
 ---
 ## Task 6: OpenObserve 本地部署（compose + README + .env.example）
 **Files:**
 - Create: `deploy/openobserve/docker-compose.yml`
 - Create: `deploy/openobserve/README.md`
 - Modify: `.env.example`（追加 `OBSERVE_*` 段）
 - [ ] **Step 1: 写 docker-compose**
 新建 `deploy/openobserve/docker-compose.yml`：
 ```yaml
 # 本地开发用 OpenObserve(单容器 = local 模式)。用于接收 app-server 的接口指标。
 # 启动: cd deploy/openobserve && docker compose up -d
 # Web UI: http://localhost:5080  (账号见下方 env)
 services:
  openobserve:
    image: public.ecr.aws/zinclabs/openobserve:latest
    container_name: openobserve
    ports:
      - "5080:5080"
    environment:
      ZO_ROOT_USER_EMAIL: "admin@shaguabijia.local"
      ZO_ROOT_USER_PASSWORD: "Complexpass#123"
      ZO_DATA_DIR: "/data"
    volumes:
      - ./data:/data
    restart: unless-stopped
 ```
 - [ ] **Step 2: 写 README**
 新建 `deploy/openobserve/README.md`：
 ````markdown
 # OpenObserve 本地部署（接口 QPS / 耗时可观测）
 app-server 通过中间件采集每个接口的 QPS + 耗时 + 错误率，批量上报到这里。
 ## 启动
 ```bash
 cd deploy/openobserve
 docker compose up -d
 ```
 - Web UI：http://localhost:5080
 - 登录：`admin@shaguabijia.local` / `Complexpass#123`（见 `docker-compose.yml`）
 - 数据落 `deploy/openobserve/data/`（已挂卷持久化；`data/` 建议 gitignore）
 ## 让 app-server 上报
 在项目根的 `.env` 打开观测（`OBSERVE_*`，账号密码与 compose 里 root 一致）：
 ```dotenv
 OBSERVE_ENABLED=true
 OBSERVE_ENDPOINT=http://localhost:5080
 OBSERVE_ORG=default
 OBSERVE_STREAM=app_requests
 OBSERVE_USER=admin@shaguabijia.local
 OBSERVE_PASSWORD=Complexpass#123
 ```
 重启 app-server，随便打几个接口。stream `app_requests` **首次上报自动创建**，
 在 UI 的 Logs → 选 `app_requests` 就能看到逐条请求事件。
 ## 查询（Logs 页 SQL，或建 Dashboard 面板）
 各接口 QPS（1 分钟分桶，面板里再除 60 得每秒）：
 ```sql
 SELECT route, histogram(_timestamp, '1 minute') AS ts, count(*) AS cnt
 FROM app_requests GROUP BY route, ts ORDER BY ts
 ```
 各接口 P95 耗时（毫秒）：
 ```sql
 SELECT route, approx_percentile_cont(duration_ms, 0.95) AS p95_ms
 FROM app_requests GROUP BY route ORDER BY p95_ms DESC
 ```
 各接口错误率（5xx 占比）：
 ```sql
 SELECT route,
       count(*) FILTER (WHERE status >= 500) * 100.0 / count(*) AS err_pct
 FROM app_requests GROUP BY route ORDER BY err_pct DESC
 ```
 ## 停止 / 清数据
 ```bash
 docker compose down            # 停止(保留数据)
 docker compose down -v && rm -rf data   # 停止并清空数据
 ```
 > 生产部署（持久化规格、独立 ingest 账号、鉴权收紧）见 spec 第 9 节，本期不做。
 ````
 - [ ] **Step 3: 追加 `.env.example`**
 在 `.env.example` 末尾追加：
 ```dotenv
 # ===== 可观测(OpenObserve 接口指标)=====
 # 采集每个接口 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker,见 deploy/openobserve/)。
 # 默认关;开启需 ENABLED=true 且填 USER/PASSWORD(与 docker-compose 里 root 账号一致)。
 OBSERVE_ENABLED=false
 OBSERVE_ENDPOINT=http://localhost:5080
 OBSERVE_ORG=default
 OBSERVE_STREAM=app_requests
 OBSERVE_USER=admin@shaguabijia.local
 OBSERVE_PASSWORD=Complexpass#123
 # 进阶(一般不用改):攒批间隔秒 / 单批最大条数 / 有界队列上限(满则丢) / 上报超时秒
 OBSERVE_FLUSH_INTERVAL_SEC=5
 OBSERVE_BATCH_MAX=200
 OBSERVE_QUEUE_MAX=10000
 OBSERVE_TIMEOUT_SEC=5
 ```
 - [ ] **Step 4: 校验 compose 语法（不需真拉镜像）**
 Run: `docker compose -f deploy/openobserve/docker-compose.yml config`
 Expected: 打印规整后的配置、无报错（若本机无 docker，可跳过，标注为手动验证项）
 - [ ] **Step 5: 提交**
 ```bash
 git add deploy/openobserve/docker-compose.yml deploy/openobserve/README.md .env.example
 git commit -m "feat(observe): 加 OpenObserve 本地 compose + README + .env.example"
 ```
 ---
 ## Task 7: 端到端手动验证 + 全量 lint/test 收尾
 **Files:** 无（验证 + 收尾）
 - [ ] **Step 1: 起 OpenObserve**
 Run: `cd deploy/openobserve && docker compose up -d`
 Expected: 容器起来，浏览器打开 http://localhost:5080 能登录
 - [ ] **Step 2: 本地开观测起 app-server**
 在根 `.env` 设 `OBSERVE_ENABLED=true` + `OBSERVE_USER/PASSWORD`（同 compose），然后：
 Run: `./run.sh`（Windows 用 `python -m uvicorn app.main:app --port 8770 --reload --reload-dir app`）
 Expected: 启动日志出现 `observe worker started endpoint=http://localhost:5080 ...`
 - [ ] **Step 3: 打几个接口产生数据**
 Run: `curl http://localhost:8770/health && curl http://localhost:8770/things-does-not-exist -i`（或正常业务接口若干）
 Expected: 稍等 ≤5s（flush 间隔），OpenObserve UI 的 Logs → `app_requests` 出现事件；`/health` 不应出现；不存在的路径 route 为 `__unmatched__`
 - [ ] **Step 4: 验证三条查询**
 在 OpenObserve UI 分别粘贴 README 里的 QPS / P95 / 错误率 SQL，确认能出数。
 - [ ] **Step 5: lint(仅本改动涉及文件)+ 全量测试**
 > 说明:仓库基线有 ~558 个既有 ruff 错误、且未强制 ruff 通过。不要去清历史欠债(范围蔓延)。只要求**本次新增/改动的文件**零 ruff 错误。
 Run: `ruff check app/core/observe.py app/core/observe_worker.py tests/test_observe.py && python -m pytest -q`
 Expected: 上述三个新文件 ruff 无错;测试里 `tests/test_observe.py` 全 PASS,且**全量失败数不超过基线的 4 个**(test_compare_proxy ×2 / test_coupon_proxy ×1 / test_invite ×1,均与本功能无关)。
 额外确认我对既有文件的改动没有引入**新的** ruff 错误:`ruff check app/core/config.py app/main.py`(数量应与基线一致,不因本改动增加)。
 - [ ] **Step 6: 关观测复跑一次,确认降级**
 把 `.env` 的 `OBSERVE_ENABLED` 改回 `false`，`ruff check .` 不涉及，直接 `pytest -q`
 Expected: 全 PASS（验证 observe 关闭时零副作用）
 - [ ] **Step 7: 收尾提交（如有 .env 之外的改动）**
 ```bash
 git add -A
 git commit -m "chore(observe): 端到端验证与收尾" --allow-empty
 ```
 > `.env` 不入 git（已 gitignore）；本任务只验证，不提交 `.env`。
 ---
 ## Self-Review（写完计划后自查）
 - **Spec 覆盖**：Docker 部署→Task 6/7；事件 schema→Task 3（`record_event` 事件字段）；中间件→Task 3；worker→Task 4；配置→Task 1；main 接线→Task 5；查询/仪表盘→Task 6 README；测试→Task 1-5；决策(a)队列满丢→Task 2；(b)失败不重试→Task 4；(c)跳过 /health→Task 3。全覆盖。
 - **占位符**：无 TBD/TODO；每个代码步骤含完整代码。
 - **类型/命名一致**：`get_queue` / `record_event` / `take_dropped` / `RequestMetricsMiddleware` / `start_observe_worker` / `stop_observe_worker` / `observe_configured` / 事件字段名，跨 Task 1-5 与文件结构表一致。
@@ -0,0 +1,236 @@
 # 接口 QPS + 耗时可观测（OpenObserve）设计
 - **日期**：2026-07-06
 - **状态**：已评审通过，待写实现计划
 - **范围**：仅 app-server（8770）；admin（8771）暂不接入
 - **方案**：A —— 轻量自研 ASGI 中间件 + 后台 worker 批量直采到 OpenObserve
 ## 1. 背景与目标
 app-server 目前除 CORS 外无任何中间件，也无接口级可观测。需要按**每个接口**采集：
 - **QPS**（每秒请求数，可按接口/时间分桶）
 - **耗时**（P50/P95/P99 等分位）
 顺带低成本拿到**错误率**（`status >= 500` 占比）。落地目标是：本地 Docker 跑一个 OpenObserve 实例接收数据，服务侧加埋点上报，在 OpenObserve 仪表盘上看各接口 QPS + 耗时。
 ### 非目标（YAGNI）
 - 不做分布式 trace / span 关联（只要接口聚合指标）。
 - 不引入 OpenTelemetry / Prometheus 客户端等重依赖。
 - 不采集请求体 / query / 用户身份等，任何 PII 都不进上报。
 - admin（8771）本期不接（中间件写成可复用，未来一行挂载即可）。
 - 上报失败不做持久化重试 / 落盘补偿（best-effort）。
 ## 2. 方案选型
 对比过三条路（详见评审记录）：
 - **A 轻量自研中间件 + JSON 直采**（选中）：零新依赖（`httpx` 已在依赖里），完全贴合本仓库「后台 worker + JSON 事件 + `*_configured` 优雅降级」的既有习惯，恰好满足「每接口 QPS + 耗时 + 错误率」并保留原始事件下钻能力。
 - B OpenTelemetry 自动埋点 + OTLP：行业标准、顺带 trace，但多 5–6 个依赖、概念多、数据量/成本高于需求，与精简代码库风格相悖。
 - C Prometheus 进程内聚合 + remote_write/抓取：数据量最小，但 remote_write 编码复杂或需额外抓取进程，丢失单请求下钻，最不贴合 OpenObserve 的 log-first 强项。
 **结论：A。**
 ## 3. 架构与数据流
 ```
 每个 HTTP 请求
  → RequestMetricsMiddleware（最外层：测总耗时 / 抓路由模板 + 状态码）
  → record_event()  非阻塞入队（有界队列，满则丢最旧，绝不阻塞、绝不 OOM）
  → observe_worker（后台 asyncio.Task，随 lifespan 启停）批量 drain
  → httpx POST  {ENDPOINT}/api/{ORG}/{STREAM}/_json  → OpenObserve
                                                        → 仪表盘 SQL 聚合出 QPS / 分位耗时 / 错误率
 ```
 **核心不变量**：
 1. 请求路径上只做「测时 + 构建一个小 dict + `put_nowait`」，**无任何网络/磁盘 I/O**。
 2. 所有上报 I/O 在后台 worker；worker 捕获全部异常，绝不让埋点影响请求。
 3. 未配置观测（`observe_configured=False`）→ 中间件透传、worker 不启动，整套 no-op。
 4. OpenObserve 不可用 → 队列填满后丢弃事件 + 限流告警，业务零影响。
 ## 4. 组件设计
 ### 4.1 OpenObserve 本地部署 —— `deploy/openobserve/docker-compose.yml`（新增）
 ```yaml
 services:
  openobserve:
    image: public.ecr.aws/zinclabs/openobserve:latest
    container_name: openobserve
    ports: ["5080:5080"]
    environment:
      ZO_ROOT_USER_EMAIL: "admin@shaguabijia.local"
      ZO_ROOT_USER_PASSWORD: "Complexpass#123"
      ZO_DATA_DIR: "/data"
    volumes: ["./data:/data"]
    restart: unless-stopped
 ```
 - `docker compose up -d` 启动；Web UI `http://localhost:5080`，用上面邮箱/密码登录。
 - 单容器 = local 模式，数据落 `./data`（已挂卷持久化）。
 - **stream 首次上报自动创建**，无需预建 `app_requests`。
 - 上报鉴权：HTTP Basic auth（`email:password`），本地直接用 root 账号；生产应另建仅具 ingest 权限的用户/服务账号（本期不涉及）。
 ### 4.2 事件 schema（一请求一行 JSON）
 ```json
 {
  "_timestamp": 1720000000000000,        // 微秒(µs)整数，请求完成时刻。OpenObserve 默认时间列 _timestamp 以微秒计
  "service": "app-server",               // 取 LOG_SERVICE_NAME / 固定值
  "env": "dev",                          // settings.APP_ENV
  "method": "POST",
  "route": "/api/v1/coupon/step",        // 路由模板（非实际 path）
  "status": 200,
  "duration_ms": 42.7                    // float 毫秒
 }
 ```
 - **只存路由模板**（如 `/c/{code}`、`/media` 静态归一），避免 path 参数把维度打爆。
 - 未匹配路由（404 / 扫描器）归一到常量 `__unmatched__`。
 - 只采 method / route / status / duration —— 无 body、无 query、无 PII。
 ### 4.3 埋点中间件 —— `app/core/observe.py`（新增）
 **纯 ASGI 中间件**（比 `BaseHTTPMiddleware` 开销低；能可靠读到路由与最终状态码；scope 按引用透传，内层 router 的 `scope["route"]` 外层可见）。
 职责：
 1. 非 `http` 请求、或 `not settings.observe_configured` → 直接透传，不测。
 2. `perf_counter()` 记起点；包一层 `send` 抓 `http.response.start` 的 `status`（默认兜底 500，覆盖下游抛异常未产出 response 的情况）。
 3. `finally` 里算 `duration_ms`，从 `scope` 取路由模板（见下），构建事件，调 `record_event()`。
 4. 跳过路径集合 `_SKIP_PATHS = {"/health"}`（纯噪音）。
 **路由模板解析（跨 Starlette 版本稳健）**：
 ```python
 route = scope.get("route")
 template = getattr(route, "path", None)
 if template is None:            # 未匹配 / 老版本未写 scope["route"]
    template = "__unmatched__"
 ```
 （若实测某 Starlette 版本不写 `scope["route"]`，回退用 `request.app.router.routes` 逐个 `route.matches(scope)==Match.FULL` 找模板；实现时以实际版本为准，优先 `scope["route"]`。）
 **入队（`record_event`）**：模块级 `asyncio.Queue(maxsize=OBSERVE_QUEUE_MAX)`。用 `put_nowait`，`QueueFull` 则丢弃并累加一个 `_dropped` 计数（每累计 N 条限流打一条 WARNING）。**永不 `await put()`、永不阻塞请求**。
 > 决策(a)：队列满 → **丢弃**（不阻塞请求）。
 ### 4.4 上报 worker —— `app/core/observe_worker.py`（新增）
 对齐现有 `heartbeat_monitor_worker.py` / `daily_exchange_worker.py` / `withdraw_reconcile_worker.py` 的 `start_*` / `stop_*` 形态。
 - `start_observe_worker() -> asyncio.Task | None`
  - `not settings.observe_configured` → 返回 `None`（no-op）。
  - 否则建专用 `httpx.AsyncClient`（`base_url=ENDPOINT`，`auth=(USER, PASSWORD)`，`timeout=OBSERVE_TIMEOUT_SEC`），起 `_run_loop` task。
 - `_run_loop()`：循环
  1. `_collect_batch()`：`await asyncio.wait_for(queue.get(), timeout=FLUSH_INTERVAL)` 拿到首条（超时且空 → 返回空，continue）；再 `get_nowait()` 连抽到 `BATCH_MAX` 条或抽空。
  2. `POST /api/{ORG}/{STREAM}/_json`，body 为事件数组。
  3. **catch 所有异常**：失败限流打 WARNING，**直接丢弃该批，不重试**。
 - `stop_observe_worker(task)`：best-effort 收尾 flush（短超时）→ `task.cancel()` → `await`（吞 `CancelledError`）→ 关 client。
 > 决策(b)：上报失败 → **直接丢弃，不重试**（best-effort 遥测）。
 ### 4.5 配置 —— `app/core/config.py`（改）
 新增一段 `# ===== 可观测（OpenObserve 接口指标）=====`，默认全关（prod 安全）：
 | 配置 | 默认 | 说明 |
 |---|---|---|
 | `OBSERVE_ENABLED` | `False` | 总开关；默认关，opt-in |
 | `OBSERVE_ENDPOINT` | `http://localhost:5080` | OpenObserve base URL |
 | `OBSERVE_ORG` | `default` | 组织名 |
 | `OBSERVE_STREAM` | `app_requests` | stream 名 |
 | `OBSERVE_USER` | `""` | Basic auth 邮箱 |
 | `OBSERVE_PASSWORD` | `""` | Basic auth 密码/token |
 | `OBSERVE_FLUSH_INTERVAL_SEC` | `5.0` | worker 最长攒批间隔 |
 | `OBSERVE_BATCH_MAX` | `200` | 单批最大事件数 |
 | `OBSERVE_QUEUE_MAX` | `10000` | 有界队列上限，满则丢 |
 | `OBSERVE_TIMEOUT_SEC` | `5.0` | 上报 HTTP 超时 |
 ```python
@property
 def observe_configured(self) -> bool:
    return bool(self.OBSERVE_ENABLED and self.OBSERVE_ENDPOINT
                and self.OBSERVE_USER and self.OBSERVE_PASSWORD)
 ```
 `.env.example` 同步补一段带注释的 `OBSERVE_*`（沿用该文件重注释风格），`OBSERVE_ENABLED=false`。
 ### 4.6 接线 —— `app/main.py`（改）
 - import `RequestMetricsMiddleware`、`start_observe_worker` / `stop_observe_worker`。
 - `app.add_middleware(RequestMetricsMiddleware)`：放在 CORS `add_middleware` **之后** → 成为最外层，测到含 CORS 的完整耗时。无条件挂载（内部自 no-op）。
 - `lifespan`：启动 `observe_task = start_observe_worker()`；`finally` 里 `await stop_observe_worker(observe_task)`，与现有 worker 并列。
 ### 4.7 OpenObserve 查询 / 仪表盘 —— `deploy/openobserve/README.md`（新增）
 含：compose 启停、登录、stream 自动创建说明、`.env` 接线，以及可直接粘的示例 SQL：
 - **各接口 QPS**（1 分钟分桶）：
  ```sql
  SELECT route, histogram(_timestamp, '1 minute') AS ts, count(*) AS cnt
  FROM app_requests GROUP BY route, ts ORDER BY ts
  ```
  （面板按 `cnt/60` 展示每秒；或用 OpenObserve 图表的 rate 能力。）
 - **各接口 P95 耗时**：
  ```sql
  SELECT route, approx_percentile_cont(duration_ms, 0.95) AS p95_ms
  FROM app_requests GROUP BY route ORDER BY p95_ms DESC
  ```
 - **各接口错误率**：
  ```sql
  SELECT route,
         count(*) FILTER (WHERE status >= 500) * 100.0 / count(*) AS err_pct
  FROM app_requests GROUP BY route ORDER BY err_pct DESC
  ```
 ## 5. 关键设计决策汇总
 - **(a) 队列满 → 丢弃**（不阻塞请求）：遥测让路于业务可用性。
 - **(b) 上报失败 → 不重试**：best-effort；避免 poison batch 堆积与队列无限增长。
 - **(c) 跳过 `/health`**：健康检查是纯噪音，硬编码在 `_SKIP_PATHS`。
 - **只存路由模板 + `__unmatched__`**：防维度爆炸。
 - **默认 OFF、opt-in**：prod 安全默认；开启后仍全异步 + 有界。
 - **纯 ASGI 中间件 + `perf_counter`**：请求路径开销微秒级，无 I/O。
 ## 6. 安全 / 性能保证
 - 请求路径新增开销 ≈ 一次 `perf_counter` 差 + 一个小 dict + 一次 `put_nowait`（微秒级），无锁竞争的显著热点。
 - 失败隔离：入队丢弃 + worker 全异常捕获；OpenObserve 宕机不影响任何请求。
 - 有界内存：队列 `maxsize` 封顶，最坏丢事件不涨内存。
 - 无 PII：仅 method / route / status / duration。
 ## 7. 测试策略 —— `tests/test_observe.py`（新增）
 沿用仓库约定（`TestClient` + `monkeypatch`，绝不打真网络；`conftest` 在 import 前设 env）：
 1. 埋点入队字段正确：模板路由、`status`、`duration_ms > 0`。
 2. 参数化路由 → 取到**模板**而非实际 path。
 3. 未匹配路径（404）→ `route == "__unmatched__"`。
 4. `OBSERVE_ENABLED=false` → 零入队、零 HTTP（现有测试不受影响）。
 5. 队列满 → `record_event` 不抛异常（走丢弃分支）。
 6. worker 批量 POST 的 URL / payload 正确（monkeypatch httpx client / `_post`，不打网络）。
 7. `/health` 被跳过 → 不入队。
 > `settings` 是 `lru_cache` 单例；需要开启观测的用例通过 monkeypatch `settings` 属性或直接调 `record_event` / 中间件并 patch `observe_configured` 实现，避免全局 env 改动波及他用例。
 ## 8. 文件清单
 | 文件 | 动作 |
 |---|---|
 | `deploy/openobserve/docker-compose.yml` | 新增（OpenObserve 容器）|
 | `deploy/openobserve/README.md` | 新增（部署步骤 + 查询/仪表盘）|
 | `app/core/observe.py` | 新增（中间件 + 有界队列 + `record_event` + 路由解析）|
 | `app/core/observe_worker.py` | 新增（后台批量上报 worker）|
 | `app/core/config.py` | 改（`OBSERVE_*` + `observe_configured`）|
 | `app/main.py` | 改（挂中间件 + lifespan 启停 worker）|
 | `.env.example` | 改（新增 `OBSERVE_*` 注释段）|
 | `tests/test_observe.py` | 新增 |
 ## 9. 未来工作（本期不做）
 - admin（8771）接入同一套中间件（`service` 字段区分）。
 - 生产部署 OpenObserve（持久化、独立 ingest 账号、资源规格、鉴权收紧）。
 - 上报字段扩展（如按 user/设备维度、上游 pricebot 透传耗时拆分）。
@@ -1,12 +1,13 @@
 """ad_ecpm_record.trace_id 落库 + 按 trace 聚合广告收益(元)。"""
 from __future__ import annotations
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 from sqlalchemy import delete
 from app.db.session import SessionLocal
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.coupon_state import CouponSession
 from app.repositories import ad_ecpm as crud_ecpm
@@ -57,6 +58,39 @@ def test_revenue_yuan_by_trace_empty() -> None:
        db.close()
 def test_terminal_coupon_trace_is_not_attributable_to_late_impression() -> None:
    """领券失败或被放弃后才到达的广告展示保留收益记录，但不再关联死亡 trace。"""
    db = SessionLocal()
    try:
        db.add_all([
            CouponSession(
                trace_id="failed-before-ad", device_id="d-late-ad", status="failed", app_env="prod",
                started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
            ),
            CouponSession(
                trace_id="abandoned-before-ad", device_id="d-late-ad", status="abandoned", app_env="prod",
                started_at=datetime(2020, 1, 2, tzinfo=UTC), started_date=date(2020, 1, 2),
            ),
        ])
        db.flush()
        assert crud_ecpm.attributable_trace_id(
            db, feed_scene="coupon", trace_id="failed-before-ad"
        ) is None
        assert crud_ecpm.attributable_trace_id(
            db, feed_scene="coupon", trace_id="abandoned-before-ad"
        ) is None
        assert crud_ecpm.attributable_trace_id(
            db, feed_scene="comparison", trace_id="failed-before-ad"
        ) == "failed-before-ad"
        assert crud_ecpm.attributable_trace_id(
            db, feed_scene="coupon", trace_id="unknown-trace"
        ) == "unknown-trace"
    finally:
        db.rollback()
        db.close()
 def test_create_ecpm_record_persists_trace_id() -> None:
    """create_ecpm_record 落 trace_id。"""
    db = SessionLocal()
@@ -1,12 +1,15 @@
 """Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
 from __future__ import annotations
 from datetime import datetime
 import pytest
 from fastapi.testclient import TestClient
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
 from app.models.wallet import CashTransaction, WithdrawOrder
 from app.repositories import user as user_repo
@@ -69,6 +72,49 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
    assert "jd_order_count" in data["cps"]
 def test_dashboard_period_comparison_is_aggregated_by_backend(
    admin_client: TestClient, admin_token: str
 ) -> None:
    created_at = datetime(2037, 1, 15, 12)
    rows = [
        ("dashboard-aggregate-success", "success", 101, 0.1),
        ("dashboard-aggregate-failed", "failed", 200, 0.2),
        ("dashboard-aggregate-cancelled", "cancelled", 300, 0.3),
        ("dashboard-aggregate-running", "running", 400, 0.4),
    ]
    db = SessionLocal()
    try:
        for trace_id, status, total_ms, llm_cost_yuan in rows:
            db.add(
                ComparisonRecord(
                    trace_id=trace_id,
                    status=status,
                    total_ms=total_ms,
                    llm_cost_yuan=llm_cost_yuan,
                    created_at=created_at,
                )
            )
        db.commit()
    finally:
        db.close()
    response = admin_client.get(
        "/admin/api/stats/overview",
        params={"date_from": "2037-01-15", "date_to": "2037-01-15"},
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    comparison = response.json()["period"]["comparison"]
    assert comparison["total"] == 4
    assert comparison["completed"] == 2
    assert comparison["cancelled"] == 1
    assert comparison["success"] == 1
    assert comparison["success_rate"] == 0.3333
    assert comparison["median_duration_ms"] == 151
    assert comparison["p95_duration_ms"] == 195
    assert comparison["token_cost_total_yuan"] == pytest.approx(1.0)
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
    uid = _seed_user_with_data("13800000002")
    r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
@@ -123,7 +123,8 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
    # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
    assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
    assert set(roles["tech"]["pages"]) == {
-        "dashboard", "device-liveness", "config", "ad-revenue", "event-logs", "audit-logs",
+        "dashboard", "device-liveness", "config", "ad-revenue", "huawei-review",
        "event-logs", "audit-logs",
    }
@@ -0,0 +1,130 @@
 """华为审核开关:admin 读写 + 客户端公开端点 + 审计 + 空库回退。
 背景:华为应用市场审核要求新手引导的「快速设置」权限步必须可被用户关闭。开关切到 review 后
 客户端(仅华为 ROM)在该步左上角显示退出按钮。默认 default = 上线至今的现状(不可关闭)。
 autouse 清理每个用例后清空 app_config,避免污染其他文件里假设默认值的用例(同 test_admin_config)。
 """
 from __future__ import annotations
 from collections.abc import Iterator
 import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy import delete, select
 from app.admin.main import admin_app
 from app.admin.repositories import admin_user as admin_repo
 from app.db.session import SessionLocal
 from app.models.admin import AdminAuditLog
 from app.models.app_config import AppConfig
@pytest.fixture()
 def admin_client() -> TestClient:
    return TestClient(admin_app)
@pytest.fixture()
 def token() -> str:
    db = SessionLocal()
    try:
        if admin_repo.get_by_username(db, "hw_admin") is None:
            admin_repo.create_admin(
                db, username="hw_admin", password="hwpass123", role="super_admin"
            )
    finally:
        db.close()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": "hw_admin", "password": "hwpass123"}
    ).json()["access_token"]
@pytest.fixture(autouse=True)
 def _clean_config() -> Iterator[None]:
    yield
    db = SessionLocal()
    try:
        db.execute(delete(AppConfig))
        # 审计行同样要清:同库跨用例累积会让 test_switch_writes_audit 数到前面用例写的行
        # (按 action 限定,不碰其他模块可能已写入的审计)。
        db.execute(delete(AdminAuditLog).where(AdminAuditLog.action == "huawei_review.set"))
        db.commit()
    finally:
        db.close()
 def _auth(t: str) -> dict:
    return {"Authorization": f"Bearer {t}"}
 def test_public_default_not_closable(client: TestClient) -> None:
    """空库(从未切过)→ 客户端拿到 default / 不可关闭 = 维持现状;且不需要鉴权。"""
    r = client.get("/api/v1/platform/huawei-review")
    assert r.status_code == 200, r.text
    assert r.json() == {"mode": "default", "onboarding_closable": False}
 def test_admin_get_default(admin_client: TestClient, token: str) -> None:
    r = admin_client.get("/admin/api/huawei-review", headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["mode"] == "default"
    assert body["updated_at"] is None  # 从未切过
 def test_switch_to_review_takes_effect(
    admin_client: TestClient, client: TestClient, token: str
 ) -> None:
    """admin 切 review → 公开端点立刻下发可关闭(跨进程/跨 app 一致,因为落在 DB 而非内存)。"""
    r = admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}, headers=_auth(token)
    )
    assert r.status_code == 200, r.text
    assert r.json()["mode"] == "review"
    assert r.json()["updated_at"] is not None
    pub = client.get("/api/v1/platform/huawei-review").json()
    assert pub == {"mode": "review", "onboarding_closable": True}
    # 切回 default → 客户端恢复不可关闭(审核过了要能一键收回)
    admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "default"}, headers=_auth(token)
    )
    assert client.get("/api/v1/platform/huawei-review").json()["onboarding_closable"] is False
 def test_switch_writes_audit(admin_client: TestClient, token: str) -> None:
    admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}, headers=_auth(token)
    )
    db = SessionLocal()
    try:
        logs = db.execute(
            select(AdminAuditLog).where(AdminAuditLog.action == "huawei_review.set")
        ).scalars().all()
        assert len(logs) == 1
        assert logs[0].detail == {"before": "default", "after": "review"}
    finally:
        db.close()
 def test_invalid_mode_rejected(admin_client: TestClient, token: str) -> None:
    """mode 是 Literal,非法值由 FastAPI 校验挡在 422(不会落库)。"""
    r = admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "nope"}, headers=_auth(token)
    )
    assert r.status_code == 422, r.text
    db = SessionLocal()
    try:
        assert db.get(AppConfig, "huawei_review") is None
    finally:
        db.close()
 def test_requires_admin_auth(admin_client: TestClient) -> None:
    assert admin_client.get("/admin/api/huawei-review").status_code == 401
    assert admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}
    ).status_code == 401
@@ -0,0 +1,234 @@
 """接口指标可观测(observe)单测:配置门槛 / 队列 / 中间件 / worker。
 沿用仓库约定:TestClient + monkeypatch,绝不打真网络。observe 默认关(conftest 未设
 OBSERVE_*),需要开启的用例用 monkeypatch 改 settings 单例属性。
 """
 from __future__ import annotations
 import asyncio
 import httpx
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from app.core import observe, observe_worker
 from app.core.config import settings
 def test_observe_configured_requires_switch_and_creds(monkeypatch):
    # 开关开 + endpoint(默认 localhost)+ user + password 齐全 → True
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    assert settings.observe_configured is True
    # 缺密码 → False
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "")
    assert settings.observe_configured is False
    # 缺用户名 → False
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    monkeypatch.setattr(settings, "OBSERVE_USER", "")
    assert settings.observe_configured is False
    # 开关关 → False(即便凭证齐全)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert settings.observe_configured is False
 def test_record_event_enqueues(monkeypatch):
    q = asyncio.Queue(maxsize=10)
    monkeypatch.setattr(observe, "_queue", q)
    observe.record_event({"route": "/x"})
    assert q.get_nowait() == {"route": "/x"}
 def test_record_event_drops_when_full(monkeypatch):
    q = asyncio.Queue(maxsize=1)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(observe, "_dropped", 0)
    observe.record_event({"n": 1})   # 占满
    observe.record_event({"n": 2})   # 满 → 丢弃当前,不抛异常
    assert observe.take_dropped() == 1
    assert observe.take_dropped() == 0   # 取出后清零
    assert q.get_nowait() == {"n": 1}    # 保留的是先到的
 def _make_probe_app() -> FastAPI:
    """独立最小 app:只挂中间件 + 两个无鉴权路由,不碰真业务 DB/auth。"""
    app = FastAPI()
    app.add_middleware(observe.RequestMetricsMiddleware)
    @app.get("/things/{tid}")
    def get_thing(tid: str):
        return {"tid": tid}
    @app.get("/health")
    def health():
        return {"ok": True}
    return app
@pytest.fixture
 def observe_on(monkeypatch):
    """开启观测 + 换一个干净小队列,返回该队列供断言。"""
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", True)
    monkeypatch.setattr(settings, "OBSERVE_USER", "u")
    monkeypatch.setattr(settings, "OBSERVE_PASSWORD", "p")
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    return q
 def test_middleware_records_route_template(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/things/42")
    assert r.status_code == 200
    evt = observe_on.get_nowait()
    assert evt["route"] == "/things/{tid}"   # 模板,不是 /things/42
    assert evt["method"] == "GET"
    assert evt["status"] == 200
    assert evt["duration_ms"] >= 0
    assert evt["service"] and "env" in evt and isinstance(evt["_timestamp"], int)
 def test_middleware_skips_health(observe_on):
    client = TestClient(_make_probe_app())
    client.get("/health")
    assert observe_on.empty()
 def test_middleware_unmatched_route_is_normalized(observe_on):
    client = TestClient(_make_probe_app())
    r = client.get("/definitely-not-a-route")
    assert r.status_code == 404
    evt = observe_on.get_nowait()
    assert evt["route"] == "__unmatched__"
    assert evt["status"] == 404
 def test_middleware_noop_when_disabled(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    client = TestClient(_make_probe_app())
    client.get("/things/1")
    assert q.empty()   # 未配置观测 → 零入队
 async def test_collect_batch_drains_up_to_batch_max(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.1)
    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
    for i in range(3):
        q.put_nowait({"n": i})
    batch = await observe_worker._collect_batch()
    assert [e["n"] for e in batch] == [0, 1, 2]
 async def test_collect_batch_timeout_returns_empty(monkeypatch):
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.05)
    batch = await observe_worker._collect_batch()
    assert batch == []
 async def test_post_batch_hits_json_ingest_url(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ORG", "default")
    monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests")
    captured = {}
    def handler(request: httpx.Request) -> httpx.Response:
        captured["url"] = str(request.url)
        captured["json"] = request.content
        return httpx.Response(200, json={"code": 200})
    client = httpx.AsyncClient(
        base_url="http://oo", transport=httpx.MockTransport(handler)
    )
    await observe_worker._post_batch(client, [{"route": "/x", "status": 200}])
    await client.aclose()
    assert captured["url"] == "http://oo/api/default/app_requests/_json"
    assert b"/x" in captured["json"]
 def test_start_observe_worker_noop_when_not_configured(monkeypatch):
    monkeypatch.setattr(settings, "OBSERVE_ENABLED", False)
    assert observe_worker.start_observe_worker() is None
 async def test_run_loop_survives_post_failure(monkeypatch):
    """_post_batch 抛异常时,loop 不崩溃、继续处理后续批次(best-effort 契约)。"""
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_FLUSH_INTERVAL_SEC", 0.02)
    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
    seen: list[list[int]] = []
    async def boom(client, batch):
        seen.append([e["n"] for e in batch])
        raise RuntimeError("boom")
    monkeypatch.setattr(observe_worker, "_post_batch", boom)
    q.put_nowait({"n": 1})
    task = asyncio.create_task(observe_worker._run_loop(None))
    try:
        for _ in range(50):  # 轮询直到第 1 批被处理(失败),最多等 0.5s
            await asyncio.sleep(0.01)
            if seen:
                break
        q.put_nowait({"n": 2})
        for _ in range(50):  # 第 2 批被处理 → 证明失败后 loop 仍存活
            await asyncio.sleep(0.01)
            if len(seen) >= 2:
                break
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
    assert seen == [[1], [2]]
 async def test_stop_flushes_remaining_and_closes_client(monkeypatch):
    """stop:cancel 后把剩余事件 best-effort 发出最后一批,并关闭 + 置空 client。"""
    q = asyncio.Queue(maxsize=100)
    monkeypatch.setattr(observe, "_queue", q)
    monkeypatch.setattr(settings, "OBSERVE_ORG", "default")
    monkeypatch.setattr(settings, "OBSERVE_STREAM", "app_requests")
    monkeypatch.setattr(settings, "OBSERVE_BATCH_MAX", 200)
    q.put_nowait({"n": 1})
    q.put_nowait({"n": 2})
    posted: dict = {}
    def handler(request: httpx.Request) -> httpx.Response:
        posted["body"] = request.content
        return httpx.Response(200, json={"code": 200})
    client = httpx.AsyncClient(
        base_url="http://oo", transport=httpx.MockTransport(handler)
    )
    monkeypatch.setattr(observe_worker, "_client", client)
    async def _noop() -> None:
        return None
    task = asyncio.create_task(_noop())
    await observe_worker.stop_observe_worker(task)
    assert b'"n"' in posted["body"]        # 关停时把剩余事件发了出去
    assert observe_worker._client is None  # client 已关闭并置空
 def test_app_has_metrics_middleware():
    from app.main import app
    names = [m.cls.__name__ for m in app.user_middleware]
    assert "RequestMetricsMiddleware" in names
Author	SHA1	Message	Date
unknown	64872b8919	fix(ad): detach late impressions from abandoned coupons	2026-07-21 11:33:21 +08:00
unknown	207e189fb3	fix(ad): detach late impressions from failed coupon traces	2026-07-21 11:33:21 +08:00
zuochenyong	c53ce896f7	feat(huawei-review): 华为审核开关(admin 可切 + 客户端下发 + 审计) (#147 ) 华为应用市场审核要求新手引导的「快速设置」权限步必须可被用户关闭,平时又要保住权限开启率,故做成后台可切的两态开关,送审期间切开、过审后收回。 - app_config 新增 huawei_review 行(default / review),空库与脏值一律回退 default = 上线至今的现状,宁可不给退出按钮也不误放开 - admin: GET/PATCH /admin/api/huawei-review,权限 operator/tech,切换写审计 - 客户端: GET /api/v1/platform/huawei-review 不鉴权(引导页在登录前就展示), 下发 onboarding_closable;机型 gate 由客户端做,故此处不判 ROM Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: 左辰勇 <exinglang@gmail.com> Reviewed-on: #147 Co-authored-by: zuochenyong <zuochenyong@wonderable.ai> Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>	2026-07-21 10:43:58 +08:00
linkeyu	ed26935b14	feat(admin): 数据大盘比价指标改为后端聚合 (#146 ) ## 修改内容 - 数据大盘比价指标改由后端按日期区间聚合 - 增加完成数、中途退出数、成功率、中位数、P95 和 TOKEN 总成本 - 成功率分母排除中途退出，耗时仅统计 success/failed - 增加后端聚合口径测试 --------- Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #146 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-21 10:11:39 +08:00
guke	48037f03fd	docs: OpenObserve 接口 QPS/耗时可观测设计 spec (#145 ) openobserve上报 --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #145	2026-07-20 18:55:38 +08:00