SSE Patch 增量推送架构全链路实现

- web/sse_manager.py: asyncio.Queue 连接池 + broadcast + event_stream
- web/routers/sse_router.py: GET /api/events + POST /api/internal/collector-patch
- weather_sources.py: 采集后 POST patch 给 web
- Nginx polyweather.conf: /api/events proxy_buffering off
- frontend/api/events/route.ts: Next.js SSE 代理
- frontend/hooks/use-sse-patches.ts: EventSource + useLatestPatch + 重连
- LiveTemperatureThresholdChart: mergePatchIntoHourly 增量合并
- scan-terminal-query: 接入 SSE patch 更新行数据
- 新增 ssePatchArchitecture.test.ts 架构测试

deploy/nginx/polyweather.conf

@@ -29,6 +29,19 @@ server {
     proxy_buffers 8 16k;
     proxy_busy_buffers_size 32k;
 
+    location /api/events {
+        proxy_pass http://127.0.0.1:8000;
+        proxy_http_version 1.1;
+        proxy_buffering off;
+        proxy_cache off;
+        proxy_read_timeout 86400s;
+        proxy_set_header Connection '';
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+    }
+
     location / {
         proxy_pass http://127.0.0.1:8000;
         proxy_http_version 1.1;

frontend/app/api/events/route.ts

@@ -0,0 +1,40 @@
+import { NextRequest, NextResponse } from "next/server";
+
+const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+export async function GET(req: NextRequest) {
+  if (!API_BASE) {
+    return NextResponse.json(
+      { error: "POLYWEATHER_API_BASE_URL is not configured" },
+      { status: 500 },
+    );
+  }
+
+  const upstream = await fetch(`${API_BASE.replace(/\/+$/, "")}/api/events`, {
+    cache: "no-store",
+    headers: {
+      Accept: "text/event-stream",
+      Cookie: req.headers.get("cookie") || "",
+    },
+  });
+
+  if (!upstream.ok || !upstream.body) {
+    return NextResponse.json(
+      { error: `SSE upstream failed with HTTP ${upstream.status}` },
+      { status: upstream.status || 502 },
+    );
+  }
+
+  return new Response(upstream.body, {
+    status: 200,
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache, no-transform",
+      Connection: "keep-alive",
+      "X-Accel-Buffering": "no",
+    },
+  });
+}

frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx

@@ -1,7 +1,7 @@
 "use client";
 
 import clsx from "clsx";
-import { useEffect, useMemo, useState } from "react";
+import { useEffect, useMemo, useRef, useState } from "react";
 import {
   CartesianGrid,
   Line,
@@ -22,6 +22,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 { Panel } from "@/components/dashboard/scan-terminal/Panel";
 import { rowName, temp } from "@/components/dashboard/scan-terminal/utils";
 
@@ -484,19 +485,51 @@ function seedHourlyForecastFromRow(row: ScanOpportunityRow | null): HourlyForeca
   };
 }
 
-async function fetchHourlyForecastForCity(city: string): Promise<HourlyForecast> {
-  const cached = _hourlyCache.get(city);
-  if (cached && Date.now() - cached.ts < HOURLY_CACHE_TTL_MS) {
-    return cached.data;
+type HourlyForecastFetchOptions = {
+  ignoreCache?: boolean;
+};
+
+function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForecast {
+  const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly;
+  if (!json || !hourlySource) return null;
+  return {
+    forecastTodayHigh: json.forecast?.today_high ?? null,
+    debPrediction: json.deb?.prediction ?? (json as any)?.overview?.deb_prediction ?? null,
+    localTime: json.local_time || null,
+    times: hourlySource.times || [],
+    temps: hourlySource.temps || [],
+    modelCurves: (json.models_hourly ?? (json as any)?.timeseries?.models_hourly)?.curves || undefined,
+    runwayPlateHistory: (json as any)?.runway_plate_history || (json.amos as any)?.runway_plate_history || undefined,
+    amos: json.amos || null,
+    airportCurrent: json.airport_current || null,
+    airportPrimary: json.airport_primary || null,
+    forecastDaily: json.forecast?.daily || [],
+    multiModelDaily: json.multi_model_daily || {},
+    settlementTodayObs: (json as any).timeseries?.settlement_today_obs || (json as any)?.settlement_today_obs || undefined,
+    metarTodayObs: (json as any).timeseries?.metar_today_obs || (json as any)?.metar_today_obs || undefined,
+    airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs || (json as any)?.airport_primary_today_obs || undefined,
+  };
+}
+
+async function fetchHourlyForecastForCity(
+  city: string,
+  options: HourlyForecastFetchOptions = {},
+): Promise<HourlyForecast> {
+  if (!options.ignoreCache) {
+    const cached = _hourlyCache.get(city);
+    if (cached && Date.now() - cached.ts < HOURLY_CACHE_TTL_MS) {
+      return cached.data;
+    }
+
+    const sessionCached = readSessionCache(city);
+    if (sessionCached) {
+      _hourlyCache.set(city, sessionCached);
+      return sessionCached.data;
+    }
   }
 
-  const sessionCached = readSessionCache(city);
-  if (sessionCached) {
-    _hourlyCache.set(city, sessionCached);
-    return sessionCached.data;
-  }
-
-  const pending = _hourlyRequestCache.get(city);
+  const requestKey = options.ignoreCache ? `${city}:live` : city;
+  const pending = _hourlyRequestCache.get(requestKey);
   if (pending) return pending;
 
   const request = fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false`, {
@@ -507,37 +540,96 @@ async function fetchHourlyForecastForCity(city: string): Promise<HourlyForecast>
       return res.json() as Promise<CityDetail>;
     })
     .then((json) => {
-      const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly;
-      if (!json || !hourlySource) return null;
-      const data: HourlyForecast = {
-        forecastTodayHigh: json.forecast?.today_high ?? null,
-        debPrediction: json.deb?.prediction ?? (json as any)?.overview?.deb_prediction ?? null,
-        localTime: json.local_time || null,
-        times: hourlySource.times || [],
-        temps: hourlySource.temps || [],
-        modelCurves: (json.models_hourly ?? (json as any)?.timeseries?.models_hourly)?.curves || undefined,
-        runwayPlateHistory: (json as any)?.runway_plate_history || (json.amos as any)?.runway_plate_history || undefined,
-        amos: json.amos || null,
-        airportCurrent: json.airport_current || null,
-        airportPrimary: json.airport_primary || null,
-        forecastDaily: json.forecast?.daily || [],
-        multiModelDaily: json.multi_model_daily || {},
-        settlementTodayObs: (json as any).timeseries?.settlement_today_obs || (json as any)?.settlement_today_obs || undefined,
-        metarTodayObs: (json as any).timeseries?.metar_today_obs || (json as any)?.metar_today_obs || undefined,
-        airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs || (json as any)?.airport_primary_today_obs || undefined,
-      };
+      const data = parseHourlyForecastFromCityDetail(json);
+      if (!data) return null;
       _hourlyCache.set(city, { ts: Date.now(), data });
       writeSessionCache(city, data);
       return data;
     })
     .finally(() => {
-      _hourlyRequestCache.delete(city);
+      _hourlyRequestCache.delete(requestKey);
     });
 
-  _hourlyRequestCache.set(city, request);
+  _hourlyRequestCache.set(requestKey, request);
   return request;
 }
 
+function shouldPollLiveChart({
+  city,
+  compact,
+  isActive,
+  isMaximized,
+}: {
+  city: string;
+  compact: boolean;
+  isActive: boolean;
+  isMaximized: boolean;
+}) {
+  return Boolean(city) && (compact || isActive || isMaximized);
+}
+
+function mergePatchIntoHourly(
+  prev: HourlyForecast,
+  patch: CityPatch,
+): HourlyForecast {
+  const changes = patch.changes || {};
+  const tempValue = validNumber(changes.temp);
+  const obsTime = typeof changes.obs_time === "string" ? changes.obs_time : null;
+  const source = typeof changes.source === "string" ? changes.source : "";
+  const explicitHourlyPatch = changes.hourly && typeof changes.hourly === "object"
+    ? changes.hourly as Partial<NonNullable<HourlyForecast>>
+    : {};
+
+  const next: NonNullable<HourlyForecast> = {
+    ...(prev || {
+      forecastTodayHigh: null,
+      debPrediction: null,
+      localTime: null,
+      times: [],
+      temps: [],
+      forecastDaily: [],
+      multiModelDaily: {},
+    }),
+    ...explicitHourlyPatch,
+  };
+
+  if (changes.amos && typeof changes.amos === "object") {
+    next.amos = changes.amos as AmosData;
+  }
+
+  if (tempValue !== null) {
+    next.airportCurrent = {
+      ...(next.airportCurrent || {}),
+      obs_time: next.airportCurrent?.obs_time ?? null,
+      temp: tempValue,
+      max_so_far: Math.max(
+        tempValue,
+        validNumber(next.airportCurrent?.max_so_far) ?? tempValue,
+      ),
+    };
+    next.airportPrimary = {
+      ...(next.airportPrimary || {}),
+      obs_time: next.airportPrimary?.obs_time ?? null,
+      temp: tempValue,
+      max_so_far: Math.max(
+        tempValue,
+        validNumber(next.airportPrimary?.max_so_far) ?? tempValue,
+      ),
+      source_label: next.airportPrimary?.source_label || source || undefined,
+    };
+  }
+
+  if (tempValue !== null && obsTime) {
+    const obsPoint: RawObsPoint = [obsTime, tempValue];
+    const currentObs = Array.isArray(next.airportPrimaryTodayObs)
+      ? next.airportPrimaryTodayObs
+      : [];
+    next.airportPrimaryTodayObs = [...currentObs, obsPoint].slice(-MAX_OBS_POINTS);
+  }
+
+  return next;
+}
+
 function parseRunwayHistoryValue(point: Record<string, unknown>) {
   return validNumber(point.max_temp_c) ?? validNumber(point.temp_c) ?? validNumber(point.temp) ?? validNumber(point.value);
 }
@@ -1155,12 +1247,17 @@ export function LiveTemperatureThresholdChart({
 }) {
   const [hourly, setHourly] = useState<HourlyForecast>(null);
   const city = String(row?.city || "").toLowerCase().trim();
+  const latestPatch = useLatestPatch(city);
   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);
 
   useEffect(() => {
     setUserToggledKeys({});
+    lastPatchAtRef.current = Date.now();
+    lastAppliedPatchRevisionRef.current = 0;
   }, [city, timeframe]);
 
   useEffect(() => {
@@ -1203,53 +1300,53 @@ export function LiveTemperatureThresholdChart({
     };
   }, [city, row, isActive, slotIndex]);
 
-  // ── 60s live poll: summary (number) + detail amos/runway (curves) ──
   useEffect(() => {
-    if (!city) return;
-    const poll = () => {
-      // Lightweight: current temp number
+    if (!latestPatch || latestPatch.revision <= lastAppliedPatchRevisionRef.current) return;
+    lastAppliedPatchRevisionRef.current = latestPatch.revision;
+    lastPatchAtRef.current = Date.now();
+    const tempValue = validNumber(latestPatch.changes.temp);
+    if (tempValue !== null) setLiveTemp(tempValue);
+    setHourly((prev) => mergePatchIntoHourly(prev ?? seedHourlyForecastFromRow(row), latestPatch));
+  }, [latestPatch, row]);
+
+  // ── SSE fallback: only full-fetch if a visible chart has seen no patch for 2 minutes ──
+  useEffect(() => {
+    if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return;
+    let cancelled = false;
+
+    const refreshFullDetail = () => {
+      lastPatchAtRef.current = Date.now();
+
+      fetchHourlyForecastForCity(city, { ignoreCache: true })
+        .then((data) => {
+          if (cancelled || !data) return;
+          setHourly(data);
+        })
+        .catch(() => {});
+    };
+
+    const checkFallback = () => {
+      if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
+      if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return;
+
       fetch(`/api/city/${encodeURIComponent(city)}/summary`)
         .then((res) => (res.ok ? res.json() : null))
         .then((payload) => {
-          if (!payload) return;
+          if (cancelled || !payload) return;
           const temp = validNumber(payload?.current?.temp);
           if (temp !== null) setLiveTemp(temp);
         })
         .catch(() => {});
 
-      // Refresh runway curves and station observations
-      fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false`)
-        .then((res) => (res.ok ? res.json() : null))
-        .then((json) => {
-          if (!json) return;
-          setHourly((prev) => {
-            if (!prev) return prev;
-            return {
-              ...prev,
-              amos: json.amos ?? prev.amos,
-              airportPrimary: json.airport_primary ?? prev.airportPrimary,
-              airportCurrent: json.airport_current ?? prev.airportCurrent,
-              airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs
-                || json.airport_primary_today_obs
-                || prev.airportPrimaryTodayObs,
-              runwayPlateHistory: (json as any)?.runway_plate_history
-                || (json.amos as any)?.runway_plate_history
-                || prev.runwayPlateHistory,
-              settlementTodayObs: (json as any).timeseries?.settlement_today_obs
-                || (json as any)?.settlement_today_obs
-                || prev.settlementTodayObs,
-              metarTodayObs: (json as any).timeseries?.metar_today_obs
-                || (json as any)?.metar_today_obs
-                || prev.metarTodayObs,
-            };
-          });
-        })
-        .catch(() => {});
+      refreshFullDetail();
     };
-    poll();
-    const id = setInterval(poll, 60_000);
-    return () => clearInterval(id);
-  }, [city]);
+
+    const id = setInterval(checkFallback, 60_000);
+    return () => {
+      cancelled = true;
+      clearInterval(id);
+    };
+  }, [city, compact, isActive, isMaximized]);
 
   const { data, series } = useMemo(() => {
     if (timeframe === "3D") {
@@ -1300,8 +1397,11 @@ export function LiveTemperatureThresholdChart({
   const isTokyo = normalizedKey === 'tokyo';
   const isSingapore = normalizedKey === 'singapore';
   const isParis = normalizedKey === 'paris';
+  const stationSourceCode = hourly?.airportPrimary?.source_code || (row as any)?.station_source_code || '';
+  const hasRealStationNetwork = /^(mgm|jma_amedas|fmi|knmi|cowin_obs|ims|ncm|aeroweb|madis_hfmetar|singapore_mss)$/.test(stationSourceCode);
   const isWeatherStation = !runwaySensorCities.has(normalizedKey)
-    && !isHKO && !isShenzhen && !isTokyo && !isSingapore && !isParis;
+    && !isHKO && !isShenzhen && !isTokyo && !isSingapore && !isParis
+    && hasRealStationNetwork;
 
   const runwayHeaderLabel = isShenzhen ? '天文台实测 (10分钟)'
     : isHKO ? '参考站点 (1分钟)'
@@ -1311,8 +1411,7 @@ export function LiveTemperatureThresholdChart({
     : isWeatherStation ? '气象站实测'
     : '跑道实测 (1分钟)';
 
-  const metarHeaderLabel = isShenzhen ? '天文台实测 (10分钟)'
-    : isHKO ? '天文台实测 (10分钟)'
+  const metarHeaderLabel = (isShenzhen || isHKO) ? '天文台实测 (10分钟)'
     : 'METAR 结算 (30分钟)';
 
   const runwayHighLabel = isShenzhen ? '天文台实测'
@@ -1800,3 +1899,4 @@ export function __buildTemperatureChartDataForTest(
 export const __isTemperatureSeriesVisibleByDefaultForTest = isTemperatureSeriesVisibleByDefault;
 export const __getVisibleTemperatureSeriesForTest = getVisibleTemperatureSeries;
 export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics;
+export const __shouldPollLiveChartForTest = shouldPollLiveChart;

frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts

@@ -5,6 +5,7 @@ import {
   DASHBOARD_REFRESH_POLICY_SEC,
 } from "@/lib/refresh-policy";
 import { scanTerminalQueryPolicy } from "@/components/dashboard/scan-terminal/scan-terminal-client";
+import { __shouldPollLiveChartForTest } from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart";
 
 function assert(condition: unknown, message: string) {
   if (!condition) throw new Error(message);
@@ -16,7 +17,7 @@ export function runTests() {
   assert(DASHBOARD_REFRESH_POLICY_MS.marketOverview === 10 * 60_000, "market overview should refresh every 10 minutes");
   assert(DASHBOARD_REFRESH_POLICY_MS.model === 30 * 60_000, "DEB and multi-model data should refresh every 30 minutes");
   assert(DASHBOARD_REFRESH_POLICY_SEC.metar === 5 * 60, "METAR polling should be 5 minutes");
-  assert(scanTerminalQueryPolicy.autoRefreshMs === DASHBOARD_REFRESH_POLICY_MS.scanRows, "scan terminal auto refresh should use the shared row cadence");
+  assert(scanTerminalQueryPolicy.autoRefreshMs === null, "scan terminal auto refresh should be disabled after SSE patch migration");
 
   const projectRoot = process.cwd();
   const querySource = fs.readFileSync(
@@ -29,8 +30,9 @@ export function runTests() {
   );
 
   assert(
-    querySource.includes("DASHBOARD_REFRESH_POLICY_MS.scanRows"),
-    "scan list local cache should use the shared 5-minute row cadence",
+    querySource.includes("useSsePatchVersion") &&
+      !querySource.includes("window.setInterval"),
+    "scan list should subscribe to SSE patch state instead of running a 5-minute interval",
   );
   assert(
     chartSource.includes("DASHBOARD_REFRESH_POLICY_MS.metar") &&
@@ -38,8 +40,23 @@ export function runTests() {
     "selected city detail chart cache should align with 5-minute scan/metar cadence",
   );
   assert(
-    chartSource.includes("setInterval(poll, 60_000)"),
-    "selected city chart should poll live temperature and runway curves every 60 seconds",
+    chartSource.includes("useLatestPatch") &&
+      chartSource.includes("latestPatch") &&
+      chartSource.includes("2 * 60_000"),
+    "selected city chart should consume SSE patches and use a 2-minute no-patch fallback",
+  );
+  assert(
+    chartSource.includes("fetchHourlyForecastForCity(city, { ignoreCache: true })") &&
+      chartSource.includes("setHourly(data)"),
+    "visible chart fallback must refresh the full city detail payload when SSE patches stop",
+  );
+  assert(
+    __shouldPollLiveChartForTest({ city: "shanghai", compact: true, isActive: false, isMaximized: false }) === true,
+    "compact grid slots are visible charts and should run the no-patch fallback guard",
+  );
+  assert(
+    __shouldPollLiveChartForTest({ city: "shanghai", compact: false, isActive: false, isMaximized: false }) === false,
+    "inactive non-compact charts should not run live polling",
   );
   assert(
     chartSource.includes("_hourlyRequestCache") &&

frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts

@@ -0,0 +1,69 @@
+import fs from "node:fs";
+import path from "node:path";
+
+function assert(condition: unknown, message: string) {
+  if (!condition) throw new Error(message);
+}
+
+function readRepoFile(...parts: string[]) {
+  return fs.readFileSync(path.join(process.cwd(), "..", ...parts), "utf8");
+}
+
+function readFrontendFile(...parts: string[]) {
+  return fs.readFileSync(path.join(process.cwd(), ...parts), "utf8");
+}
+
+export function runTests() {
+  const repoRoot = path.join(process.cwd(), "..");
+
+  const sseManagerPath = path.join(repoRoot, "web", "sse_manager.py");
+  assert(fs.existsSync(sseManagerPath), "FastAPI backend must define web/sse_manager.py");
+  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("event_stream("), "SSE manager must expose an async event_stream(user_id)");
+  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 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('"/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");
+
+  const appFactory = readRepoFile("web", "app_factory.py");
+  assert(appFactory.includes("sse_router"), "FastAPI app factory must register the SSE router");
+
+  const nginx = readRepoFile("deploy", "nginx", "polyweather.conf");
+  assert(nginx.includes("location /api/events"), "Nginx deploy config must route /api/events separately");
+  assert(nginx.includes("proxy_buffering off"), "Nginx /api/events must disable proxy buffering for SSE");
+  assert(nginx.includes("proxy_read_timeout 86400s"), "Nginx /api/events must keep SSE connections open");
+
+  const weatherSources = readRepoFile("src", "data_collection", "weather_sources.py");
+  assert(weatherSources.includes("_emit_temperature_patch_if_changed"), "collector must centralize temperature patch emission");
+  assert(weatherSources.includes("requests.post"), "collector must POST patches to the internal endpoint");
+  assert(weatherSources.includes("/api/internal/collector-patch"), "collector must POST to /api/internal/collector-patch");
+  assert(weatherSources.includes("threading.Thread"), "collector patch POST must run in a separate thread");
+
+  const hookPath = path.join(process.cwd(), "hooks", "use-sse-patches.ts");
+  assert(fs.existsSync(hookPath), "frontend must define hooks/use-sse-patches.ts");
+  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("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 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("2 * 60_000"), "temperature chart must wait two minutes without patches before full-fetch fallback");
+  assert(
+    !chart.includes("setInterval(poll, 60_000)"),
+    "temperature chart must not use unconditional 60-second full-detail polling after SSE patch migration",
+  );
+}

frontend/components/dashboard/scan-terminal/scan-terminal-client.ts

@@ -18,7 +18,7 @@ export type RemoteData<T> =
   | { status: "error"; error: string; previous?: T };
 
 export const scanTerminalQueryPolicy = {
-  autoRefreshMs: DASHBOARD_REFRESH_POLICY_MS.scanRows,
+  autoRefreshMs: null,
   manualForceRefreshCooldownMs: 2 * 60_000,
 } as const;
 

frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts

@@ -1,15 +1,18 @@
 "use client";
 
-import { useCallback, useEffect, useRef, useState } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import {
-  scanTerminalQueryPolicy,
   scanTerminalClient,
-  shouldRunAutoTerminalRefresh,
   shouldSkipManualTerminalRefresh,
 } from "@/components/dashboard/scan-terminal/scan-terminal-client";
 import { useRemoteDataQuery } from "@/components/dashboard/scan-terminal/use-remote-data-query";
 import { REGIONS } from "@/components/dashboard/scan-terminal/continent-grouping";
 import { DASHBOARD_REFRESH_POLICY_MS } from "@/lib/refresh-policy";
+import {
+  getLatestPatchesSnapshot,
+  useSsePatchVersion,
+  type CityPatch,
+} from "@/hooks/use-sse-patches";
 import type { ScanTerminalResponse } from "@/lib/dashboard-types";
 
 const SCAN_CACHE_PREFIX = "polyweather_scan_v2";
@@ -35,6 +38,44 @@ function writeScanCache(data: ScanTerminalResponse, tradingRegion: string) {
   try { localStorage.setItem(scanCacheKey(tradingRegion), JSON.stringify({ ts: Date.now(), data })); } catch { /* ignore */ }
 }
 
+function normalizeCityKey(city: string | null | undefined) {
+  return String(city || "").trim().toLowerCase();
+}
+
+function applyPatchToScanRow(row: any, patch: CityPatch) {
+  const temp = typeof patch.changes.temp === "number" && Number.isFinite(patch.changes.temp)
+    ? patch.changes.temp
+    : null;
+  if (temp === null) return row;
+  return {
+    ...row,
+    current_temp: temp,
+    current_max_so_far: Math.max(
+      temp,
+      typeof row.current_max_so_far === "number" && Number.isFinite(row.current_max_so_far)
+        ? row.current_max_so_far
+        : temp,
+    ),
+    local_time: typeof patch.changes.obs_time === "string" ? patch.changes.obs_time : row.local_time,
+    sse_revision: patch.revision,
+  };
+}
+
+function applyTerminalPatches(
+  data: ScanTerminalResponse | null | undefined,
+  patches: Map<string, CityPatch>,
+): ScanTerminalResponse | null | undefined {
+  if (!data?.rows?.length || !patches.size) return data;
+  let changed = false;
+  const rows = data.rows.map((row: any) => {
+    const patch = patches.get(normalizeCityKey(row.city));
+    if (!patch) return row;
+    changed = true;
+    return applyPatchToScanRow(row, patch);
+  });
+  return changed ? { ...data, rows } : data;
+}
+
 export function useScanTerminalQuery({
   isPro,
   proAccessLoading,
@@ -49,7 +90,6 @@ export function useScanTerminalQuery({
   const {
     data: terminalData,
     error: scanError,
-    isLoading,
     loading: scanLoading,
     remote: scanRemote,
     reset,
@@ -57,6 +97,7 @@ export function useScanTerminalQuery({
   } = useRemoteDataQuery<ScanTerminalResponse>();
 
   const lastForcedScanRefreshAtRef = useRef(0);
+  const patchVersion = useSsePatchVersion();
   const [cachedRows, setCachedRows] = useState<ScanTerminalResponse | null>(() => {
     if (typeof window !== "undefined") return readScanCache(tradingRegion || "");
     return null;
@@ -101,7 +142,10 @@ export function useScanTerminalQuery({
     void fetchScanTerminal({ forceRefresh: false, showLoading: true });
   }, [fetchScanTerminal, isPro, proAccessLoading, reset, timezoneOffsetSeconds, tradingRegion]);
 
-  const effectiveData = terminalData || cachedRows;
+  const effectiveData = useMemo(
+    () => applyTerminalPatches(terminalData || cachedRows, getLatestPatchesSnapshot()),
+    [terminalData, cachedRows, patchVersion],
+  );
 
   const refreshScanTerminalManually = useCallback(() => {
     if (
@@ -115,29 +159,6 @@ export function useScanTerminalQuery({
     void fetchScanTerminal({ forceRefresh: true, showLoading: true });
   }, [fetchScanTerminal, terminalData]);
 
-  useEffect(() => {
-    if (proAccessLoading || !isPro) return;
-    if (
-      typeof window === "undefined" ||
-      typeof window.setInterval !== "function" ||
-      typeof window.clearInterval !== "function"
-    ) {
-      return;
-    }
-    const intervalId = window.setInterval(() => {
-      if (
-        !shouldRunAutoTerminalRefresh({
-          documentHidden: document.hidden,
-          isLoading: isLoading(),
-        })
-      ) {
-        return;
-      }
-      void fetchScanTerminal({ forceRefresh: false, showLoading: false });
-    }, scanTerminalQueryPolicy.autoRefreshMs);
-    return () => window.clearInterval(intervalId);
-  }, [fetchScanTerminal, isLoading, isPro, proAccessLoading]);
-
   // Preload adjacent regions in idle time for instant tab switches
   useEffect(() => {
     if (typeof window === "undefined" || !tradingRegion || !isPro) return;

frontend/hooks/use-sse-patches.ts

@@ -0,0 +1,146 @@
+"use client";
+
+import { useEffect, useSyncExternalStore } from "react";
+import { resolveBackendApiUrl } from "@/lib/backend-api";
+
+export type CityPatch = {
+  type?: string;
+  city: string;
+  changes: Record<string, unknown>;
+  revision: number;
+  ts?: number;
+};
+
+const latestPatches = new Map<string, CityPatch>();
+const latestRevisions = new Map<string, number>();
+const cityListeners = new Map<string, Set<() => void>>();
+const globalListeners = new Set<() => void>();
+
+let eventSource: EventSource | null = null;
+let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+let reconnectAttempt = 0;
+let patchVersion = 0;
+
+function normalizeCityKey(city: string | null | undefined) {
+  return String(city || "").trim().toLowerCase();
+}
+
+function notify(city: string) {
+  patchVersion += 1;
+  cityListeners.get(city)?.forEach((listener) => listener());
+  globalListeners.forEach((listener) => listener());
+}
+
+function scheduleReconnect() {
+  if (reconnectTimer || typeof window === "undefined") return;
+  const delayMs = Math.min(30_000, 1_000 * Math.max(1, 2 ** reconnectAttempt));
+  reconnectAttempt += 1;
+  reconnectTimer = setTimeout(() => {
+    reconnectTimer = null;
+    connectSsePatches();
+  }, delayMs);
+}
+
+function closeEventSource() {
+  if (!eventSource) return;
+  eventSource.close();
+  eventSource = null;
+}
+
+function connectSsePatches() {
+  if (typeof window === "undefined" || eventSource) return;
+  closeEventSource();
+  eventSource = new EventSource(resolveBackendApiUrl("/api/events"));
+
+  eventSource.onopen = () => {
+    reconnectAttempt = 0;
+  };
+
+  eventSource.onmessage = (event) => {
+    try {
+      applySsePatch(JSON.parse(event.data));
+    } catch {
+      // Ignore malformed frames; the stream stays alive.
+    }
+  };
+
+  eventSource.onerror = () => {
+    closeEventSource();
+    scheduleReconnect();
+  };
+}
+
+export function ensureSsePatchConnection() {
+  connectSsePatches();
+}
+
+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;
+  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;
+  }
+
+  const previousRevision = latestRevisions.get(city) ?? 0;
+  if (revision <= previousRevision) return false;
+
+  const normalizedPatch: CityPatch = {
+    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);
+  return true;
+}
+
+export function getLatestPatchesSnapshot() {
+  return latestPatches;
+}
+
+export function useSsePatchVersion() {
+  useEffect(() => {
+    ensureSsePatchConnection();
+  }, []);
+
+  return useSyncExternalStore(
+    (listener) => {
+      globalListeners.add(listener);
+      return () => globalListeners.delete(listener);
+    },
+    () => patchVersion,
+    () => 0,
+  );
+}
+
+export function useLatestPatch(city: string | null | undefined) {
+  const cityKey = normalizeCityKey(city);
+
+  useEffect(() => {
+    ensureSsePatchConnection();
+  }, []);
+
+  return useSyncExternalStore(
+    (listener) => {
+      if (!cityKey) return () => {};
+      const listeners = cityListeners.get(cityKey) ?? new Set<() => void>();
+      listeners.add(listener);
+      cityListeners.set(cityKey, listeners);
+      return () => {
+        listeners.delete(listener);
+        if (!listeners.size) cityListeners.delete(cityKey);
+      };
+    },
+    () => (cityKey ? latestPatches.get(cityKey) ?? null : null),
+    () => null,
+  );
+}
+
+export const __applySsePatchForTest = applySsePatch;

src/data_collection/weather_sources.py

@@ -1,6 +1,7 @@
 import os
 import httpx
 import re
+import requests
 import threading
 import time
 from typing import Any, Optional, Dict, List
@@ -268,6 +269,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         )
         self._disk_cache_lock = threading.Lock()
         self._disk_cache_last_mtime: float = 0.0
+        self.collector_patch_endpoint = str(
+            os.getenv(
+                "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT",
+                "http://127.0.0.1:8000/api/internal/collector-patch",
+            )
+        ).strip()
+        self._last_emitted_temperature_patch: Dict[str, tuple[float, str]] = {}
+        self._temperature_patch_lock = threading.Lock()
         self._load_open_meteo_disk_cache()
         logger.info(
             f"Open-Meteo 磁盘缓存路径: {self._disk_cache_path} (max_age={self._disk_cache_max_age_sec}s)"
@@ -338,6 +347,62 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                 with lock:
                     for key in stale:
                         cache.pop(key, None)
+
+    def _emit_temperature_patch_if_changed(
+        self,
+        city: str,
+        temp: Any,
+        obs_time: Any = None,
+        *,
+        source: str = "",
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        if temp is None or not self.collector_patch_endpoint:
+            return
+        try:
+            temp_value = round(float(temp), 2)
+        except (TypeError, ValueError):
+            return
+
+        obs_time_value = str(obs_time or datetime.now(timezone.utc).isoformat())
+        source_value = str(source or "weather").strip().lower() or "weather"
+        city_value = str(city or "").strip().lower()
+        if not city_value:
+            return
+
+        dedupe_key = f"{city_value}:{source_value}"
+        signature = (temp_value, obs_time_value)
+        with self._temperature_patch_lock:
+            if self._last_emitted_temperature_patch.get(dedupe_key) == signature:
+                return
+            self._last_emitted_temperature_patch[dedupe_key] = signature
+
+        changes: Dict[str, Any] = {
+            "temp": temp_value,
+            "obs_time": obs_time_value,
+            "source": source_value,
+        }
+        if extra:
+            changes.update(extra)
+
+        payload = {"city": city_value, "changes": changes}
+
+        def post_patch() -> None:
+            try:
+                requests.post(
+                    self.collector_patch_endpoint,
+                    json=payload,
+                    timeout=2,
+                )
+            except Exception as exc:
+                logger.debug(
+                    "collector patch POST failed city={} source={}: {}",
+                    city_value,
+                    source_value,
+                    exc,
+                )
+
+        threading.Thread(target=post_patch, daemon=True).start()
         # MADIS cache is a single list, not a keyed dict — expire on age
         with self._madis_cache_lock:
             if self._madis_cache is not None and now - self._madis_cache_ts > self.madis_cache_ttl_sec * 2:
@@ -878,6 +943,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         settlement_current = self.fetch_settlement_current(city_lower)
         if settlement_current:
             results["settlement_current"] = settlement_current
+            current = settlement_current.get("current") or {}
+            self._emit_temperature_patch_if_changed(
+                city_lower,
+                current.get("temp"),
+                settlement_current.get("observation_time") or current.get("obs_time"),
+                source=str(settlement_current.get("source") or "settlement"),
+            )
 
         city_meta = self.CITY_REGISTRY.get(str(city_lower or "").strip().lower()) or {}
         settlement_source = str(city_meta.get("settlement_source") or "").strip().lower()
@@ -905,9 +977,15 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         if not mgm_data:
             return
         results["mgm"] = mgm_data
+        mgm_current = mgm_data.get("current") or {}
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            mgm_current.get("temp"),
+            mgm_data.get("obs_time"),
+            source="mgm",
+        )
         # Log airport obs for high-freq monitoring (Ankara 17128)
         try:
-            mgm_current = mgm_data.get("current") or {}
             DBManager().append_airport_obs(
                 icao=str(istno),
                 city=city_lower,
@@ -966,8 +1044,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         ims_data = self.fetch_from_ims(self.IMS_LOD_AIRPORT_STATION)
         if ims_data:
             results["ims"] = ims_data
+            ims_current = ims_data.get("current") or {}
+            self._emit_temperature_patch_if_changed(
+                city_lower,
+                ims_current.get("temp"),
+                ims_data.get("obs_time"),
+                source="ims",
+            )
             try:
-                ims_current = ims_data.get("current") or {}
                 DBManager().append_airport_obs(
                     icao="LLBG",
                     city=city_lower,
@@ -984,8 +1068,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         ncm_data = self.fetch_from_ncm()
         if ncm_data:
             results["ncm"] = ncm_data
+            ncm_current = ncm_data.get("current") or {}
+            self._emit_temperature_patch_if_changed(
+                city_lower,
+                ncm_current.get("temp"),
+                ncm_data.get("obs_time"),
+                source="ncm",
+            )
             try:
-                ncm_current = ncm_data.get("current") or {}
                 DBManager().append_airport_obs(
                     icao="OEJN",
                     city=city_lower,
@@ -1002,8 +1092,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         aw_data = self.fetch_from_aeroweb()
         if aw_data:
             results["aeroweb"] = aw_data
+            aw_current = aw_data.get("current") or {}
+            self._emit_temperature_patch_if_changed(
+                city_lower,
+                aw_current.get("temp"),
+                aw_data.get("obs_time"),
+                source="aeroweb",
+            )
             try:
-                aw_current = aw_data.get("current") or {}
                 DBManager().append_airport_obs(
                     icao="LFPB",
                     city=city_lower,
@@ -1031,6 +1127,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             return
         results["jma_official_nearby"] = official_rows
         results["jma_current"] = official_rows[0]  # primary station for airport_primary
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            official_rows[0].get("temp"),
+            official_rows[0].get("obs_time"),
+            source="jma_amedas",
+        )
         if "mgm_nearby" not in results:
             results["mgm_nearby"] = official_rows
         results["nearby_source"] = "jma"
@@ -1058,6 +1160,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             return
         results["knmi_official_nearby"] = rows
         results["knmi_current"] = rows[0]  # primary station for airport_primary
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            rows[0].get("temp"),
+            rows[0].get("obs_time"),
+            source="knmi",
+        )
         if "mgm_nearby" not in results:
             results["mgm_nearby"] = rows
         results["nearby_source"] = "knmi"
@@ -1084,6 +1192,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             return
         results["fmi_official_nearby"] = rows
         results["fmi_current"] = rows[0]  # primary station for airport_primary
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            rows[0].get("temp"),
+            rows[0].get("obs_time"),
+            source="fmi",
+        )
         if "mgm_nearby" not in results:
             results["mgm_nearby"] = rows
         results["nearby_source"] = "fmi"
@@ -1110,6 +1224,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         if not rows:
             return
         results["hko_obs_nearby"] = rows
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            rows[0].get("temp"),
+            rows[0].get("obs_time"),
+            source="hko_obs",
+        )
         if "mgm_nearby" not in results:
             results["mgm_nearby"] = rows
         results["nearby_source"] = "hko_obs"
@@ -1136,6 +1256,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             return
         results["cowin_obs_nearby"] = rows
         results["cowin_current"] = rows[0]
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            rows[0].get("temp"),
+            rows[0].get("obs_time"),
+            source="cowin_obs",
+        )
         results["mgm_nearby"] = rows
         results["nearby_source"] = "cowin_obs"
         try:
@@ -1192,6 +1318,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
         }
         results["mgm_nearby"] = [row]
         results["nearby_source"] = "cwa"
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            temp,
+            sc.get("observation_time"),
+            source="cwa",
+        )
 
     def _attach_korean_amos_data(
         self, results: Dict, city_lower: str, use_fahrenheit: bool
@@ -1209,6 +1341,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
                             city_lower, amos_data.get("temp_c"), amos_data.get("temp_source"),
                             len(amos_data.get("runway_obs", {}).get("runway_pairs", []) or []))
                 results["amos"] = amos_data
+                self._emit_temperature_patch_if_changed(
+                    city_lower,
+                    amos_data.get("temp_c"),
+                    amos_data.get("observation_time"),
+                    source="amos",
+                    extra={"amos": amos_data},
+                )
                 try:
                     DBManager().append_airport_obs(
                         icao=amos_data.get("icao") or "",
@@ -1255,6 +1394,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             )
             # Reuse the existing `amos` detail shape consumed by dashboard runway panels.
             results["amos"] = amsc_data
+            self._emit_temperature_patch_if_changed(
+                city_lower,
+                amsc_data.get("temp_c"),
+                amsc_data.get("observation_time"),
+                source="amsc_awos",
+                extra={"amos": amsc_data},
+            )
             try:
                 icao = amsc_data.get("icao") or ""
                 obs_time = amsc_data.get("observation_time") or datetime.now().isoformat()
@@ -1332,6 +1478,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             if obs["icao"] == icao:
                 # Inject into results so it flows through to airport_primary
                 results["madis_hfmetar_current"] = obs
+                self._emit_temperature_patch_if_changed(
+                    city_lower,
+                    obs.get("temp_c"),
+                    obs.get("obs_time"),
+                    source="madis_hfmetar",
+                )
                 try:
                     DBManager().append_airport_obs(
                         icao=icao,
@@ -1359,6 +1511,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
             return
 
         results["singapore_mss_current"] = obs
+        self._emit_temperature_patch_if_changed(
+            city_lower,
+            obs.get("temp_c"),
+            obs.get("obs_time"),
+            source="singapore_mss",
+        )
         try:
             DBManager().append_airport_obs(
                 icao="WSSS",

web/app_factory.py

@@ -16,6 +16,7 @@ from web.routers.auth import router as auth_router
 from web.routers.ops import router as ops_router
 from web.routers.payments import router as payments_router
 from web.routers.scan import router as scan_router
+from web.routers.sse_router import router as sse_router
 from web.routers.system import router as system_router
 from web.routes import router as legacy_router
 from web.scan_terminal_service import start_scan_terminal_prewarm
@@ -37,6 +38,7 @@ def create_app() -> FastAPI:
         core_app.include_router(auth_router)
         core_app.include_router(analytics_router)
         core_app.include_router(scan_router)
+        core_app.include_router(sse_router)
         core_app.include_router(payments_router)
         core_app.include_router(ops_router)
         core_app.include_router(legacy_router)

web/routers/sse_router.py

@@ -0,0 +1,39 @@
+"""SSE endpoints for live terminal patch delivery."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import StreamingResponse
+
+from web.sse_manager import sse_manager
+
+
+router = APIRouter(tags=["events"])
+
+
+@router.get("/api/events")
+async def sse_events(request: Request):
+    user_id = getattr(request.state, "auth_user_id", None) or "anon"
+    return StreamingResponse(
+        sse_manager.event_stream(user_id),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache, no-transform",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+@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)
+    return {"ok": True, "revision": event["revision"]}

web/sse_manager.py

@@ -0,0 +1,92 @@
+"""In-process SSE patch broadcaster for live terminal updates."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+import time
+from collections import defaultdict
+from typing import Any, AsyncIterator, DefaultDict
+
+
+HEARTBEAT_INTERVAL_SECONDS = 30
+QUEUE_MAXSIZE = 256
+
+
+class SseManager:
+    def __init__(self) -> None:
+        self._queues: DefaultDict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set)
+        self._lock = threading.RLock()
+        self._revision = 0
+
+    def _next_revision(self) -> int:
+        with self._lock:
+            self._revision += 1
+            return self._revision
+
+    def broadcast(self, city: str, changes: dict[str, Any]) -> dict[str, Any]:
+        event = {
+            "type": "city_patch",
+            "city": str(city or "").strip().lower(),
+            "changes": changes or {},
+            "revision": self._next_revision(),
+            "ts": int(time.time() * 1000),
+        }
+        if not event["city"]:
+            return event
+
+        with self._lock:
+            queues = [queue for queue_set in self._queues.values() for queue in queue_set]
+
+        for queue in queues:
+            try:
+                queue.put_nowait(event)
+            except asyncio.QueueFull:
+                try:
+                    queue.get_nowait()
+                except asyncio.QueueEmpty:
+                    pass
+                try:
+                    queue.put_nowait(event)
+                except asyncio.QueueFull:
+                    pass
+        return event
+
+    async def event_stream(self, user_id: str) -> AsyncIterator[str]:
+        user_key = str(user_id or "anon")
+        queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_MAXSIZE)
+        with self._lock:
+            self._queues[user_key].add(queue)
+
+        try:
+            yield self._format_event({
+                "type": "connected",
+                "revision": self._revision,
+                "ts": int(time.time() * 1000),
+            })
+            while True:
+                try:
+                    event = await asyncio.wait_for(
+                        queue.get(),
+                        timeout=HEARTBEAT_INTERVAL_SECONDS,
+                    )
+                except asyncio.TimeoutError:
+                    event = {
+                        "type": "heartbeat",
+                        "revision": self._revision,
+                        "ts": int(time.time() * 1000),
+                    }
+                yield self._format_event(event)
+        finally:
+            with self._lock:
+                self._queues[user_key].discard(queue)
+                if not self._queues[user_key]:
+                    self._queues.pop(user_key, None)
+
+    @staticmethod
+    def _format_event(event: dict[str, Any]) -> str:
+        return f"data: {json.dumps(event, ensure_ascii=False, separators=(',', ':'))}\n\n"
+
+
+sse_manager = SseManager()