diff --git a/app/service/config_release_service.py b/app/service/config_release_service.py index be6da76..9f3925a 100644 --- a/app/service/config_release_service.py +++ b/app/service/config_release_service.py @@ -1,8 +1,9 @@ import logging from datetime import UTC, datetime +from typing import Any from uuid import uuid4 -from sqlalchemy import select +from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.model.audit import InteractionAudit @@ -11,6 +12,25 @@ from app.model.platform import DomainEventOutbox logger = logging.getLogger(__name__) +# 受 `config_release` **整版本替换**影响的表，以及各自的**逻辑键**。 +# +# 逻辑键用来判断"同一份配置"在新版本里还在不在：它不含 `release_id`、不含自增 id， +# 也**不含 version** —— 同名提示词在不同版本里可以用不同 version，那仍是同一份配置。 +# +# 这张清单是**穷举**的：`information_schema` 里带 `release_id` 列的表只有这三张。 +# 漏掉任何一张的后果都是**静默失效**，而且不会报错：客服闲聊提示词就这么失效过一次 +# —— 它挂在 release 174，active 变成 181 后 `load_active_prompt` 读不到，而 Agent 侧 +# 有逐字段兜底、回落到代码默认值，于是功能看着正常、没有任何人发现、也没有任何告警。 +RELEASE_SCOPED_TABLES: tuple[tuple[str, tuple[str, ...]], ...] = ( + ("platform_config_item", ("namespace", "config_key")), + ("prompt_template_version", ("prompt_code", "task_type", "agent_type")), + ("model_routing_rule", ("rule_code",)), +) + +# 搬运配置项时要剥掉的列：`id` 与 `release_id` 由新版本自己生成， +# 带过去要么主键冲突、要么把内容挂到旧版本上。 +NOT_PORTABLE_COLUMNS = frozenset({"id", "release_id"}) + class ConfigReleaseError(ValueError): pass @@ -48,43 +68,89 @@ class ConfigReleaseService: await self.session.flush() return release + async def active_release_id(self) -> int | None: + """当前生效版本的 id；没有任何生效版本时返回 None。""" + found = await self.session.scalar( + select(ConfigRelease.id).where(ConfigRelease.status == "active") + ) + return int(found) if found is not None else None + + async def effective_snapshot( + self, release_id: int | None = None + ) -> dict[str, list[dict[str, Any]]]: + """读某个发布版本在**全部受管表**里的内容；默认读当前生效版本。 + + 发布脚本应当**先取这份快照**，把它原样搬到新版本、再追加本次变更 —— 因为 + `config_release` 是整版本替换，不搬就等于删（详见 RELEASE_SCOPED_TABLES 的说明）。 + 每行已剥掉 `id` 与 `release_id`（见 NOT_PORTABLE_COLUMNS），可直接作为写入载荷。 + """ + if release_id is None: + release_id = await self.active_release_id() + snapshot: dict[str, list[dict[str, Any]]] = {} + for table, _keys in RELEASE_SCOPED_TABLES: + if release_id is None: + snapshot[table] = [] + continue + rows = (await self.session.execute( + text(f"SELECT * FROM {table} WHERE release_id = :rid"), {"rid": release_id} + )).mappings().all() + snapshot[table] = [ + { + name: value + for name, value in dict(row).items() + if name not in NOT_PORTABLE_COLUMNS + } + for row in rows + ] + return snapshot + + async def _table_keys( + self, table: str, keys: tuple[str, ...], release_ids: list[int] + ) -> dict[int, set[tuple[Any, ...]]]: + """按 release 分组取出某张表的逻辑键集合。""" + columns = ", ".join(keys) + grouped: dict[int, set[tuple[Any, ...]]] = {} + for release_id in release_ids: + rows = (await self.session.execute( + text(f"SELECT {columns} FROM {table} WHERE release_id = :rid"), + {"rid": release_id}, + )).mappings().all() + grouped[release_id] = {tuple(row[key] for key in keys) for row in rows} + return grouped + async def _warn_dropped_items( self, release: ConfigRelease, previous_active: list[ConfigRelease] ) -> None: - """列出"旧版本有、新版本没有"的配置项——它们在激活后会静默失效。 + """点名"旧版本有、新版本没有"的配置 —— 它们在激活后会静默失效。 - `config_release` 是**整版本替换**语义：激活新版本后，旧版本的所有配置项都不再生效。 - 所以只要新版本漏了某一项，它就是**无声消失**的——工具白名单随之变成空集、相关 - Agent 的工具被 fail-closed 拒掉，而现场表现只是"客服/风控什么都答不了"， - 没人会想到是发布配置少了一条。 + **逐张覆盖全部受管表**。原先只比对 `platform_config_item`，于是客服闲聊提示词在 + `prompt_template_version` 里被静默丢掉时，连一行告警都没有：它挂在 release 174， + active 变成 181 后读不到，而 Agent 侧有兜底、回落到代码默认值，功能看着正常， + 于是没有人发现（详见 RELEASE_SCOPED_TABLES 的说明）。 - 本项目已经两次靠"发布前手工继承"规避（客服与风控的发布脚本里各写了一遍继承逻辑）， - 这说明风险是真实且反复出现的。这里不阻断激活（有时确实是要撤下某项配置）， - 但把它显式写进日志，让运维事后能查到"是谁把它弄没的"。 + 不阻断激活 —— 有时确实是要主动撤下某项配置；这里要的是"事后能查到是谁弄没的"。 """ if not previous_active: return - rows = (await self.session.execute( - select( - PlatformConfigItem.namespace, - PlatformConfigItem.config_key, - PlatformConfigItem.release_id, - ).where(PlatformConfigItem.release_id.in_( - [previous.id for previous in previous_active] + [release.id] - )) - )).all() - new_keys = {(namespace, key) for namespace, key, rid in rows if rid == release.id} - for previous in previous_active: - old_keys = {(namespace, key) for namespace, key, rid in rows if rid == previous.id} - dropped = sorted(old_keys - new_keys) - if dropped: - logger.warning( - "配置发布 %s 取代 %s：以下 %d 条配置项在新版本中不存在，激活后即失效 → %s", - release.release_no, - previous.release_no, - len(dropped), - [f"{namespace}/{key}" for namespace, key in dropped], + release_ids = [previous.id for previous in previous_active] + [release.id] + for table, keys in RELEASE_SCOPED_TABLES: + grouped = await self._table_keys(table, keys, release_ids) + new_keys = grouped.get(release.id, set()) + for previous in previous_active: + dropped = sorted( + (key for key in grouped.get(previous.id, set()) if key not in new_keys), + key=str, ) + if dropped: + logger.warning( + "配置发布 %s 取代 %s：%s 里有 %d 条配置在新版本中不存在，" + "激活后即失效 → %s", + release.release_no, + previous.release_no, + table, + len(dropped), + [dict(zip(keys, key, strict=True)) for key in dropped], + ) async def activate(self, release_id: int, actor_id: int) -> ConfigRelease: release = await self._get(release_id) diff --git a/tests/unit/service/test_config_release_dropped_items.py b/tests/unit/service/test_config_release_dropped_items.py index 12d6c96..e3b674d 100644 --- a/tests/unit/service/test_config_release_dropped_items.py +++ b/tests/unit/service/test_config_release_dropped_items.py @@ -1,12 +1,15 @@ -"""配置发布激活时的"配置项被静默丢掉"警告。 +"""配置发布激活时的"配置被静默丢掉"告警。 -**背景**：`config_release` 是**整版本替换**语义 —— 激活新版本后，旧版本的配置项全部不再 -生效。所以新版本只要漏了某项，它就是**无声消失**的：`agent_tools` 里的工具白名单一少， -相关 Agent 的工具就被 fail-closed 拒掉，而现场表现只是"客服/风控什么都答不了"， -没人会想到是发布配置少了一条。 +**背景**：`config_release` 是**整版本替换**语义 —— 激活新版本后，旧版本在 +`platform_config_item` / `prompt_template_version` / `model_routing_rule` 三张表里的内容 +**全部失效**。新版本只要漏了某项，它就是**无声消失**的，而且不会报错。 -本项目已经两次靠"发布前手工继承"规避（客服与风控的发布脚本里各写了一遍继承逻辑）， -说明风险真实且反复出现。这里锁住：只要发生丢项，就必须在日志里点名。 +这不是假想：客服闲聊提示词就这么失效过一次 —— 它挂在 release 174，active 变成 181 后 +`load_active_prompt` 读不到，而 Agent 侧有逐字段兜底、回落到代码默认值，于是功能看着正常， +**没有任何人发现，也没有任何告警**。第一版告警只比对 `platform_config_item`， +正是那个盲区让这件事发生。 + +所以这里逐张表锁住：只要发生丢项，就必须在日志里点名，并且说清是哪张表。 """ import logging @@ -14,25 +17,41 @@ from typing import Any, cast import pytest -from app.service.config_release_service import ConfigReleaseService +from app.service.config_release_service import RELEASE_SCOPED_TABLES, ConfigReleaseService + +_KNOWN_TABLES = tuple(table for table, _keys in RELEASE_SCOPED_TABLES) class _FakeResult: - def __init__(self, rows: list[tuple[Any, ...]]) -> None: + def __init__(self, rows: list[dict[str, Any]]) -> None: self._rows = rows - def all(self) -> list[tuple[Any, ...]]: + def mappings(self) -> "_FakeResult": + return self + + def all(self) -> list[dict[str, Any]]: return self._rows class _FakeSession: - """只提供 `execute(...).all()`，因为被测方法只用到这一条路径。""" + """按「表 + release_id」返回预设行。 - def __init__(self, rows: list[tuple[Any, ...]]) -> None: + 被测代码逐表、逐 release 查询（`SELECT FROM WHERE release_id = :rid`）， + 所以这里从 SQL 文本里认出表名，再按 `params["rid"]` 取行。不这么做的话， + 查提示词表时会拿到平台配置项的行，报 `KeyError: 'prompt_code'` + —— 第一版 fake 就踩了这个（真实代码是按表查的，问题只在 fake）。 + """ + + def __init__(self, rows: dict[tuple[str, int], list[dict[str, Any]]]) -> None: self._rows = rows - async def execute(self, statement: Any) -> _FakeResult: - return _FakeResult(self._rows) + async def execute(self, statement: Any, params: dict[str, Any] | None = None) -> _FakeResult: + sql = str(statement) + table = next((name for name in _KNOWN_TABLES if name in sql), "") + release_id = (params or {}).get("rid") + if not table or release_id is None: + return _FakeResult([]) + return _FakeResult(list(self._rows.get((table, int(release_id)), []))) class _Release: @@ -41,36 +60,67 @@ class _Release: self.release_no = release_no -def _service(rows: list[tuple[Any, ...]]) -> ConfigReleaseService: +def _service(rows: dict[tuple[str, int], list[dict[str, Any]]]) -> ConfigReleaseService: return ConfigReleaseService(cast(Any, _FakeSession(rows))) +def _item(key: str, release_id: int) -> tuple[tuple[str, int], list[dict[str, Any]]]: + return ( + ("platform_config_item", release_id), + [{"namespace": "agent_tools", "config_key": key}], + ) + + +def _prompt(code: str, release_id: int) -> tuple[tuple[str, int], list[dict[str, Any]]]: + return ( + ("prompt_template_version", release_id), + [{"prompt_code": code, "task_type": "chat", "agent_type": "customer_service"}], + ) + + @pytest.mark.asyncio async def test_warns_when_new_release_drops_config_items( caplog: pytest.LogCaptureFixture, ) -> None: - """新版本少了一条配置项时必须点名警告——它会静默失效。""" - service = _service([ - ("agent_tools", "customer_service:faq", 1), # 旧版本有 - ("agent_tools", "risk:risk_overview", 2), # 新版本新增 - ]) + """新版本少了一条平台配置项时必须点名——它会静默失效。""" + service = _service(dict([_item("customer_service:faq", 1), _item("risk:risk_overview", 2)])) with caplog.at_level(logging.WARNING): await service._warn_dropped_items(_Release(2, "new"), [_Release(1, "old")]) + assert "agent_tools" in caplog.text assert "customer_service:faq" in caplog.text - assert "agent_tools/customer_service:faq" in caplog.text + + +@pytest.mark.asyncio +async def test_warns_when_new_release_drops_a_prompt_template( + caplog: pytest.LogCaptureFixture, +) -> None: + """提示词模板被丢掉时同样要点名，并说清是哪张表。 + + 这条是本次修复的核心：第一版告警只比对 `platform_config_item`，于是客服闲聊提示词 + 在 `prompt_template_version` 里被静默丢掉时**一行告警都没有**，直到有人手工查库才发现。 + """ + service = _service(dict([_prompt("customer_service_chitchat", 1)])) + + with caplog.at_level(logging.WARNING): + await service._warn_dropped_items(_Release(2, "new"), [_Release(1, "old")]) + + assert "prompt_template_version" in caplog.text + assert "customer_service_chitchat" in caplog.text @pytest.mark.asyncio async def test_no_warning_when_everything_is_carried_over( caplog: pytest.LogCaptureFixture, ) -> None: - """配置项被完整继承时不该有噪音——否则运维会习惯性忽略这条日志。""" - service = _service([ - ("agent_tools", "customer_service:faq", 1), - ("agent_tools", "customer_service:faq", 2), - ]) + """配置被完整继承时不该有噪音——否则运维会习惯性忽略这条日志。""" + service = _service(dict([ + _item("customer_service:faq", 1), + _item("customer_service:faq", 2), + _prompt("customer_service_chitchat", 1), + _prompt("customer_service_chitchat", 2), + ])) with caplog.at_level(logging.WARNING): await service._warn_dropped_items(_Release(2, "new"), [_Release(1, "old")]) @@ -83,7 +133,7 @@ async def test_first_release_warns_about_nothing( caplog: pytest.LogCaptureFixture, ) -> None: """平台首个版本没有"前一个版本"，不该报丢项。""" - service = _service([("agent_tools", "customer_service:faq", 1)]) + service = _service(dict([_item("customer_service:faq", 1)])) with caplog.at_level(logging.WARNING): await service._warn_dropped_items(_Release(1, "first"), [])