group_xinghuo_jinrong/docs/项目框架设计/开发计划-架构改进.md

改进实施计划 · 全量版

文档状态：实施计划，尚未动代码。待审核通过后按批次实施。 代码基线：分支 risk-control-agent，HEAD 2d0e2fa，pytest 503 绿。 覆盖范围：《改进方案评审-问题清单与对比.md》中全部 24 项（21 项原始 + 第一轮审核补充 3 项），另含用户本轮指定的 Redis 分布式锁与告警补充。 关键约束：用户明确"马上合并代码了" → 合并前只做零/低风险项，结构性重构一律后置。 上游文档：《架构设计说明书.md》（事实来源）、《改进方案评审-问题清单与对比.md》（已过第一轮审核，无阻断级错误）。

审核记录

第二轮（2026-09-09，独立 AI 全新上下文审核本计划）

总体结论：需修改后实施（小改）。 计划主体可行、降级链成立、24 项覆盖完整；Lua 脚本形式、FakeGateway 降级、分批策略均经代码核验成立。发现 1 处事实错误 + 1 处设计 caveat + 2 项可选，本轮已全部吸收。

#	审核意见	档位	本轮处理
1	run_locked 调用点是三处不是两处，漏了 alert_service.py:239 的 agg:suitability:	重要	已更正为三处（§4.1.1、§8.2）
2	TTL=10s 引入"临界区超时导致互斥丢失"这一当前不存在的新失败模式；计划称"远超毫秒级实测"但无实测数据	重要	TTL 上调至 30 秒，并补入风险表（§4.1.5、§4.1.7）
3	合并前须确认 locks.py / redis_gateway.py 在目标分支无在途改动	重要	已补入合并前检查（§8.4）
4	C4 改动需在 main.py 新增 agent_service 引用	可选	已补提醒（§3.3）
5	新增锁测试应覆盖三处 key 形态，不能只测两处	可选	已补充（§8.2）

未吸收意见：无。

关于意见 1 的说明：此处是本文初版的实际疏漏——grep run_locked 的输出中本已包含 alert_service.py:239，撰写时漏计。该调用点是 R-02 适当性阻断预警的聚合锁，属最关键业务路径之一，若实施时漏改会导致"唯一阻断点"的聚合失去保护。审核此项有实质价值。

---

0. 结论速览

批次	项数	内容	时机	风险
第 1 批	9	A 类文档口径 ×5、G 类文档标注 ×3、C4 无 key 告警	合并前	极低（文档 + 1 处日志）
第 2 批	2	B6 Redis 分布式锁、B2 中间件测试守卫	合并前	中 / 低
第 3 批	5	B1 唯一索引+重试、B5 限流 EXPIRE、C1 只读账号、F1 convert、F2 意图识别	合并后（需拍板）	中~高
第 4 批	8	D1 日志基建、E1~E4 可维护性、F3 动态评分、B3/B4 判据固化	合并后	中~高
合计	24

本批（合并前）只做第 1、2 批共 11 项，理由是其余项或需拍板、或涉及 DDL、或属结构性重构，放在合并窗口内风险不可控。

---

1. 背景与约束

1.1 决策链

1. 本轮产出《架构设计说明书》，§7.3 列出 11 项改进空间、§9.4 列出 5 处文档口径不符
2. 补充发现 3 项（B1 seq 并发重号、D1 日志占位、F1 convert 能力收窄）
3. 第一轮独立 AI 审核：六项核心断言全属实，无阻断级错误，提 3 重要 + 2 可选，已全部吸收
4. 用户拍板：补告警、换 Redis 分布式锁，理由是"马上合并代码"
5. 用户补充：更新文档需覆盖所有改动项，不止这两项

1.2 硬约束

约束	影响
马上合并代码	合并前改动必须保守、可回滚、不触碰结构
五条红线	Core 只读、审计只 INSERT、不自动冻结/改评级、仅 R-02 可阻断、四 Agent 不互调 LLM
pytest 503 绿	任何改动后必须仍全绿
改路由需同步	tests/test_main.py::test_all_routers_mounted 硬编码路由清单
测试不依赖 Redis	redis_gateway 一律 fake 注入，新代码不得假设 Redis 可用
新增依赖需确认	技术选型硬阀门

---

2. 分批策略与理由

合并前 ── 第1批：文档口径/标注 + 告警      （不动逻辑，零回归风险）
      └─ 第2批：Redis锁 + 测试守卫        （用户指定；双层降级保证不劣化）

合并后 ── 第3批：需拍板或涉及 DDL 的项
      └─ 第4批：结构性重构

为什么 B6（Redis 锁）放在合并前：用户明确指定。设计上采用双层降级，最坏情况退回当前行为，不会比现状差。

为什么 B1（唯一索引）不放合并前：涉及 DDL 变更，且需先核查历史数据有无重号；合并窗口内做 DDL 风险过高。

为什么 D1（日志基建）放最后：它是全局性改动，会触及所有模块的日志调用，最适合在合并完成、分支稳定后单独做。

---

3. 第 1 批：合并前必做（极低风险，9 项）

3.1 A 类 · 文档口径修正（5 项，纯文档）

ID	文件	现状 → 修正
A1	docs/memory/MEMORY.md §0	"注入词表 42 条" → 45 条（实测：指令覆盖 18 + 角色重置 11 + 系统提示泄露 9 + 越权诱导 7）
A2	docs/memory/MEMORY.md 仓库地图	"风控四 Tool" → 5 个（query_overdue_alerts / alert_query / customer_context / suitability_check / aml_lookup；query_agent_behavior 定义未注册）
A3	docs/项目框架设计/表设计/02-mysql-agent专用.sql 注释	"5 张" → 6 张（多出 risk_aml_list）
A4	架构说明书 §4.3.2	补明 scoring.py 为 NotImplementedError 预留桩，L3 risk_score 一期恒 NULL
A5	架构说明书 §4.3.4	补明 locks.py 为进程内锁，多实例失效为 TODO（本方案第 2 批解决）

验证：人工核对文档与代码计数一致，无需跑测试。

3.2 G 类 · 非缺陷标注（3 项，纯文档）

这三项看起来像问题但设计正确，写进架构说明书可避免后续被重复提出：

ID	事项	要写清什么
G1	X-Trace-Id 可前端伪造	有格式白名单 ^[A-Za-z0-9._-]{1,64}$ 防响应头注入；身份只认 JWT，trace_id 绝不用于权限判定
G2	trace_id 概率唯一	uuid4().hex[:16] = 64 bit；请求级 ID，约 2^32 次请求才 50% 碰撞；真撞仅两条日志串一起，无需数据库唯一校验
G3	admin.py / knowledge.py 空壳	T-21 拍板一期只做脚本入库；未 include_router 不影响启动；knowledge.py 空 ≠ RAG 缺失（能力在 service/rag_service.py）

3.3 C4 · 无 DeepSeek key 启动告警

现状（agent_service.py:148-150）：无 key 时静默返回 _degraded_reply，不抛异常不告警。生产漏配 key 会"看起来正常"。

改动（app/main.py lifespan，紧跟现有 JWT_DEV_SECRET 告警之后）：

if not settings.deepseek_api_key:
    logger.warning(
        "DEEPSEEK_API_KEY 未配置，对话将走降级回复（前缀 %s），LLM 能力不可用",
        agent_service._DEGRADED_PREFIX,
    )

设计要点：启动期一次性告警，不阻塞启动，沿用 main.py:61-65 现有 dev secret 告警的模式与位置。不在请求路径告警（会刷屏）。

实施提醒（第二轮审核补充）：main.py 需新增对 app.service.agent_service 的引用以取 _DEGRADED_PREFIX（该常量位于 agent_service.py:37）。属低风险 import，但需确认不引入循环导入——若出现，改为直接写字面量常量。

风险：无。仅新增日志。

3.4 审计降级告警强化

现状：utils/authz.py:70-76 与 audit_middleware.py:78 已有 logger.exception/warning，但未打印 trace_id——审计失败时连是哪次请求都不知道。

改动：两处补 trace_id 与关键上下文：

# utils/authz.py
logger.exception(
    "authz 审计失败（已降级，拒绝语义不变）：code=%s agent=%s actor=%s trace_id=%s",
    code, agent_type, actor_id, current_trace(),
)

# api/audit_middleware.py
logger.warning(
    "http_access 审计失败（降级不阻塞）：status=%s path=%s trace_id=%s",
    status, request.url.path, current_request_id(),
    exc_info=True,
)

说明：Python 默认 root logger 会输出 WARNING 及以上到 stderr，uvicorn 控制台能看到，所以本改动即使不落地 D1 也有价值。D1（格式化 + 落盘）属第 4 批。

风险：无。仅改日志内容。

3.5 本批不涉及代码逻辑改动

第 1 批 9 项中，只有 C4 和 3.4 动代码，且都只加日志，不改任何业务分支。

---

4. 第 2 批：合并前做（2 项）

4.1 B6 · Redis 分布式锁（用户指定）

4.1.1 现状

app/service/risk/locks.py:18-40 用 threading.Lock，只活在单进程内存里，跨进程不可见。多实例部署时两实例各有各的锁，同一客户并发请求会同时进入临界区。

调用点共三处（第二轮审核更正，初版漏计第三处）：

#	位置	key 形态	保护对象
1	alert_service.py:180	agg:event:{customer_id}:{date}	交易事件预警聚合首单
2	alert_service.py:239	agg:suitability:{customer_id}:{product_id}:{date}	R-02 适当性阻断预警聚合
3	profile_l3.py:155	l3:{customer_id}	L3 监测层更新

第 2 处尤其不能漏：它是 record_suitability_alert（R-02 阻断路径）的聚合锁，而 R-02 是全系统唯一能阻断交易的规则。若实施时漏改，这条最关键路径的聚合将失去并发保护——同客户同产品同日可能出多张重复阻断单。

现有兜底：L3 侧另有数据库乐观锁（computed_at 比对 + 3 次重试），跨进程天然有效，最坏是"多出一张重复预警单"，非数据错乱。

4.1.2 三条硬约束（决定设计）

① 测试完全不依赖本机 Redis —— tests/test_integration_risk.py:6 明确写"redis_gateway fake（不依赖本机 Redis）"，多测试用 monkeypatch.setattr(redis_gateway, "_gateway", FakeGateway()) 注入。新代码绝不能假设 Redis 可用。

② FakeGateway 只有部分方法 —— 只实现 publish/delete/exists/incr/rpush/lrange/ltrim/expire，不会有 acquire_lock/release_lock。调用会抛 AttributeError，设计上须归类为"Redis 不可用"。

③ 测试无直接 run_locked 用例 —— 锁是间接被测的，需新增针对性测试。

4.1.3 设计：双层锁，Redis 为主、进程内为备

run_locked(key, fn)
   ├─ 尝试 Redis 锁（SET key token NX EX ttl）
   │     ├─ 拿到            → fn(locked=True)，finally 释放   ✅ 跨进程互斥
   │     ├─ 等待超时未拿到   → 告警 → fn(locked=False)        ⚠ 保持原降级语义
   │     └─ Redis 不可用     → 落到进程内锁 ↓                  ⚠ 单进程内仍互斥
   └─ 进程内锁（threading.Lock，原逻辑原样保留）
         ├─ 拿到 → fn(locked=True)
         └─ 超时 → fn(locked=False)

为什么保留进程内锁：Redis 挂掉时若直接无锁，连"单进程串行保护"也丢了。保留双层，最坏退回当前行为，不会更差。

4.1.4 代码设计

# app/service/risk/redis_gateway.py —— 新增两个方法（纯新增，不动现有方法）

class RedisGateway:
    def acquire_lock(self, key: str, token: str, ttl_seconds: int) -> bool:
        """SET key token NX EX ttl：抢到返回 True，已存在返回 False。"""
        return bool(self._ensure().set(key, token, nx=True, ex=ttl_seconds))

    def release_lock(self, key: str, token: str) -> bool:
        """Lua 脚本释放：只删自己持有的锁，防误删他人锁。"""
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return bool(self._ensure().eval(script, 1, key, token))

为什么释放必须用 Lua：GET 判断 + DEL 删除两步，在 A 锁超时后 B 拿到锁时，A 的 DEL 会误删 B 的锁。Lua 保证"读-比对-删"原子。

# app/service/risk/locks.py —— 改造 run_locked

LOCK_TIMEOUT_SECONDS = 2.0     # 获取锁的等待上限（保持原值、原语义）
LOCK_TTL_SECONDS = 30          # 新增：锁持有上限，防进程崩溃后死锁（取值理由见 §4.1.5）
_LOCK_KEY_PREFIX = "lock:"     # 新增：与 sess:/ratelimit:/auth: 命名空间隔离
_RETRY_INTERVAL = 0.05         # 新增：抢锁重试间隔

def run_locked(key: str, fn: Callable[[bool], Any]) -> Any:
    """锁内执行 fn(locked=True)；Redis 不可用时退回进程内锁；超时降级 fn(locked=False)。"""
    token = uuid4().hex
    outcome = _acquire_redis(_LOCK_KEY_PREFIX + key, token)  # acquired/timeout/unavailable

    if outcome == "acquired":
        try:
            return fn(locked=True)
        finally:
            _release_redis(_LOCK_KEY_PREFIX + key, token)

    if outcome == "timeout":
        logger.warning("分布式锁等待超时，降级执行（冲突由调用方兜底）：%s", key)
        return fn(locked=False)                               # 与原语义一致

    # unavailable：Redis 不可用 → 退回进程内锁（原逻辑原样保留）
    lock = lock_for(key)
    if not lock.acquire(timeout=LOCK_TIMEOUT_SECONDS):
        logger.warning("进程内锁超时，降级执行（冲突由调用方兜底）：%s", key)
        return fn(locked=False)
    try:
        return fn(locked=True)
    finally:
        lock.release()

_acquire_redis 内部轮询至 LOCK_TIMEOUT_SECONDS 耗尽；任何异常（含 FakeGateway 的 AttributeError）一律归类为 unavailable 并记日志，绝不向上抛。

4.1.5 参数取值

参数	值	理由
LOCK_TIMEOUT_SECONDS	2.0	保持原值，不改变现有降级时序
LOCK_TTL_SECONDS	30	见下方专项说明（第二轮审核由 10 上调）
_RETRY_INTERVAL	0.05	2 秒窗口约 40 次尝试，兼顾响应与 Redis 压力
key 前缀	lock:	与既有 key 命名空间隔离，避免误删业务 key

LOCK_TTL_SECONDS 取值专项说明（第二轮审核补充）

原进程内锁 threading.Lock 一旦拿到就持有到 release()，永不自动过期。改用 Redis 锁引入 TTL 后，就引入了一个当前不存在的新失败模式：

若临界区执行耗时超过 TTL，Redis 自动删除 key，另一实例可抢到锁 → 两实例并发进入临界区。

这是 TTL 的固有代价（也是防"进程崩溃后锁永不释放"必须付的对价）。评估与缓解：

• 兜底仍在：L3 有数据库乐观锁，预警聚合有"锁内重查锚点"，最坏后果是重复 pending 单，非数据损坏，与"超时降级"同类
• 取值：初版取 10 秒且未提供实测数据，本轮上调至 30 秒留安全边际
• 后续：取得实测 p99 后按 max(30, p99 × 5) 调整

4.1.6 涉及文件

文件	改动
app/service/risk/redis_gateway.py	新增 acquire_lock / release_lock（纯新增）
app/service/risk/locks.py	改造 run_locked 为双层；新增常量与内部函数；lock_for 保留
tests/	新增锁测试（见 §8.2）

不改：alert_service.py / profile_l3.py 的调用方式，run_locked(key, fn) 签名与 fn(locked) 语义完全不变。

4.1.7 风险

风险	缓解
测试连不上 Redis 累积延迟	不可用即降级，不重试到超时；必要时加熔断标记（§9 待确认）
FakeGateway 缺方法抛 AttributeError	归类 unavailable，降级进程内锁，测试行为不变
临界区耗时超 TTL（30 秒）→ Redis 自动解锁，两实例并发进入	这是 TTL 的固有代价，当前单进程行为没有此模式。兜底仍在（L3 乐观锁 / 锁内重查锚点），最坏为重复 pending 单；取得 p99 后按 max(30, p99×5) 调整（详见 §4.1.5 专项说明）
测试期产生大量 unavailable 日志噪声	三处调用点在 503 测试中均走 Fake→AttributeError→unavailable 路径，被捕获不导致失败，但日志量会上升。实施前后对比基线确认不漂移即可
Redis 极端延迟下双持锁	业务侧已有兜底：L3 乐观锁 + 预警锚点在锁内重查

4.2 B2 · 中间件执行顺序测试守卫

现状：main.py audit 先注册（81 行）、trace 后注册（87 行）。Starlette add_middleware 用 insert(0) 且 build_middleware_stack 反向包裹 → 数组越靠前越外层 → trace 在外层先执行，audit 在内层，与其注释"执行序在 trace 之内"吻合。

风险：调换两个装饰器后，audit 会先执行，current_trace() 返回空串，全站审计静默丢失 trace_id，不报错不失败。

改动：新增测试，发起请求后断言 audit_log 中该行 trace_id 非空。

def test_audit_middleware_runs_inside_trace_middleware(client, captured_audit):
    """守卫：audit 必须在 trace 之内执行，否则 trace_id 全空且不报错。"""
    resp = client.get("/health")           # 或任一非 _SKIP_PATHS 路径
    rows = captured_audit()
    assert rows, "应落 http_access 审计"
    assert rows[-1]["trace_id"], "audit 若先于 trace 执行，此处会静默为空"

为什么不放 ensure_trace() 兜底：那会掩盖顺序错误，导致 trace_id 与响应头不一致，更难查。

验证：临时调换装饰器 → 测试应变红（确认测试有效）→ 恢复 → 转绿。

---

5. 第 3 批：合并后 · 需拍板或涉及 DDL（5 项）

以下各项不在合并窗口内实施，此处给出设计以供届时使用。

5.1 B1 · agent_message 唯一索引 + 重试

问题：01-mysql-共用底座.sql 中 agent_message 只有 KEY idx_session_seq (session_id, seq_no)（普通索引，非 UNIQUE）；insert_turn（session_repository.py:242）在事务内取 MAX(seq_no)+1 但 SELECT 无 FOR UPDATE → 并发同会话静默产生重号消息，LLM 上下文乱序。

实施（两步必须同时做）：

1. DDL：先执行 SELECT session_id, seq_no, COUNT(*) FROM agent_message GROUP BY 1,2 HAVING COUNT(*)>1 确认无历史重号，有则先清洗；再加 UNIQUE KEY uk_session_seq (session_id, seq_no)
2. 代码：insert_turn 捕获 IntegrityError → 重读 MAX(seq_no) → 重试（建议 3 次，与 L3 乐观锁一致）

审核强调：只加索引不加重试，会把"静默脏数据"变成并发 500 硬失败，等于换个形式没解决。

附带清理：session_repository.py:134 next_seq_no()（非事务 .connect() 取号）+ :142 insert_message() 单独 INSERT 是更危险的并行取号路径（连事务都没有）。建议统一走 insert_turn，废弃该组合。

5.2 B5 · 限流 INCR + EXPIRE 非原子

首次命中先 INCR 再 EXPIRE，两步之间崩溃会留下无 TTL 计数键 → 该 actor 被永久限流。

改法：不做"首命中"判断，每次 INCR 后无条件 EXPIRE（一行级改动）。

5.3 C1 · core_ro DB 级只读账号

状态（2026-09-10）：✅ 已由基金转换线接手定案，并扩展为 D20（账号矩阵 xh_core_ro/xh_core_rw/xh_agent_rw + get_engine(db, role) + 3 个权限断言 + T-0b 阻断前置）。落地清单见 docs/项目框架设计/架构设计-基金转换交易.md §11.1。本节原始评估保留备查，勿重复设计。

需运维配合。注意：core_ro.py:55 与 gateway_repository.py:24 共用 get_engine(settings.mysql_core_database)，要拆成两个连接串（只读账号 + 可写账号）。

5.4 F1 · 支持 convert

状态（2026-09-10）：✅ 已立项为独立线（PRD v0.9 + 架构 v1.0 定稿，代码未动），开工入口 项目根 交接文档.md §B。 本节指出的两处硬编码正是该线 D7 _amount_view 要解决的问题（core_ro.py:390/:432 的 trade_type 白名单， 不改则 convert 在 RISK-001/002/003 聚合中隐形且不报错）。本线不再跟进，本节保留备查。

除网关外还有两处硬编码（最易漏）：core_ro.py:390（list_trades_range）与 :432（sum_trades_on_date）写死 AND trade_type IN ('subscribe','redeem')。不改的话 convert 在 RISK-001/002/003 聚合中完全隐形且不报错。

前置：产品需明确三条口径（校验对象、是否计入当日累计、RISK-003 算几笔），且属接口协议变更。

5.5 F2 · 意图识别

当前 tool_service.py:100-112 关键词匹配、命中即停、单意图。若升级 LLM 路由，需权衡"不可复现"对审计的影响。建议先扩同义词表 + 支持多意图，保持零延迟与可单测。

---

6. 第 4 批：合并后 · 重构类（8 项）

ID	项	要点
D1	日志基建	utils/logger.py 仅一行 docstring。用 logging.basicConfig + 文件 handler + logging.Filter 注入 trace_id（无需新依赖）
E1	权限逻辑分散三处	建议维持三层纵深（矩阵=能力声明 / chat=业务线约束 / tool=兜底），只补文档，不收拢
E2	Tool 注册表分裂	抽统一 ALL_TOOLS 聚合器，低风险低收益
E3	model/ 闲置	渐进式：给 insert_alert/insert_audit_log 等高频入口加 TypedDict
E4	注入词表硬编码 45 条	配置化 + 持续红队补充
F3	R-05 动态评分	实现 scoring.recompute_customer_score（签名已冻结），同步放开 L3 risk_score
B3	deny() 双写无事务	建议维持现状，在文档固化判据
B4	出单双写无事务	同上

B3/B4 判据（写入文档即可，不改代码）：

新增可丢留痕，变更不可丢。 出单/越权是新增，最多少一条审计，业务单据仍在；处置是变更，handle_alert_with_audit 必须同事务，否则"改了状态没记录"是合规红线。

---

7. 明确不改的范围

不做	原因
C3 审计 fail-open → fail-closed 切换	用户已选"保持 fail-open + 补告警"
C2 dev debug 通道改造	现状有双闸门 + lifespan 校验，仅需在部署文档加警示
五条红线相关行为	任何改动不得触碰
run_locked 对外契约	保持不变，避免波及两个调用点
新增第三方依赖	Redis 库已在 requirements.txt，无需新增

---

8. 回归验证

8.1 基线（每次改动前后）

python -m pytest -q          # 基线 503 passed

改动后必须仍全绿，新增测试计入总数。

8.2 第 2 批新增测试（须同步交付）

用例	断言
Redis 可用时抢锁成功	fn(locked=True)，Redis 侧 key 带 TTL
Redis 锁被占用	超时后 fn(locked=False)，降级告警出现
Redis 抛 ConnectionError	退回进程内锁，不抛异常，业务完成
FakeGateway 缺方法（AttributeError）	归类 unavailable，测试不红
释放只删自己的锁	错误 token 释放返回 False，锁仍在
并发互斥（可选，需真 Redis）	两个进程抢同一 key，仅一个进入
三处调用点 key 形态全覆盖	agg:event: / agg:suitability: / l3: 三种 key 均带 lock: 前缀且互不冲突（第二轮审核补充，避免只测两处）

8.3 手工冒烟

uvicorn app.main:app --reload
# 1. 清空 .env 的 DEEPSEEK_API_KEY → 启动日志出现降级告警
# 2. 恢复 key → 无该告警
# 3. POST /api/simulate/trade 阻断 + 放行各一次
# 4. 预警聚合仍为"同客户同日仅一张 pending 单"
# 5. 停 Redis → 业务仍可完成（退进程内锁）
# 6. 恢复 Redis → 行为与停服前一致

8.4 合并前专项检查

• tests/test_main.py::test_all_routers_mounted 路由清单未受影响（本批不改路由）
• 未新增配置项 → 不改 .env.example（若决定配置化 LOCK_TTL_SECONDS 则需同步）
• 未新增第三方依赖
• 确认 locks.py / redis_gateway.py 在目标合并分支无在途改动（第二轮审核补充：这两个是本次被改文件，有在途改动易冲突）

---

9. 待确认

#	事项	说明	倾向
1	LOCK_TTL_SECONDS 是否配置化	硬编码 10 秒简单，但环境差异	先硬编码，合并后按需配置化
2	是否加"Redis 熔断"标记	Redis 长期不可用时累积连接延迟	先不做，实测变慢再补
3	生产部署拓扑（单进程/多实例）	决定 B1/B6 的紧迫性	已按"可能多实例"处理（B6 本批做）
4	审计 fail-open/fail-closed	用户已选 fail-open + 告警	已定
5	是否支持 convert	需产品三条口径	合并后再议
6	日志是否落盘	D1 范围	合并后

---

10. 给审核 AI 的检查清单

分批合理性

• "合并前只做第 1、2 批"的划分是否合理？有无应提前或推迟的项？
• B6 放合并前是否风险偏高？双层降级是否真能保证"不劣化"？

设计正确性

• Redis 锁 TTL（10 秒）与两个临界区耗时是否匹配？
• "Redis 不可用"与"超时"分开处理是否合理？超时直接 fn(locked=False) 会否在高并发下放大冲突？
• Lua 释放脚本在 redis-py 中的调用形式（eval(script, 1, key, token)）是否正确？
• uuid4().hex 作为 token 是否足够（同 key 两次持有必须不同）？

兼容性

• FakeGateway 缺方法抛 AttributeError → 归类 unavailable，这个判断稳吗？Fake 若实现 __getattr__ 会怎样？
• 503 测试中有多少会走到 run_locked？改动后耗时是否显著变长？
• 是否影响 test_integration_risk.py（真 MySQL + TRD-TEST- 前缀隔离）？

范围

• 24 项是否覆盖完整？有无遗漏此前记录的问题？
• §7 "不改的范围"是否列全？

结论要求：明确给出 可实施 / 需修改后实施 / 不建议实施。若发现设计缺陷，请直接给出替代方案。