fix: ignore stale weather data

README.md

@@ -107,6 +107,8 @@ You can set `WEATHER_PROVIDERS` to override the regional default order. Include
 
 When NEA or JMA content conflicts with Open-Meteo for the same time and region, the briefing prioritizes the latest data from the local official agency and retains conflicting sources for your verification.
 
+Current air-quality observations from multiple sources are compared only when they are within two hours of the latest available data. Older observations are excluded from the current conflict while the source's weather forecast remains available.
+
 ### JMA office codes
 
 See [`docs/jma-office-codes.md`](docs/jma-office-codes.md) for the forecast office codes covering all 47 prefectures and usage instructions.

docs/design.md

@@ -135,6 +135,8 @@ JMA 没有 office code 时不会猜测东京或其他预报区。
 
 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。
 
+多个来源同时提供当前空气质量观测时，应用层以最新资料为基准，只把落后不超过两小时的观测转换为本轮模型来源文档。来源提供空气质量有效时刻时直接使用；没有提供时，使用包含该数据的天气快照更新时间进行比较。被淘汰的来源仍保留天气文档，筛选不处理指定日期的空气质量预报。
+
 Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。
 
 ## 语言

docs/notes.md

@@ -16,6 +16,14 @@
 
 如果厂商以后提供具体花粉种类和浓度，再把它声明为结构化过敏原能力。
 
+## 当前空气质量冲突使用固定两小时窗口
+
+空气质量会在数小时内明显变化。把相隔很久的观测并列为冲突，会让旧数值获得与最新资料相同的权重；只保留单个最新来源又会隐藏同一时段不同标准或模型之间值得核验的真实差异。
+
+因此多个当前观测只在落后最新资料不超过两小时时共同进入本轮模型上下文。两小时覆盖当前按小时更新和调度的正常时间偏差，同时能排除晚间仍返回的午后数据。阈值属于产品规则，不开放为运行时配置。
+
+这项选择成立的条件是主要空气质量来源仍按小时级频率更新。如果来源更新周期明显改变，或运行数据表明两小时会频繁误删有效资料，再根据观测到的延迟分布调整窗口。未来日期的预报不使用这项筛选。
+
 ## 默认简报语言是英文
 
 简报语言以前是隐含行为。变成配置项后，产品级默认值选择英文，现有中国大陆示例则显式填写 `zh-CN`。

docs/requirements.md

@@ -53,6 +53,7 @@
 - 较低精度匹配在用户确认前不能自动写入地点配置。
 - 中国大陆、新加坡和日本应使用适合当地的天气信息。
 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。
+- 多个来源的当前空气质量观测只有在时刻接近时才形成冲突；明显落后于最新资料的旧观测不得继续参与本轮结论。
 - 用户也可以明确指定天气来源及备用顺序。
 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。
 

tests/test_collection.py

@@ -0,0 +1,148 @@
+import pendulum
+
+from weather_briefing.application.collection import collect_weather_documents
+from weather_briefing.capabilities import CapabilityName, CapabilityProviderSet, ProviderCapabilities
+from weather_briefing.models import (
+    AirQualitySnapshot,
+    AirQualityTimeKind,
+    ResolvedLocation,
+    WeatherContextSnapshot,
+)
+
+
+def _air_quality(
+    source: str,
+    *,
+    effective_at: pendulum.DateTime | None,
+    time_kind: AirQualityTimeKind = AirQualityTimeKind.OBSERVATION,
+) -> AirQualitySnapshot:
+    return AirQualitySnapshot(
+        source_id=f"air-quality:{source}",
+        source_name=source,
+        source_url=f"https://example.invalid/{source}/air-quality",
+        effective_at=effective_at,
+        time_kind=time_kind,
+        aqi=50,
+        aqi_display="50",
+        aqi_standard="test",
+        pm25_aqi=None,
+        pm25_concentration=10,
+        pm25_unit="μg/m³",
+        category="good",
+        health_guidance="normal activity",
+    )
+
+
+def _snapshot(
+    source: str,
+    observed_at: pendulum.DateTime,
+    *,
+    air_quality_effective_at: pendulum.DateTime | None,
+    time_kind: AirQualityTimeKind = AirQualityTimeKind.OBSERVATION,
+) -> WeatherContextSnapshot:
+    return WeatherContextSnapshot(
+        source_id=f"weather:{source}",
+        source_name=source,
+        source_url=f"https://example.invalid/{source}/weather",
+        observed_at=observed_at,
+        weather_forecast=("forecast",),
+        air_quality=_air_quality(source, effective_at=air_quality_effective_at, time_kind=time_kind),
+    )
+
+
+class _Provider:
+    def __init__(self, snapshot: WeatherContextSnapshot) -> None:
+        self._snapshot = snapshot
+
+    async def fetch(self, latitude: float, longitude: float) -> WeatherContextSnapshot:
+        return self._snapshot
+
+    async def fetch_for_date(
+        self,
+        latitude: float,
+        longitude: float,
+        forecast_date: pendulum.Date,
+    ) -> WeatherContextSnapshot:
+        return self._snapshot
+
+
+def _provider_set(*snapshots: WeatherContextSnapshot) -> CapabilityProviderSet:
+    metadata = ProviderCapabilities(
+        provider_id="test",
+        provider_name="Test",
+        capabilities=frozenset({CapabilityName.WEATHER, CapabilityName.AIR_QUALITY}),
+    )
+    return CapabilityProviderSet(
+        weather=_Provider(snapshots[0]),
+        weather_metadata=metadata,
+        supplements=tuple(_Provider(snapshot) for snapshot in snapshots[1:]),
+        supplement_metadata=tuple(metadata for _ in snapshots[1:]),
+    )
+
+
+_LOCATION = ResolvedLocation(
+    id="test",
+    name="Test",
+    latitude=39.9,
+    longitude=116.3,
+    country_code="CN",
+    administrative_area="Beijing",
+    timezone="Asia/Shanghai",
+    is_mainland_china=True,
+)
+
+
+async def test_collection_drops_air_quality_observations_more_than_two_hours_behind() -> None:
+    latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai")
+    provider = _provider_set(
+        _snapshot("qweather", latest, air_quality_effective_at=None),
+        _snapshot("open-meteo", latest, air_quality_effective_at=latest.subtract(hours=8)),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert {document.id for document in documents} == {
+        "weather:qweather",
+        "air-quality:qweather",
+        "weather:open-meteo",
+    }
+
+
+async def test_collection_keeps_air_quality_observations_at_two_hour_boundary() -> None:
+    latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai")
+    provider = _provider_set(
+        _snapshot("qweather", latest, air_quality_effective_at=None),
+        _snapshot("open-meteo", latest, air_quality_effective_at=latest.subtract(hours=2)),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert {document.id for document in documents if document.id.startswith("air-quality:")} == {
+        "air-quality:qweather",
+        "air-quality:open-meteo",
+    }
+
+
+async def test_collection_does_not_filter_air_quality_forecasts() -> None:
+    latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai")
+    provider = _provider_set(
+        _snapshot(
+            "qweather",
+            latest,
+            air_quality_effective_at=latest,
+            time_kind=AirQualityTimeKind.FORECAST,
+        ),
+        _snapshot(
+            "open-meteo",
+            latest,
+            air_quality_effective_at=latest.subtract(hours=8),
+            time_kind=AirQualityTimeKind.FORECAST,
+        ),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, pendulum.date(2026, 7, 28))
+
+    assert {document.id for document in documents if document.id.startswith("air-quality:")} == {
+        "air-quality:qweather",
+        "air-quality:open-meteo",
+    }

weather_briefing/application/collection.py

@@ -4,16 +4,25 @@
 
 import asyncio
 import logging
+from dataclasses import replace
 
 import pendulum
 
 from ..capabilities import CapabilityProviderSet
-from ..models import Article, FeedConfig, ResolvedLocation, SourceDocument
+from ..models import (
+    AirQualityTimeKind,
+    Article,
+    FeedConfig,
+    ResolvedLocation,
+    SourceDocument,
+    WeatherContextSnapshot,
+)
 from ..sources import RSSFeedSource
 from ..state import SQLiteStateStore
 from ..weather import WeatherContextProvider, fetch_weather_context, snapshot_to_documents
 
 _LOGGER = logging.getLogger("weather_briefing.service")
+_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS = 2
 
 
 async def collect_rss_articles(
@@ -65,4 +74,42 @@ async def collect_weather_documents(
         )
     else:
         snapshots = (await fetch_weather_context(provider, location.latitude, location.longitude, forecast_date),)
+    snapshots = _filter_stale_air_quality_observations(snapshots)
     return tuple(document for snapshot in snapshots for document in snapshot_to_documents(snapshot))
+
+
+def _filter_stale_air_quality_observations(
+    snapshots: tuple[WeatherContextSnapshot, ...],
+) -> tuple[WeatherContextSnapshot, ...]:
+    """Remove observations too old to support a current air-quality conflict."""
+    observation_times = tuple(_air_quality_observation_time(snapshot) for snapshot in snapshots)
+    available_times = tuple(value for value in observation_times if value is not None)
+    if len(available_times) < 2:
+        return snapshots
+
+    latest_time = max(available_times)
+    earliest_retained_time = latest_time.subtract(hours=_AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS)
+    filtered: list[WeatherContextSnapshot] = []
+    for snapshot, observation_time in zip(snapshots, observation_times, strict=True):
+        air_quality = snapshot.air_quality
+        if observation_time is not None and observation_time < earliest_retained_time and air_quality is not None:
+            _LOGGER.info(
+                "Discarding stale air-quality observation source_id=%s comparison_time=%s "
+                "latest_time=%s max_lag_hours=%d",
+                air_quality.source_id,
+                observation_time.to_iso8601_string(),
+                latest_time.to_iso8601_string(),
+                _AIR_QUALITY_OBSERVATION_MAX_LAG_HOURS,
+            )
+            filtered.append(replace(snapshot, air_quality=None))
+        else:
+            filtered.append(snapshot)
+    return tuple(filtered)
+
+
+def _air_quality_observation_time(snapshot: WeatherContextSnapshot) -> pendulum.DateTime | None:
+    """Return the best available time for comparing a current observation."""
+    air_quality = snapshot.air_quality
+    if air_quality is None or air_quality.time_kind is not AirQualityTimeKind.OBSERVATION:
+        return None
+    return air_quality.effective_at or snapshot.observed_at