diff --git a/.env.example b/.env.example index a7f92f8..6a0191f 100644 --- a/.env.example +++ b/.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 之类 diff --git a/app/api/v1/compare.py b/app/api/v1/compare.py index ff1ba7e..fcb1e03 100644 --- a/app/api/v1/compare.py +++ b/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 diff --git a/app/api/v1/compare_record.py b/app/api/v1/compare_record.py index a2652a6..3d2b484 100644 --- a/app/api/v1/compare_record.py +++ b/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( diff --git a/app/core/config.py b/app/core/config.py index 81149db..72b54e6 100644 --- a/app/core/config.py +++ b/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。 diff --git a/app/core/llm_cost_backfill_worker.py b/app/core/llm_cost_backfill_worker.py new file mode 100644 index 0000000..6dc65cf --- /dev/null +++ b/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 diff --git a/app/main.py b/app/main.py index 799c38b..5cc5f62 100644 --- a/app/main.py +++ b/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") diff --git a/app/services/comparison_llm_backfill.py b/app/services/comparison_llm_backfill.py new file mode 100644 index 0000000..f59806d --- /dev/null +++ b/app/services/comparison_llm_backfill.py @@ -0,0 +1,145 @@ +"""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.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 _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(UTC) - 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 >= 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, + } diff --git a/app/services/pricebot_llm_calls.py b/app/services/pricebot_llm_calls.py index 1c44bac..f8fce5a 100644 --- a/app/services/pricebot_llm_calls.py +++ b/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 [] diff --git a/deploy/comparison-llm-cost.md b/deploy/comparison-llm-cost.md new file mode 100644 index 0000000..d72c1fc --- /dev/null +++ b/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 仍保留时才能准确回填； +原始调用已经清理的记录不能用估算值冒充真实成本。 diff --git a/tests/test_compare_harvest.py b/tests/test_compare_harvest.py index b13028e..9172823 100644 --- a/tests/test_compare_harvest.py +++ b/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: diff --git a/tests/test_comparison_llm_backfill.py b/tests/test_comparison_llm_backfill.py new file mode 100644 index 0000000..61b3646 --- /dev/null +++ b/tests/test_comparison_llm_backfill.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from datetime import datetime + +from app.db.session import SessionLocal +from app.models.comparison import ComparisonRecord +from app.services import comparison_llm_backfill + + +def _record(trace_id: str, *, status: str = "success") -> int: + with SessionLocal() as db: + rec = ComparisonRecord( + trace_id=trace_id, + status=status, + created_at=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) diff --git a/tests/test_llm_cost.py b/tests/test_llm_cost.py index 1e1c1c3..dbf95e0 100644 --- a/tests/test_llm_cost.py +++ b/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: diff --git a/tests/test_pricebot_llm_calls.py b/tests/test_pricebot_llm_calls.py new file mode 100644 index 0000000..d50b5fc --- /dev/null +++ b/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