Add observation collector health status

2026-06-08 14:39:49 +08:00
parent 505b75d6ce
commit 96dde3b50e
10 changed files with 725 additions and 4 deletions
@@ -0,0 +1,47 @@
+import { NextRequest, NextResponse } from "next/server";
+import { buildProxyExceptionResponse } from "@/lib/api-proxy";
+import {
+  applyAuthResponseCookies,
+  buildBackendRequestHeaders,
+} from "@/lib/backend-auth";
+import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth";
+
+const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
+
+export async function GET(req: NextRequest) {
+  if (!API_BASE) {
+    return NextResponse.json(
+      { error: "POLYWEATHER_API_BASE_URL is not configured" },
+      { status: 500 },
+    );
+  }
+
+  try {
+    const auth = await buildBackendRequestHeaders(req);
+    const authError = requireOpsProxyAuth(req, auth);
+    if (authError) return authError;
+
+    const upstream = new URL(`${API_BASE}/api/ops/observation-collector-status`);
+    req.nextUrl.searchParams.forEach((value, key) => {
+      upstream.searchParams.set(key, value);
+    });
+
+    const res = await fetch(upstream.toString(), {
+      cache: "no-store",
+      headers: auth.headers,
+    });
+    const raw = await res.text();
+    const response = new NextResponse(raw, {
+      headers: {
+        "Cache-Control": "no-store",
+        "Content-Type": "application/json",
+      },
+      status: res.status,
+    });
+    return applyAuthResponseCookies(response, auth.response);
+  } catch (error) {
+    return buildProxyExceptionResponse(error, {
+      publicMessage: "Observation collector status check failed",
+    });
+  }
+}
@@ -16,6 +16,10 @@ export function runTests() {
    path.join(projectRoot, "app", "api", "ops", "source-health", "route.ts"),
    "utf8",
  );
+  const collectorRoute = fs.readFileSync(
+    path.join(projectRoot, "app", "api", "ops", "observation-collector-status", "route.ts"),
+    "utf8",
+  );

  assert(
    opsApi.includes("sourceHealth") &&
@@ -40,4 +44,23 @@ export function runTests() {
      nextRoute.includes("no-store"),
    "source health proxy must stay ops-admin protected and uncached",
  );
+  assert(
+    opsApi.includes("observationCollectorStatus") &&
+      opsApi.includes("/api/ops/observation-collector-status"),
+    "ops client must expose observation collector status endpoint",
+  );
+  assert(
+    systemPage.includes("观测采集器") &&
+      systemPage.includes("collectorStatus") &&
+      systemPage.includes("failure_count") &&
+      systemPage.includes("last_latency_ms") &&
+      systemPage.includes("冷却"),
+    "ops system page must show observation collector failures, latency, and cooldown status",
+  );
+  assert(
+    collectorRoute.includes("requireOpsProxyAuth") &&
+      collectorRoute.includes("/api/ops/observation-collector-status") &&
+      collectorRoute.includes("no-store"),
+    "observation collector proxy must stay ops-admin protected and uncached",
+  );
 }
@@ -1,12 +1,17 @@
 "use client";

 import { useEffect, useState } from "react";
-import { AlertTriangle, RefreshCcw, ShieldCheck, Database, Cpu, HardDrive, RadioTower } from "lucide-react";
+import { Activity, AlertTriangle, RefreshCcw, ShieldCheck, Database, Cpu, HardDrive, RadioTower } from "lucide-react";
 import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
 import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
 import { opsApi } from "@/lib/ops-api";
-import type { SourceHealthPayload, SystemStatusPayload, HealthPayload } from "@/types/ops";
+import type {
+  ObservationCollectorStatusPayload,
+  SourceHealthPayload,
+  SystemStatusPayload,
+  HealthPayload,
+} from "@/types/ops";

 function sourceStatusTone(status?: string) {
  if (status === "fresh") return "text-emerald-500";
@@ -25,12 +30,48 @@ function sourceStatusLabel(status?: string) {
  return "未知";
 }

+function collectorStatusTone(status?: string) {
+  if (status === "ok") return "text-emerald-500";
+  if (status === "due") return "text-blue-500";
+  if (status === "cooldown") return "text-amber-500";
+  if (status === "failed" || status === "never_run") return "text-red-500";
+  return "text-slate-500";
+}
+
+function collectorStatusLabel(status?: string) {
+  if (status === "ok") return "正常";
+  if (status === "due") return "到期";
+  if (status === "cooldown") return "冷却";
+  if (status === "failed") return "失败";
+  if (status === "never_run") return "未采集";
+  return "未知";
+}
+
 function formatAge(ageMin?: number | null) {
  if (ageMin == null) return "—";
  if (ageMin < 60) return `${Math.round(ageMin)}m`;
  return `${(ageMin / 60).toFixed(1)}h`;
 }

+function formatSeconds(seconds?: number | null) {
+  if (seconds == null) return "—";
+  if (seconds <= 0) return "已到期";
+  if (seconds < 60) return `${Math.round(seconds)}s`;
+  if (seconds < 3600) return `${Math.round(seconds / 60)}m`;
+  return `${(seconds / 3600).toFixed(1)}h`;
+}
+
+function formatLatency(ms?: number | null) {
+  if (ms == null) return "—";
+  if (ms < 1000) return `${Math.round(ms)}ms`;
+  return `${(ms / 1000).toFixed(1)}s`;
+}
+
+function formatTimestamp(value?: string | null) {
+  if (!value) return "—";
+  return value.replace("T", " ").replace("Z", "").slice(0, 19);
+}
+
 function sourceReasonLabel(reason?: string | null) {
  const key = String(reason || "").trim().toLowerCase();
  if (key === "observation_time_missing") return "观测时间缺失";
@@ -65,20 +106,23 @@ export function SystemPageClient() {
  const [health, setHealth] = useState<HealthPayload | null>(null);
  const [status, setStatus] = useState<SystemStatusPayload | null>(null);
  const [sourceHealth, setSourceHealth] = useState<SourceHealthPayload | null>(null);
+  const [collectorStatus, setCollectorStatus] = useState<ObservationCollectorStatusPayload | null>(null);
  const [error, setError] = useState("");

  const load = async () => {
    setLoading(true);
    setError("");
    try {
-      const [h, s, sh] = await Promise.all([
+      const [h, s, sh, cs] = await Promise.all([
        opsApi.health(),
        opsApi.systemStatus() as Promise<SystemStatusPayload>,
        opsApi.sourceHealth(80) as Promise<SourceHealthPayload>,
+        opsApi.observationCollectorStatus(200) as Promise<ObservationCollectorStatusPayload>,
      ]);
      setHealth(h);
      setStatus(s);
      setSourceHealth(sh);
+      setCollectorStatus(cs);
    } catch (e) {
      setError(String(e).slice(0, 200));
    } finally {
@@ -100,6 +144,12 @@ export function SystemPageClient() {

  const dbOk = status?.db?.ok ?? health?.db?.ok;
  const cacheAnalysis = status?.cache?.analysis;
+  const collectorIssues = (collectorStatus?.entries || [])
+    .filter((entry) => {
+      const state = String(entry.status || "");
+      return ["failed", "cooldown", "never_run", "due"].includes(state) || (entry.failure_count ?? 0) > 0;
+    })
+    .slice(0, 12);
  const sourceIssues = (sourceHealth?.cities || [])
    .flatMap((city) =>
      (city.sources || [])
@@ -240,6 +290,96 @@ export function SystemPageClient() {
        </Card>
      ) : null}

+      <Card>
+        <CardHeader>
+          <CardTitle className="flex items-center gap-2">
+            <Activity className="h-4 w-4 text-cyan-500" />
+            观测采集器
+          </CardTitle>
+        </CardHeader>
+        <CardContent>
+          <div className="mb-4 grid grid-cols-2 gap-3 text-sm sm:grid-cols-5">
+            {["ok", "due", "cooldown", "failed", "never_run"].map((key) => (
+              <div key={key} className="rounded-lg border border-slate-200 bg-slate-50 px-3 py-2">
+                <div className="text-[11px] text-slate-500">{collectorStatusLabel(key)}</div>
+                <div className={`text-lg font-black ${collectorStatusTone(key)}`}>
+                  {collectorStatus?.status_counts?.[key] ?? 0}
+                </div>
+              </div>
+            ))}
+          </div>
+
+          {(collectorStatus?.sources || []).length ? (
+            <div className="mb-4 grid grid-cols-1 gap-3 text-xs md:grid-cols-2 xl:grid-cols-3">
+              {(collectorStatus?.sources || []).map((source) => (
+                <div key={source.source} className="rounded-lg border border-slate-200 bg-white px-3 py-2">
+                  <div className="mb-2 flex items-center justify-between gap-2">
+                    <span className="font-mono text-sm font-black text-slate-800">{source.source}</span>
+                    <span className={`font-bold ${collectorStatusTone(source.worst_status)}`}>
+                      {collectorStatusLabel(source.worst_status)}
+                    </span>
+                  </div>
+                  <div className="grid grid-cols-2 gap-x-3 gap-y-1 text-slate-600">
+                    <span>城市 {source.city_count ?? 0}</span>
+                    <span>间隔 {source.min_interval_sec ?? source.interval_sec ?? "—"}s</span>
+                    <span>失败 {source.failure_count ?? 0}</span>
+                    <span>冷却 {source.cooldown_count ?? 0}</span>
+                    <span>延迟 {formatLatency(source.avg_latency_ms)}</span>
+                    <span title={source.last_success_at || ""}>成功 {formatTimestamp(source.last_success_at)}</span>
+                  </div>
+                </div>
+              ))}
+            </div>
+          ) : (
+            <div className="mb-4 rounded-lg border border-slate-200 bg-slate-50 px-3 py-2 text-sm text-slate-500">
+              暂无后台采集状态；collector 首次写入后会显示每个 source/city 的最近采集时间、失败次数、延迟和冷却状态。
+            </div>
+          )}
+
+          {collectorIssues.length ? (
+            <div className="overflow-x-auto rounded-lg border border-slate-200">
+              <table className="w-full min-w-[860px] text-left text-xs">
+                <thead className="bg-slate-50 text-slate-500">
+                  <tr>
+                    <th className="px-3 py-2">Source</th>
+                    <th className="px-3 py-2">城市</th>
+                    <th className="px-3 py-2">状态</th>
+                    <th className="px-3 py-2">最近成功</th>
+                    <th className="px-3 py-2">失败次数</th>
+                    <th className="px-3 py-2">延迟</th>
+                    <th className="px-3 py-2">下次</th>
+                    <th className="px-3 py-2">错误</th>
+                  </tr>
+                </thead>
+                <tbody>
+                  {collectorIssues.map((entry) => (
+                    <tr key={`${entry.source}-${entry.city}`} className="border-t border-slate-100">
+                      <td className="px-3 py-2 font-mono font-bold text-slate-800">{entry.source}</td>
+                      <td className="px-3 py-2 font-mono text-slate-700">{entry.city}</td>
+                      <td className={`px-3 py-2 font-bold ${collectorStatusTone(entry.status)}`}>
+                        {collectorStatusLabel(entry.status)}
+                      </td>
+                      <td className="px-3 py-2 font-mono text-slate-600">{formatTimestamp(entry.last_success_at)}</td>
+                      <td className="px-3 py-2 font-mono text-slate-600">{entry.failure_count ?? 0}</td>
+                      <td className="px-3 py-2 font-mono text-slate-600">{formatLatency(entry.last_latency_ms)}</td>
+                      <td className="px-3 py-2 font-mono text-slate-600">{formatSeconds(entry.due_in_sec)}</td>
+                      <td className="max-w-[260px] truncate px-3 py-2 text-slate-500" title={entry.last_error || ""}>
+                        {entry.last_error || "—"}
+                      </td>
+                    </tr>
+                  ))}
+                </tbody>
+              </table>
+            </div>
+          ) : (
+            <div className="flex items-center gap-2 rounded-lg border border-emerald-200 bg-emerald-50 px-3 py-2 text-sm font-semibold text-emerald-700">
+              <ShieldCheck className="h-4 w-4" />
+              当前后台采集器没有失败、冷却或到期积压。
+            </div>
+          )}
+        </CardContent>
+      </Card>
+
      <Card>
        <CardHeader>
          <CardTitle className="flex items-center gap-2">
@@ -19,6 +19,9 @@ export const opsApi = {
  sourceHealth(limit = 80) {
    return opsFetch<Record<string, unknown>>(`/api/ops/source-health?limit=${limit}`);
  },
+  observationCollectorStatus(limit = 200) {
+    return opsFetch<Record<string, unknown>>(`/api/ops/observation-collector-status?limit=${limit}`);
+  },
  paymentRuntime() {
    return opsFetch<Record<string, unknown>>("/api/payments/runtime");
  },
@@ -84,6 +84,57 @@ export type SourceHealthPayload = {
  total_cities?: number;
 };

+export type ObservationCollectorStatus =
+  | "ok"
+  | "due"
+  | "cooldown"
+  | "failed"
+  | "never_run"
+  | string;
+
+export type ObservationCollectorEntry = {
+  source: string;
+  city: string;
+  interval_sec?: number;
+  last_due_at?: string | null;
+  last_started_at?: string | null;
+  last_success_at?: string | null;
+  last_failure_at?: string | null;
+  last_latency_ms?: number | null;
+  failure_count?: number;
+  last_error?: string | null;
+  updated_at?: string | null;
+  next_due_at?: string | null;
+  due_in_sec?: number | null;
+  age_sec?: number | null;
+  in_cooldown?: boolean;
+  cooldown_until_at?: string | null;
+  status?: ObservationCollectorStatus;
+};
+
+export type ObservationCollectorSourceSummary = {
+  source: string;
+  city_count?: number;
+  interval_sec?: number;
+  min_interval_sec?: number;
+  max_interval_sec?: number;
+  failure_count?: number;
+  cooldown_count?: number;
+  status_counts?: Record<string, number>;
+  avg_latency_ms?: number | null;
+  last_success_at?: string | null;
+  last_failure_at?: string | null;
+  worst_status?: ObservationCollectorStatus;
+};
+
+export type ObservationCollectorStatusPayload = {
+  checked_at?: string;
+  entries?: ObservationCollectorEntry[];
+  sources?: ObservationCollectorSourceSummary[];
+  status_counts?: Record<string, number>;
+  total_entries?: number;
+};
+
 export type PaymentRuntimePayload = {
  rpc?: Record<string, unknown> | string;
  chain_id?: number;
@@ -6,6 +6,7 @@ import os
 import sqlite3
 import threading
 import time
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional

@@ -239,9 +240,264 @@ class RuntimeStateDB:
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_intraday_path_snapshots_city_date ON intraday_path_snapshots_store(city, target_date, id DESC)"
            )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS observation_collector_status_store (
+                    source TEXT NOT NULL,
+                    city TEXT NOT NULL,
+                    interval_sec INTEGER NOT NULL,
+                    last_due_ts REAL,
+                    last_started_ts REAL,
+                    last_success_ts REAL,
+                    last_failure_ts REAL,
+                    last_latency_ms REAL,
+                    failure_count INTEGER NOT NULL DEFAULT 0,
+                    last_error TEXT,
+                    updated_at REAL NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    PRIMARY KEY (source, city)
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_observation_collector_status_source ON observation_collector_status_store(source, updated_at DESC)"
+            )
            conn.commit()


+def _ts_to_utc_iso(value: Any) -> Optional[str]:
+    try:
+        ts = float(value)
+    except (TypeError, ValueError):
+        return None
+    if ts <= 0:
+        return None
+    return datetime.fromtimestamp(ts, timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+class ObservationCollectorStatusRepository:
+    def __init__(self, db: Optional[RuntimeStateDB] = None):
+        self.db = db or RuntimeStateDB.instance()
+
+    def record_result(
+        self,
+        *,
+        source: str,
+        city: str,
+        interval_sec: int,
+        due_ts: float,
+        started_ts: float,
+        completed_ts: float,
+        ok: bool,
+        error: Optional[str] = None,
+    ) -> None:
+        source_key = str(source or "").strip().lower()
+        city_key = str(city or "").strip().lower()
+        if not source_key or not city_key:
+            return
+        safe_interval = max(1, int(interval_sec or 60))
+        due = float(due_ts or completed_ts)
+        started = float(started_ts or due)
+        completed = float(completed_ts or time.time())
+        latency_ms = round(max(0.0, completed - started) * 1000.0, 1)
+        error_text = None if ok else str(error or "no_results").strip()[:500]
+        payload = {
+            "source": source_key,
+            "city": city_key,
+            "ok": bool(ok),
+            "error": error_text,
+            "interval_sec": safe_interval,
+            "due_ts": due,
+            "started_ts": started,
+            "completed_ts": completed,
+        }
+        with self.db.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO observation_collector_status_store (
+                    source, city, interval_sec, last_due_ts, last_started_ts,
+                    last_success_ts, last_failure_ts, last_latency_ms,
+                    failure_count, last_error, updated_at, payload_json
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(source, city) DO UPDATE SET
+                    interval_sec = excluded.interval_sec,
+                    last_due_ts = excluded.last_due_ts,
+                    last_started_ts = excluded.last_started_ts,
+                    last_success_ts = COALESCE(excluded.last_success_ts, observation_collector_status_store.last_success_ts),
+                    last_failure_ts = COALESCE(excluded.last_failure_ts, observation_collector_status_store.last_failure_ts),
+                    last_latency_ms = excluded.last_latency_ms,
+                    failure_count = CASE
+                        WHEN excluded.last_failure_ts IS NOT NULL
+                        THEN observation_collector_status_store.failure_count + 1
+                        ELSE observation_collector_status_store.failure_count
+                    END,
+                    last_error = excluded.last_error,
+                    updated_at = excluded.updated_at,
+                    payload_json = excluded.payload_json
+                """,
+                (
+                    source_key,
+                    city_key,
+                    safe_interval,
+                    due,
+                    started,
+                    completed if ok else None,
+                    completed if not ok else None,
+                    latency_ms,
+                    0 if ok else 1,
+                    error_text,
+                    completed,
+                    json.dumps(payload, ensure_ascii=False),
+                ),
+            )
+            conn.commit()
+
+    def load_snapshot(self, *, now_ts: Optional[float] = None, limit: int = 500) -> Dict[str, Any]:
+        now = float(time.time() if now_ts is None else now_ts)
+        safe_limit = max(1, min(int(limit or 500), 1000))
+        with self.db.connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT source, city, interval_sec, last_due_ts, last_started_ts,
+                       last_success_ts, last_failure_ts, last_latency_ms,
+                       failure_count, last_error, updated_at
+                FROM observation_collector_status_store
+                ORDER BY source ASC, city ASC
+                """
+            ).fetchall()
+
+        entries: List[Dict[str, Any]] = []
+        status_counts: Dict[str, int] = {}
+        source_summary: Dict[str, Dict[str, Any]] = {}
+
+        for row in rows:
+            source = str(row["source"] or "")
+            city = str(row["city"] or "")
+            interval = max(1, int(row["interval_sec"] or 60))
+            last_due_ts = _float_or_none(row["last_due_ts"])
+            last_started_ts = _float_or_none(row["last_started_ts"])
+            last_success_ts = _float_or_none(row["last_success_ts"])
+            last_failure_ts = _float_or_none(row["last_failure_ts"])
+            updated_at_ts = _float_or_none(row["updated_at"])
+            next_due_ts = last_due_ts + interval if last_due_ts is not None else None
+            due_in_sec = round(next_due_ts - now, 1) if next_due_ts is not None else None
+            latest_failure = (
+                last_failure_ts is not None
+                and (last_success_ts is None or last_failure_ts >= last_success_ts)
+            )
+            in_cooldown = bool(latest_failure and next_due_ts is not None and next_due_ts > now)
+            if last_started_ts is None:
+                status = "never_run"
+            elif latest_failure:
+                status = "cooldown" if in_cooldown else "failed"
+            elif next_due_ts is not None and next_due_ts <= now:
+                status = "due"
+            else:
+                status = "ok"
+
+            latency = _float_or_none(row["last_latency_ms"])
+            failure_count = int(row["failure_count"] or 0)
+            entry = {
+                "source": source,
+                "city": city,
+                "interval_sec": interval,
+                "last_due_ts": last_due_ts,
+                "last_due_at": _ts_to_utc_iso(last_due_ts),
+                "last_started_ts": last_started_ts,
+                "last_started_at": _ts_to_utc_iso(last_started_ts),
+                "last_success_ts": last_success_ts,
+                "last_success_at": _ts_to_utc_iso(last_success_ts),
+                "last_failure_ts": last_failure_ts,
+                "last_failure_at": _ts_to_utc_iso(last_failure_ts),
+                "last_latency_ms": latency,
+                "failure_count": failure_count,
+                "last_error": row["last_error"],
+                "updated_at_ts": updated_at_ts,
+                "updated_at": _ts_to_utc_iso(updated_at_ts),
+                "next_due_ts": next_due_ts,
+                "next_due_at": _ts_to_utc_iso(next_due_ts),
+                "due_in_sec": due_in_sec,
+                "age_sec": round(now - last_success_ts, 1) if last_success_ts is not None else None,
+                "in_cooldown": in_cooldown,
+                "cooldown_until_ts": next_due_ts if in_cooldown else None,
+                "cooldown_until_at": _ts_to_utc_iso(next_due_ts) if in_cooldown else None,
+                "status": status,
+            }
+            entries.append(entry)
+            status_counts[status] = status_counts.get(status, 0) + 1
+
+            summary = source_summary.setdefault(
+                source,
+                {
+                    "source": source,
+                    "city_count": 0,
+                    "interval_sec": interval,
+                    "min_interval_sec": interval,
+                    "max_interval_sec": interval,
+                    "failure_count": 0,
+                    "cooldown_count": 0,
+                    "status_counts": {},
+                    "_latencies": [],
+                    "_last_success_ts": None,
+                    "_last_failure_ts": None,
+                },
+            )
+            summary["city_count"] += 1
+            summary["min_interval_sec"] = min(int(summary["min_interval_sec"]), interval)
+            summary["max_interval_sec"] = max(int(summary["max_interval_sec"]), interval)
+            summary["failure_count"] += failure_count
+            if status == "cooldown":
+                summary["cooldown_count"] += 1
+            summary["status_counts"][status] = summary["status_counts"].get(status, 0) + 1
+            if latency is not None:
+                summary["_latencies"].append(latency)
+            if last_success_ts is not None:
+                current_success = summary["_last_success_ts"]
+                summary["_last_success_ts"] = (
+                    last_success_ts if current_success is None else max(current_success, last_success_ts)
+                )
+            if last_failure_ts is not None:
+                current_failure = summary["_last_failure_ts"]
+                summary["_last_failure_ts"] = (
+                    last_failure_ts if current_failure is None else max(current_failure, last_failure_ts)
+                )
+
+        source_priority = {"failed": 5, "cooldown": 4, "never_run": 3, "due": 2, "ok": 1}
+        sources: List[Dict[str, Any]] = []
+        for summary in source_summary.values():
+            latencies = summary.pop("_latencies")
+            last_success_ts = summary.pop("_last_success_ts")
+            last_failure_ts = summary.pop("_last_failure_ts")
+            summary["avg_latency_ms"] = (
+                round(sum(latencies) / len(latencies), 1) if latencies else None
+            )
+            summary["last_success_at"] = _ts_to_utc_iso(last_success_ts)
+            summary["last_failure_at"] = _ts_to_utc_iso(last_failure_ts)
+            summary["worst_status"] = max(
+                summary["status_counts"],
+                key=lambda key: source_priority.get(str(key), 0),
+                default="unknown",
+            )
+            sources.append(summary)
+
+        return {
+            "checked_at": _ts_to_utc_iso(now),
+            "entries": entries[:safe_limit],
+            "sources": sorted(sources, key=lambda item: str(item.get("source") or "")),
+            "status_counts": status_counts,
+            "total_entries": len(entries),
+        }
+
+
+def _float_or_none(value: Any) -> Optional[float]:
+    try:
+        if value is None or value == "":
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
 class DailyRecordRepository:
    def __init__(self, db: Optional[RuntimeStateDB] = None):
        self.db = db or RuntimeStateDB.instance()
@@ -133,6 +133,132 @@ def test_observation_collector_run_due_once_refreshes_panel_cache():
    assert refreshed == ["qingdao", "qingdao"]


+def test_observation_collector_records_source_status_to_runtime_state(tmp_path):
+    from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB
+    from web.observation_collector_service import (
+        ObservationCollector,
+        ObservationSourceProfile,
+    )
+
+    class FakeWeather:
+        def _uses_fahrenheit(self, city):
+            return False
+
+        def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit):
+            results["amsc_awos"] = {"source": "amsc_awos", "temp_c": 22.8}
+
+    db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
+    status_repo = ObservationCollectorStatusRepository(db)
+    collector = ObservationCollector(
+        weather=FakeWeather(),
+        profiles=[ObservationSourceProfile("amsc_awos", ("beijing",), 180)],
+        status_recorder=status_repo,
+    )
+
+    assert collector.run_due_once(now_ts=1000.0) == 1
+
+    payload = status_repo.load_snapshot(now_ts=1001.0)
+    assert payload["total_entries"] == 1
+    assert payload["status_counts"] == {"ok": 1}
+
+    entry = payload["entries"][0]
+    assert entry["source"] == "amsc_awos"
+    assert entry["city"] == "beijing"
+    assert entry["interval_sec"] == 180
+    assert entry["failure_count"] == 0
+    assert entry["last_error"] is None
+    assert entry["last_success_at"] is not None
+    assert entry["last_failure_at"] is None
+    assert entry["last_latency_ms"] is not None
+    assert entry["next_due_ts"] == 1180.0
+    assert entry["in_cooldown"] is False
+    assert entry["status"] == "ok"
+
+    source = payload["sources"][0]
+    assert source["source"] == "amsc_awos"
+    assert source["city_count"] == 1
+    assert source["failure_count"] == 0
+    assert source["avg_latency_ms"] is not None
+    assert source["status_counts"] == {"ok": 1}
+
+
+def test_observation_collector_records_failure_and_cooldown(tmp_path):
+    from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB
+    from web.observation_collector_service import (
+        ObservationCollector,
+        ObservationSourceProfile,
+    )
+
+    class FakeWeather:
+        def _uses_fahrenheit(self, city):
+            return False
+
+        def _attach_china_amsc_awos_data(self, results, city, use_fahrenheit):
+            raise RuntimeError("upstream timeout")
+
+    db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
+    status_repo = ObservationCollectorStatusRepository(db)
+    collector = ObservationCollector(
+        weather=FakeWeather(),
+        profiles=[ObservationSourceProfile("amsc_awos", ("seoul",), 180)],
+        status_recorder=status_repo,
+    )
+
+    assert collector.run_due_once(now_ts=2000.0) == 0
+
+    payload = status_repo.load_snapshot(now_ts=2010.0)
+    assert payload["total_entries"] == 1
+    assert payload["status_counts"] == {"cooldown": 1}
+
+    entry = payload["entries"][0]
+    assert entry["source"] == "amsc_awos"
+    assert entry["city"] == "seoul"
+    assert entry["failure_count"] == 1
+    assert entry["last_success_at"] is None
+    assert entry["last_failure_at"] is not None
+    assert entry["last_error"] == "upstream timeout"
+    assert entry["next_due_ts"] == 2180.0
+    assert entry["in_cooldown"] is True
+    assert entry["status"] == "cooldown"
+
+    source = payload["sources"][0]
+    assert source["failure_count"] == 1
+    assert source["cooldown_count"] == 1
+
+
+def test_ops_observation_collector_status_returns_runtime_snapshot(monkeypatch, tmp_path):
+    from src.database.runtime_state import ObservationCollectorStatusRepository, RuntimeStateDB
+    from web.services import ops_api
+
+    db = RuntimeStateDB(str(tmp_path / "polyweather.db"))
+    status_repo = ObservationCollectorStatusRepository(db)
+    now = time.time()
+    status_repo.record_result(
+        source="cowin_obs",
+        city="hong kong",
+        interval_sec=60,
+        due_ts=now,
+        started_ts=now,
+        completed_ts=now + 0.25,
+        ok=True,
+    )
+
+    monkeypatch.setattr(
+        ops_api.legacy_routes,
+        "_require_ops_admin",
+        lambda request: {"email": "ops@example.com"},
+    )
+    monkeypatch.setattr(ops_api, "ObservationCollectorStatusRepository", lambda: status_repo)
+
+    payload = ops_api.get_ops_observation_collector_status(object(), limit=10)
+
+    assert payload["total_entries"] == 1
+    assert payload["entries"][0]["source"] == "cowin_obs"
+    assert payload["entries"][0]["city"] == "hong kong"
+    assert payload["sources"][0]["source"] == "cowin_obs"
+    assert payload["status_counts"] == {"ok": 1}
+
+
 def test_observation_collector_worker_entrypoint_exists():
    from web import observation_collector_worker

@@ -14,6 +14,7 @@ from src.data_collection.amos_station_sources import AMOS_AIRPORT_CODES
 from src.data_collection.amsc_awos_sources import AMSC_AWOS_AIRPORTS
 from src.data_collection.city_registry import CITY_REGISTRY
 from src.data_collection.hko_obs_sources import HKO_STATIONS
+from src.database.runtime_state import ObservationCollectorStatusRepository


 def _env_bool(name: str, default: bool) -> bool:
@@ -51,10 +52,12 @@ class ObservationCollector:
        weather: Any,
        profiles: Sequence[ObservationSourceProfile],
        cache_refresher: Optional[Callable[[str], Any]] = None,
+        status_recorder: Optional[ObservationCollectorStatusRepository] = None,
    ) -> None:
        self.weather = weather
        self.profiles = list(profiles)
        self.cache_refresher = cache_refresher
+        self.status_recorder = status_recorder
        self._last_run_ts: dict[tuple[str, str], float] = {}
        self._lock = threading.Lock()

@@ -73,19 +76,70 @@ class ObservationCollector:

        completed = 0
        for profile, city in due:
+            started_wall = time.time()
+            started_ts = now if now_ts is not None else started_wall
+            ok = False
+            error: Optional[str] = None
            try:
-                if self._collect_city_source(profile.source, city):
+                ok = self._collect_city_source(profile.source, city)
+                if ok:
                    completed += 1
                    self._refresh_city_cache(city)
+                else:
+                    error = "no_results"
            except Exception as exc:
+                error = str(exc) or exc.__class__.__name__
                logger.warning(
                    "observation collector source failed source={} city={}: {}",
                    profile.source,
                    city,
                    exc,
                )
+            finally:
+                completed_ts = started_ts + max(0.0, time.time() - started_wall)
+                self._record_source_status(
+                    profile=profile,
+                    city=city,
+                    due_ts=now,
+                    started_ts=started_ts,
+                    completed_ts=completed_ts,
+                    ok=ok,
+                    error=error,
+                )
        return completed

+    def _record_source_status(
+        self,
+        *,
+        profile: ObservationSourceProfile,
+        city: str,
+        due_ts: float,
+        started_ts: float,
+        completed_ts: float,
+        ok: bool,
+        error: Optional[str],
+    ) -> None:
+        if not self.status_recorder:
+            return
+        try:
+            self.status_recorder.record_result(
+                source=profile.source,
+                city=city,
+                interval_sec=profile.interval_sec,
+                due_ts=due_ts,
+                started_ts=started_ts,
+                completed_ts=completed_ts,
+                ok=ok,
+                error=error,
+            )
+        except Exception as exc:
+            logger.warning(
+                "observation collector status write failed source={} city={}: {}",
+                profile.source,
+                city,
+                exc,
+            )
+
    def _collect_city_source(self, source: str, city: str) -> bool:
        normalized_source = str(source or "").strip().lower()
        normalized_city = str(city or "").strip().lower()
@@ -162,6 +216,7 @@ def start_observation_collector_loop(
    weather: Any,
    cache_refresher: Optional[Callable[[str], Any]] = None,
    profiles: Optional[Sequence[ObservationSourceProfile]] = None,
+    status_recorder: Optional[ObservationCollectorStatusRepository] = None,
 ) -> Optional[threading.Thread]:
    if not _env_bool("POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED", True):
        return None
@@ -175,6 +230,7 @@ def start_observation_collector_loop(
        weather=weather,
        profiles=selected_profiles,
        cache_refresher=cache_refresher,
+        status_recorder=status_recorder or ObservationCollectorStatusRepository(),
    )

    global _COLLECTOR_THREAD
@@ -13,6 +13,7 @@ from web.services.ops_api import (
    get_ops_memberships_overview,
    get_ops_health_check,
    get_ops_logs,
+    get_ops_observation_collector_status,
    get_ops_source_health,
    get_ops_truth_history,
    get_ops_weekly_leaderboard,
@@ -266,6 +267,14 @@ async def ops_source_health(
    return get_ops_source_health(request, cities=cities, limit=limit)


+@router.get("/api/ops/observation-collector-status")
+async def ops_observation_collector_status(
+    request: Request,
+    limit: int = 200,
+):
+    return get_ops_observation_collector_status(request, limit=limit)
+
+
@router.get("/api/ops/training/accuracy")
 async def ops_training_accuracy(request: Request):
    return get_ops_training_accuracy(request)
@@ -10,6 +10,7 @@ from fastapi import HTTPException, Request
 import requests as _requests

 from src.database.db_manager import DBManager
+from src.database.runtime_state import ObservationCollectorStatusRepository
 from src.utils.runtime_secrets import get_runtime_secret, get_runtime_secret_status
 from web.services.observation_freshness import (
    build_observation_freshness,
@@ -1960,6 +1961,15 @@ def get_ops_source_health(
    }


+def get_ops_observation_collector_status(
+    request: Request,
+    limit: int = 200,
+) -> dict[str, Any]:
+    _require_ops(request)
+    safe_limit = max(1, min(int(limit or 200), 500))
+    return ObservationCollectorStatusRepository().load_snapshot(limit=safe_limit)
+
+
 def get_ops_health_check(request: Request) -> dict[str, Any]:
    _require_ops(request)
    import os