diff --git a/AGENTS.md b/AGENTS.md index e5f9dcc0..62a91089 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -63,6 +63,7 @@ Usage notes: - Keep current requirements and architecture in `docs/`. Describe the current contract directly rather than narrating superseded designs. - Keep this file limited to development judgment, workflow constraints, and lessons that future agents could otherwise miss. Link to the other documents instead of duplicating them. - Update the document that owns a changed decision in the same change as the implementation. +- Use `docs/notes.md` to explain the rationale, trade-offs, and operating boundaries behind key architecture choices when the current contract alone would not make them clear. For accepted design concerns that remain intentionally unresolved, also state the assumptions that make the choice acceptable and concrete triggers for reevaluation; update or remove the note when those assumptions change. ## Engineering judgment diff --git a/docs/notes.md b/docs/notes.md new file mode 100644 index 00000000..48d2a216 --- /dev/null +++ b/docs/notes.md @@ -0,0 +1,35 @@ +# 架构备忘 + +本文补充 [design.md](design.md)，解释关键架构选择背后的理由、权衡和适用边界，尤其是仅从当前功能契约中不容易看出的运行规模或环境假设。它不是待办清单，也不复述正式设计；每条备忘应帮助维护者理解为什么采用当前方案，以及什么变化会使这个理由不再成立。涉及有意暂缓的有效设计问题时，还需要明确成立条件和重新评估触发点；假设变化时，应同步更新实现、正式设计文档和本文件。 + +## 异步编排中的同步本地持久化 + +### 当前选择 + +`SQLiteStateStore`、`SQLiteRuntimeDiagnostics` 和 `CachedLocationResolver` 继续使用同步 SQLite 或文件系统调用。它们直接运行在异步任务所在的事件循环线程中，不使用 `asyncio.to_thread`，也不引入异步 SQLite 依赖。 + +这个选择接受了短时间阻塞事件循环的代价。当前任务按小时级频率调度，每次运行按顺序处理地点；状态历史受保留窗口限制，定位缓存也只保存少量配置地点。正常情况下，本地 SQLite 事务和缓存文件读写远短于天气、RSS、地理编码和 LLM 网络请求，因此额外的线程或依赖不会改善用户可感知的主要延迟。 + +### 成立条件 + +- 状态数据库和定位缓存位于低延迟的本地磁盘或等效持久卷，而不是高延迟网络文件系统。 +- 单次运行仍按顺序处理地点，调度任务维持低频；应用不承担高并发请求服务的职责。 +- 历史保留窗口继续限制数据库规模，写事务保持短小。 +- 没有观测到持续的事件循环停顿、SQLite 锁等待或任务错过调度窗口。 + +手动一次性运行可以与 daemon 访问同一状态文件，因此 SQLite 锁竞争仍可能发生。当前实现依赖短事务和 SQLite 自身的进程间锁，而不是承诺并发写入吞吐量。 + +### 重新评估触发点 + +出现以下任一情况时，应测量事件循环延迟和数据库操作耗时，并重新设计持久化边界： + +- 地点开始并行处理，或 forecast、briefing、手动运行之间出现常态化并发； +- 调度频率、地点数量、历史保留量或单次写入量显著增加； +- 状态文件迁移到网络存储，或监控显示文件 I/O 延迟不可忽略； +- 出现可复现的调度漂移、消息投递延迟、数据库锁超时或其他事件循环阻塞证据。 + +### 未来改造约束 + +不能只把现有方法直接包进 `asyncio.to_thread`。`sqlite3` 连接默认具有线程亲和性，连接的创建、使用和关闭必须位于兼容的线程边界；并发写操作还需要保持现有事务原子性和确定顺序。定位缓存的临时文件替换同样需要防止并发写入互相覆盖。 + +若触发改造，优先把同步持久化封装到专用串行 worker 或重新定义状态存储 adapter，并用实际延迟数据判断是否值得引入新依赖。改造必须保留每轮成功记录与历史清理的事务边界、多地点状态隔离、缓存原子替换以及现有错误语义。 diff --git a/tests/test_llm.py b/tests/test_llm.py index d85a8cde..24d261fa 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -206,6 +206,35 @@ def test_rejects_warning_entry_not_a_dict() -> None: ) +@pytest.mark.parametrize("field", ("id", "title", "status", "detail")) +@pytest.mark.parametrize("value", (None, "", " ", 42)) +def test_rejects_warning_without_required_text(field: str, value: object) -> None: + warning = { + "id": "w1", + "title": "Warning", + "status": "active", + "detail": "Details", + "source_ids": ["source"], + } + warning[field] = value + payload = { + "headline": "Briefing", + "headline_source_ids": ["source"], + "conclusions": [], + "active_warnings": [warning], + "resolved_warning_ids": [], + "advice": [], + "disaster_tracking": [], + } + + with pytest.raises(LLMError, match=rf"active_warnings entries: {field} must be a non-empty string"): + parse_result( + payload, + pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai"), + {"source"}, + ) + + def test_rejects_warning_without_source_ids() -> None: payload = { "headline": "Briefing", @@ -261,6 +290,26 @@ def test_rejects_warning_with_unknown_source_id() -> None: ) +@pytest.mark.parametrize("headline", (None, "", " ", 42)) +def test_rejects_result_without_headline_text(headline: object) -> None: + payload = { + "headline": headline, + "headline_source_ids": ["source"], + "conclusions": [], + "active_warnings": [], + "resolved_warning_ids": [], + "advice": [], + "disaster_tracking": [], + } + + with pytest.raises(LLMError, match="headline must be a non-empty string"): + parse_result( + payload, + pendulum.datetime(2026, 7, 13, 9, tz="Asia/Shanghai"), + {"source"}, + ) + + @pytest.mark.parametrize("resolved_warning_ids", [None, "warning", [1], [""]]) def test_rejects_malformed_resolved_warning_ids(resolved_warning_ids: object) -> None: payload = { diff --git a/tests/test_service.py b/tests/test_service.py index 4913be2c..2c5edcd6 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -1135,16 +1135,17 @@ async def test_stale_feed_triggers_ops_alert(tmp_path: Path) -> None: class FailingOnceLLM: - def __init__(self, *, fail_before_response: bool = False) -> None: + def __init__(self, *, fail_before_response: bool = False, omit_headline: bool = False) -> None: self.attempts = 0 self._fail_before_response = fail_before_response + self._omit_headline = omit_headline async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dict[str, object]: self.attempts += 1 if self.attempts == 1: if self._fail_before_response: raise LLMError("request failed before a response was available") - return { + invalid_result = { "headline": "Briefing", "headline_source_ids": ["invented"], "conclusions": [{"text": "Claim", "source_ids": ["invented"]}], @@ -1153,6 +1154,9 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic "advice": [], "disaster_tracking": [], } + if self._omit_headline: + del invalid_result["headline"] + return invalid_result assert ("previous_invalid_response" in payload) is not self._fail_before_response allowed_source_ids = payload["allowed_source_ids"] assert isinstance(allowed_source_ids, list) @@ -1168,8 +1172,11 @@ async def summarize(self, system_prompt: str, payload: dict[str, object]) -> dic } -@pytest.mark.parametrize("fail_before_response", (False, True)) -async def test_llm_retry_on_validation_failure(tmp_path: Path, fail_before_response: bool) -> None: +@pytest.mark.parametrize( + ("fail_before_response", "omit_headline"), + ((False, False), (True, False), (False, True)), +) +async def test_llm_retry_on_validation_failure(tmp_path: Path, fail_before_response: bool, omit_headline: bool) -> None: timezone = pendulum.timezone("Asia/Shanghai") now = pendulum.datetime(2026, 7, 13, 9, tz=timezone) article = Article( @@ -1192,7 +1199,7 @@ async def test_llm_retry_on_validation_failure(tmp_path: Path, fail_before_respo briefing_max_characters=3500, llm_max_attempts=2, ) - llm = FailingOnceLLM(fail_before_response=fail_before_response) + llm = FailingOnceLLM(fail_before_response=fail_before_response, omit_headline=omit_headline) publisher = RecordingPublisher() delivery = DeliveryProvider(PlainTextRenderer(), publisher) diff --git a/tests/test_weather_context.py b/tests/test_weather_context.py index 5ade7bcb..191a5574 100644 --- a/tests/test_weather_context.py +++ b/tests/test_weather_context.py @@ -1,4 +1,5 @@ import base64 +from typing import TypeGuard import httpx import jwt @@ -25,6 +26,10 @@ def authorization_header(self) -> str: return "Bearer runtime-token" +def _is_string_keyed_dict(value: object) -> TypeGuard[dict[str, object]]: + return isinstance(value, dict) and all(isinstance(key, str) for key in value) + + _QWEATHER_DAILY_ITEM = { "fxDate": "2026-07-13", "textDay": "晴", @@ -747,6 +752,59 @@ async def test_open_meteo_rejects_empty_forecast() -> None: await OpenMeteoProvider(client).fetch(1, 2) +@pytest.mark.parametrize( + "missing_field", + ( + "time", + "weather_code", + "temperature_2m_max", + "temperature_2m_min", + "apparent_temperature_max", + "apparent_temperature_min", + "precipitation_sum", + "precipitation_probability_max", + "wind_speed_10m_max", + "wind_gusts_10m_max", + "wind_direction_10m_dominant", + "uv_index_max", + ), +) +async def test_open_meteo_forecast_identifies_missing_required_field(missing_field: str) -> None: + response = _open_meteo_weather_response() + daily = response["daily"] + assert _is_string_keyed_dict(daily) + del daily[missing_field] + + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=response))) as client: + with pytest.raises( + WeatherContextError, + match=f"weather forecast parsing failed: daily forecast missing required field: {missing_field}", + ): + await OpenMeteoProvider(client).fetch(1, 2) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + ( + ("daily", [], "daily forecast must be an object"), + ("weather_code", 1, "daily forecast field must be an array: weather_code"), + ("weather_code", [], "daily forecast field has no value at index 0: weather_code"), + ), +) +async def test_open_meteo_forecast_identifies_invalid_field_shape(field: str, value: object, message: str) -> None: + response = _open_meteo_weather_response() + if field == "daily": + response[field] = value + else: + daily = response["daily"] + assert _is_string_keyed_dict(daily) + daily[field] = value + + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, json=response))) as client: + with pytest.raises(WeatherContextError, match=f"weather forecast parsing failed: {message}"): + await OpenMeteoProvider(client).fetch(1, 2) + + async def test_open_meteo_air_quality_failure_is_logged_without_failing_weather(caplog) -> None: def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/v1/forecast": diff --git a/weather_briefing/llm.py b/weather_briefing/llm.py index fc20607d..f64d2aaa 100644 --- a/weather_briefing/llm.py +++ b/weather_briefing/llm.py @@ -123,6 +123,12 @@ def sourced_text(value: Mapping[str, Any], key: str) -> str: raise LLMError(f"{key} entries must contain non-empty text") return text + def required_text(value: Mapping[str, Any], field: str, *, context: str = "") -> str: + text = value.get(field) + if not isinstance(text, str) or not text.strip(): + raise LLMError(f"{context}{field} must be a non-empty string") + return text + def parse_sourced_text_items(key: str) -> tuple[Conclusion, ...]: values = payload.get(key, []) if not isinstance(values, list): @@ -170,10 +176,10 @@ def advice() -> tuple[Advice, ...]: raise LLMError("active_warnings entries must be objects") warnings.append( Warning( - id=str(value["id"]), - title=str(value["title"]), - status=str(value["status"]), - detail=str(value["detail"]), + id=required_text(value, "id", context="active_warnings entries: "), + title=required_text(value, "title", context="active_warnings entries: "), + status=required_text(value, "status", context="active_warnings entries: "), + detail=required_text(value, "detail", context="active_warnings entries: "), source_ids=cited_source_ids(value, "source_ids"), last_confirmed_at=now, ) @@ -185,7 +191,7 @@ def advice() -> tuple[Advice, ...]: parsed_advice = advice() parsed_disaster_tracking = parse_sourced_text_items("disaster_tracking") return BriefingResult( - headline=str(payload["headline"]), + headline=required_text(payload, "headline"), headline_source_ids=cited_source_ids(payload, "headline_source_ids"), conclusions=parsed_conclusions, active_warnings=tuple(warnings), diff --git a/weather_briefing/weather_context.py b/weather_briefing/weather_context.py index 5a26d8ec..936edd69 100644 --- a/weather_briefing/weather_context.py +++ b/weather_briefing/weather_context.py @@ -34,6 +34,10 @@ class _QWeatherResponseError(ValueError): """Raised for safe, code-defined QWeather response contract errors.""" +class _OpenMeteoResponseError(ValueError): + """Raised for safe, code-defined Open-Meteo response contract errors.""" + + class WeatherContextProvider(Protocol): async def fetch(self, latitude: float, longitude: float) -> WeatherContextSnapshot: ... @@ -375,8 +379,10 @@ async def fetch( ) response.raise_for_status() payload = response.json() - daily: dict[str, list[object]] = payload["daily"] - times = daily["time"] + daily = payload["daily"] + if not _is_string_keyed_dict(daily): + raise _OpenMeteoResponseError("daily forecast must be an object") + times = _open_meteo_daily_values(daily, "time") forecast_count = min(2, len(times)) if forecast_date is None else len(times) weather_forecast = tuple(_format_open_meteo_day(daily, index) for index in range(forecast_count)) if not weather_forecast: @@ -389,6 +395,8 @@ async def fetch( ) except WeatherContextError: raise + except _OpenMeteoResponseError as exc: + raise WeatherContextError(f"Open-Meteo weather forecast parsing failed: {exc}") from None except (httpx.HTTPError, KeyError, TypeError, ValueError) as exc: raise WeatherContextError(f"Open-Meteo weather forecast failed: {_safe_provider_error(exc)}") from None @@ -706,6 +714,10 @@ def _is_string_keyed_dict(value: object) -> TypeGuard[dict[str, object]]: return isinstance(value, dict) and all(isinstance(key, str) for key in value) +def _is_object_list(value: object) -> TypeGuard[list[object]]: + return isinstance(value, list) + + def _format_qweather_day(item: object) -> str: if not _is_string_keyed_dict(item): raise TypeError("daily forecast entries must be objects") @@ -730,16 +742,34 @@ def _format_qweather_day(item: object) -> str: ) -def _format_open_meteo_day(daily: dict[str, list[object]], index: int) -> str: +def _open_meteo_daily_values(daily: dict[str, object], field: str) -> list[object]: + if field not in daily: + raise _OpenMeteoResponseError(f"daily forecast missing required field: {field}") + values = daily[field] + if not _is_object_list(values): + raise _OpenMeteoResponseError(f"daily forecast field must be an array: {field}") + return values + + +def _open_meteo_daily_value(daily: dict[str, object], field: str, index: int) -> object: + values = _open_meteo_daily_values(daily, field) + if index >= len(values): + raise _OpenMeteoResponseError(f"daily forecast field has no value at index {index}: {field}") + return values[index] + + +def _format_open_meteo_day(daily: dict[str, object], index: int) -> str: return ( - f"{daily['time'][index]}：WMO天气代码{daily['weather_code'][index]}，" - f"{daily['temperature_2m_min'][index]}~{daily['temperature_2m_max'][index]}℃，" - f"体感{daily['apparent_temperature_min'][index]}~" - f"{daily['apparent_temperature_max'][index]}℃，" - f"预计降水{daily['precipitation_sum'][index]}毫米，" - f"最高降水概率{daily['precipitation_probability_max'][index]}%，" - f"最大风速{daily['wind_speed_10m_max'][index]}千米/小时，" - f"最大阵风{daily['wind_gusts_10m_max'][index]}千米/小时，" - f"主导风向{daily['wind_direction_10m_dominant'][index]}°，" - f"最高紫外线指数{daily['uv_index_max'][index]}" + f"{_open_meteo_daily_value(daily, 'time', index)}：" + f"WMO天气代码{_open_meteo_daily_value(daily, 'weather_code', index)}，" + f"{_open_meteo_daily_value(daily, 'temperature_2m_min', index)}~" + f"{_open_meteo_daily_value(daily, 'temperature_2m_max', index)}℃，" + f"体感{_open_meteo_daily_value(daily, 'apparent_temperature_min', index)}~" + f"{_open_meteo_daily_value(daily, 'apparent_temperature_max', index)}℃，" + f"预计降水{_open_meteo_daily_value(daily, 'precipitation_sum', index)}毫米，" + f"最高降水概率{_open_meteo_daily_value(daily, 'precipitation_probability_max', index)}%，" + f"最大风速{_open_meteo_daily_value(daily, 'wind_speed_10m_max', index)}千米/小时，" + f"最大阵风{_open_meteo_daily_value(daily, 'wind_gusts_10m_max', index)}千米/小时，" + f"主导风向{_open_meteo_daily_value(daily, 'wind_direction_10m_dominant', index)}°，" + f"最高紫外线指数{_open_meteo_daily_value(daily, 'uv_index_max', index)}" )