Move AI city stream concurrency into the scan client

The AI city forecast hook still owned stream queueing and in-flight request dedupe after the first request-client pass. Moving that policy into scanTerminalClient keeps network concurrency, queued progress, and requestKey reuse in the request layer while leaving the hook responsible only for cached UI state and progress rendering.

Constraint: Preserve existing two-stream concurrency limit and queued user-facing progress copy.

Rejected: Move localStorage cache at the same time | cache policy should be separated from stream transport policy to keep this refactor reviewable.

Confidence: high

Scope-risk: narrow

Reversibility: clean

Tested: npm run build

Not-tested: Live multi-city SSE under production latency.

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

@@ -54,10 +54,16 @@ type AiCityReadOptions = {
   city: string;
   forceRefresh?: boolean;
   locale: string;
+  requestKey?: string;
   onProgress?: (progress: AiCityStreamProgress) => void;
   signal?: AbortSignal;
 };
 
+const AI_CITY_READ_MAX_CONCURRENT_STREAMS = 2;
+const pendingAiCityReadRequests = new Map<string, Promise<AiCityForecastPayload>>();
+const queuedAiCityReadTasks: Array<() => void> = [];
+let activeAiCityReadStreams = 0;
+
 function getRemoteError(error: unknown) {
   return error instanceof Error ? error.message : String(error);
 }
@@ -265,13 +271,34 @@ async function getMarketScan(city: string, options: MarketScanQueryOptions = {})
   );
 }
 
-async function streamAiCityRead({
+function runQueuedAiCityReadTask<T>(task: () => Promise<T>, onQueued?: () => void) {
+  return new Promise<T>((resolve, reject) => {
+    const run = () => {
+      activeAiCityReadStreams += 1;
+      task()
+        .then(resolve, reject)
+        .finally(() => {
+          activeAiCityReadStreams = Math.max(0, activeAiCityReadStreams - 1);
+          const next = queuedAiCityReadTasks.shift();
+          if (next) next();
+        });
+    };
+    if (activeAiCityReadStreams < AI_CITY_READ_MAX_CONCURRENT_STREAMS) {
+      run();
+    } else {
+      onQueued?.();
+      queuedAiCityReadTasks.push(run);
+    }
+  });
+}
+
+async function streamAiCityReadRequest({
   city,
   forceRefresh = false,
   locale,
   onProgress,
   signal,
-}: AiCityReadOptions) {
+}: Omit<AiCityReadOptions, "requestKey">) {
   const headers = await buildBrowserBackendHeaders({
     Accept: "text/event-stream",
     "Content-Type": "application/json",
@@ -317,6 +344,31 @@ async function streamAiCityRead({
   return readAiCityForecastStream(response, locale, onProgress);
 }
 
+function streamAiCityRead(options: AiCityReadOptions) {
+  const pendingKey = options.requestKey || "";
+  const pending = pendingKey ? pendingAiCityReadRequests.get(pendingKey) : null;
+  if (pending) return pending;
+
+  const request = runQueuedAiCityReadTask(
+    () => streamAiCityReadRequest(options),
+    () => {
+      options.onProgress?.({
+        stage: "queued",
+        message_en:
+          "AI observation read is queued behind the cities already streaming...",
+        message_zh: "AI 观测解读已排队，正在等待前面的城市完成流式生成…",
+      });
+    },
+  ).finally(() => {
+    if (pendingKey) pendingAiCityReadRequests.delete(pendingKey);
+  });
+
+  if (pendingKey) {
+    pendingAiCityReadRequests.set(pendingKey, request);
+  }
+  return request;
+}
+
 export const scanTerminalClient = {
   getCityDetail,
   getMarketScan,

frontend/components/dashboard/scan-terminal/use-ai-city-card-data.ts

@@ -14,19 +14,12 @@ import { normalizeCityKey } from "./decision-utils";
 
 const AI_CITY_FORECAST_CACHE_PREFIX = "polyWeather_aiCityForecast_v6";
 const AI_CITY_FORECAST_CACHE_TTL_MS = 60 * 60 * 1000;
-const AI_CITY_FORECAST_MAX_CONCURRENT_STREAMS = 2;
 const CITY_MARKET_SCAN_CACHE_PREFIX = "polyWeather_cityMarketScan_v3";
 const CITY_MARKET_SCAN_CACHE_TTL_MS = 10 * 60 * 1000;
-const pendingAiCityForecastRequests = new Map<
-  string,
-  Promise<AiCityForecastPayload>
->();
-const queuedAiCityForecastTasks: Array<() => void> = [];
 const aiCityForecastStateCache = new Map<
   string,
   { state: AiCityForecastState; updatedAt: number }
 >();
-let activeAiCityForecastStreams = 0;
 
 function getStorage() {
   if (typeof window === "undefined") return null;
@@ -110,72 +103,6 @@ function writeCachedAiForecastState(key: string, state: AiCityForecastState) {
   });
 }
 
-function runQueuedAiCityForecastTask<T>(
-  task: () => Promise<T>,
-  onQueued?: () => void,
-) {
-  return new Promise<T>((resolve, reject) => {
-    const run = () => {
-      activeAiCityForecastStreams += 1;
-      task()
-        .then(resolve, reject)
-        .finally(() => {
-          activeAiCityForecastStreams = Math.max(
-            0,
-            activeAiCityForecastStreams - 1,
-          );
-          const next = queuedAiCityForecastTasks.shift();
-          if (next) next();
-        });
-    };
-    if (activeAiCityForecastStreams < AI_CITY_FORECAST_MAX_CONCURRENT_STREAMS) {
-      run();
-    } else {
-      onQueued?.();
-      queuedAiCityForecastTasks.push(run);
-    }
-  });
-}
-
-function requestAiCityForecast({
-  city,
-  forceRefresh,
-  locale,
-  onProgress,
-  requestKey,
-}: {
-  city: string;
-  forceRefresh: boolean;
-  locale: string;
-  onProgress?: (progress: AiCityStreamProgress) => void;
-  requestKey: string;
-}) {
-  const pending = pendingAiCityForecastRequests.get(requestKey);
-  if (pending) return pending;
-
-  const request = runQueuedAiCityForecastTask(async () => {
-    return scanTerminalClient.streamAiCityRead({
-      city,
-      forceRefresh,
-      locale,
-      onProgress,
-    });
-  }, () => {
-    onProgress?.({
-      stage: "queued",
-      message_en:
-        "AI observation read is queued behind the cities already streaming...",
-      message_zh: "AI 观测解读已排队，正在等待前面的城市完成流式生成…",
-    });
-  })
-    .finally(() => {
-      pendingAiCityForecastRequests.delete(requestKey);
-    });
-
-  pendingAiCityForecastRequests.set(requestKey, request);
-  return request;
-}
-
 function getAiCityStreamProgressText(progress: AiCityStreamProgress, isEn: boolean) {
   const localizedMessage = String(
     (isEn ? progress.message_en : progress.message_zh) ||
@@ -379,7 +306,7 @@ export function useAiCityForecast({
           };
     writeCachedAiForecastState(cacheKey, loadingState);
     setAiForecast(loadingState);
-    void requestAiCityForecast({
+    void scanTerminalClient.streamAiCityRead({
         city: detailCityName,
         forceRefresh: aiRefreshToken > 0,
         locale,