Merge branch 'main' into codex/fix-comparison-token-cost

将回填窗口和价格配置生效时间统一转换为北京墙钟时间，避免改价前记录使用新价格重算，并补充改价前后边界回归测试。

.env.example

@@ -139,6 +139,11 @@ PRICEBOT_COMPARE_TIMEOUT_SEC=60
 # 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
 # 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
 INTERNAL_API_SECRET=
+# 新版比价 harvest 完成后即时回填；以下 worker 再补偿短暂故障期间漏掉的记录。
+LLM_COST_BACKFILL_ENABLED=true
+LLM_COST_BACKFILL_INTERVAL_SEC=300
+LLM_COST_BACKFILL_BATCH_SIZE=100
+LLM_COST_BACKFILL_LOOKBACK_DAYS=30
 
 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类

app/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,6 +36,7 @@ from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare")
 
@@ -80,7 +81,7 @@ def _harvest_running_blocking(
 def _harvest_done_blocking(
     trace_id: str, user_id: int | None, done_params: dict, business_type: str,
     device_id: str | None, device_info: dict | None, trace_url: str | None,
-) -> None:
+) -> int:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -102,6 +103,7 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
+        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -243,7 +245,12 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+async def price_step(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
+) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -252,12 +259,15 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            await run_in_threadpool(
+            record_id = await run_in_threadpool(
                 _harvest_done_blocking, trace_id, (user.id if user else None),
                 done_params, "food",
                 meta.get("device_id"), meta.get("device_info"),
                 resp.get("trace_url") or done_params.get("trace_url"),
             )
+            background_tasks.add_task(
+                backfill_comparison_llm_cost, record_id, trace_id
+            )
         except Exception as e:  # noqa: BLE001
             logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,8 +16,6 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.db.session import SessionLocal
-from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -30,8 +28,7 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare_record")
 
@@ -121,32 +118,7 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    calls = fetch_llm_calls(trace_id)
-    if not calls:
-        return
-    db = SessionLocal()
-    try:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is None:
-            return
-        rec.llm_calls = calls
-        rec.llm_call_count = len(calls)
-        rec.retry_count = sum(1 for c in calls if c.get("error"))
-        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
-        # error 的调用 usage 可能为 None,or {} 兜底)
-        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
-        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
-        # 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
-        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
-        db.commit()
-        logger.info(
-            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
-            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
-        )
-    except Exception as e:  # noqa: BLE001 best-effort
-        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
-    finally:
-        db.close()
+    backfill_comparison_llm_cost(record_id, trace_id)
 
 
 @router.get(

app/core/config.py

@@ -377,6 +377,12 @@ class Settings(BaseSettings):
     # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
     # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
     INTERNAL_API_SECRET: str = ""
+    # Current compare clients are persisted by server-side harvest. Repair any
+    # recent terminal rows left without LLM usage by transient upstream/auth failures.
+    LLM_COST_BACKFILL_ENABLED: bool = True
+    LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
+    LLM_COST_BACKFILL_BATCH_SIZE: int = 100
+    LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
 
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。

app/core/llm_cost_backfill_worker.py

@@ -0,0 +1,63 @@
+"""Periodic repair worker for comparison records with missing LLM token cost."""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+from app.core.config import settings
+from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
+from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
+
+logger = logging.getLogger("shagua.llm_cost_backfill_worker")
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
+    logger.info(
+        "LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
+        interval,
+        settings.LLM_COST_BACKFILL_BATCH_SIZE,
+        settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+    )
+    try:
+        while True:
+            try:
+                auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
+                if auth_ready:
+                    result = await asyncio.to_thread(
+                        repair_missing_comparison_llm_costs,
+                        limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
+                        lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+                    )
+                    logger.info("LLM cost backfill batch result=%s", result)
+                else:
+                    logger.error(
+                        "LLM cost backfill skipped: PriceBot internal auth is not ready"
+                    )
+            except Exception:  # noqa: BLE001
+                logger.exception("LLM cost backfill batch failed")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("LLM cost backfill worker stopped")
+        raise
+
+
+def start_llm_cost_backfill_worker() -> asyncio.Task | None:
+    if not settings.LLM_COST_BACKFILL_ENABLED:
+        logger.info("LLM cost backfill worker disabled")
+        return None
+    if not settings.INTERNAL_API_SECRET:
+        logger.warning(
+            "LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
+        )
+        return None
+    return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
+
+
+async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/main.py

@@ -60,6 +60,10 @@ from app.core.inactivity_reset_worker import (
     start_inactivity_reset_worker,
     stop_inactivity_reset_worker,
 )
+from app.core.llm_cost_backfill_worker import (
+    start_llm_cost_backfill_worker,
+    stop_llm_cost_backfill_worker,
+)
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
@@ -103,6 +107,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
+    llm_cost_backfill_task = start_llm_cost_backfill_worker()
     try:
         yield
     finally:
@@ -112,6 +117,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
+        await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")

app/services/comparison_llm_backfill.py

@@ -0,0 +1,156 @@
+"""Persist and repair comparison-record LLM token costs."""
+from __future__ import annotations
+
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services.llm_cost import compute_llm_cost, get_llm_prices
+from app.services.pricebot_llm_calls import fetch_llm_calls
+
+logger = logging.getLogger("shagua.comparison_llm_backfill")
+
+
+def _utc_to_beijing_naive(value: datetime) -> datetime:
+    """Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=UTC)
+    return value.astimezone(CN_TZ).replace(tzinfo=None)
+
+
+def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
+    """Store calls and all derived fields atomically."""
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is None or rec.trace_id != trace_id:
+            logger.warning(
+                "LLM cost backfill record mismatch record_id=%s trace=%s",
+                record_id,
+                trace_id,
+            )
+            return False
+
+        # Never recalculate a frozen historical cost with a newer price config.
+        if rec.llm_cost_yuan is not None and rec.llm_calls:
+            return False
+
+        rec.llm_calls = calls
+        rec.llm_call_count = len(calls)
+        rec.retry_count = sum(1 for call in calls if call.get("error"))
+        rec.input_tokens = sum(
+            (call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
+        )
+        rec.output_tokens = sum(
+            (call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
+        )
+        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
+            calls, get_llm_prices(db)
+        )
+        db.commit()
+        logger.info(
+            "LLM cost backfilled trace=%s calls=%d input_tokens=%d "
+            "output_tokens=%d cost=%s",
+            trace_id,
+            len(calls),
+            rec.input_tokens,
+            rec.output_tokens,
+            rec.llm_cost_yuan,
+        )
+        return True
+
+
+def backfill_comparison_llm_cost(
+    record_id: int,
+    trace_id: str,
+    *,
+    attempts: int = 3,
+    retry_delays: tuple[float, ...] = (1.0, 3.0),
+) -> bool:
+    """Fetch and persist one record, retrying short-lived upstream races."""
+    if not settings.INTERNAL_API_SECRET or not trace_id:
+        logger.warning(
+            "LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
+            trace_id,
+        )
+        return False
+    total_attempts = max(1, attempts)
+    for attempt in range(total_attempts):
+        calls = fetch_llm_calls(trace_id)
+        if calls:
+            try:
+                return _store_calls(record_id, trace_id, calls)
+            except Exception:  # noqa: BLE001 - background repair must stay alive
+                logger.exception(
+                    "LLM cost store failed trace=%s record_id=%s",
+                    trace_id,
+                    record_id,
+                )
+                return False
+
+        if attempt + 1 < total_attempts:
+            delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
+            if delay > 0:
+                time.sleep(delay)
+
+    logger.warning(
+        "LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
+        trace_id,
+        record_id,
+        total_attempts,
+    )
+    return False
+
+
+def repair_missing_comparison_llm_costs(
+    *,
+    limit: int = 100,
+    lookback_days: int = 30,
+) -> dict[str, int]:
+    """Repair a bounded batch of recent terminal records with missing cost."""
+    cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
+        days=max(1, lookback_days)
+    )
+    with SessionLocal() as db:
+        # app_config has no price history. Repricing a record from before the
+        # current config became effective would fabricate a historical cost, so
+        # only repair records at/after that timestamp.
+        price_config_updated_at = db.execute(
+            select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
+        ).scalar_one_or_none()
+        date_conditions = [ComparisonRecord.created_at >= cutoff]
+        if price_config_updated_at is not None:
+            date_conditions.append(
+                ComparisonRecord.created_at
+                >= _utc_to_beijing_naive(price_config_updated_at)
+            )
+        candidates = list(
+            db.execute(
+                select(ComparisonRecord.id, ComparisonRecord.trace_id)
+                .where(
+                    *date_conditions,
+                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.llm_cost_yuan.is_(None),
+                )
+                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
+                .limit(max(1, limit))
+            ).all()
+        )
+
+    repaired = 0
+    for record_id, trace_id in candidates:
+        if backfill_comparison_llm_cost(
+            record_id, trace_id, attempts=1, retry_delays=()
+        ):
+            repaired += 1
+    return {
+        "candidates": len(candidates),
+        "repaired": repaired,
+        "unresolved": len(candidates) - repaired,
+    }

app/services/pricebot_llm_calls.py

@@ -20,19 +20,67 @@ from app.core.pricebot_router import pick_pricebot
 logger = logging.getLogger("shagua.pricebot_llm")
 
 
+def pricebot_llm_auth_ready() -> bool:
+    """Verify every configured PriceBot instance accepts the shared secret."""
+    secret = settings.INTERNAL_API_SECRET
+    if not secret:
+        logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
+        return False
+    for base in settings.pricebot_instances:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
+        try:
+            resp = httpx.get(
+                url, headers={"X-Internal-Secret": secret}, timeout=3.0
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
+            return False
+        if resp.status_code != 200:
+            logger.error(
+                "PriceBot LLM auth check rejected base=%s status=%s; "
+                "verify both services use the same INTERNAL_API_SECRET",
+                base,
+                resp.status_code,
+            )
+            return False
+    return True
+
+
 def fetch_llm_calls(trace_id: str) -> list[dict]:
     """返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
     未配密钥 / 无 trace_id / 拉取失败 → []。"""
     secret = settings.INTERNAL_API_SECRET
     if not secret or not trace_id:
         return []
-    base = pick_pricebot(trace_id).rstrip("/")
-    url = f"{base}/api/internal/llm_calls/{trace_id}"
-    try:
-        resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
-        if resp.status_code == 200:
-            return resp.json().get("calls", []) or []
-        logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
-    except Exception as e:  # noqa: BLE001 — best-effort,任何异常都不该影响上报
-        logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
+    preferred = pick_pricebot(trace_id)
+    # LLM JSONL is instance-local. If the cluster topology changed after a
+    # historical trace was created, consistent hashing may now point elsewhere;
+    # probe the remaining configured instances only when the preferred one is empty.
+    bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
+    for base in bases:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
+        try:
+            resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
+            if resp.status_code == 200:
+                calls = resp.json().get("calls", []) or []
+                if calls:
+                    return calls
+                continue
+            if resp.status_code in (401, 403):
+                logger.error(
+                    "fetch_llm_calls rejected trace=%s base=%s status=%s; "
+                    "INTERNAL_API_SECRET differs between app-server and PriceBot",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+            else:
+                logger.warning(
+                    "fetch_llm_calls trace=%s base=%s status=%s",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+        except Exception as e:  # noqa: BLE001 — best-effort
+            logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
     return []

deploy/comparison-llm-cost.md

@@ -0,0 +1,40 @@
+# 比价 TOKEN 成本采集与补偿
+
+## 部署前置
+
+App Server 与 PriceBot 使用各自独立的 `.env`，但下面的值必须完全一致：
+
+- `/opt/shaguabijia-app-server/.env`
+- `/opt/pricebot-backend/.env`
+- 配置项：`INTERNAL_API_SECRET`
+
+不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
+
+App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
+`PriceBot LLM auth check rejected`，并跳过本轮补偿，避免对所有缺失记录重复发送失败请求。
+
+## 数据链路
+
+1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
+2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用，冻结 Token、成本和单价快照。
+3. 周期 worker 扫描近期 `success/failed` 且 `llm_cost_yuan IS NULL` 的记录进行补偿；
+   为避免用现价伪造历史成本，只处理当前单价配置生效时间之后的记录。
+4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
+
+## 上线验收（只读 SQL）
+
+```sql
+SELECT
+  (created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
+  count(*) AS records,
+  count(llm_cost_yuan) AS cost_records,
+  round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
+FROM comparison_record
+WHERE created_at >= now() - interval '3 days'
+GROUP BY 1
+ORDER BY 1 DESC;
+```
+
+新产生的正常终态比价记录应在短时间内写入 `input_tokens`、`output_tokens` 和
+`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填；
+原始调用已经清理的记录不能用估算值冒充真实成本。

tests/test_compare_harvest.py

@@ -9,17 +9,16 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
 """
 from __future__ import annotations
 
-import json
 import uuid
 from unittest.mock import MagicMock, patch
 
 import httpx
+from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.schemas.compare_record import ComparisonRecordIn
-from sqlalchemy import select
 
 
 def _tid() -> str:
@@ -216,7 +215,9 @@ def test_price_step_done_harvests_success(client) -> None:
                   "action": {"command": "done", "params": _done_params()},
                   "trace_url": "https://price.shaguabijia.com/traces/done2/"}
     p, _cap = _mock_pricebot(done_frame)
-    with p:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -224,6 +225,7 @@ def test_price_step_done_harvests_success(client) -> None:
         assert rec is not None and rec.status == "success"
         assert rec.best_platform_id == "meituan"
         assert rec.saved_amount_cents == 500
+        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_trace_finalize_harvests_abort(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services import comparison_llm_backfill
+
+
+def _record(
+    trace_id: str,
+    *,
+    status: str = "success",
+    created_at: datetime | None = None,
+) -> int:
+    with SessionLocal() as db:
+        rec = ComparisonRecord(
+            trace_id=trace_id,
+            status=status,
+            created_at=created_at or datetime.now(),
+        )
+        db.add(rec)
+        db.commit()
+        return rec.id
+
+
+def _delete(record_id: int) -> None:
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is not None:
+            db.delete(rec)
+            db.commit()
+
+
+def test_backfill_retries_then_persists_cost(monkeypatch):
+    record_id = _record("llm-retry-1")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 1000, "completion_tokens": 500},
+        }
+    ]
+    responses = iter([[], calls])
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "fetch_llm_calls",
+        lambda trace_id: next(responses),
+    )
+    sleeps: list[float] = []
+    monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+
+    try:
+        assert comparison_llm_backfill.backfill_comparison_llm_cost(
+            record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
+        )
+        assert sleeps == [0.25]
+        with SessionLocal() as db:
+            rec = db.get(ComparisonRecord, record_id)
+            assert rec.input_tokens == 1000
+            assert rec.output_tokens == 500
+            assert rec.llm_cost_yuan is not None
+            assert rec.llm_calls == calls
+    finally:
+        _delete(record_id)
+
+
+def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
+    missing_id = _record("llm-repair-missing")
+    running_id = _record("llm-repair-running", status="running")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 100, "completion_tokens": 20},
+        }
+    ]
+    seen: list[str] = []
+
+    def fetch(trace_id: str) -> list[dict]:
+        seen.append(trace_id)
+        return calls
+
+    monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10, lookback_days=1
+        )
+        assert result["repaired"] >= 1
+        assert "llm-repair-missing" in seen
+        assert "llm-repair-running" not in seen
+        with SessionLocal() as db:
+            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
+            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
+    finally:
+        _delete(missing_id)
+        _delete(running_id)
+
+
+def test_repair_excludes_records_before_current_price_config(monkeypatch):
+    price_changed_at = datetime.now(UTC) - timedelta(hours=1)
+    before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
+    after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
+    before_id = _record(
+        "llm-before-price-change",
+        created_at=before_change.replace(tzinfo=None),
+    )
+    after_id = _record(
+        "llm-after-price-change",
+        created_at=after_change.replace(tzinfo=None),
+    )
+    with SessionLocal() as db:
+        existing = db.get(AppConfig, "llm_token_price")
+        if existing is not None:
+            db.delete(existing)
+            db.flush()
+        db.add(
+            AppConfig(
+                key="llm_token_price",
+                value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
+                updated_at=price_changed_at,
+            )
+        )
+        db.commit()
+
+    seen: list[str] = []
+
+    def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
+        seen.append(trace_id)
+        return True
+
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "backfill_comparison_llm_cost",
+        backfill,
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10_000,
+            lookback_days=1,
+        )
+        assert "llm-after-price-change" in seen
+        assert "llm-before-price-change" not in seen
+        assert result["repaired"] == len(seen)
+        assert result["unresolved"] == 0
+    finally:
+        _delete(before_id)
+        _delete(after_id)
+        with SessionLocal() as db:
+            config = db.get(AppConfig, "llm_token_price")
+            if config is not None:
+                db.delete(config)
+                db.commit()

tests/test_llm_cost.py

@@ -132,6 +132,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
     from app.models.app_config import AppConfig
     from app.models.comparison import ComparisonRecord
     from app.repositories import app_config
+    from app.services import comparison_llm_backfill
 
     sample = [
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -139,7 +140,12 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
     ]
-    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
+    monkeypatch.setattr(
+        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
+    )
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -0,0 +1,55 @@
+from unittest.mock import MagicMock
+
+from app.core.config import settings
+from app.services import pricebot_llm_calls
+
+
+def test_auth_probe_accepts_matching_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    response = MagicMock(status_code=200)
+    get = MagicMock(return_value=response)
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
+    assert get.call_count == len(settings.pricebot_instances)
+    assert all(
+        call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
+        for call in get.call_args_list
+    )
+
+
+def test_auth_probe_rejects_mismatched_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
+    monkeypatch.setattr(
+        pricebot_llm_calls.httpx,
+        "get",
+        MagicMock(return_value=MagicMock(status_code=403)),
+    )
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
+
+
+def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    monkeypatch.setattr(
+        settings,
+        "PRICEBOT_INSTANCES",
+        "http://pricebot-1:8000,http://pricebot-2:8000",
+    )
+    monkeypatch.setattr(
+        pricebot_llm_calls,
+        "pick_pricebot",
+        lambda trace_id: "http://pricebot-1:8000",
+    )
+    calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
+
+    def get(url, **kwargs):
+        response = MagicMock(status_code=200)
+        response.json.return_value = {
+            "calls": [] if "pricebot-1" in url else calls
+        }
+        return response
+
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls