fix(platform): 配置发布激活时点名"将被丢掉的配置项"

config_release 是**整版本替换**语义:激活新版本后,旧版本的配置项全部不再生效。
所以新版本只要漏了某项,它就是**无声消失**的——agent_tools 里的工具白名单一少,
相关 Agent 的工具就被 fail-closed 拒掉,而现场表现只是"客服/风控什么都答不了",
没人会想到是发布配置少了一条。

本项目已经两次靠"发布前手工继承"规避(客服与风控的发布脚本里各写了一遍继承逻辑),
说明这个风险真实且反复出现。

改为在 activate 时先比对"被取代版本的配置项"与"新版本的配置项",把将被丢掉的逐条
写进 warning 日志。**不阻断激活**——有时确实是要主动撤下某项配置,拒绝会让正常运维
做不了事;这里要的是"事后能查到是谁把它弄没的"。

新增 tests/unit/service/test_config_release_dropped_items.py(3 条):丢项时点名、
完整继承时无噪音(否则运维会习惯性忽略这条日志)、首个版本不报丢项。

ruff / mypy(135 文件) / 610 unit+contract 全绿。
This commit is contained in:
2026-09-11 13:01:58 +08:00
parent a7ac1b6a1c
commit 38a9bc8285
2 changed files with 134 additions and 0 deletions
+43
View File
@@ -1,3 +1,4 @@
import logging
from datetime import UTC, datetime
from uuid import uuid4
@@ -8,6 +9,8 @@ from app.model.audit import InteractionAudit
from app.model.configuration import ConfigRelease, PlatformConfigItem
from app.model.platform import DomainEventOutbox
logger = logging.getLogger(__name__)
class ConfigReleaseError(ValueError):
pass
@@ -45,6 +48,44 @@ class ConfigReleaseService:
await self.session.flush()
return release
async def _warn_dropped_items(
self, release: ConfigRelease, previous_active: list[ConfigRelease]
) -> None:
"""列出"旧版本有、新版本没有"的配置项——它们在激活后会静默失效。
`config_release` 是**整版本替换**语义:激活新版本后,旧版本的所有配置项都不再生效。
所以只要新版本漏了某一项,它就是**无声消失**的——工具白名单随之变成空集、相关
Agent 的工具被 fail-closed 拒掉,而现场表现只是"客服/风控什么都答不了",
没人会想到是发布配置少了一条。
本项目已经两次靠"发布前手工继承"规避(客服与风控的发布脚本里各写了一遍继承逻辑),
这说明风险是真实且反复出现的。这里不阻断激活(有时确实是要撤下某项配置),
但把它显式写进日志,让运维事后能查到"是谁把它弄没的"。
"""
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],
)
async def activate(self, release_id: int, actor_id: int) -> ConfigRelease:
release = await self._get(release_id)
if release.status != "approved":
@@ -54,6 +95,8 @@ class ConfigReleaseService:
)
now = self._now()
previous_active = [previous for previous in active if previous.id != release.id]
# 把"本次激活会让哪些配置项失效"显式记进日志,见 _warn_dropped_items 的说明。
await self._warn_dropped_items(release, previous_active)
for previous in previous_active:
previous.status = "superseded"
previous.updated_at = now
@@ -0,0 +1,91 @@
"""配置发布激活时的"配置项被静默丢掉"警告。
**背景**:`config_release` 是**整版本替换**语义 —— 激活新版本后,旧版本的配置项全部不再
生效。所以新版本只要漏了某项,它就是**无声消失**的:`agent_tools` 里的工具白名单一少,
相关 Agent 的工具就被 fail-closed 拒掉,而现场表现只是"客服/风控什么都答不了",
没人会想到是发布配置少了一条。
本项目已经两次靠"发布前手工继承"规避(客服与风控的发布脚本里各写了一遍继承逻辑),
说明风险真实且反复出现。这里锁住:只要发生丢项,就必须在日志里点名。
"""
import logging
from typing import Any, cast
import pytest
from app.service.config_release_service import ConfigReleaseService
class _FakeResult:
def __init__(self, rows: list[tuple[Any, ...]]) -> None:
self._rows = rows
def all(self) -> list[tuple[Any, ...]]:
return self._rows
class _FakeSession:
"""只提供 `execute(...).all()`,因为被测方法只用到这一条路径。"""
def __init__(self, rows: list[tuple[Any, ...]]) -> None:
self._rows = rows
async def execute(self, statement: Any) -> _FakeResult:
return _FakeResult(self._rows)
class _Release:
def __init__(self, release_id: int, release_no: str) -> None:
self.id = release_id
self.release_no = release_no
def _service(rows: list[tuple[Any, ...]]) -> ConfigReleaseService:
return ConfigReleaseService(cast(Any, _FakeSession(rows)))
@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), # 新版本新增
])
with caplog.at_level(logging.WARNING):
await service._warn_dropped_items(_Release(2, "new"), [_Release(1, "old")])
assert "customer_service:faq" in caplog.text
assert "agent_tools/customer_service:faq" 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),
])
with caplog.at_level(logging.WARNING):
await service._warn_dropped_items(_Release(2, "new"), [_Release(1, "old")])
assert caplog.text == ""
@pytest.mark.asyncio
async def test_first_release_warns_about_nothing(
caplog: pytest.LogCaptureFixture,
) -> None:
"""平台首个版本没有"前一个版本",不该报丢项。"""
service = _service([("agent_tools", "customer_service:faq", 1)])
with caplog.at_level(logging.WARNING):
await service._warn_dropped_items(_Release(1, "first"), [])
assert caplog.text == ""