feat: implement SSE-based real-time event distribution system with replay support and heartbeat functionality

2026-05-26 22:27:55 +08:00
parent 5c6ffa4742
commit 645e304b3e
20 changed files with 1449 additions and 119 deletions
@@ -1,5 +1,8 @@
 {
  "css.validate": false,
  "scss.validate": false,
-  "less.validate": false
+  "less.validate": false,
+  "python.analysis.extraPaths": [
+    "./"
+  ]
 }
@@ -0,0 +1,71 @@
+# Production Realtime SSE Patch Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Upgrade PolyWeather terminal charts from in-process best-effort SSE patches to a replayable, city-scoped, versioned realtime observation stream for PM highest-temperature prediction workflows.
+
+**Architecture:** Keep HTTP APIs as the full snapshot/source-of-truth layer. Add a short-window SQLite event log for SSE replay, version observations as `city_observation_patch.v1`, fan out live events through the existing SSE manager, and let the frontend subscribe only to visible cities with `since_revision` reconnect replay.
+
+**Tech Stack:** FastAPI, SQLite/WAL via `DBManager`, in-process `asyncio.Queue` SSE fanout, React/Next.js `EventSource`, TypeScript external store hooks.
+
+---
+
+## Constraints
+
+- Default replay retention is 6 hours because this event log is not the business history store.
+- The retention can be raised with `POLYWEATHER_PATCH_EVENT_RETENTION_HOURS`, but the product does not need all-day patch retention.
+- First production step is SQLite-only; Redis/Postgres pub/sub remains a later multi-instance extension.
+- Existing legacy `city_patch` ingest and frontend handling must keep working during rollout.
+
+## Tasks
+
+- [ ] Backend schema tests
+  - Add tests proving legacy collector payloads normalize to `city_observation_patch.v1`.
+  - Cover runway point conversion from `amos.runway_obs.point_temperatures`.
+  - Cover invalid payload rejection when city and useful observation data are missing.
+
+- [ ] Backend schema implementation
+  - Add `web/realtime_patch_schema.py`.
+  - Normalize city/source/obs time/temp/max/runway payload fields.
+  - Keep payload small and JSON-serializable.
+
+- [ ] Event store tests
+  - Add tests for monotonic SQLite revisions.
+  - Add city-filtered replay tests.
+  - Add retention cleanup tests for stale replay rows.
+
+- [ ] Event store implementation
+  - Add `observation_patch_events` table and indexes in `DBManager`.
+  - Add `web/realtime_event_store.py` for append, replay, latest revision, and cleanup.
+  - Use `POLYWEATHER_PATCH_EVENT_RETENTION_HOURS=6` as the default.
+
+- [ ] SSE replay tests
+  - Add tests for `/api/events?cities=...&since_revision=...`.
+  - Verify replay only returns subscribed cities.
+  - Verify replay over limit emits `resync_required`.
+
+- [ ] SSE implementation
+  - Update router to parse `cities`, `since_revision`, and bounded `replay_limit`.
+  - Write normalized events to SQLite before broadcasting.
+  - Update manager to track per-connection city subscriptions while keeping heartbeat behavior.
+
+- [ ] Frontend SSE tests
+  - Extend architecture tests for v1 schema, `cities`, `since_revision`, replay/resync handling.
+  - Add chart merge coverage for v1 runway point payloads.
+
+- [ ] Frontend SSE implementation
+  - Update `use-sse-patches.ts` to normalize v1 and legacy patch events.
+  - Track global `lastRevision`.
+  - Reconnect with visible-city `cities` and `since_revision`.
+  - Expose a resync signal for charts when the server cannot replay.
+
+- [ ] Chart/list integration
+  - Ensure visible charts register their city subscription.
+  - Keep terminal row patching lightweight: current temp, current max, local time, revision.
+  - Append v1 temp/runway points into existing chart series without forcing full-detail polling.
+
+- [ ] Verification
+  - Run targeted backend pytest files.
+  - Run `npm run test:business`.
+  - Run `npm run typecheck`.
+  - Run `npm run build`.
@@ -13,7 +13,12 @@ export async function GET(req: NextRequest) {
    );
  }

-  const upstream = await fetch(`${API_BASE.replace(/\/+$/, "")}/api/events`, {
+  const upstreamUrl = new URL(`${API_BASE.replace(/\/+$/, "")}/api/events`);
+  req.nextUrl.searchParams.forEach((value, key) => {
+    upstreamUrl.searchParams.append(key, value);
+  });
+
+  const upstream = await fetch(upstreamUrl.toString(), {
    cache: "no-store",
    headers: {
      Accept: "text/event-stream",
@@ -24,7 +24,7 @@ import type {
 } from "@/lib/dashboard-types";
 import { buildDebBaselinePath } from "@/lib/temperature-chart-paths";
 import { DASHBOARD_REFRESH_POLICY_MS } from "@/lib/refresh-policy";
-import { useLatestPatch, type CityPatch } from "@/hooks/use-sse-patches";
+import { useLatestPatch, useSseResyncVersion, type CityPatch } from "@/hooks/use-sse-patches";
 import { Panel } from "@/components/dashboard/scan-terminal/Panel";
 import { rowName, temp } from "@/components/dashboard/scan-terminal/utils";

@@ -82,6 +82,28 @@ function getVisibleTemperatureSeries(
  });
 }

+function getActiveTemperatureSeries(
+  city: string,
+  chartSeries: EvidenceSeries[],
+  userToggledKeys: Record<string, boolean>,
+  showRunwayDetails: boolean,
+) {
+  const rawVisible = getVisibleTemperatureSeries(city, chartSeries, userToggledKeys);
+  const hasRunwayMax = rawVisible.some((item) => item.key === "runway_max");
+
+  return rawVisible.filter((item) => {
+    const isIndividualRunway =
+      item.key.startsWith("runway_") && item.key !== "runway_max";
+    if (showRunwayDetails) {
+      return item.key !== "runway_max";
+    }
+    if (!hasRunwayMax) {
+      return true;
+    }
+    return !isIndividualRunway;
+  });
+}
+
 function buildRunwayPlates(
  amos: AmosData | null | undefined,
  row: ScanOpportunityRow | null,
@@ -194,8 +216,8 @@ type RunwayHistorySeries = {

 const MAX_OBS_POINTS = 1440;
 const HOURLY_CACHE_TTL_MS = DASHBOARD_REFRESH_POLICY_MS.metar;
-const FULL_DAY_SLOT_MINUTES = 30;
-const FULL_DAY_SLOTS = 48;
+const FULL_DAY_SLOT_MINUTES = 1;
+const FULL_DAY_SLOTS = (24 * 60) / FULL_DAY_SLOT_MINUTES;
 const SLOT_INTERVAL_MS = FULL_DAY_SLOT_MINUTES * 60 * 1000;
 const _hourlyCache = new Map<string, { ts: number; data: HourlyForecast }>();
 const _hourlyRequestCache = new Map<string, Promise<HourlyForecast>>();
@@ -622,7 +644,12 @@ function mergePatchIntoHourly(
  const amosChanges = changes.amos as Record<string, any> | undefined;
  const obsTimeVal = obsTime || amosChanges?.observation_time || amosChanges?.observation_time_local;
  const runwayObs = amosChanges?.runway_obs;
-  if (runwayObs && Array.isArray(runwayObs.point_temperatures) && obsTimeVal) {
+  const runwayPoints = Array.isArray(changes.runway_points)
+    ? changes.runway_points
+    : runwayObs && Array.isArray(runwayObs.point_temperatures)
+      ? runwayObs.point_temperatures
+      : [];
+  if (runwayPoints.length && obsTimeVal) {
    const history: Record<string, Array<Record<string, unknown>>> = {};
    const sourceHistory = next.runwayPlateHistory || (next.amos as any)?.runway_plate_history || {};
    
@@ -634,10 +661,10 @@ function mergePatchIntoHourly(
    });

    // Append new points from point_temperatures
-    runwayObs.point_temperatures.forEach((pt: any) => {
+    runwayPoints.forEach((pt: any) => {
      const rwy = pt.runway || "";
      if (!rwy) return;
-      const tempVal = validNumber(pt.target_runway_max) ?? validNumber(pt.tdz_temp) ?? validNumber(pt.end_temp);
+      const tempVal = validNumber(pt.temp) ?? validNumber(pt.target_runway_max) ?? validNumber(pt.tdz_temp) ?? validNumber(pt.end_temp);
      if (tempVal === null) return;

      const rwyHistory = history[rwy] || [];
@@ -653,6 +680,13 @@ function mergePatchIntoHourly(
    });

    next.runwayPlateHistory = history;
+    next.amos = {
+      ...(next.amos || {}),
+      runway_obs: {
+        ...((next.amos as any)?.runway_obs || {}),
+        point_temperatures: runwayPoints,
+      },
+    } as any;
    if (next.amos) {
      (next.amos as any).runway_plate_history = history;
    }
@@ -1405,13 +1439,14 @@ export function LiveTemperatureThresholdChart({
  const [hourly, setHourly] = useState<HourlyForecast>(null);
  const city = String(row?.city || "").toLowerCase().trim();
  const latestPatch = useLatestPatch(city);
+  const resyncVersion = useSseResyncVersion();
  const [timeframe, setTimeframe] = useState<"1D" | "3D">("1D");
  const [userToggledKeys, setUserToggledKeys] = useState<Record<string, boolean>>({});
  const [liveTemp, setLiveTemp] = useState<number | null>(null);
  const lastPatchAtRef = useRef<number>(Date.now());
  const lastAppliedPatchRevisionRef = useRef<number>(0);

-  const [showRunwayDetails, setShowRunwayDetails] = useState<boolean>(false);
+  const [showRunwayDetails, setShowRunwayDetails] = useState<boolean>(true);
  const [refAreaLeft, setRefAreaLeft] = useState<number | null>(null);
  const [refAreaRight, setRefAreaRight] = useState<number | null>(null);
  const [zoomRange, setZoomRange] = useState<[number, number] | null>(null);
@@ -1420,6 +1455,7 @@ export function LiveTemperatureThresholdChart({
  useEffect(() => {
    setUserToggledKeys({});
    setZoomRange(null);
+    setShowRunwayDetails(true);
    lastPatchAtRef.current = Date.now();
    lastAppliedPatchRevisionRef.current = 0;
  }, [city, timeframe]);
@@ -1480,6 +1516,20 @@ export function LiveTemperatureThresholdChart({
    setHourly((prev) => mergePatchIntoHourly(prev ?? seedHourlyForecastFromRow(row), latestPatch));
  }, [latestPatch, row]);

+  useEffect(() => {
+    if (!resyncVersion || !city) return;
+    let cancelled = false;
+    fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })
+      .then((data) => {
+        if (cancelled || !data) return;
+        setHourly(data);
+      })
+      .catch(() => {});
+    return () => {
+      cancelled = true;
+    };
+  }, [resyncVersion, city, targetResolution]);
+
  // ── SSE fallback: only full-fetch if a visible chart has seen no patch for 2 minutes ──
  useEffect(() => {
    if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return;
@@ -1581,15 +1631,12 @@ export function LiveTemperatureThresholdChart({
  };

  const activeSeries = useMemo(() => {
-    const rawVisible = getVisibleTemperatureSeries(city, chartSeries, userToggledKeys);
-    return rawVisible.filter((s) => {
-      const isIndividualRunway = s.key.startsWith("runway_") && s.key !== "runway_max";
-      if (showRunwayDetails) {
-        return s.key !== "runway_max";
-      } else {
-        return !isIndividualRunway;
-      }
-    });
+    return getActiveTemperatureSeries(
+      city,
+      chartSeries,
+      userToggledKeys,
+      showRunwayDetails,
+    );
  }, [chartSeries, userToggledKeys, city, showRunwayDetails]);

  const normalizedKey = normalizeCityKey(row?.city);
@@ -1903,7 +1950,7 @@ export function LiveTemperatureThresholdChart({
                  </div>
                  <div className="flex flex-col">
                    <span className="text-[11px] font-semibold text-slate-500 uppercase tracking-wider">
-                      {isEn ? "METAR Settlement (30m) · Daily High" : `${metarHeaderLabel} · 当日最高`}
+                      {isEn ? "METAR Settlement · Daily High" : `${metarHeaderLabel} · 当日最高`}
                    </span>
                    <span className="text-2xl font-bold font-mono text-blue-600 mt-1">
                      {temp(observedHighMetar)}
@@ -2226,5 +2273,7 @@ export function __buildTemperatureChartDataForTest(

 export const __isTemperatureSeriesVisibleByDefaultForTest = isTemperatureSeriesVisibleByDefault;
 export const __getVisibleTemperatureSeriesForTest = getVisibleTemperatureSeries;
+export const __getActiveTemperatureSeriesForTest = getActiveTemperatureSeries;
 export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics;
 export const __shouldPollLiveChartForTest = shouldPollLiveChart;
+export const __mergePatchIntoHourlyForTest = mergePatchIntoHourly;
@@ -21,15 +21,35 @@ export function runTests() {
  const sseManager = fs.readFileSync(sseManagerPath, "utf8");
  assert(sseManager.includes("asyncio.Queue"), "SSE manager must keep asyncio.Queue connections");
  assert(sseManager.includes("broadcast("), "SSE manager must expose broadcast(city, changes)");
+  assert(sseManager.includes("broadcast_event"), "SSE manager must broadcast stored replayable events");
  assert(sseManager.includes("event_stream("), "SSE manager must expose an async event_stream(user_id)");
+  assert(sseManager.includes("_queue_cities"), "SSE manager must track per-connection city subscriptions");
  assert(sseManager.includes("revision"), "SSE patches must carry monotonic revision numbers");
  assert(sseManager.includes("30"), "SSE stream must include a 30-second heartbeat");
  assert(sseManager.includes("data: "), "SSE stream must emit data: JSON frames");

+  const schemaPath = path.join(repoRoot, "web", "realtime_patch_schema.py");
+  assert(fs.existsSync(schemaPath), "backend must define a versioned realtime patch schema module");
+  const schema = fs.readFileSync(schemaPath, "utf8");
+  assert(schema.includes("city_observation_patch.v1"), "patch schema must expose city_observation_patch.v1");
+  assert(schema.includes("normalize_observation_patch"), "patch schema must normalize collector payloads");
+  assert(schema.includes("runway_points"), "patch schema must preserve runway point observations");
+
+  const storePath = path.join(repoRoot, "web", "realtime_event_store.py");
+  assert(fs.existsSync(storePath), "backend must define a realtime event replay store");
+  const store = fs.readFileSync(storePath, "utf8");
+  assert(store.includes("observation_patch_events"), "event store must use the SQLite observation_patch_events table");
+  assert(store.includes("replay_events"), "event store must expose replay_events");
+  assert(store.includes("replay_requires_resync"), "event store must detect incomplete replay windows");
+
  const sseRouterPath = path.join(repoRoot, "web", "routers", "sse_router.py");
  assert(fs.existsSync(sseRouterPath), "FastAPI backend must define web/routers/sse_router.py");
  const sseRouter = fs.readFileSync(sseRouterPath, "utf8");
  assert(sseRouter.includes('"/api/events"'), "SSE router must expose GET /api/events");
+  assert(sseRouter.includes("cities"), "SSE route must accept a cities query parameter");
+  assert(sseRouter.includes("since_revision"), "SSE route must accept since_revision for replay");
+  assert(sseRouter.includes("replay_limit"), "SSE route must bound replay batches");
+  assert(sseRouter.includes("resync_required"), "SSE route must emit resync_required when replay is incomplete");
  assert(sseRouter.includes('"/api/internal/collector-patch"'), "SSE router must expose collector patch ingest endpoint");
  assert(sseRouter.includes("StreamingResponse"), "SSE route must return StreamingResponse");
  assert(sseRouter.includes('"text/event-stream"'), "SSE route must use text/event-stream media type");
@@ -53,14 +73,24 @@ export function runTests() {
  const hook = fs.readFileSync(hookPath, "utf8");
  assert(hook.includes("new EventSource"), "frontend patch hook must connect with EventSource");
  assert(hook.includes("/api/events"), "frontend patch hook must subscribe to /api/events");
+  assert(hook.includes("city_observation_patch.v1"), "frontend patch hook must accept v1 observation patch events");
+  assert(hook.includes("subscribedCities"), "frontend patch hook must track the visible city subscription set");
+  assert(hook.includes("since_revision"), "frontend patch hook must reconnect with since_revision");
+  assert(hook.includes("resync_required"), "frontend patch hook must react to server resync_required events");
+  assert(hook.includes("lastRevision"), "frontend patch hook must track the global last processed revision");
  assert(hook.includes("Map<"), "frontend patch hook must keep latest patches in a Map");
  assert(hook.includes("useLatestPatch"), "frontend patch hook must export useLatestPatch(city)");
  assert(hook.includes("revision"), "frontend patch hook must track revisions and skip stale patches");
  assert(hook.includes("setTimeout"), "frontend patch hook must implement explicit reconnect backoff");

+  const bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts");
+  assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI");
+
  const chart = readFrontendFile("components", "dashboard", "scan-terminal", "LiveTemperatureThresholdChart.tsx");
  assert(chart.includes("useLatestPatch"), "temperature chart must consume useLatestPatch(city)");
  assert(chart.includes("latestPatch"), "temperature chart must react to incoming SSE patches");
+  assert(chart.includes("useSseResyncVersion"), "temperature chart must resync full detail when SSE replay is incomplete");
+  assert(chart.includes("runway_points"), "temperature chart must merge v1 runway_points into runway history");
  assert(chart.includes("2 * 60_000"), "temperature chart must wait two minutes without patches before full-fetch fallback");
  assert(
    !chart.includes("setInterval(poll, 60_000)"),
@@ -1,8 +1,10 @@
 import {
  __buildTemperatureChartDataForTest,
+  __getActiveTemperatureSeriesForTest,
  __getObservationDisplayMetricsForTest,
  __getVisibleTemperatureSeriesForTest,
  __isTemperatureSeriesVisibleByDefaultForTest,
+  __mergePatchIntoHourlyForTest,
 } from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart";

 function assert(condition: unknown, message: string) {
@@ -57,6 +59,7 @@ export function runTests() {

  const { series } = __buildTemperatureChartDataForTest(guangzhou, hourly, "1D");
  const defaultVisibleSeries = __getVisibleTemperatureSeriesForTest("guangzhou", series, {});
+  const activeDefaultSeries = __getActiveTemperatureSeriesForTest("guangzhou", series, {}, true);

  const settlementRunway = seriesByKey(series, "runway_02L_20R") as any;
  assert(settlementRunway, "settlement runway should use a stable runway-pair key");
@@ -77,6 +80,14 @@ export function runTests() {
    __isTemperatureSeriesVisibleByDefaultForTest("guangzhou", "runway_02L_20R"),
    "runway series should be visible by default",
  );
+  assert(
+    activeDefaultSeries.some((item) => item.key === "runway_02L_20R"),
+    "settlement runway should remain in the active chart series by default",
+  );
+  assert(
+    activeDefaultSeries.some((item) => item.key === "runway_01L_19R"),
+    "auxiliary runway should remain in the active chart series by default",
+  );
  assert(
    __isTemperatureSeriesVisibleByDefaultForTest("guangzhou", "settlement"),
    "settlement/HKO observations should be visible by default",
@@ -284,6 +295,87 @@ export function runTests() {
  assert(madisSeries.label.includes("MADIS"), "US MADIS series should be labeled as NOAA MADIS instead of plain METAR");
  assert(madisSeries.values.filter((value: number | null) => value !== null).length >= 2, "MADIS series should keep sub-hourly observations");

+  const newYorkMinuteStream = __buildTemperatureChartDataForTest(
+    {
+      city: "new york",
+      local_date: "2026-05-25",
+      local_time: "10:04",
+      tz_offset_seconds: -4 * 60 * 60,
+      airport: "KLGA",
+    } as any,
+    {
+      localTime: "10:04",
+      times: ["00:00", "06:00", "12:00", "18:00"],
+      temps: [55, 57, 65, 72],
+      airportPrimary: {
+        source_code: "madis_hfmetar",
+        source_label: "NOAA MADIS",
+      },
+      airportPrimaryTodayObs: [
+        ["2026-05-25T14:01:00Z", 73.1],
+        ["2026-05-25T14:02:00Z", 73.4],
+        ["2026-05-25T14:03:00Z", 73.8],
+      ],
+    } as any,
+    "1D",
+  );
+  const minuteLabels = newYorkMinuteStream.data
+    .filter((point) => point.madis !== null)
+    .map((point) => point.label);
+  assert(
+    minuteLabels.includes("10:01") &&
+      minuteLabels.includes("10:02") &&
+      minuteLabels.includes("10:03"),
+    "live observation chart should preserve minute-level SSE patch points instead of collapsing them into a 30-minute bucket",
+  );
+
+  const chengduMergedHourly = __mergePatchIntoHourlyForTest(
+    {
+      localTime: "05:25",
+      times: ["00:00", "06:00", "12:00", "18:00"],
+      temps: [24, 28, 31, 27],
+      runwayPlateHistory: {
+        "02L/20R": [{ time: "05:20", temp: 24.2 }],
+      },
+    } as any,
+    {
+      type: "city_observation_patch.v1",
+      city: "chengdu",
+      revision: 12,
+      changes: {
+        temp: 24.8,
+        obs_time: "2026-05-26 05:26:00",
+        source: "amsc_awos",
+        runway_points: [
+          {
+            runway: "02L/20R",
+            temp: 25.1,
+            tdz_temp: 24.7,
+            mid_temp: 24.9,
+            end_temp: 25.1,
+            target_runway_max: 25.1,
+          },
+        ],
+      },
+    } as any,
+  );
+  const chengduMergedChart = __buildTemperatureChartDataForTest(
+    {
+      city: "chengdu",
+      local_date: "2026-05-26",
+      local_time: "05:26",
+      tz_offset_seconds: 8 * 60 * 60,
+    } as any,
+    chengduMergedHourly as any,
+    "1D",
+  );
+  const chengduMergedRunway = seriesByKey(chengduMergedChart.series, "runway_02L_20R") as any;
+  assert(chengduMergedRunway, "v1 runway_points patch should update the runway series");
+  assert(
+    chengduMergedRunway.values.some((value: number | null) => value === 25.1),
+    "v1 runway_points patch should append the latest runway max point to the chart",
+  );
+
  const shanghaiDebFromDetail = __buildTemperatureChartDataForTest(
    {
      city: "shanghai",
@@ -3,6 +3,8 @@
 import { useEffect, useSyncExternalStore } from "react";
 import { resolveBackendApiUrl } from "@/lib/backend-api";

+const V1_EVENT_TYPE = "city_observation_patch.v1";
+
 export type CityPatch = {
  type?: string;
  city: string;
@@ -11,29 +13,82 @@ export type CityPatch = {
  ts?: number;
 };

+type ObservationPatchV1 = {
+  type?: string;
+  city?: string;
+  source?: string;
+  obs_time?: string | null;
+  revision?: number;
+  ts?: number;
+  payload?: Record<string, unknown>;
+};
+
 const latestPatches = new Map<string, CityPatch>();
 const latestRevisions = new Map<string, number>();
 const cityListeners = new Map<string, Set<() => void>>();
 const globalListeners = new Set<() => void>();
+const resyncListeners = new Set<() => void>();
+const subscribedCities = new Map<string, number>();

 let eventSource: EventSource | null = null;
 let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
 let reconnectAttempt = 0;
 let patchVersion = 0;
+let resyncVersion = 0;
+let lastRevision = 0;
 let useFallbackUrl = false;
+let activeConnectionKey = "";

 function normalizeCityKey(city: string | null | undefined) {
  return String(city || "").trim().toLowerCase();
 }

+function subscribedCityList() {
+  return Array.from(subscribedCities.keys()).sort();
+}
+
 function notify(city: string) {
  patchVersion += 1;
  cityListeners.get(city)?.forEach((listener) => listener());
  globalListeners.forEach((listener) => listener());
 }

+function notifyResync(latestServerRevision: number | null) {
+  if (latestServerRevision !== null) {
+    lastRevision = Math.max(lastRevision, latestServerRevision);
+  }
+  resyncVersion += 1;
+  resyncListeners.forEach((listener) => listener());
+  globalListeners.forEach((listener) => listener());
+}
+
+function clearReconnectTimer() {
+  if (!reconnectTimer) return;
+  clearTimeout(reconnectTimer);
+  reconnectTimer = null;
+}
+
+function buildSseUrl(baseUrl: string) {
+  const params = new URLSearchParams();
+  const cities = subscribedCityList();
+  if (cities.length) {
+    params.set("cities", cities.join(","));
+  }
+  if (lastRevision > 0) {
+    params.set("since_revision", String(lastRevision));
+  }
+  params.set("replay_limit", "500");
+
+  const query = params.toString();
+  return query ? `${baseUrl}?${query}` : baseUrl;
+}
+
+function currentConnectionKey() {
+  return `${useFallbackUrl ? "fallback" : "direct"}:${subscribedCityList().join("|")}:${lastRevision}`;
+}
+
 function scheduleReconnect() {
-  if (reconnectTimer || typeof window === "undefined") return;
+  if (reconnectTimer || typeof window === "undefined" || subscribedCities.size === 0) return;
  const delayMs = Math.min(30_000, 1_000 * Math.max(1, 2 ** reconnectAttempt));
  reconnectAttempt += 1;
  reconnectTimer = setTimeout(() => {
@@ -46,30 +101,31 @@ function closeEventSource() {
  if (!eventSource) return;
  eventSource.close();
  eventSource = null;
+  activeConnectionKey = "";
+}
+
+function reconnectNow() {
+  if (typeof window === "undefined") return;
+  clearReconnectTimer();
+  closeEventSource();
+  connectSsePatches();
 }

 function connectSsePatches() {
-  if (typeof window === "undefined" || eventSource) return;
+  if (typeof window === "undefined" || eventSource || subscribedCities.size === 0) return;

-  let url = resolveBackendApiUrl("/api/events");
-  if (useFallbackUrl) {
-    url = "/api/events";
-    console.log("[SSE] Falling back to same-origin BFF proxy URL:", url);
-  } else {
-    console.log("[SSE] Attempting to connect to direct URL:", url);
-  }
+  const baseUrl = useFallbackUrl ? "/api/events" : resolveBackendApiUrl("/api/events");
+  const url = buildSseUrl(baseUrl);
+  activeConnectionKey = currentConnectionKey();

  try {
-    closeEventSource();
    eventSource = new EventSource(url, { withCredentials: true });

    eventSource.onopen = () => {
-      console.log("[SSE] Connection established successfully to:", url);
      reconnectAttempt = 0;
    };

    eventSource.onmessage = (event) => {
-      console.log("[SSE] Received patch message:", event.data);
      try {
        applySsePatch(JSON.parse(event.data));
      } catch (err) {
@@ -81,48 +137,127 @@ function connectSsePatches() {
      console.error("[SSE] Connection error or stream closed:", err);
      closeEventSource();
      if (!useFallbackUrl && url !== "/api/events") {
-        console.warn("[SSE] Direct connection failed. Switching to same-origin BFF proxy fallback for next attempt.");
        useFallbackUrl = true;
      }
      scheduleReconnect();
    };
  } catch (err) {
    console.error("[SSE] Exception thrown while instantiating EventSource:", err);
-    if (!useFallbackUrl && url !== "/api/events") {
+    closeEventSource();
+    if (!useFallbackUrl && baseUrl !== "/api/events") {
      useFallbackUrl = true;
    }
    scheduleReconnect();
  }
 }

-export function ensureSsePatchConnection() {
-  connectSsePatches();
+function ensureSsePatchConnection() {
+  if (subscribedCities.size === 0) {
+    closeEventSource();
+    clearReconnectTimer();
+    return;
+  }
+  if (eventSource && activeConnectionKey === currentConnectionKey()) return;
+  reconnectNow();
 }

-export function applySsePatch(payload: unknown) {
-  if (!payload || typeof payload !== "object") return false;
-  const patch = payload as Partial<CityPatch>;
-  if (patch.type && patch.type !== "city_patch") return false;
+function registerCitySubscription(city: string) {
+  const cityKey = normalizeCityKey(city);
+  if (!cityKey) return () => {};
+
+  const previousCount = subscribedCities.get(cityKey) ?? 0;
+  subscribedCities.set(cityKey, previousCount + 1);
+  if (previousCount === 0) {
+    ensureSsePatchConnection();
+  }
+
+  return () => {
+    const nextCount = (subscribedCities.get(cityKey) ?? 1) - 1;
+    if (nextCount <= 0) {
+      subscribedCities.delete(cityKey);
+    } else {
+      subscribedCities.set(cityKey, nextCount);
+    }
+    ensureSsePatchConnection();
+  };
+}
+
+function normalizeLegacyPatch(patch: Partial<CityPatch>): CityPatch | null {
  const city = normalizeCityKey(patch.city);
  const changes = patch.changes;
  const revision = Number(patch.revision);
  if (!city || !changes || typeof changes !== "object" || !Number.isFinite(revision)) {
-    return false;
+    return null;
  }
-
-  const previousRevision = latestRevisions.get(city) ?? 0;
-  if (revision <= previousRevision) return false;
-
-  const normalizedPatch: CityPatch = {
+  return {
    type: "city_patch",
    city,
    changes: changes as Record<string, unknown>,
    revision,
    ts: typeof patch.ts === "number" ? patch.ts : Date.now(),
  };
-  latestRevisions.set(city, revision);
-  latestPatches.set(city, normalizedPatch);
-  notify(city);
+}
+
+function normalizeV1Patch(patch: ObservationPatchV1): CityPatch | null {
+  const city = normalizeCityKey(patch.city);
+  const revision = Number(patch.revision);
+  const payload = patch.payload;
+  if (!city || !payload || typeof payload !== "object" || !Number.isFinite(revision)) {
+    return null;
+  }
+
+  const changes: Record<string, unknown> = {
+    ...payload,
+    source: typeof patch.source === "string" ? patch.source : payload.source,
+    obs_time: typeof patch.obs_time === "string" ? patch.obs_time : payload.obs_time,
+    schema_type: V1_EVENT_TYPE,
+  };
+
+  return {
+    type: V1_EVENT_TYPE,
+    city,
+    changes,
+    revision,
+    ts: typeof patch.ts === "number" ? patch.ts : Date.now(),
+  };
+}
+
+function normalizeIncomingPatch(payload: unknown): CityPatch | null {
+  if (!payload || typeof payload !== "object") return null;
+  const patch = payload as Partial<CityPatch> & ObservationPatchV1;
+  if (patch.type === "city_patch" || !patch.type) {
+    return normalizeLegacyPatch(patch);
+  }
+  if (patch.type === V1_EVENT_TYPE) {
+    return normalizeV1Patch(patch);
+  }
+  return null;
+}
+
+export function applySsePatch(payload: unknown) {
+  if (!payload || typeof payload !== "object") return false;
+  const event = payload as { type?: string; latest_revision?: number };
+
+  if (event.type === "connected" || event.type === "heartbeat") {
+    return false;
+  }
+
+  if (event.type === "resync_required") {
+    const latestServerRevision = Number(event.latest_revision);
+    notifyResync(Number.isFinite(latestServerRevision) ? latestServerRevision : null);
+    return true;
+  }
+
+  const normalizedPatch = normalizeIncomingPatch(payload);
+  if (!normalizedPatch) return false;
+
+  const previousRevision = latestRevisions.get(normalizedPatch.city) ?? 0;
+  if (normalizedPatch.revision <= previousRevision) return false;
+
+  latestRevisions.set(normalizedPatch.city, normalizedPatch.revision);
+  latestPatches.set(normalizedPatch.city, normalizedPatch);
+  lastRevision = Math.max(lastRevision, normalizedPatch.revision);
+  notify(normalizedPatch.city);
  return true;
 }

@@ -131,14 +266,6 @@ export function getLatestPatchesSnapshot() {
 }

 export function useSsePatchVersion() {
-  if (typeof window !== "undefined") {
-    ensureSsePatchConnection();
-  }
-
-  useEffect(() => {
-    ensureSsePatchConnection();
-  }, []);
-
  return useSyncExternalStore(
    (listener) => {
      globalListeners.add(listener);
@@ -149,16 +276,24 @@ export function useSsePatchVersion() {
  );
 }

+export function useSseResyncVersion() {
+  return useSyncExternalStore(
+    (listener) => {
+      resyncListeners.add(listener);
+      return () => resyncListeners.delete(listener);
+    },
+    () => resyncVersion,
+    () => 0,
+  );
+}
+
 export function useLatestPatch(city: string | null | undefined) {
  const cityKey = normalizeCityKey(city);

-  if (typeof window !== "undefined") {
-    ensureSsePatchConnection();
-  }
-
  useEffect(() => {
-    ensureSsePatchConnection();
-  }, []);
+    if (!cityKey) return undefined;
+    return registerCitySubscription(cityKey);
+  }, [cityKey]);

  return useSyncExternalStore(
    (listener) => {
@@ -177,3 +312,4 @@ export function useLatestPatch(city: string | null | undefined) {
 }

 export const __applySsePatchForTest = applySsePatch;
+export const __buildSseUrlForTest = buildSseUrl;
@@ -57,3 +57,7 @@ skip-magic-trailing-comma = false

 # Like Black, automatically detect the appropriate line ending.
 line-ending = "auto"
+
+[tool.pyright]
+include = ["src", "web", "tests"]
+extraPaths = ["."]
@@ -11,11 +11,12 @@ import csv
 import os
 import io
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Any, Dict, Optional

 from loguru import logger

+# pyrefly: ignore [missing-import]
 from src.utils.metrics import record_source_call

 HKO_BASE_URL = os.getenv("HKO_BASE_URL", "").strip() or "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather"
@@ -34,11 +35,17 @@ HKO_STATIONS = {


 class HkoObsSourceMixin:
+    session: Any
+    timeout: Any
+    _hko_obs_cache: Dict[str, Any]
+    _hko_obs_cache_lock: Any
+    hko_obs_cache_ttl_sec: Any
+
    def _hko_http_get(self, url: str) -> str:
        getter = getattr(self, "_http_get", None)
        if callable(getter):
            resp = getter(url)
-            return resp.text if hasattr(resp, "text") else resp
+            return str(resp.text) if hasattr(resp, "text") else str(resp)
        resp = self.session.get(url, timeout=self.timeout)
        resp.raise_for_status()
        return resp.text
@@ -49,7 +56,7 @@ class HkoObsSourceMixin:
        use_fahrenheit: bool = False,
    ) -> Optional[Dict[str, Any]]:
        started = time.perf_counter()
-        city_key = str(city or "").strip().lower()
+        city_key = (city or "").strip().lower()
        meta = HKO_STATIONS.get(city_key) or {}
        if not meta:
            return None
@@ -95,11 +102,11 @@ class HkoObsSourceMixin:

            result = {
                "source": "hko_obs",
-                "timestamp": datetime.utcnow().isoformat(),
+                "timestamp": datetime.now(timezone.utc).isoformat(),
                "station_code": meta["code"],
                "station_name": meta["label"],
                "icao": meta["icao"],
-                "obs_time": obs_iso or datetime.utcnow().isoformat(),
+                "obs_time": obs_iso or datetime.now(timezone.utc).isoformat(),
                "current": {
                    "temp": temp,
                },
@@ -132,7 +139,7 @@ class HkoObsSourceMixin:
        current = self.fetch_hko_obs_current(city, use_fahrenheit=use_fahrenheit)
        if not current:
            return []
-        meta = HKO_STATIONS.get(str(city or "").strip().lower()) or {}
+        meta = HKO_STATIONS.get((city or "").strip().lower()) or {}
        return [
            {
                "name": meta.get("label") or "HKO Station",
@@ -339,6 +339,30 @@ class DBManager:
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_payment_audit_events_created_at ON payment_audit_events(created_at DESC)"
            )
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS observation_patch_events (
+                    revision INTEGER PRIMARY KEY AUTOINCREMENT,
+                    schema_type TEXT NOT NULL,
+                    schema_version INTEGER NOT NULL,
+                    city TEXT NOT NULL,
+                    source TEXT NOT NULL,
+                    obs_time TEXT,
+                    payload_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL
+                )
+            """)
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_observation_patch_events_city_revision
+                ON observation_patch_events(city, revision)
+                """
+            )
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_observation_patch_events_created_at
+                ON observation_patch_events(created_at)
+                """
+            )
            conn.execute("""
                CREATE TABLE IF NOT EXISTS app_analytics_events (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -3,8 +3,9 @@ import os
 from src.database.db_manager import DBManager

 def test_full_cache_lifecycle():
-    with tempfile.TemporaryDirectory() as tmpdir:
-        db_path = os.path.join(tmpdir, "test.db")
+    tmpdir = tempfile.TemporaryDirectory()
+    try:
+        db_path = os.path.join(tmpdir.name, "test.db")
        if hasattr(DBManager, "_initialized_paths"):
            DBManager._initialized_paths.clear()
            
@@ -31,3 +32,18 @@ def test_full_cache_lifecycle():
        assert cached["payload"] == payload
        assert cached["version"] == "v1"
        assert cached["source_fingerprint"] == "testcity:full"
+
+        # Release database file handles for Windows temp directory cleanup
+        del db
+        import gc
+        gc.collect()
+    finally:
+        try:
+            tmpdir.cleanup()
+        except PermissionError:
+            import gc
+            gc.collect()
+            try:
+                tmpdir.cleanup()
+            except Exception:
+                pass
@@ -0,0 +1,88 @@
+import sqlite3
+from datetime import datetime, timezone
+
+from src.database.db_manager import DBManager
+from web.realtime_event_store import RealtimeEventStore
+from web.realtime_patch_schema import normalize_observation_patch
+
+
+def _event(city: str, temp: float, source: str = "metar"):
+    return normalize_observation_patch(
+        {
+            "city": city,
+            "changes": {
+                "temp": temp,
+                "obs_time": "2026-05-26T08:15:00Z",
+                "source": source,
+            },
+        }
+    )
+
+
+def test_event_store_appends_monotonic_revisions_and_replays_by_city(tmp_path):
+    db_path = str(tmp_path / "polyweather.db")
+    DBManager._initialized_paths.clear()
+    store = RealtimeEventStore(db_path=db_path)
+
+    taipei = store.append_event(_event("taipei", 31.2))
+    seoul = store.append_event(_event("seoul", 27.8))
+    taipei_next = store.append_event(_event("taipei", 31.5))
+
+    assert taipei["revision"] == 1
+    assert seoul["revision"] == 2
+    assert taipei_next["revision"] == 3
+    assert store.latest_revision() == 3
+
+    replay = store.replay_events(cities={"taipei"}, since_revision=1, limit=10)
+
+    assert [event["revision"] for event in replay] == [3]
+    assert replay[0]["city"] == "taipei"
+    assert replay[0]["payload"]["temp"] == 31.5
+
+
+def test_event_store_cleanup_uses_short_replay_retention(tmp_path):
+    db_path = str(tmp_path / "polyweather.db")
+    DBManager._initialized_paths.clear()
+    store = RealtimeEventStore(db_path=db_path)
+
+    old_event = store.append_event(_event("taipei", 30.0))
+    fresh_event = store.append_event(_event("taipei", 31.0))
+
+    with sqlite3.connect(db_path) as conn:
+        conn.execute(
+            "UPDATE observation_patch_events SET created_at = ? WHERE revision = ?",
+            ("2026-05-26T00:00:00+00:00", old_event["revision"]),
+        )
+        conn.execute(
+            "UPDATE observation_patch_events SET created_at = ? WHERE revision = ?",
+            ("2026-05-26T11:30:00+00:00", fresh_event["revision"]),
+        )
+        conn.commit()
+
+    deleted = store.cleanup_old_events(
+        retention_hours=6,
+        now=datetime(2026, 5, 26, 12, 0, tzinfo=timezone.utc),
+    )
+
+    assert deleted == 1
+    replay = store.replay_events(cities={"taipei"}, since_revision=0, limit=10)
+    assert [event["revision"] for event in replay] == [fresh_event["revision"]]
+
+
+def test_event_store_reports_replay_gap_when_limit_is_exceeded(tmp_path):
+    db_path = str(tmp_path / "polyweather.db")
+    DBManager._initialized_paths.clear()
+    store = RealtimeEventStore(db_path=db_path)
+
+    for temp in [30.0, 30.5, 31.0]:
+        store.append_event(_event("hong kong", temp, source="hko"))
+
+    replay = store.replay_events(cities={"hong kong"}, since_revision=0, limit=2)
+
+    assert len(replay) == 2
+    assert store.replay_requires_resync(
+        cities={"hong kong"},
+        since_revision=0,
+        replay_count=len(replay),
+        limit=2,
+    )
@@ -0,0 +1,86 @@
+import pytest
+
+from web.realtime_patch_schema import (
+    PatchValidationError,
+    normalize_observation_patch,
+)
+
+
+def test_legacy_temperature_patch_normalizes_to_v1_with_runway_points():
+    event = normalize_observation_patch(
+        {
+            "city": " Seoul ",
+            "changes": {
+                "temp": "31.25",
+                "max_so_far": 32.1,
+                "obs_time": "2026-05-26T08:15:00Z",
+                "source": "amos",
+                "amos": {
+                    "icao": "RKSS",
+                    "station_label": "Gimpo Airport",
+                    "runway_obs": {
+                        "point_temperatures": [
+                            {
+                                "runway": "14L/32R",
+                                "tdz_temp": 31.2,
+                                "mid_temp": 31.6,
+                                "end_temp": 31.8,
+                                "target_runway_max": 31.8,
+                            }
+                        ]
+                    },
+                },
+            },
+        }
+    )
+
+    assert event["type"] == "city_observation_patch.v1"
+    assert event["schema_type"] == "city_observation_patch"
+    assert event["schema_version"] == 1
+    assert event["city"] == "seoul"
+    assert event["source"] == "amos"
+    assert event["obs_time"] == "2026-05-26T08:15:00Z"
+    assert event["payload"]["temp"] == 31.25
+    assert event["payload"]["max_so_far"] == 32.1
+    assert event["payload"]["station_code"] == "RKSS"
+    assert event["payload"]["station_label"] == "Gimpo Airport"
+    assert event["payload"]["unit"] == "celsius"
+    assert event["payload"]["runway_points"] == [
+        {
+            "runway": "14L/32R",
+            "temp": 31.8,
+            "tdz_temp": 31.2,
+            "mid_temp": 31.6,
+            "end_temp": 31.8,
+            "target_runway_max": 31.8,
+        }
+    ]
+
+
+def test_v1_patch_payload_is_accepted_and_normalized():
+    event = normalize_observation_patch(
+        {
+            "type": "city_observation_patch.v1",
+            "city": "Taipei",
+            "source": "cwa",
+            "obs_time": "2026-05-26T07:01:00Z",
+            "payload": {
+                "temp": 29.4,
+                "station_code": "46692",
+                "runway_points": [{"runway": "05/23", "temp": 30.2}],
+            },
+        }
+    )
+
+    assert event["city"] == "taipei"
+    assert event["source"] == "cwa"
+    assert event["payload"]["temp"] == 29.4
+    assert event["payload"]["runway_points"][0]["temp"] == 30.2
+
+
+def test_invalid_patch_without_city_or_observation_data_is_rejected():
+    with pytest.raises(PatchValidationError):
+        normalize_observation_patch({"changes": {"temp": 21.0}})
+
+    with pytest.raises(PatchValidationError):
+        normalize_observation_patch({"city": "taipei", "changes": {"source": "metar"}})
@@ -0,0 +1,133 @@
+import json
+
+from fastapi.testclient import TestClient
+
+from web.app import app
+from web.routers import sse_router
+
+
+def _decode_sse_events(text: str):
+    events = []
+    for frame in text.strip().split("\n\n"):
+        if not frame.startswith("data: "):
+            continue
+        events.append(json.loads(frame[len("data: "):]))
+    return events
+
+
+def test_events_endpoint_replays_only_requested_cities(monkeypatch):
+    captured = {}
+
+    class FakeStore:
+        def latest_revision(self):
+            return 44
+
+        def replay_events(self, *, cities, since_revision, limit):
+            captured["cities"] = cities
+            captured["since_revision"] = since_revision
+            captured["limit"] = limit
+            return [
+                {
+                    "type": "city_observation_patch.v1",
+                    "revision": 43,
+                    "city": "taipei",
+                    "source": "cwa",
+                    "obs_time": "2026-05-26T08:15:00Z",
+                    "ts": 1780000000000,
+                    "payload": {"temp": 31.2},
+                }
+            ]
+
+        def replay_requires_resync(self, *, cities, since_revision, replay_count, limit):
+            return False
+
+    async def finite_stream(
+        user_id,
+        *,
+        cities=None,
+        replay_events=None,
+        connected_revision=0,
+        resync_event=None,
+    ):
+        yield sse_router.sse_manager._format_event(
+            {"type": "connected", "revision": connected_revision}
+        )
+        for event in replay_events or []:
+            yield sse_router.sse_manager._format_event(event)
+
+    monkeypatch.setattr(sse_router, "event_store", FakeStore())
+    monkeypatch.setattr(sse_router.sse_manager, "event_stream", finite_stream)
+
+    response = TestClient(app).get(
+        "/api/events?cities=taipei,hong%20kong&since_revision=42&replay_limit=25"
+    )
+
+    assert response.status_code == 200
+    assert captured == {
+        "cities": {"taipei", "hong kong"},
+        "since_revision": 42,
+        "limit": 25,
+    }
+    events = _decode_sse_events(response.text)
+    assert [event["type"] for event in events] == [
+        "connected",
+        "city_observation_patch.v1",
+    ]
+    assert events[1]["city"] == "taipei"
+
+
+def test_events_endpoint_emits_resync_when_replay_is_incomplete(monkeypatch):
+    class FakeStore:
+        def latest_revision(self):
+            return 99
+
+        def replay_events(self, *, cities, since_revision, limit):
+            return [
+                {
+                    "type": "city_observation_patch.v1",
+                    "revision": 98,
+                    "city": "taipei",
+                    "source": "cwa",
+                    "obs_time": "2026-05-26T08:15:00Z",
+                    "ts": 1780000000000,
+                    "payload": {"temp": 31.2},
+                }
+            ]
+
+        def replay_requires_resync(self, *, cities, since_revision, replay_count, limit):
+            return True
+
+    async def finite_stream(
+        user_id,
+        *,
+        cities=None,
+        replay_events=None,
+        connected_revision=0,
+        resync_event=None,
+    ):
+        yield sse_router.sse_manager._format_event(
+            {"type": "connected", "revision": connected_revision}
+        )
+        for event in replay_events or []:
+            yield sse_router.sse_manager._format_event(event)
+        if resync_event:
+            yield sse_router.sse_manager._format_event(resync_event)
+
+    monkeypatch.setattr(sse_router, "event_store", FakeStore())
+    monkeypatch.setattr(sse_router.sse_manager, "event_stream", finite_stream)
+
+    response = TestClient(app).get(
+        "/api/events?cities=taipei&since_revision=1&replay_limit=1"
+    )
+
+    assert response.status_code == 200
+    events = _decode_sse_events(response.text)
+    assert events[-1]["type"] == "resync_required"
+    assert events[-1]["reason"] == "replay_window_exceeded"
+    assert events[-1]["latest_revision"] == 99
+
+
+def test_replay_limit_is_bounded():
+    assert sse_router._bounded_replay_limit(0) == 1
+    assert sse_router._bounded_replay_limit(500) == 500
+    assert sse_router._bounded_replay_limit(5000) == 2000
@@ -146,11 +146,11 @@ def _is_plausible_city_temp(city: str, value: Any, unit: str = "°C") -> bool:
    temp = _sf(value)
    if temp is None:
        return False
-    meta = CITY_REGISTRY.get(str(city or "").strip().lower(), {}) or {}
+    meta = CITY_REGISTRY.get((city or "").strip().lower(), {}) or {}
    min_c = _sf(meta.get("min_plausible_metar_temp_c"))
    if min_c is None:
        return True
-    min_value = min_c * 9 / 5 + 32 if str(unit or "").upper().endswith("F") else min_c
+    min_value = min_c * 9 / 5 + 32 if (unit or "").upper().endswith("F") else min_c
    return temp >= min_value


@@ -158,7 +158,7 @@ def _parse_local_hour(local_time_str: Optional[str]) -> Optional[int]:
    if not local_time_str:
        return None
    try:
-        parts = str(local_time_str).strip().split(":")
+        parts = local_time_str.strip().split(":")
        hour = int(parts[0])
        if 0 <= hour <= 23:
            return hour
@@ -204,7 +204,7 @@ def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) -
    now = datetime.now(timezone.utc).isoformat()
    with _ANALYSIS_CACHE_STATS_LOCK:
        _ANALYSIS_CACHE_STATS["total_requests"] = int(_ANALYSIS_CACHE_STATS.get("total_requests") or 0) + 1
-        _ANALYSIS_CACHE_STATS["last_city"] = str(city or "")
+        _ANALYSIS_CACHE_STATS["last_city"] = city or ""
        if force_refresh:
            _ANALYSIS_CACHE_STATS["force_refresh_requests"] = int(_ANALYSIS_CACHE_STATS.get("force_refresh_requests") or 0) + 1
        if hit:
@@ -243,7 +243,7 @@ def _analysis_ttl_for_city(city: str) -> int:


 def _analysis_cache_key(city: str, detail_mode: str = "full") -> str:
-    normalized_raw = str(detail_mode or "").strip().lower()
+    normalized_raw = (detail_mode or "").strip().lower()
    if normalized_raw == "panel":
        normalized_mode = "panel"
    elif normalized_raw == "market":
@@ -666,7 +666,7 @@ def _analyze(
    """
    # Check cache – skip when explicitly refreshing observations
    ttl = _analysis_ttl_for_city(city)
-    normalized_detail_mode_raw = str(detail_mode or "full").strip().lower()
+    normalized_detail_mode_raw = (detail_mode or "full").strip().lower()
    if normalized_detail_mode_raw == "panel":
        normalized_detail_mode = "panel"
    elif normalized_detail_mode_raw == "market":
@@ -750,8 +750,9 @@ def _analyze(
                first_start = nws_periods[0].get("start_time")
                if first_start:
                    maybe_dt = datetime.fromisoformat(str(first_start))
-                    if maybe_dt.utcoffset() is not None:
-                        utc_offset = int(maybe_dt.utcoffset().total_seconds())
+                    offset_td = maybe_dt.utcoffset()
+                    if offset_td is not None:
+                        utc_offset = int(offset_td.total_seconds())
        except Exception:
            utc_offset = None
    if utc_offset is None:
@@ -770,7 +771,7 @@ def _analyze(
    metar_current_is_today = _metar_is_current_local_day(
        metar,
        local_date=local_date_str,
-        utc_offset=int(utc_offset or 0),
+        utc_offset=utc_offset,
    )

    # ── 2. Current conditions (settlement > AMOS runway sensors > METAR > MGM > NMC fallback) ──
@@ -878,12 +879,13 @@ def _analyze(
    if not obs_time_str and current_source == "amos":
        amos_obs_time = amos_data.get("observation_time")
        if amos_obs_time:
-            obs_time_str = _format_observation_time_local(amos_obs_time, int(utc_offset or 0))
+            obs_time_str = _format_observation_time_local(amos_obs_time, utc_offset)
+    nmc_fallback = None
    if not obs_time_str and current_source == "nmc":
        nmc_fallback = _fetch_nmc_current_fallback(city, use_fahrenheit=is_f)
        obs_time_str = _format_observation_time_local(
            nmc_fallback.get("publish_time") or nmc_fallback.get("timestamp"),
-            int(utc_offset or 0),
+            utc_offset,
        )

    current_obs_raw = obs_t
@@ -972,9 +974,9 @@ def _analyze(
            if (
                max_temp_time
                and max_so_far is not None
-                and str(max_temp_time) != str(obs_time_str)
+                and max_temp_time != obs_time_str
            ):
-                settlement_today_obs.append({"time": str(max_temp_time), "temp": max_so_far})
+                settlement_today_obs.append({"time": max_temp_time, "temp": max_so_far})

    metar_today_obs_payload = [
        {"time": t, "temp": v}
@@ -1044,7 +1046,7 @@ def _analyze(
            else cur_temp
        )
        if fallback_high is not None:
-            om_today = float(fallback_high)
+            om_today = fallback_high
            if not forecast_daily:
                forecast_daily = [{"date": local_date_str, "max_temp": om_today}]
    sunrise = (
@@ -1065,7 +1067,9 @@ def _analyze(
        current_forecasts["Open-Meteo"] = om_today
    for m, v in mm.get("forecasts", {}).items():
        if v is not None and not _is_excluded_model_name(m):
-            current_forecasts[m] = _sf(v)
+            temp_val = _sf(v)
+            if temp_val is not None:
+                current_forecasts[m] = temp_val
    nws_high = _sf(raw.get("nws", {}).get("today_high"))
    if nws_high is not None:
        current_forecasts["NWS"] = nws_high
@@ -1244,8 +1248,8 @@ def _analyze(
    # between the current observed temperature and the model's hourly path.
    # Uses cur_temp / max_so_far already resolved at lines 1052-1095 above.
    _local_hour = _parse_local_hour(local_time_str)
-    peak_first = int(first_peak_h or 14)
-    peak_last_h = int(last_peak_h or 17)
+    peak_first = first_peak_h or 14
+    peak_last_h = last_peak_h or 17

    if (
        deb_val is not None
@@ -1377,7 +1381,7 @@ def _analyze(
            taf if isinstance(taf, dict) else {},
            city,
            local_date_str,
-            int(utc_offset or 0),
+            utc_offset,
            first_peak_h,
            last_peak_h,
        )
@@ -1481,6 +1485,8 @@ def _analyze(
    multi_model_daily = {}
    mm_daily_raw = mm.get("daily_forecasts", {})
    for i, d_str in enumerate(dates):
+        d_probs = []
+        d_probs_all = []
        if i == 0:
            day_m = current_forecasts.copy()
            d_val, d_winfo = deb_val, deb_weights
@@ -1534,7 +1540,7 @@ def _analyze(
    # ── Assemble result ──
    runway_plate_history = {}
    icao = risk.get("icao", "")
-    if icao:
+    if isinstance(icao, str) and icao:
        try:
            from src.database.db_manager import DBManager
            raw_runway_obs = DBManager().get_runway_obs_recent(icao, minutes=36 * 60)
@@ -1581,7 +1587,7 @@ def _analyze(
        "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()),
        "lat": lat,
        "lon": lon,
-        "utc_offset_seconds": int(utc_offset or 0),
+        "utc_offset_seconds": utc_offset,
        "temp_symbol": sym,
        "local_time": local_time_str,
        "local_date": local_date_str,
@@ -1779,7 +1785,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
        except Exception:
            return None

-    jobs = {
+    jobs: Dict[str, Any] = {
        "settlement_current": lambda: _weather.fetch_settlement_current(city) or {},
        "open_meteo": lambda: _weather.fetch_from_open_meteo(lat, lon, use_fahrenheit=is_f) or {},
        "multi_model": lambda: _weather.fetch_multi_model(lat, lon, city=city, use_fahrenheit=is_f) or {},
@@ -1832,7 +1838,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
    metar_current_is_today = _metar_is_current_local_day(
        metar,
        local_date=local_date_str,
-        utc_offset=int(utc_offset or 0),
+        utc_offset=utc_offset,
    )

    sc_cur = settlement_current.get("current") or {}
@@ -1922,13 +1928,13 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
                (datetime.now(timezone.utc) - dt.astimezone(timezone.utc)).total_seconds() / 60
            )
        except Exception:
-            obs_time_str = str(obs_t)[:16]
+            obs_time_str = obs_t[:16]
    if not obs_time_str and current_source == "nmc":
        if not nmc_fallback:
            nmc_fallback = _fetch_nmc_current_fallback(city, use_fahrenheit=is_f)
        obs_time_str = _format_observation_time_local(
            nmc_fallback.get("publish_time") or nmc_fallback.get("timestamp"),
-            int(utc_offset or 0),
+            utc_offset,
        )

    om_daily = (open_meteo.get("daily") or {}) if isinstance(open_meteo, dict) else {}
@@ -1965,14 +1971,16 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
            else cur_temp
        )
        if fallback_high is not None:
-            om_today = float(fallback_high)
+            om_today = fallback_high

    current_forecasts: Dict[str, float] = {}
    if om_today is not None:
        current_forecasts["Open-Meteo"] = om_today
    for m, v in mm.get("forecasts", {}).items():
        if v is not None and not _is_excluded_model_name(m):
-            current_forecasts[m] = _sf(v)
+            temp_val = _sf(v)
+            if temp_val is not None:
+                current_forecasts[m] = temp_val
    if nws_high is not None:
        current_forecasts["NWS"] = nws_high
    if mgm_high is not None:
@@ -1980,7 +1988,9 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
    elif mgm_hourly_high is not None:
        current_forecasts["MGM Hourly"] = mgm_hourly_high
    if hko_forecast is not None:
-        current_forecasts["HKO"] = _sf(hko_forecast)
+        temp_hko = _sf(hko_forecast)
+        if temp_hko is not None:
+            current_forecasts["HKO"] = temp_hko
    current_forecasts = {
        model_name: value
        for model_name, value in current_forecasts.items()
@@ -2011,8 +2021,8 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
                settlement_today_obs.append({"time": raw_time, "temp": raw_temp})
        if not settlement_today_obs and obs_time_str and cur_temp is not None:
            settlement_today_obs.append({"time": obs_time_str, "temp": cur_temp})
-        if max_temp_time and max_so_far is not None and str(max_temp_time) != str(obs_time_str):
-            settlement_today_obs.append({"time": str(max_temp_time), "temp": max_so_far})
+        if max_temp_time and max_so_far is not None and max_temp_time != obs_time_str:
+            settlement_today_obs.append({"time": max_temp_time, "temp": max_so_far})

    metar_today_obs_payload = [
        {"time": obs_time, "temp": obs_temp}
@@ -2042,7 +2052,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]:
        "name": city,
        "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()),
        "temp_symbol": sym,
-        "utc_offset_seconds": int(utc_offset or 0),
+        "utc_offset_seconds": utc_offset,
        "local_time": local_time_str,
        "local_date": local_date_str,
        "risk": {
@@ -0,0 +1,251 @@
+"""SQLite-backed replay log for realtime observation SSE patches."""
+
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, Optional, Set
+
+from src.database.db_manager import DBManager
+from web.realtime_patch_schema import EVENT_TYPE
+
+
+DEFAULT_RETENTION_HOURS = 6
+MAX_REPLAY_LIMIT = 2000
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _iso(dt: datetime) -> str:
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc).isoformat()
+
+
+def _created_at_to_ms(value: str) -> int:
+    raw = str(value or "").strip()
+    if raw.endswith("Z"):
+        raw = f"{raw[:-1]}+00:00"
+    try:
+        dt = datetime.fromisoformat(raw)
+    except ValueError:
+        return int(_utc_now().timestamp() * 1000)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return int(dt.astimezone(timezone.utc).timestamp() * 1000)
+
+
+def _normalize_city_set(cities: Optional[Set[str]]) -> Set[str]:
+    return {str(city or "").strip().lower() for city in (cities or set()) if str(city or "").strip()}
+
+
+def _retention_hours_from_env() -> int:
+    raw = os.getenv("POLYWEATHER_PATCH_EVENT_RETENTION_HOURS", "").strip()
+    if not raw:
+        return DEFAULT_RETENTION_HOURS
+    try:
+        return max(1, int(float(raw)))
+    except ValueError:
+        return DEFAULT_RETENTION_HOURS
+
+
+class RealtimeEventStore:
+    def __init__(self, db_path: Optional[str] = None) -> None:
+        self._db = DBManager(db_path)
+        self.db_path = self._db.db_path
+        self._ensure_table()
+
+    def _connect(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path, timeout=10)
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("PRAGMA busy_timeout=5000")
+        return conn
+
+    def _ensure_table(self) -> None:
+        db_dir = os.path.dirname(self.db_path)
+        if db_dir:
+            os.makedirs(db_dir, exist_ok=True)
+        with self._connect() as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS observation_patch_events (
+                    revision INTEGER PRIMARY KEY AUTOINCREMENT,
+                    schema_type TEXT NOT NULL,
+                    schema_version INTEGER NOT NULL,
+                    city TEXT NOT NULL,
+                    source TEXT NOT NULL,
+                    obs_time TEXT,
+                    payload_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL
+                )
+                """
+            )
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_observation_patch_events_city_revision
+                ON observation_patch_events(city, revision)
+                """
+            )
+            conn.execute(
+                """
+                CREATE INDEX IF NOT EXISTS idx_observation_patch_events_created_at
+                ON observation_patch_events(created_at)
+                """
+            )
+            conn.commit()
+
+    def append_event(self, event: Dict[str, Any]) -> Dict[str, Any]:
+        if event.get("type") != EVENT_TYPE:
+            raise ValueError("unsupported realtime event type")
+        payload = event.get("payload")
+        if not isinstance(payload, dict):
+            raise ValueError("event payload must be an object")
+
+        created_at = _iso(_utc_now())
+        with self._connect() as conn:
+            cursor = conn.execute(
+                """
+                INSERT INTO observation_patch_events (
+                    schema_type, schema_version, city, source, obs_time, payload_json, created_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    str(event["schema_type"]),
+                    int(event["schema_version"]),
+                    str(event["city"]),
+                    str(event["source"]),
+                    event.get("obs_time"),
+                    json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
+                    created_at,
+                ),
+            )
+            revision = int(cursor.lastrowid)
+            conn.commit()
+
+        stored = {
+            "type": event["type"],
+            "revision": revision,
+            "city": str(event["city"]),
+            "source": str(event["source"]),
+            "obs_time": event.get("obs_time"),
+            "ts": int(event.get("ts") or _created_at_to_ms(created_at)),
+            "payload": payload,
+        }
+        self.cleanup_old_events(retention_hours=_retention_hours_from_env())
+        return stored
+
+    def latest_revision(self) -> int:
+        with self._connect() as conn:
+            row = conn.execute(
+                "SELECT COALESCE(MAX(revision), 0) FROM observation_patch_events"
+            ).fetchone()
+        return int(row[0] or 0) if row else 0
+
+    def replay_events(
+        self,
+        *,
+        cities: Optional[Set[str]],
+        since_revision: int,
+        limit: int,
+    ) -> list[Dict[str, Any]]:
+        city_set = _normalize_city_set(cities)
+        bounded_limit = max(1, min(MAX_REPLAY_LIMIT, int(limit or 1)))
+        params: list[Any] = [max(0, int(since_revision or 0))]
+        where = "revision > ?"
+        if city_set:
+            placeholders = ",".join("?" for _ in city_set)
+            where += f" AND city IN ({placeholders})"
+            params.extend(sorted(city_set))
+        params.append(bounded_limit)
+
+        with self._connect() as conn:
+            conn.row_factory = sqlite3.Row
+            rows = conn.execute(
+                f"""
+                SELECT revision, schema_type, schema_version, city, source, obs_time,
+                       payload_json, created_at
+                FROM observation_patch_events
+                WHERE {where}
+                ORDER BY revision ASC
+                LIMIT ?
+                """,
+                params,
+            ).fetchall()
+        return [self._row_to_event(row) for row in rows]
+
+    def replay_requires_resync(
+        self,
+        *,
+        cities: Optional[Set[str]],
+        since_revision: int,
+        replay_count: int,
+        limit: int,
+    ) -> bool:
+        city_set = _normalize_city_set(cities)
+        since = max(0, int(since_revision or 0))
+        where_parts = []
+        params: list[Any] = []
+        if city_set:
+            placeholders = ",".join("?" for _ in city_set)
+            where_parts.append(f"city IN ({placeholders})")
+            params.extend(sorted(city_set))
+        where = f"WHERE {' AND '.join(where_parts)}" if where_parts else ""
+
+        with self._connect() as conn:
+            min_row = conn.execute(
+                f"SELECT MIN(revision) FROM observation_patch_events {where}",
+                params,
+            ).fetchone()
+            min_revision = int(min_row[0] or 0) if min_row else 0
+            if min_revision and since > 0 and since < min_revision - 1:
+                return True
+
+            if replay_count < max(1, int(limit or 1)):
+                return False
+
+            count_params = [since, *params]
+            count_where = "revision > ?"
+            if where_parts:
+                count_where += " AND " + " AND ".join(where_parts)
+            count_row = conn.execute(
+                f"SELECT COUNT(1) FROM observation_patch_events WHERE {count_where}",
+                count_params,
+            ).fetchone()
+        return int(count_row[0] or 0) > int(limit or 1)
+
+    def cleanup_old_events(
+        self,
+        *,
+        retention_hours: Optional[int] = None,
+        now: Optional[datetime] = None,
+    ) -> int:
+        hours = max(1, int(retention_hours or _retention_hours_from_env()))
+        cutoff = _iso((now or _utc_now()) - timedelta(hours=hours))
+        with self._connect() as conn:
+            cursor = conn.execute(
+                "DELETE FROM observation_patch_events WHERE created_at < ?",
+                (cutoff,),
+            )
+            deleted = int(cursor.rowcount or 0)
+            conn.commit()
+        return deleted
+
+    @staticmethod
+    def _row_to_event(row: sqlite3.Row) -> Dict[str, Any]:
+        payload = json.loads(row["payload_json"])
+        schema_type = str(row["schema_type"])
+        schema_version = int(row["schema_version"])
+        return {
+            "type": f"{schema_type}.v{schema_version}",
+            "revision": int(row["revision"]),
+            "city": str(row["city"]),
+            "source": str(row["source"]),
+            "obs_time": row["obs_time"],
+            "ts": _created_at_to_ms(row["created_at"]),
+            "payload": payload,
+        }
@@ -0,0 +1,200 @@
+"""Versioned realtime observation patch normalization."""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, Iterable, List, Optional
+
+
+SCHEMA_TYPE = "city_observation_patch"
+SCHEMA_VERSION = 1
+EVENT_TYPE = "city_observation_patch.v1"
+
+
+class PatchValidationError(ValueError):
+    """Raised when a collector patch cannot become a replayable observation event."""
+
+
+def _normalize_city(value: Any) -> str:
+    return str(value or "").strip().lower()
+
+
+def _normalize_source(value: Any) -> str:
+    source = str(value or "").strip().lower()
+    return source or "weather"
+
+
+def _finite_number(value: Any) -> Optional[float]:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    if number != number or number in {float("inf"), float("-inf")}:
+        return None
+    return round(number, 2)
+
+
+def _first_number(*values: Any) -> Optional[float]:
+    for value in values:
+        number = _finite_number(value)
+        if number is not None:
+            return number
+    return None
+
+
+def _iter_runway_points(raw_points: Any) -> Iterable[Dict[str, Any]]:
+    if isinstance(raw_points, list):
+        for item in raw_points:
+            if isinstance(item, dict):
+                yield item
+
+
+def _normalize_runway_points(raw_points: Any) -> List[Dict[str, Any]]:
+    points: List[Dict[str, Any]] = []
+    for raw in _iter_runway_points(raw_points):
+        runway = str(raw.get("runway") or raw.get("rwy") or "").strip().upper()
+        temp = _first_number(
+            raw.get("temp"),
+            raw.get("target_runway_max"),
+            raw.get("tdz_temp"),
+            raw.get("mid_temp"),
+            raw.get("end_temp"),
+        )
+        if not runway and temp is None:
+            continue
+
+        point: Dict[str, Any] = {}
+        if runway:
+            point["runway"] = runway
+        if temp is not None:
+            point["temp"] = temp
+        for key in ("tdz_temp", "mid_temp", "end_temp", "target_runway_max"):
+            value = _finite_number(raw.get(key))
+            if value is not None:
+                point[key] = value
+        if isinstance(raw.get("is_settlement"), bool):
+            point["is_settlement"] = raw["is_settlement"]
+        if point:
+            points.append(point)
+    return points
+
+
+def _legacy_changes(patch: Dict[str, Any]) -> Dict[str, Any]:
+    changes = patch.get("changes")
+    return changes if isinstance(changes, dict) else {}
+
+
+def _payload_from_legacy(changes: Dict[str, Any]) -> Dict[str, Any]:
+    amos = changes.get("amos") if isinstance(changes.get("amos"), dict) else {}
+    runway_obs = amos.get("runway_obs") if isinstance(amos.get("runway_obs"), dict) else {}
+
+    payload: Dict[str, Any] = {}
+    temp = _finite_number(changes.get("temp"))
+    if temp is not None:
+        payload["temp"] = temp
+    max_so_far = _first_number(changes.get("max_so_far"), changes.get("current_max_so_far"))
+    if max_so_far is not None:
+        payload["max_so_far"] = max_so_far
+
+    station_code = str(
+        changes.get("station_code")
+        or changes.get("icao")
+        or amos.get("icao")
+        or ""
+    ).strip().upper()
+    if station_code:
+        payload["station_code"] = station_code
+
+    station_label = str(
+        changes.get("station_label")
+        or amos.get("station_label")
+        or amos.get("station_name")
+        or ""
+    ).strip()
+    if station_label:
+        payload["station_label"] = station_label
+
+    series_key = str(changes.get("series_key") or "").strip()
+    if series_key:
+        payload["series_key"] = series_key
+    payload["unit"] = str(changes.get("unit") or "celsius").strip().lower() or "celsius"
+
+    raw_runway_points = changes.get("runway_points")
+    if raw_runway_points is None:
+        raw_runway_points = runway_obs.get("point_temperatures")
+    runway_points = _normalize_runway_points(raw_runway_points)
+    if runway_points:
+        payload["runway_points"] = runway_points
+
+    hourly = changes.get("hourly")
+    if isinstance(hourly, dict):
+        payload["hourly"] = hourly
+
+    return payload
+
+
+def _payload_from_v1(raw_payload: Any) -> Dict[str, Any]:
+    if not isinstance(raw_payload, dict):
+        return {}
+
+    payload: Dict[str, Any] = {}
+    temp = _finite_number(raw_payload.get("temp"))
+    if temp is not None:
+        payload["temp"] = temp
+    max_so_far = _finite_number(raw_payload.get("max_so_far"))
+    if max_so_far is not None:
+        payload["max_so_far"] = max_so_far
+
+    for key in ("station_code", "station_label", "series_key", "unit"):
+        value = raw_payload.get(key)
+        if isinstance(value, str) and value.strip():
+            payload[key] = value.strip()
+    if "unit" not in payload:
+        payload["unit"] = "celsius"
+
+    runway_points = _normalize_runway_points(raw_payload.get("runway_points"))
+    if runway_points:
+        payload["runway_points"] = runway_points
+    if isinstance(raw_payload.get("hourly"), dict):
+        payload["hourly"] = raw_payload["hourly"]
+    return payload
+
+
+def _has_observation(payload: Dict[str, Any]) -> bool:
+    return any(
+        key in payload
+        for key in ("temp", "max_so_far", "runway_points", "hourly")
+    )
+
+
+def normalize_observation_patch(patch: Dict[str, Any]) -> Dict[str, Any]:
+    if not isinstance(patch, dict):
+        raise PatchValidationError("patch must be an object")
+
+    if patch.get("type") == EVENT_TYPE:
+        city = _normalize_city(patch.get("city"))
+        source = _normalize_source(patch.get("source"))
+        obs_time = str(patch.get("obs_time") or "").strip() or None
+        payload = _payload_from_v1(patch.get("payload"))
+    else:
+        changes = _legacy_changes(patch)
+        city = _normalize_city(patch.get("city"))
+        source = _normalize_source(changes.get("source") or patch.get("source"))
+        obs_time = str(changes.get("obs_time") or patch.get("obs_time") or "").strip() or None
+        payload = _payload_from_legacy(changes)
+
+    if not city:
+        raise PatchValidationError("city is required")
+    if not _has_observation(payload):
+        raise PatchValidationError("patch must include temperature, max, runway, or hourly data")
+
+    return {
+        "type": EVENT_TYPE,
+        "schema_type": SCHEMA_TYPE,
+        "schema_version": SCHEMA_VERSION,
+        "city": city,
+        "source": source,
+        "obs_time": obs_time,
+        "ts": int(time.time() * 1000),
+        "payload": payload,
+    }
@@ -2,15 +2,35 @@

 from __future__ import annotations

-from typing import Any
+import time
+from typing import Any, Optional, Set

-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse

+from web.realtime_event_store import RealtimeEventStore, MAX_REPLAY_LIMIT
+from web.realtime_patch_schema import PatchValidationError, normalize_observation_patch
 from web.sse_manager import sse_manager


 router = APIRouter(tags=["events"])
+event_store = RealtimeEventStore()
+
+
+def _parse_cities_param(cities: str) -> Set[str]:
+    return {
+        item.strip().lower()
+        for item in str(cities or "").split(",")
+        if item.strip()
+    }
+
+
+def _bounded_replay_limit(value: int) -> int:
+    try:
+        limit = int(value)
+    except (TypeError, ValueError):
+        limit = 500
+    return max(1, min(MAX_REPLAY_LIMIT, limit))


@router.options("/api/events")
@@ -19,12 +39,56 @@ async def sse_events_preflight(request: Request):


@router.get("/api/events")
-async def sse_events(request: Request):
+async def sse_events(
+    request: Request,
+    cities: str = "",
+    since_revision: Optional[int] = Query(default=None),
+    replay_limit: int = Query(default=500),
+):
    user_id = getattr(request.state, "auth_user_id", None) or "anon"
    origin = request.headers.get("origin", "")
    allowed = origin in {"https://polyweather.top", "https://www.polyweather.top", "http://localhost:3000"}
+    city_set = _parse_cities_param(cities)
+    limit = _bounded_replay_limit(replay_limit)
+    latest_revision = event_store.latest_revision()
+    replay_events = []
+    resync_event = None
+
+    if since_revision is not None:
+        try:
+            replay_events = event_store.replay_events(
+                cities=city_set,
+                since_revision=max(0, int(since_revision)),
+                limit=limit,
+            )
+            if event_store.replay_requires_resync(
+                cities=city_set,
+                since_revision=max(0, int(since_revision)),
+                replay_count=len(replay_events),
+                limit=limit,
+            ):
+                resync_event = {
+                    "type": "resync_required",
+                    "reason": "replay_window_exceeded",
+                    "latest_revision": latest_revision,
+                    "ts": int(time.time() * 1000),
+                }
+        except Exception:
+            resync_event = {
+                "type": "resync_required",
+                "reason": "replay_failed",
+                "latest_revision": latest_revision,
+                "ts": int(time.time() * 1000),
+            }
+
    return StreamingResponse(
-        sse_manager.event_stream(user_id),
+        sse_manager.event_stream(
+            user_id,
+            cities=city_set,
+            replay_events=replay_events,
+            connected_revision=latest_revision,
+            resync_event=resync_event,
+        ),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
@@ -38,11 +102,15 @@ async def sse_events(request: Request):

@router.post("/api/internal/collector-patch")
 async def ingest_patch(patch: dict[str, Any]):
-    city = str(patch.get("city") or "").strip().lower()
-    changes = patch.get("changes")
-    if not city:
-        raise HTTPException(status_code=400, detail="city is required")
-    if not isinstance(changes, dict):
-        raise HTTPException(status_code=400, detail="changes must be an object")
-    event = sse_manager.broadcast(city, changes)
+    try:
+        normalized = normalize_observation_patch(patch)
+    except PatchValidationError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    try:
+        event = event_store.append_event(normalized)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail="event log write failed") from exc
+
+    sse_manager.broadcast_event(event)
    return {"ok": True, "revision": event["revision"]}
@@ -251,8 +251,8 @@ def build_city_detail_payload(
        or _build_intraday_meteorology(data),
        "vertical_profile_signal": data.get("vertical_profile_signal") or {},
        "taf": data.get("taf") or {},
-        "runway_plate_history": aggregate_runway_history(data.get("runway_plate_history") or {}, resolution),
-        "runway_band_history": build_runway_band_history(data.get("runway_plate_history") or {}, resolution),
+        "runway_plate_history": aggregate_runway_history(data.get("runway_plate_history") or {}, resolution or "10m"),
+        "runway_band_history": build_runway_band_history(data.get("runway_plate_history") or {}, resolution or "10m"),

        "risk": data.get("risk"),
        "settlement_station": data.get("settlement_station") or {},
@@ -7,7 +7,7 @@ import json
 import threading
 import time
 from collections import defaultdict
-from typing import Any, AsyncIterator, DefaultDict
+from typing import Any, AsyncIterator, DefaultDict, Iterable, Optional, Set


 HEARTBEAT_INTERVAL_SECONDS = 30
@@ -17,6 +17,7 @@ QUEUE_MAXSIZE = 256
 class SseManager:
    def __init__(self) -> None:
        self._queues: DefaultDict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set)
+        self._queue_cities: dict[int, frozenset[str]] = {}
        self._lock = threading.RLock()
        self._revision = 0

@@ -25,21 +26,57 @@ class SseManager:
            self._revision += 1
            return self._revision

+    @staticmethod
+    def _normalize_city(value: Any) -> str:
+        return str(value or "").strip().lower()
+
+    @classmethod
+    def _normalize_city_set(cls, cities: Optional[Iterable[str]]) -> Set[str]:
+        return {
+            cls._normalize_city(city)
+            for city in (cities or [])
+            if cls._normalize_city(city)
+        }
+
+    def _track_revision(self, event: dict[str, Any]) -> None:
+        try:
+            revision = int(event.get("revision") or 0)
+        except (TypeError, ValueError):
+            return
+        if revision <= 0:
+            return
+        with self._lock:
+            if revision > self._revision:
+                self._revision = revision
+
    def broadcast(self, city: str, changes: dict[str, Any]) -> dict[str, Any]:
        event = {
            "type": "city_patch",
-            "city": str(city or "").strip().lower(),
+            "city": self._normalize_city(city),
            "changes": changes or {},
            "revision": self._next_revision(),
            "ts": int(time.time() * 1000),
        }
-        if not event["city"]:
+        return self.broadcast_event(event)
+
+    def broadcast_event(self, event: dict[str, Any]) -> dict[str, Any]:
+        city = self._normalize_city(event.get("city"))
+        if city:
+            event = {**event, "city": city}
+        self._track_revision(event)
+        if not city:
            return event

        with self._lock:
-            queues = [queue for queue_set in self._queues.values() for queue in queue_set]
+            queue_items = [
+                (queue, self._queue_cities.get(id(queue), frozenset()))
+                for queue_set in self._queues.values()
+                for queue in queue_set
+            ]

-        for queue in queues:
+        for queue, subscribed_cities in queue_items:
+            if subscribed_cities and city not in subscribed_cities:
+                continue
            try:
                queue.put_nowait(event)
            except asyncio.QueueFull:
@@ -53,18 +90,37 @@ class SseManager:
                    pass
        return event

-    async def event_stream(self, user_id: str) -> AsyncIterator[str]:
+    async def event_stream(
+        self,
+        user_id: str,
+        *,
+        cities: Optional[Iterable[str]] = None,
+        replay_events: Optional[Iterable[dict[str, Any]]] = None,
+        connected_revision: Optional[int] = None,
+        resync_event: Optional[dict[str, Any]] = None,
+    ) -> AsyncIterator[str]:
        user_key = str(user_id or "anon")
+        city_set = frozenset(self._normalize_city_set(cities))
        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_MAXSIZE)
        with self._lock:
            self._queues[user_key].add(queue)
+            self._queue_cities[id(queue)] = city_set
+            if connected_revision is not None:
+                self._revision = max(self._revision, int(connected_revision or 0))

        try:
            yield self._format_event({
                "type": "connected",
                "revision": self._revision,
+                "cities": sorted(city_set),
                "ts": int(time.time() * 1000),
            })
+            for event in replay_events or []:
+                self._track_revision(event)
+                yield self._format_event(event)
+            if resync_event:
+                self._track_revision({"revision": resync_event.get("latest_revision")})
+                yield self._format_event(resync_event)
            while True:
                try:
                    event = await asyncio.wait_for(
@@ -81,6 +137,7 @@ class SseManager:
        finally:
            with self._lock:
                self._queues[user_key].discard(queue)
+                self._queue_cities.pop(id(queue), None)
                if not self._queues[user_key]:
                    self._queues.pop(user_key, None)