fix: ignore stale weather data

docs/design.md

@@ -135,6 +135,8 @@ JMA 没有 office code 时不会猜测东京或其他预报区。
 
 指定目标日期时，天气、空气质量、生活指数和过敏原要选择同一天的数据。服务不支持该日期时保留明确缺失，不复用当前观测或其他日期建议。
 
+当前来源文档在应用层关联用于新鲜度比较的时刻：天气使用快照更新时间，空气质量和过敏原优先使用各自的观测时刻，没有独立时刻时回退天气快照时间。应用层统一执行产品需求定义的新鲜度窗口；指定日期的预报和没有当前观测语义的文档不参与比较。
+
 Open-Meteo 的逐小时空气质量和花粉预报按目标日峰值生成生活建议输入。AQI 和污染物保持来源给出的标准和单位，不做跨标准换算。
 
 ## 语言

docs/requirements.md

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

tests/test_collection.py

@@ -0,0 +1,291 @@
+import pendulum
+import pytest
+
+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",
+    }
+
+
+@pytest.mark.parametrize(
+    ("snapshot", "expected_context"),
+    (
+        (
+            _snapshot(
+                "qweather",
+                pendulum.datetime(2026, 7, 27, 21, tz=None),
+                air_quality_effective_at=None,
+            ),
+            "Weather snapshot weather:qweather observation time",
+        ),
+        (
+            _snapshot(
+                "qweather",
+                pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai"),
+                air_quality_effective_at=pendulum.datetime(2026, 7, 27, 21, tz=None),
+            ),
+            "Air-quality snapshot air-quality:qweather observation time",
+        ),
+        (
+            _snapshot(
+                "qweather",
+                pendulum.datetime(2026, 7, 27, 21, tz="Asia/Shanghai"),
+                air_quality_effective_at=None,
+                include_allergen=True,
+                allergen_observed_at=pendulum.datetime(2026, 7, 27, 21, tz=None),
+            ),
+            "Allergen snapshot allergen:qweather observation time",
+        ),
+    ),
+)
+async def test_collection_rejects_ambiguous_current_observation_times(
+    snapshot: WeatherContextSnapshot,
+    expected_context: str,
+) -> None:
+    with pytest.raises(
+        ValueError,
+        match=rf"^{expected_context} must include explicit timezone information$",
+    ):
+        await collect_weather_documents(_provider_set(snapshot), _LOCATION, None)

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