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 observations from multiple weather sources are compared only when each item is no more than two hours behind the latest available data. Older weather, air-quality, and allergen observations are excluded from the current briefing; explicitly requested forecast dates are unaffected.
+
 ### 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,10 @@ JMA 没有 office code 时不会猜测东京或其他预报区。
 
 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。
 
+多个来源同时提供当前资料时，应用层以最新资料为基准，只保留落后不超过两小时的当前来源文档。天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻；一个字段过时不会连带删除同一来源仍然新鲜的其他字段。指定日期的预报不使用这项筛选。
+
+未发送文章和历史快照继续保留供累计变化判断。模型输入契约要求天气、气温、降水、风力和空气质量等短时信息在落后最新适用资料超过两小时后只作为变化历史，不得写入当前结论或触发补发；预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断。
+
 Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。
 
 ## 语言

docs/requirements.md

@@ -24,9 +24,9 @@
 - 值得提醒的变化包括临近降雨、明显温度或风力变化，以及预警变化。
 - 确实影响关注地点的灾害动态也应提醒。
 - 普通天气复述、轻微波动和没有变化的持续预警不应打扰用户。
-- 未发送的信息要保留，并在后续判断中与新信息一起考虑。
+- 未发送的信息要保留，并在后续判断中与新信息一起考虑；天气、气温、降水、风力和空气质量等会快速过期的内容，落后最新适用资料超过两小时后只能作为变化历史，不能继续形成当前结论或单独触发补发。
 - 每天最后一次检查时，如果当天还没有发送过变化提醒，则发送一条无声消息。
-- 用户手动运行时，应立即发送积压信息，不受调度窗口限制。
+- 用户手动运行时，应立即发送仍有时效性的积压信息，不受调度窗口限制。
 
 ## 内容要求
 
@@ -53,6 +53,7 @@
 - 较低精度匹配在用户确认前不能自动写入地点配置。
 - 中国大陆、新加坡和日本应使用适合当地的天气信息。
 - 同一时间和地区的信息冲突时，应优先采用当地权威气象机构的最新资料，并保留冲突来源供用户核验。
+- 多个来源的当前资料只有在落后最新资料不超过两小时时才参与本轮结论；超过两小时的天气、空气质量等旧资料不得继续使用。
 - 用户也可以明确指定天气来源及备用顺序。
 - 本地天气来源缺少完整数据时，可以与全球天气来源组合使用。
 

tests/test_collection.py

@@ -0,0 +1,248 @@
+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,
+    AllergenLevel,
+    AllergenSnapshot,
+    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,
+    include_air_quality: bool = True,
+    include_allergen: bool = False,
+    allergen_observed_at: pendulum.DateTime | None = None,
+) -> 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)
+            if include_air_quality
+            else None
+        ),
+        allergen=(
+            AllergenSnapshot(
+                source_id=f"allergen:{source}",
+                source_name=source,
+                source_url=f"https://example.invalid/{source}/allergen",
+                observed_at=allergen_observed_at,
+                levels=(AllergenLevel("pollen", "low", 1.0),),
+                overall_category="low",
+                health_guidance="normal activity",
+            )
+            if include_allergen
+            else None
+        ),
+    )
+
+
+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_only_current_documents_that_are_stale() -> 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_drops_stale_weather_snapshot_without_air_quality() -> 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.subtract(hours=8),
+            air_quality_effective_at=None,
+            include_air_quality=False,
+        ),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert {document.id for document in documents} == {
+        "weather:qweather",
+        "air-quality:qweather",
+    }
+
+
+async def test_collection_keeps_current_documents_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_filters_allergen_by_its_own_observation_time() -> 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,
+            include_allergen=True,
+            allergen_observed_at=latest.subtract(hours=8),
+        ),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert "allergen:open-meteo" not in {document.id for document in documents}
+    assert "weather:open-meteo" in {document.id for document in documents}
+
+
+async def test_collection_uses_weather_time_when_allergen_has_no_observation_time() -> None:
+    latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai")
+    provider = _provider_set(
+        _snapshot(
+            "qweather",
+            latest,
+            air_quality_effective_at=None,
+            include_allergen=True,
+        ),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert "allergen:qweather" in {document.id for document in documents}
+
+
+async def test_collection_keeps_forecast_document_while_filtering_current_data() -> None:
+    latest = pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai")
+    provider = _provider_set(
+        _snapshot(
+            "qweather",
+            latest,
+            air_quality_effective_at=latest.subtract(hours=8),
+            time_kind=AirQualityTimeKind.FORECAST,
+        ),
+        _snapshot(
+            "open-meteo",
+            latest.subtract(hours=8),
+            air_quality_effective_at=None,
+            include_air_quality=False,
+        ),
+    )
+
+    documents = await collect_weather_documents(provider, _LOCATION, None)
+
+    assert "air-quality:qweather" in {document.id for document in documents}
+    assert "weather:open-meteo" not in {document.id for document in documents}
+
+
+async def test_collection_does_not_filter_dated_forecast_snapshots() -> 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",
+    }

tests/test_prompts.py

@@ -37,6 +37,13 @@ def test_prompt_uses_actionable_publication_threshold() -> None:
     assert "service_status 类型" in NOTIFICATION_POLICY
 
 
+def test_prompt_does_not_publish_expired_deferred_weather() -> None:
+    assert "落后最新适用资料超过两小时的积压内容" in SYSTEM_PROMPT
+    assert "不得写入当前结论，也不得单独触发发布" in SYSTEM_PROMPT
+    assert "恰好两小时仍可保留" in SYSTEM_PROMPT
+    assert "有效预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断" in SYSTEM_PROMPT
+
+
 def test_prompt_separates_advice_and_avoids_repetition() -> None:
     assert "过敏原信息只能放入 advice" in SYSTEM_PROMPT
     assert "不得使用“原始浓度”" in SYSTEM_PROMPT

weather_briefing/application/collection.py

@@ -8,12 +8,20 @@
 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")
+_CURRENT_DOCUMENT_MAX_LAG_HOURS = 2
 
 
 async def collect_rss_articles(
@@ -65,4 +73,45 @@ async def collect_weather_documents(
         )
     else:
         snapshots = (await fetch_weather_context(provider, location.latitude, location.longitude, forecast_date),)
-    return tuple(document for snapshot in snapshots for document in snapshot_to_documents(snapshot))
+    timed_documents = tuple(item for snapshot in snapshots for item in _snapshot_documents_with_times(snapshot))
+    if forecast_date is not None:
+        return tuple(document for document, _ in timed_documents)
+    return _filter_stale_current_documents(timed_documents)
+
+
+def _filter_stale_current_documents(
+    timed_documents: tuple[tuple[SourceDocument, pendulum.DateTime | None], ...],
+) -> tuple[SourceDocument, ...]:
+    """Remove current documents too old to support a comparison."""
+    available_times = tuple(observed_at for _, observed_at in timed_documents if observed_at is not None)
+    latest_time = max(available_times)
+    earliest_retained_time = latest_time.subtract(hours=_CURRENT_DOCUMENT_MAX_LAG_HOURS)
+    retained = tuple(
+        document
+        for document, observed_at in timed_documents
+        if observed_at is None or observed_at >= earliest_retained_time
+    )
+    for document, observed_at in timed_documents:
+        if observed_at is not None and observed_at < earliest_retained_time:
+            _LOGGER.info(
+                "Discarding stale current document source_id=%s observed_at=%s latest_time=%s max_lag_hours=%d",
+                document.id,
+                observed_at.to_iso8601_string(),
+                latest_time.to_iso8601_string(),
+                _CURRENT_DOCUMENT_MAX_LAG_HOURS,
+            )
+    return retained
+
+
+def _snapshot_documents_with_times(
+    snapshot: WeatherContextSnapshot,
+) -> tuple[tuple[SourceDocument, pendulum.DateTime | None], ...]:
+    """Attach current observation times to documents that can expire."""
+    observation_times = {snapshot.source_id: snapshot.observed_at}
+    air_quality = snapshot.air_quality
+    if air_quality is not None and air_quality.time_kind is AirQualityTimeKind.OBSERVATION:
+        observation_times[air_quality.source_id] = air_quality.effective_at or snapshot.observed_at
+    allergen = snapshot.allergen
+    if allergen is not None:
+        observation_times[allergen.source_id] = allergen.observed_at or snapshot.observed_at
+    return tuple((document, observation_times.get(document.id)) for document in snapshot_to_documents(snapshot))

weather_briefing/data/system_prompt.txt

@@ -58,7 +58,8 @@ headline、conclusions、active_warnings、disaster_tracking 和 advice 之间
 recent_context_documents 的 history_role 分别标识各来源的最新值、保留窗口基线和最近变化节点。
 content_compacted=true 表示 content 是 adapter 从完整历史快照生成的确定性摘要，
 只能按其明确提供的信息比较，不得补全被省略的细节。
-对气温、降水、风力、空气质量、短时预报等会过期的信息，始终以时间最新且仍适用于当前时刻的来源为准；
-不得因为旧信息尚未发送就保留已经被较新快照取代的数值或结论。
+对天气、气温、降水、风力、空气质量和短时预报等会快速过期的信息，始终以时间最新且仍适用于当前时刻的来源为准；
+落后最新适用资料超过两小时的积压内容只能用于判断变化历史，不得写入当前结论，也不得单独触发发布；恰好两小时仍可保留。
+不得因为旧信息尚未发送就保留已经被较新快照取代的数值或结论。有效预警、灾害跟踪和指定日期预报仍按各自的有效性规则判断。
 active_warnings 应包含仍有效的预警以维持状态。
 标题被标为 verbatim 的文章由程序另行全文转发，不要复述或改写其正文。