fix(platform): 配置告警覆盖全部受 release 约束的表，并提供生效快照能力
**问题**：上一轮加的"配置丢失告警"只比对 platform_config_item，而受 config_release 整版本替换影响的表有**三张**（按 information_schema 核对）：platform_config_item / prompt_template_version / model_routing_rule。这个盲区造成过真实后果 —— 客服闲聊提示词 挂在 release 174，active 变成 181 后 load_active_prompt 读不到，而 Agent 侧有逐字段 兜底、回落到代码默认值，于是功能看着正常、没人发现、**一行告警都没有**。 **改动**（均在 app/service/config_release_service.py）： 1. 新增 RELEASE_SCOPED_TABLES：三张表 + 各自的**逻辑键**。逻辑键不含 release_id、 不含自增 id、也**不含 version** —— 同名提示词在不同版本里可以用不同 version， 那仍是同一份配置。清单是穷举的，并注明漏掉任何一张的后果都是静默失效。 2. 新增 effective_snapshot()：读当前生效版本在**全部三张表**里的内容，每行已剥掉 id / elease_id（见 NOT_PORTABLE_COLUMNS），可直接作为新版本的写入载荷。 发布脚本应先取它、再追加本次变更，这样"漏继承"就从"每次靠人记得"变成结构上不容易漏。 3. _warn_dropped_items 改为逐张表比对，告警里带上表名。 model_routing_rule 没有 ORM 映射，用原生 SQL 处理；它当前 0 行，但纳进来才不会将来 配了又漏。 测试：新增一条专门锁住"提示词被丢掉时也要点名"（那正是这次的盲区），并把 fake session 改成按表 + release 返回行 —— 第一版 fake 不分表，查提示词表时会拿到配置项的行、 报 KeyError，虽然真实代码是按表查的，但 fake 不真实就盖不住问题。 ruff / mypy(136 文件) / 612 unit+contract 全绿。
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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 <keys> FROM <table> 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"), [])
|
||||
|
||||
Reference in New Issue
Block a user