Optimize terminal detail loading and Redis replay

2026-05-31 04:32:48 +08:00
parent 2a6d8748f4
commit 46effcd45f
16 changed files with 948 additions and 80 deletions
@@ -0,0 +1,47 @@
+import { NextRequest, NextResponse } from "next/server";
+import { proxyBackendJsonGet } from "@/lib/api-proxy";
+import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
+
+const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
+const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
+  process.env.POLYWEATHER_CITY_DETAIL_BATCH_PROXY_TIMEOUT_MS || "12000",
+);
+
+export async function GET(req: NextRequest) {
+  if (!API_BASE) {
+    return NextResponse.json(
+      { error: "POLYWEATHER_API_BASE_URL is not configured" },
+      { status: 500 },
+    );
+  }
+
+  const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false";
+  const cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15);
+  const searchParams = new URLSearchParams({
+    cities: req.nextUrl.searchParams.get("cities") || "",
+    force_refresh: forceRefresh,
+    limit: req.nextUrl.searchParams.get("limit") || "12",
+  });
+  for (const key of ["market_slug", "target_date", "resolution"]) {
+    const value = req.nextUrl.searchParams.get(key);
+    if (value) searchParams.set(key, value);
+  }
+
+  const controller = new AbortController();
+  const timeoutId = setTimeout(() => controller.abort(), DETAIL_BATCH_PROXY_TIMEOUT_MS);
+
+  try {
+    return await proxyBackendJsonGet(req, {
+      cacheControl: cachePolicy.responseCacheControl,
+      fetchCache:
+        cachePolicy.fetchMode === "no-store" ? "no-store" : undefined,
+      publicMessage: "Failed to fetch city detail batch",
+      revalidateSeconds: cachePolicy.revalidateSeconds,
+      signal: controller.signal,
+      timeoutPublicMessage: "City detail batch request timed out",
+      url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`,
+    });
+  } finally {
+    clearTimeout(timeoutId);
+  }
+}
@@ -61,6 +61,7 @@ import {
  mergeScanRowsWithCityFallbackRows,
 } from "@/components/dashboard/scan-terminal/city-fallback-rows";
 import { markAnalyticsOnce, trackAppEvent } from "@/lib/app-analytics";
+import { STATIC_CITY_LIST } from "@/lib/static-cities";

 const TrainingDashboard = dynamic(
  () =>
@@ -1218,7 +1219,9 @@ function ScanTerminalScreen() {
    refreshScanTerminalManually();
  }, [refreshScanTerminalManually]);

-  const [cityFallbackRows, setCityFallbackRows] = useState<ScanOpportunityRow[]>([]);
+  const [cityFallbackRows, setCityFallbackRows] = useState<ScanOpportunityRow[]>(() =>
+    cityListItemsToScanRows(STATIC_CITY_LIST),
+  );
  const rows = useMemo(
    () => {
      const scanRows = terminalData?.rows || [];
@@ -121,9 +121,11 @@ export function runTests() {

  assert(
    dashboardSource.includes("cityListItemsToScanRows") &&
+      dashboardSource.includes("STATIC_CITY_LIST") &&
+      dashboardSource.includes("useState<ScanOpportunityRow[]>(() =>") &&
      dashboardSource.includes("/api/cities") &&
      dashboardSource.includes("cityFallbackRows"),
-    "terminal dashboard should use /api/cities fallback rows when scan terminal rows are not ready",
+    "terminal dashboard should seed fallback rows from the static city snapshot before refreshing /api/cities",
  );
  assert(fs.existsSync(staticCitiesPath), "/api/cities route should have a static city snapshot fallback");
  assert(
@@ -85,6 +85,17 @@ export async function runTests() {
      chartSource.includes("setHourly(seedHourlyForecastFromRow(row))"),
    "terminal charts should render from row data immediately and dedupe concurrent city detail requests",
  );
+  assert(
+    chartLogicSource.includes("/api/cities/detail-batch") &&
+      chartLogicSource.includes("flushCityDetailBatch") &&
+      chartLogicSource.includes("primeCityDetailCache"),
+    "visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache",
+  );
+  assert(
+    chartLogicSource.includes("options.ignoreCache\n      ? runQueuedHourlyDetailRequest") &&
+      chartLogicSource.includes(": queueCityDetailBatch(city, resParam)"),
+    "normal first-paint city detail requests should enter the batch queue before single-request concurrency limiting",
+  );
  assert(
    chartSource.includes("IntersectionObserver") &&
      chartSource.includes("shouldFetchCityDetailForChart") &&
@@ -965,6 +965,26 @@ type HourlyForecastFetchOptions = {
  resolution?: string;
 };

+type CityDetailBatchPayload = {
+  details?: Record<string, CityDetail | null | undefined>;
+  errors?: Record<string, string>;
+};
+
+type CityDetailBatchWaiter = {
+  resolve: (value: HourlyForecast) => void;
+  reject: (reason?: unknown) => void;
+};
+
+type CityDetailBatchQueue = {
+  cities: Set<string>;
+  waiters: Map<string, CityDetailBatchWaiter[]>;
+  timer: ReturnType<typeof setTimeout> | null;
+};
+
+const CITY_DETAIL_BATCH_WINDOW_MS = 25;
+const CITY_DETAIL_BATCH_MAX_CITIES = 12;
+const _cityDetailBatchQueues = new Map<string, CityDetailBatchQueue>();
+
 function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForecast {
  const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly;
  if (!json || !hourlySource) return null;
@@ -993,6 +1013,133 @@ function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForec
  };
 }

+function primeCityDetailCache(
+  city: string,
+  resolution: string,
+  detail: CityDetail | null | undefined,
+): HourlyForecast {
+  const data = parseHourlyForecastFromCityDetail(detail || null);
+  if (!data) return null;
+  const cacheKey = `${city}:${resolution}`;
+  _hourlyCache.set(cacheKey, { ts: Date.now(), data });
+  writeSessionCache(cacheKey, data);
+  return data;
+}
+
+async function fetchSingleHourlyForecastForCity(
+  city: string,
+  resolution: string,
+): Promise<HourlyForecast> {
+  const res = await fetchCityDetailWithTimeout(city, resolution);
+  if (!res || !res.ok) return null;
+  const json = await res.json() as CityDetail;
+  return primeCityDetailCache(city, resolution, json);
+}
+
+function queueCityDetailBatch(city: string, resolution: string): Promise<HourlyForecast> {
+  return new Promise<HourlyForecast>((resolve, reject) => {
+    const queue = _cityDetailBatchQueues.get(resolution) || {
+      cities: new Set<string>(),
+      waiters: new Map<string, CityDetailBatchWaiter[]>(),
+      timer: null,
+    };
+    _cityDetailBatchQueues.set(resolution, queue);
+
+    const cityWaiters = queue.waiters.get(city) || [];
+    cityWaiters.push({ resolve, reject });
+    queue.waiters.set(city, cityWaiters);
+    queue.cities.add(city);
+
+    if (queue.timer === null) {
+      queue.timer = setTimeout(() => flushCityDetailBatch(resolution), CITY_DETAIL_BATCH_WINDOW_MS);
+    }
+    if (queue.cities.size >= CITY_DETAIL_BATCH_MAX_CITIES) {
+      flushCityDetailBatch(resolution);
+    }
+  });
+}
+
+function resolveBatchWaiters(
+  waiters: CityDetailBatchWaiter[] | undefined,
+  value: HourlyForecast,
+) {
+  (waiters || []).forEach((waiter) => waiter.resolve(value));
+}
+
+function rejectBatchWaiters(
+  waiters: CityDetailBatchWaiter[] | undefined,
+  reason: unknown,
+) {
+  (waiters || []).forEach((waiter) => waiter.reject(reason));
+}
+
+async function flushCityDetailBatch(resolution: string) {
+  const queue = _cityDetailBatchQueues.get(resolution);
+  if (!queue) return;
+  _cityDetailBatchQueues.delete(resolution);
+  if (queue.timer !== null) {
+    clearTimeout(queue.timer);
+    queue.timer = null;
+  }
+
+  const cities = Array.from(queue.cities).sort();
+  if (!cities.length) return;
+
+  try {
+    const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
+    const details = payload?.details || {};
+    await Promise.all(
+      cities.map(async (city) => {
+        const waiters = queue.waiters.get(city);
+        const detail = details[city];
+        const data = primeCityDetailCache(city, resolution, detail);
+        if (data) {
+          resolveBatchWaiters(waiters, data);
+          return;
+        }
+        try {
+          resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution));
+        } catch (error) {
+          rejectBatchWaiters(waiters, error);
+        }
+      }),
+    );
+  } catch (error) {
+    await Promise.all(
+      cities.map(async (city) => {
+        const waiters = queue.waiters.get(city);
+        try {
+          resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution));
+        } catch (fallbackError) {
+          rejectBatchWaiters(waiters, fallbackError || error);
+        }
+      }),
+    );
+  }
+}
+
+function fetchCityDetailBatchWithTimeout(cities: string[], resolution: string) {
+  const controller = new AbortController();
+  const timeoutId = globalThis.setTimeout(() => controller.abort(), HOURLY_DETAIL_REQUEST_TIMEOUT_MS);
+  const params = new URLSearchParams({
+    cities: cities.join(","),
+    depth: "full",
+    force_refresh: "false",
+    limit: String(Math.max(cities.length, CITY_DETAIL_BATCH_MAX_CITIES)),
+    resolution,
+  });
+  return fetch(`/api/cities/detail-batch?${params.toString()}`, {
+    headers: { Accept: "application/json" },
+    signal: controller.signal,
+  })
+    .then(async (res) => {
+      if (!res.ok) return null;
+      return res.json() as Promise<CityDetailBatchPayload>;
+    })
+    .catch(() => null)
+    .finally(() => globalThis.clearTimeout(timeoutId));
+}
+
 async function fetchHourlyForecastForCity(
  city: string,
  options: HourlyForecastFetchOptions = {},
@@ -1011,19 +1158,10 @@ async function fetchHourlyForecastForCity(
  const pending = _hourlyRequestCache.get(requestKey);
  if (pending) return pending;

-  const request = runQueuedHourlyDetailRequest(() =>
-    fetchCityDetailWithTimeout(city, resParam)
-      .then(async (res) => {
-        if (!res || !res.ok) return null;
-        return res.json() as Promise<CityDetail>;
-      })
-      .then((json) => {
-        const data = parseHourlyForecastFromCityDetail(json);
-        if (!data) return null;
-        _hourlyCache.set(cacheKey, { ts: Date.now(), data });
-        writeSessionCache(cacheKey, data);
-        return data;
-      }),
+  const request = (
+    options.ignoreCache
+      ? runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resParam))
+      : queueCityDetailBatch(city, resParam)
  )
    .finally(() => {
      _hourlyRequestCache.delete(requestKey);
@@ -27,8 +27,12 @@ export function runTests() {
      systemPage.includes("sourceHealth") &&
      systemPage.includes("MGM、KNMI、IMS") &&
      systemPage.includes("断线") &&
-      systemPage.includes("延迟"),
-    "ops system page must show source latency/disconnect status for operational city sources",
+      systemPage.includes("延迟") &&
+      systemPage.includes("sourceReasonLabel") &&
+      systemPage.includes("观测时间缺失") &&
+      systemPage.includes("formatOpsValue") &&
+      systemPage.includes("强制刷新"),
+    "ops system page must show readable source reasons, object values, and cache force-refresh metrics",
  );
  assert(
    nextRoute.includes("requireOpsProxyAuth") &&
@@ -50,6 +50,25 @@ function compactDate(value?: string) {
  return value.slice(0, 19).replace("T", " ");
 }

+function paymentReasonLabel(reason?: string) {
+  const key = String(reason || "").trim().toLowerCase();
+  if (key === "receiver_mismatch") return "收款地址不匹配";
+  if (key === "amount_mismatch") return "金额不匹配";
+  if (key === "chain_mismatch") return "支付网络不匹配";
+  if (key === "tx_not_found") return "链上交易未找到";
+  if (key === "tx_reverted") return "链上交易失败";
+  if (key === "expired") return "订单已过期";
+  if (key === "unknown") return "未知原因";
+  return key || "未知原因";
+}
+
+function compactMono(value?: string, head = 10, tail = 6) {
+  const text = String(value || "").trim();
+  if (!text) return "—";
+  if (text.length <= head + tail + 3) return text;
+  return `${text.slice(0, head)}...${text.slice(-tail)}`;
+}
+
 function RiskStat({
  label,
  value,
@@ -148,7 +167,7 @@ export function PaymentsPageClient() {
          <div className="grid grid-cols-2 gap-3 md:grid-cols-4 xl:grid-cols-7">
            <RiskStat label="总异常" value={Number(riskSummary.issues ?? 0)} sub="需要人工关注" tone={hasRisk ? "text-red-600" : "text-emerald-600"} />
            <RiskStat label="Intent 卡住" value={Number(riskSummary.stuck_intents ?? 0)} sub="submitted/过期 created" />
-            <RiskStat label="试用漏开" value={Number(riskSummary.trial_gaps ?? 0)} sub="signup 无 trial_created" />
+            <RiskStat label="试用漏开" value={Number(riskSummary.trial_gaps ?? 0)} sub="后端试用/订阅证据缺失" />
            <RiskStat label="支付异常" value={Number(riskSummary.payment_incidents ?? incidents.length)} sub="未标记处理" />
            <RiskStat label="积分异常" value={Number(riskSummary.points_discount_issues ?? 0)} sub="确认后未扣/少扣" />
            <RiskStat label="推荐异常" value={Number(riskSummary.referral_settlement_issues ?? 0)} sub="转化无奖励记录" />
@@ -287,7 +306,8 @@ export function PaymentsPageClient() {
                <thead>
                  <tr className="border-b border-white/10 text-left text-slate-400">
                    <th className="py-2 pr-4 font-medium">ID</th>
-                    <th className="py-2 pr-4 font-medium">原因</th>
+                    <th className="py-2 pr-4 font-medium">原因 / 详情</th>
+                    <th className="py-2 pr-4 font-medium">用户 / Intent</th>
                    <th className="py-2 pr-4 font-medium">时间</th>
                    <th className="py-2 font-medium">操作</th>
                  </tr>
@@ -296,7 +316,16 @@ export function PaymentsPageClient() {
                  {incidents.map((inc) => (
                    <tr key={inc.id} className="border-b border-white/5">
                      <td className="py-2 pr-4 text-slate-500 font-mono">{inc.id}</td>
-                      <td className="py-2 pr-4 text-amber-300">{inc.reason ?? "—"}</td>
+                      <td className="py-2 pr-4">
+                        <div className="font-bold text-amber-600">{paymentReasonLabel(inc.reason)}</div>
+                        <div className="mt-0.5 max-w-xl truncate text-xs text-slate-500" title={inc.detail || inc.reason || ""}>
+                          {inc.detail || inc.reason || "—"}
+                        </div>
+                      </td>
+                      <td className="py-2 pr-4 text-xs text-slate-500">
+                        <div className="font-mono" title={inc.user_id || ""}>{compactMono(inc.user_id)}</div>
+                        <div className="mt-0.5 font-mono text-blue-700" title={inc.intent_id || ""}>{compactMono(inc.intent_id, 12, 6)}</div>
+                      </td>
                      <td className="py-2 pr-4 text-slate-400 text-xs">{inc.created_at?.slice(0, 19) ?? "—"}</td>
                      <td className="py-2">
                        <Button
@@ -31,6 +31,35 @@ function formatAge(ageMin?: number | null) {
  return `${(ageMin / 60).toFixed(1)}h`;
 }

+function sourceReasonLabel(reason?: string | null) {
+  const key = String(reason || "").trim().toLowerCase();
+  if (key === "observation_time_missing") return "观测时间缺失";
+  if (key === "expected_source_not_present_in_cached_detail") return "缓存详情未包含预期来源";
+  if (key === "past_expected_cadence") return "超过预期更新节奏";
+  if (key === "within_expected_cadence") return "仍在预期更新窗口";
+  if (key === "missing_observation") return "缺少观测值";
+  if (key === "no_cached_detail") return "缺少城市详情缓存";
+  return key || "—";
+}
+
+function formatOpsValue(value: unknown) {
+  if (typeof value === "boolean") {
+    return { label: value ? "TRUE" : "FALSE", active: value };
+  }
+  if (typeof value === "string" || typeof value === "number") {
+    return { label: String(value), active: Boolean(value) };
+  }
+  if (Array.isArray(value)) {
+    return { label: `${value.length} 项`, active: value.length > 0 };
+  }
+  if (value && typeof value === "object") {
+    const entries = Object.values(value as Record<string, unknown>);
+    const enabled = entries.filter(Boolean).length;
+    return { label: `${enabled}/${entries.length} 已配置`, active: enabled > 0 };
+  }
+  return { label: "—", active: false };
+}
+
 export function SystemPageClient() {
  const [loading, setLoading] = useState(true);
  const [health, setHealth] = useState<HealthPayload | null>(null);
@@ -137,12 +166,15 @@ export function SystemPageClient() {
          <CardContent>
            <dl className="space-y-2 text-sm">
              {status?.features
-                ? Object.entries(status.features).map(([k, v]) => (
-                    <div key={k} className="flex justify-between">
-                      <span className="text-slate-400">{k}</span>
-                      <Badge variant={v ? "default" : "secondary"}>{String(v)}</Badge>
-                    </div>
-                  ))
+                ? Object.entries(status.features).map(([k, v]) => {
+                    const formatted = formatOpsValue(v);
+                    return (
+                      <div key={k} className="flex justify-between">
+                        <span className="text-slate-400">{k}</span>
+                        <Badge variant={formatted.active ? "default" : "secondary"}>{formatted.label}</Badge>
+                      </div>
+                    );
+                  })
                : <span className="text-slate-500">无数据</span>}
            </dl>
          </CardContent>
@@ -155,12 +187,15 @@ export function SystemPageClient() {
          <CardContent>
            <dl className="space-y-2 text-sm">
              {status?.integrations
-                ? Object.entries(status.integrations).map(([k, v]) => (
-                    <div key={k} className="flex justify-between">
-                      <span className="text-slate-400">{k}</span>
-                      <Badge variant={v ? "default" : "secondary"}>{String(v)}</Badge>
-                    </div>
-                  ))
+                ? Object.entries(status.integrations).map(([k, v]) => {
+                    const formatted = formatOpsValue(v);
+                    return (
+                      <div key={k} className="flex justify-between">
+                        <span className="text-slate-400">{k}</span>
+                        <Badge variant={formatted.active ? "default" : "secondary"}>{formatted.label}</Badge>
+                      </div>
+                    );
+                  })
                : <span className="text-slate-500">无数据</span>}
            </dl>
          </CardContent>
@@ -174,7 +209,7 @@ export function SystemPageClient() {
            <CardTitle>缓存分析</CardTitle>
          </CardHeader>
          <CardContent>
-            <div className="grid grid-cols-2 sm:grid-cols-4 gap-4 text-sm">
+            <div className="grid grid-cols-2 gap-4 text-sm sm:grid-cols-5">
              <div>
                <div className="text-slate-500">总请求</div>
                <div className="text-lg font-bold text-white">{cacheAnalysis.total_requests ?? 0}</div>
@@ -187,6 +222,10 @@ export function SystemPageClient() {
                <div className="text-slate-500">未命中</div>
                <div className="text-lg font-bold text-amber-400">{cacheAnalysis.cache_misses ?? 0}</div>
              </div>
+              <div>
+                <div className="text-slate-500">强制刷新</div>
+                <div className="text-lg font-bold text-blue-500">{cacheAnalysis.force_refresh_requests ?? 0}</div>
+              </div>
              <div>
                <div className="text-slate-500">命中率</div>
                <div className="text-lg font-bold text-cyan-400">
@@ -194,6 +233,9 @@ export function SystemPageClient() {
                </div>
              </div>
            </div>
+            <div className="mt-3 text-xs text-slate-500">
+              这里统计当前后端进程的分析缓存请求；部署重启后会重新累计。
+            </div>
          </CardContent>
        </Card>
      ) : null}
@@ -243,7 +285,9 @@ export function SystemPageClient() {
                      </td>
                      <td className="px-3 py-2 font-mono text-slate-600">{formatAge(source.age_min)}</td>
                      <td className="px-3 py-2 font-mono text-slate-600">{source.observed_at || "—"}</td>
-                      <td className="px-3 py-2 text-slate-500">{source.reason || "—"}</td>
+                      <td className="px-3 py-2 text-slate-500" title={source.reason || ""}>
+                        {sourceReasonLabel(source.reason)}
+                      </td>
                    </tr>
                  ))}
                </tbody>
@@ -97,9 +97,15 @@ export type PaymentIncident = {
  id: number;
  event_type?: string;
  reason?: string;
+  detail?: string;
+  intent_id?: string;
+  user_id?: string;
+  tx_hash?: string;
  payload_json?: string;
  created_at?: string;
  resolved?: boolean;
+  resolved_at?: string;
+  resolved_by?: string;
 };

 export type IncidentsPayload = {
@@ -6,6 +6,7 @@ class FakeRedis:
    def __init__(self):
        self.counter = 0
        self.entries = []
+        self.full_xrange_calls = 0

    def eval(self, _script, _numkeys, stream_key, counter_key, *args):
        (
@@ -48,7 +49,32 @@ class FakeRedis:
        return None

    def xrange(self, stream_key, min="-", max="+", count=None):
+        if min == "-" and max == "+" and count is None:
+            self.full_xrange_calls += 1
        rows = list(self.entries)
+        if isinstance(min, str) and min.startswith("("):
+            exclusive = min[1:]
+            rows = [row for row in rows if row[0] > exclusive]
+        elif min not in ("-", None):
+            rows = [row for row in rows if row[0] >= min]
+        if max not in ("+", None):
+            rows = [row for row in rows if row[0] <= max]
+        if count is not None:
+            rows = rows[: int(count)]
+        return rows
+
+    def xlen(self, stream_key):
+        return len(self.entries)
+
+    def xrevrange(self, stream_key, max="+", min="-", count=None):
+        rows = list(reversed(self.entries))
+        if isinstance(max, str) and max.startswith("("):
+            exclusive = max[1:]
+            rows = [row for row in rows if row[0] < exclusive]
+        elif max not in ("+", None):
+            rows = [row for row in rows if row[0] <= max]
+        if min not in ("-", None):
+            rows = [row for row in rows if row[0] >= min]
        if count is not None:
            rows = rows[: int(count)]
        return rows
@@ -123,3 +149,16 @@ def test_redis_event_store_reports_replay_gap_when_limit_is_exceeded():
        replay_count=len(replay),
        limit=2,
    )
+
+
+def test_redis_replay_after_known_revision_does_not_scan_full_stream():
+    fake = FakeRedis()
+    store = RedisRealtimeEventStore(redis_client=fake, maxlen=50, producer_id="test")
+
+    for idx in range(20):
+        store.append_event(_event("taipei", 30.0 + idx))
+
+    replay = store.replay_events(cities={"taipei"}, since_revision=18, limit=5)
+
+    assert [event["revision"] for event in replay] == [19, 20]
+    assert fake.full_xrange_calls == 0
@@ -13,6 +13,7 @@ import web.scan_terminal_cache as scan_terminal_cache
 import web.scan_terminal_service as scan_terminal_service
 import web.services.city_api as city_api
 import web.services.city_runtime as city_runtime
+from web.services.observation_freshness import build_observation_freshness
 from web.scan_terminal_cache import scan_terminal_cache_key
 from src.database.runtime_state import TruthRecordRepository

@@ -56,6 +57,21 @@ def test_system_status_returns_summary_shape():
    assert 'cities_count' in payload


+def test_observation_freshness_accepts_epoch_seconds():
+    now = datetime.fromtimestamp(1780169100, tz=timezone.utc)
+
+    payload = build_observation_freshness(
+        source_code="mgm",
+        observed_at=1780168800,
+        now_utc=now,
+    )
+
+    assert payload["freshness_status"] == "fresh"
+    assert payload["freshness_reason"] == "within_native_fresh_window"
+    assert payload["age_sec"] == 300
+    assert payload["observed_at"].startswith("2026-")
+
+
 def test_metrics_endpoint_returns_prometheus_payload():
    response = client.get('/metrics')
    assert response.status_code == 200
@@ -341,6 +357,116 @@ def test_ops_billing_risk_surfaces_trial_payment_referral_and_points(monkeypatch
    }.issubset({issue["category"] for issue in payload["issues"]})


+def test_ops_billing_risk_does_not_flag_signup_when_backend_trial_exists(monkeypatch):
+    from src.database.db_manager import DBManager
+
+    now = datetime.now(timezone.utc)
+    recent = now.isoformat()
+
+    def fake_supabase_rows(table, params, *, timeout=10):
+        if table == "trial_claims":
+            return [
+                {
+                    "id": 31,
+                    "user_id": "user-with-trial",
+                    "email": "trial@example.com",
+                    "telegram_user_id": None,
+                    "claimed_at": recent,
+                    "created_at": recent,
+                }
+            ]
+        if table == "subscriptions":
+            return [
+                {
+                    "id": 41,
+                    "user_id": "user-with-trial",
+                    "plan_code": "signup_trial_3d",
+                    "source": "signup_trial",
+                    "status": "active",
+                    "starts_at": recent,
+                    "expires_at": (now + timedelta(days=3)).isoformat(),
+                    "created_at": recent,
+                }
+            ]
+        return []
+
+    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
+    monkeypatch.setattr(ops_api, "_supabase_rest_rows", fake_supabase_rows)
+    monkeypatch.setattr(
+        DBManager,
+        "list_app_analytics_events",
+        lambda self, limit=20000, since_iso=None: [
+            {
+                "id": 51,
+                "event_type": "signup_success",
+                "user_id": "user-with-trial",
+                "client_id": "",
+                "session_id": "session-trial",
+                "created_at": recent,
+                "payload": {"user_id": "user-with-trial"},
+            }
+        ],
+    )
+    monkeypatch.setattr(
+        DBManager,
+        "list_payment_audit_events",
+        lambda self, limit=50, event_type=None: [],
+    )
+
+    payload = ops_api.get_ops_billing_risk(None, days=30, limit=20)
+
+    assert payload["summary"]["trial_gaps"] == 0
+    assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
+
+
+def test_ops_payment_incidents_expose_top_level_reason_and_filters_resolved(monkeypatch):
+    from src.database.db_manager import DBManager
+
+    recent = datetime.now(timezone.utc).isoformat()
+
+    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
+    monkeypatch.setattr(
+        DBManager,
+        "list_payment_audit_events",
+        lambda self, limit=50, event_type=None: [
+            {
+                "id": 71,
+                "event_type": "payment_intent_failed",
+                "created_at": recent,
+                "payload": {
+                    "reason": "receiver_mismatch",
+                    "detail": "receiver address differs",
+                    "intent_id": "intent-71",
+                    "user_id": "user-71",
+                    "tx_hash": "0x" + "7" * 64,
+                },
+            },
+            {
+                "id": 72,
+                "event_type": "payment_intent_failed",
+                "created_at": recent,
+                "payload": {
+                    "reason": "receiver_mismatch",
+                    "resolved_at": recent,
+                    "resolved_by": "ops@example.com",
+                },
+            },
+        ],
+    )
+
+    payload = ops_api.list_ops_payment_incidents(None, limit=20)
+
+    assert len(payload["incidents"]) == 1
+    incident = payload["incidents"][0]
+    assert incident["id"] == 71
+    assert incident["reason"] == "receiver_mismatch"
+    assert incident["detail"] == "receiver address differs"
+    assert incident["intent_id"] == "intent-71"
+    assert incident["user_id"] == "user-71"
+    assert incident["tx_hash"].startswith("0x777")
+    assert incident["resolved"] is False
+
+
 def test_cities_endpoint_uses_denver_display_name_for_aurora_market():
    response = client.get("/api/cities")
    assert response.status_code == 200
@@ -392,6 +518,55 @@ def test_cities_endpoint_does_not_block_on_recent_deb_index(monkeypatch):
    assert denver["deb_recent_sample_count"] == 0


+def test_city_detail_batch_endpoint_builds_multiple_cached_details(monkeypatch):
+    calls = []
+
+    monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
+    monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
+    monkeypatch.setattr(
+        city_api.legacy_routes,
+        "_city_cache_is_fresh",
+        lambda entry, ttl: True,
+    )
+    monkeypatch.setattr(
+        city_api.legacy_routes,
+        "_overlay_latest_wunderground_current",
+        lambda city, payload: {**payload, "overlay_city": city},
+    )
+
+    class FakeCache:
+        def get_city_cache(self, kind, city):
+            assert kind == "full"
+            return {
+                "payload": {
+                    "city": city,
+                    "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
+                }
+            }
+
+    def build_detail(data, market_slug, target_date, resolution):
+        calls.append((data["city"], resolution))
+        return {
+            "city": data["city"],
+            "hourly": data["hourly"],
+            "resolution": resolution,
+            "overlay_city": data["overlay_city"],
+        }
+
+    monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
+    monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
+
+    response = client.get("/api/cities/detail-batch?cities=Shanghai,Paris&resolution=10m")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["cities"] == ["shanghai", "paris"]
+    assert sorted(payload["details"]) == ["paris", "shanghai"]
+    assert payload["details"]["shanghai"]["resolution"] == "10m"
+    assert payload["details"]["paris"]["overlay_city"] == "paris"
+    assert calls == [("shanghai", "10m"), ("paris", "10m")]
+
+
 def test_payment_runtime_endpoint_returns_shape():
    response = client.get('/api/payments/runtime')
    assert response.status_code == 200
@@ -20,6 +20,8 @@ DEFAULT_COUNTER_KEY = "counter:city_observation_revision"
 DEFAULT_MAXLEN = 50000
 DEFAULT_SOCKET_TIMEOUT_SECONDS = 15.0
 DEFAULT_SOCKET_CONNECT_TIMEOUT_SECONDS = 5.0
+DEFAULT_REPLAY_CHUNK_SIZE = 512
+DEFAULT_REPLAY_SCAN_MAX = 5000

 APPEND_EVENT_SCRIPT = """
 local revision = redis.call('INCR', KEYS[2])
@@ -69,6 +71,14 @@ def _float_env(name: str, default: float) -> float:
        return default


+def _int_env(name: str, default: int, *, min_value: int, max_value: int) -> int:
+    try:
+        value = int(os.getenv(name) or default)
+    except (TypeError, ValueError):
+        value = default
+    return max(min_value, min(max_value, int(value)))
+
+
 class RedisRealtimeEventStore:
    """Persist replayable observation patch events in a Redis Stream."""

@@ -87,6 +97,18 @@ class RedisRealtimeEventStore:
        self.stream_key = stream_key or os.getenv("POLYWEATHER_REDIS_STREAM_KEY") or DEFAULT_STREAM_KEY
        self.counter_key = counter_key or os.getenv("POLYWEATHER_REDIS_COUNTER_KEY") or DEFAULT_COUNTER_KEY
        self.maxlen = max(1, int(maxlen or os.getenv("POLYWEATHER_REDIS_STREAM_MAXLEN") or DEFAULT_MAXLEN))
+        self.replay_chunk_size = _int_env(
+            "POLYWEATHER_REDIS_REPLAY_CHUNK_SIZE",
+            DEFAULT_REPLAY_CHUNK_SIZE,
+            min_value=50,
+            max_value=2000,
+        )
+        self.replay_scan_max = _int_env(
+            "POLYWEATHER_REDIS_REPLAY_SCAN_MAX",
+            DEFAULT_REPLAY_SCAN_MAX,
+            min_value=500,
+            max_value=50000,
+        )
        self.producer_id = producer_id or os.getenv("POLYWEATHER_INSTANCE_ID") or socket.gethostname()
        self._client = redis_client or self._build_client(redis_url)
        self._subscriber_lock = threading.Lock()
@@ -157,7 +179,9 @@ class RedisRealtimeEventStore:
        revision = _int_or_zero(_decode(value))
        if revision:
            return revision
-        return max((event["revision"] for event in self._all_events()), default=0)
+        rows = self._client.xrevrange(self.stream_key, max="+", min="-", count=1)
+        events = self._rows_to_events(rows)
+        return max((event["revision"] for event in events), default=0)

    def status(self) -> Dict[str, Any]:
        out: Dict[str, Any] = {
@@ -180,9 +204,7 @@ class RedisRealtimeEventStore:
            xlen = getattr(self._client, "xlen", None)
            if callable(xlen):
                out["stream_len"] = int(xlen(self.stream_key))
-            events = self._all_events()
-            if events:
-                out["oldest_revision"] = min(int(event["revision"]) for event in events)
+            out["oldest_revision"] = self._oldest_revision()
        except Exception as exc:
            out["error"] = str(exc)
        return out
@@ -197,16 +219,17 @@ class RedisRealtimeEventStore:
        city_set = _normalize_city_set(cities)
        since = max(0, int(since_revision or 0))
        bounded_limit = max(1, min(MAX_REPLAY_LIMIT, int(limit or 1)))
-        replay: List[Dict[str, Any]] = []
-        for event in self._all_events():
-            if int(event.get("revision") or 0) <= since:
-                continue
-            if city_set and str(event.get("city") or "").strip().lower() not in city_set:
-                continue
-            replay.append(event)
-            if len(replay) >= bounded_limit:
-                break
-        return replay
+        if since <= 0:
+            return self._replay_events_forward(
+                city_set=city_set,
+                since=since,
+                limit=bounded_limit,
+            )[0]
+        return self._replay_events_reverse(
+            city_set=city_set,
+            since=since,
+            limit=bounded_limit,
+        )[0]

    def replay_requires_resync(
        self,
@@ -218,21 +241,29 @@ class RedisRealtimeEventStore:
    ) -> bool:
        city_set = _normalize_city_set(cities)
        since = max(0, int(since_revision or 0))
-        matching_events = [
-            event
-            for event in self._all_events()
-            if not city_set or str(event.get("city") or "").strip().lower() in city_set
-        ]
-        if not matching_events:
-            return False
-
-        min_revision = min(int(event["revision"]) for event in matching_events)
-        if since > 0 and since < min_revision - 1:
+        oldest_revision = self._oldest_revision()
+        if since > 0 and oldest_revision and since < oldest_revision - 1:
            return True
        bounded_limit = max(1, int(limit or 1))
        if int(replay_count or 0) < bounded_limit:
            return False
-        return sum(1 for event in matching_events if int(event["revision"]) > since) > bounded_limit
+        probe_limit = min(MAX_REPLAY_LIMIT + 1, bounded_limit + 1)
+        if since <= 0:
+            probe_events, _, _ = self._replay_events_forward(
+                city_set=city_set,
+                since=since,
+                limit=probe_limit,
+            )
+            return len(probe_events) > bounded_limit
+
+        probe_events, hit_boundary, scanned = self._replay_events_reverse(
+            city_set=city_set,
+            since=since,
+            limit=probe_limit,
+        )
+        if len(probe_events) > bounded_limit:
+            return True
+        return not hit_boundary and scanned >= self.replay_scan_max

    def start_live_subscription(self, callback: Callable[[Dict[str, Any]], None]) -> None:
        with self._subscriber_lock:
@@ -273,8 +304,81 @@ class RedisRealtimeEventStore:

    def _all_events(self) -> List[Dict[str, Any]]:
        rows = self._client.xrange(self.stream_key, min="-", max="+")
+        return self._rows_to_events(rows)
+
+    def _oldest_revision(self) -> Optional[int]:
+        rows = self._client.xrange(self.stream_key, min="-", max="+", count=1)
+        events = self._rows_to_events(rows)
+        if not events:
+            return None
+        return int(events[0].get("revision") or 0) or None
+
+    def _rows_to_events(self, rows: Any) -> List[Dict[str, Any]]:
        return [self._entry_to_event(entry_id, fields) for entry_id, fields in rows or []]

+    @staticmethod
+    def _matches_city(event: Dict[str, Any], city_set: Set[str]) -> bool:
+        return not city_set or str(event.get("city") or "").strip().lower() in city_set
+
+    def _replay_events_forward(
+        self,
+        *,
+        city_set: Set[str],
+        since: int,
+        limit: int,
+    ) -> tuple[List[Dict[str, Any]], bool, int]:
+        replay: List[Dict[str, Any]] = []
+        scanned = 0
+        min_id = "-"
+        hit_boundary = False
+        while len(replay) < limit and scanned < self.replay_scan_max:
+            count = min(self.replay_chunk_size, self.replay_scan_max - scanned)
+            rows = self._client.xrange(self.stream_key, min=min_id, max="+", count=count)
+            if not rows:
+                hit_boundary = True
+                break
+            scanned += len(rows)
+            for entry_id, fields in rows:
+                event = self._entry_to_event(entry_id, fields)
+                if int(event.get("revision") or 0) <= since:
+                    continue
+                if self._matches_city(event, city_set):
+                    replay.append(event)
+                    if len(replay) >= limit:
+                        break
+            min_id = f"({_decode(rows[-1][0])}"
+        return replay, hit_boundary, scanned
+
+    def _replay_events_reverse(
+        self,
+        *,
+        city_set: Set[str],
+        since: int,
+        limit: int,
+    ) -> tuple[List[Dict[str, Any]], bool, int]:
+        replay_desc: List[Dict[str, Any]] = []
+        scanned = 0
+        max_id = "+"
+        hit_boundary = False
+        while len(replay_desc) < limit and scanned < self.replay_scan_max:
+            count = min(self.replay_chunk_size, self.replay_scan_max - scanned)
+            rows = self._client.xrevrange(self.stream_key, max=max_id, min="-", count=count)
+            if not rows:
+                hit_boundary = True
+                break
+            scanned += len(rows)
+            for entry_id, fields in rows:
+                event = self._entry_to_event(entry_id, fields)
+                if int(event.get("revision") or 0) <= since:
+                    hit_boundary = True
+                    return list(reversed(replay_desc)), hit_boundary, scanned
+                if self._matches_city(event, city_set):
+                    replay_desc.append(event)
+                    if len(replay_desc) >= limit:
+                        break
+            max_id = f"({_decode(rows[-1][0])}"
+        return list(reversed(replay_desc)), hit_boundary, scanned
+
    @staticmethod
    def _entry_to_event(entry_id: Any, fields: Dict[Any, Any]) -> Dict[str, Any]:
        normalized = {_decode(key): _decode(value) for key, value in dict(fields or {}).items()}
@@ -5,6 +5,7 @@ from typing import Any, Dict, List, Optional
 from fastapi import APIRouter, BackgroundTasks, Query, Request

 from web.services.city_api import (
+    get_city_detail_batch_payload,
    get_city_detail_aggregate_payload,
    get_city_detail_payload,
    get_city_summary_payload,
@@ -104,6 +105,27 @@ async def cities_model_range(
    return {"cities": rows}


+@router.get("/api/cities/detail-batch")
+async def city_detail_batch(
+    request: Request,
+    cities: str = "",
+    force_refresh: bool = False,
+    market_slug: Optional[str] = None,
+    target_date: Optional[str] = None,
+    resolution: Optional[str] = "10m",
+    limit: int = 12,
+):
+    return await get_city_detail_batch_payload(
+        request,
+        cities=cities,
+        force_refresh=force_refresh,
+        market_slug=market_slug,
+        target_date=target_date,
+        resolution=resolution,
+        limit=limit,
+    )
+
+
@router.get("/api/city/{name}")
 async def city_detail(
    request: Request,
@@ -99,8 +99,21 @@ def add_signal(

 def parse_utc_datetime(value: Any) -> Optional[datetime]:
    raw = str(value or "").strip()
-    if not raw or "T" not in raw:
+    if not raw:
        return None
+    if "T" not in raw:
+        try:
+            epoch = float(raw)
+        except Exception:
+            return None
+        if epoch <= 1_000_000_000:
+            return None
+        if epoch > 10_000_000_000:
+            epoch = epoch / 1000.0
+        try:
+            return datetime.fromtimestamp(epoch, tz=timezone.utc)
+        except Exception:
+            return None
    try:
        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except Exception:
@@ -3,9 +3,10 @@
 from __future__ import annotations

 import os
+import asyncio
 import threading
 import time
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional, Tuple

 from fastapi import HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
@@ -253,4 +254,95 @@ async def get_city_detail_aggregate_payload(
    )


+def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
+    seen = set()
+    out: List[str] = []
+    for item in str(raw_cities or "").split(","):
+        raw = item.strip()
+        if not raw:
+            continue
+        city = legacy_routes._normalize_city_or_404(raw)
+        if city in seen:
+            continue
+        seen.add(city)
+        out.append(city)
+        if len(out) >= limit:
+            break
+    return out
+
+
+def _build_city_detail_batch_item(
+    city: str,
+    *,
+    force_refresh: bool,
+    market_slug: Optional[str],
+    target_date: Optional[str],
+    resolution: Optional[str],
+) -> Tuple[str, Dict[str, Any]]:
+    if force_refresh:
+        data = legacy_routes._refresh_city_full_cache(city, True)
+    else:
+        cached_entry = legacy_routes._CACHE_DB.get_city_cache("full", city)
+        if cached_entry and legacy_routes._city_cache_is_fresh(
+            cached_entry,
+            legacy_routes.CITY_FULL_CACHE_TTL_SEC,
+        ):
+            data = legacy_routes._overlay_latest_wunderground_current(
+                city,
+                cached_entry.get("payload") or {},
+            )
+        else:
+            data = legacy_routes._refresh_city_full_cache(city, False)
+
+    detail = legacy_routes._build_city_detail_payload(
+        data,
+        market_slug,
+        target_date,
+        resolution,
+    )
+    return city, detail
+
+
+async def get_city_detail_batch_payload(
+    request: Request,
+    *,
+    cities: str,
+    force_refresh: bool = False,
+    market_slug: Optional[str] = None,
+    target_date: Optional[str] = None,
+    resolution: Optional[str] = "10m",
+    limit: int = 12,
+) -> Dict[str, Any]:
+    legacy_routes._assert_entitlement(request)
+    city_names = _parse_batch_city_names(cities, limit=max(1, min(24, int(limit or 12))))
+    if not city_names:
+        return {"cities": [], "details": {}, "errors": {}}
+
+    tasks = [
+        run_in_threadpool(
+            _build_city_detail_batch_item,
+            city,
+            force_refresh=force_refresh,
+            market_slug=market_slug,
+            target_date=target_date,
+            resolution=resolution,
+        )
+        for city in city_names
+    ]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    details: Dict[str, Any] = {}
+    errors: Dict[str, str] = {}
+    for city, result in zip(city_names, results):
+        if isinstance(result, Exception):
+            errors[city] = str(result)
+            continue
+        result_city, payload = result
+        details[result_city] = payload
+
+    return {
+        "cities": city_names,
+        "details": details,
+        "errors": errors,
+    }
+

@@ -380,6 +380,51 @@ def get_ops_memberships_overview(
    }


+def _normalize_payment_incident(item: Dict[str, Any]) -> Dict[str, Any]:
+    payload = item.get("payload") if isinstance(item, dict) else {}
+    payload = payload if isinstance(payload, dict) else {}
+    confirm_failure = (
+        payload.get("confirm_failure")
+        if isinstance(payload.get("confirm_failure"), dict)
+        else {}
+    )
+    reason = str(
+        payload.get("reason")
+        or confirm_failure.get("reason")
+        or payload.get("error")
+        or "unknown"
+    ).strip().lower()
+    detail = str(
+        payload.get("detail")
+        or confirm_failure.get("detail")
+        or payload.get("message")
+        or payload.get("error")
+        or ""
+    ).strip()
+    resolved_at = str(payload.get("resolved_at") or "").strip()
+    return {
+        **item,
+        "payload": payload,
+        "reason": reason or "unknown",
+        "detail": detail,
+        "intent_id": str(
+            payload.get("intent_id")
+            or payload.get("payment_intent_id")
+            or confirm_failure.get("intent_id")
+            or ""
+        ).strip(),
+        "user_id": str(payload.get("user_id") or "").strip(),
+        "tx_hash": str(
+            payload.get("tx_hash")
+            or confirm_failure.get("tx_hash")
+            or ""
+        ).strip(),
+        "resolved": bool(resolved_at),
+        "resolved_at": resolved_at,
+        "resolved_by": str(payload.get("resolved_by") or "").strip(),
+    }
+
+
 def list_ops_payment_incidents(
    request: Request,
    limit: int = 50,
@@ -395,16 +440,15 @@ def list_ops_payment_incidents(
    normalized_reason = str(reason or "").strip().lower()
    filtered = []
    for item in incidents:
-        payload = item.get("payload") if isinstance(item, dict) else {}
-        payload = payload if isinstance(payload, dict) else {}
-        item_reason = str(payload.get("reason") or "").strip().lower()
-        resolved_at = str(payload.get("resolved_at") or "").strip()
+        normalized_item = _normalize_payment_incident(item)
+        item_reason = str(normalized_item.get("reason") or "").strip().lower()
+        resolved = bool(normalized_item.get("resolved"))
        if normalized_reason and item_reason != normalized_reason:
            continue
-        if not include_resolved and resolved_at:
+        if not include_resolved and resolved:
            continue
-        filtered.append(item)
-    return {"incidents": filtered}
+        filtered.append(normalized_item)
+    return {"incidents": filtered, "total": len(filtered)}


 def resolve_ops_payment_incident(request: Request, event_id: int) -> Dict[str, Any]:
@@ -498,7 +542,28 @@ def get_ops_billing_risk(
        {
            "select": "id,user_id,email,telegram_user_id,claimed_at,created_at",
            "order": "created_at.desc",
-            "limit": str(min(safe_limit, 80)),
+            "limit": str(max(safe_limit * 10, 500)),
+        },
+    )
+    subscription_rows = collect(
+        "subscriptions",
+        {
+            "select": (
+                "id,user_id,plan_code,source,status,starts_at,expires_at,"
+                "created_at,updated_at"
+            ),
+            "or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d,status.eq.active)",
+            "order": "created_at.desc",
+            "limit": str(max(safe_limit * 20, 1000)),
+        },
+    )
+    entitlement_trial_events = collect(
+        "entitlement_events",
+        {
+            "select": "id,user_id,action,payload,created_at",
+            "action": "in.(signup_trial_claimed,signup_trial_granted)",
+            "order": "created_at.desc",
+            "limit": str(max(safe_limit * 10, 500)),
        },
    )

@@ -683,22 +748,58 @@ def get_ops_billing_risk(
        if str(row.get("event_type") or "").strip().lower()
        in {"signup_success", "signup_completed"}
    ]
+    def normalize_user_key(value: Any) -> str:
+        return str(value or "").strip().lower()
+
    trial_actor_keys = {
        _app_analytics_actor_key(row)
        for row in events
        if str(row.get("event_type") or "").strip().lower() == "trial_created"
    }
-    trial_gaps: List[Dict[str, Any]] = []
-    for row in signup_rows[:300]:
-        actor_key = _app_analytics_actor_key(row)
-        if actor_key in trial_actor_keys:
-            continue
+    subscription_user_keys = {
+        normalize_user_key(row.get("user_id"))
+        for row in subscription_rows
+        if normalize_user_key(row.get("user_id"))
+    }
+    trial_subscription_user_keys = {
+        normalize_user_key(row.get("user_id"))
+        for row in subscription_rows
+        if normalize_user_key(row.get("user_id"))
+        and (
+            str(row.get("plan_code") or "").strip().lower() == "signup_trial_3d"
+            or str(row.get("source") or "").strip().lower() == "signup_trial"
+        )
+    }
+    trial_claim_user_keys = {
+        normalize_user_key(row.get("user_id"))
+        for row in trial_claims
+        if normalize_user_key(row.get("user_id"))
+    }
+    trial_event_user_keys: set[str] = set()
+    for row in entitlement_trial_events:
+        event_user_id = normalize_user_key(row.get("user_id"))
        payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
+        payload_user_id = normalize_user_key(payload.get("user_id"))
+        if event_user_id:
+            trial_event_user_keys.add(event_user_id)
+        if payload_user_id:
+            trial_event_user_keys.add(payload_user_id)
+
+    backend_trial_user_keys = (
+        trial_subscription_user_keys | trial_claim_user_keys | trial_event_user_keys
+    )
+    trial_gaps: List[Dict[str, Any]] = []
+
+    for claim in trial_claims:
+        claim_user_id = normalize_user_key(claim.get("user_id"))
+        if not claim_user_id or claim_user_id in trial_subscription_user_keys:
+            continue
        gap = {
-            "event_id": row.get("id"),
-            "actor_key": actor_key,
-            "user_id": row.get("user_id") or payload.get("user_id"),
-            "created_at": row.get("created_at"),
+            "claim_id": claim.get("id"),
+            "user_id": claim.get("user_id"),
+            "email": claim.get("email"),
+            "created_at": claim.get("created_at") or claim.get("claimed_at"),
+            "reason": "trial_claim_without_subscription",
        }
        trial_gaps.append(gap)
        if len(trial_gaps) <= 20:
@@ -706,8 +807,46 @@ def get_ops_billing_risk(
                _risk_issue(
                    category="signup_trial",
                    severity="high",
-                    title="注册成功后未记录试用开通",
-                    detail="该用户进入 signup_success，但同窗口内没有 trial_created 事件。",
+                    title="试用 claim 已写入但订阅缺失",
+                    detail="trial_claims 已记录该用户领取试用，但 subscriptions 中没有 signup_trial_3d 记录。",
+                    user_id=gap.get("user_id"),
+                    created_at=gap.get("created_at"),
+                    reference=str(gap.get("claim_id") or ""),
+                    payload=gap,
+                )
+            )
+
+    for row in signup_rows[:300]:
+        actor_key = _app_analytics_actor_key(row)
+        if actor_key in trial_actor_keys:
+            continue
+        payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
+        signup_user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
+        if not signup_user_id:
+            continue
+        if (
+            signup_user_id in backend_trial_user_keys
+            or signup_user_id in subscription_user_keys
+        ):
+            continue
+        gap = {
+            "event_id": row.get("id"),
+            "actor_key": actor_key,
+            "user_id": signup_user_id,
+            "created_at": row.get("created_at"),
+            "reason": "signup_without_backend_trial_evidence",
+        }
+        trial_gaps.append(gap)
+        if len(trial_gaps) <= 20:
+            issues.append(
+                _risk_issue(
+                    category="signup_trial",
+                    severity="high",
+                    title="注册成功后未发现后端试用记录",
+                    detail=(
+                        "该用户进入 signup_success，但没有 trial_created、trial_claims、"
+                        "signup_trial subscription 或其他有效订阅证据。"
+                    ),
                    user_id=gap.get("user_id"),
                    created_at=gap.get("created_at"),
                    reference=str(gap.get("event_id") or ""),