From 6195349dd62c65cfb008a9273bcd9b19ee36e7d3 Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 20:48:50 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20main=20=E9=9B=86=E6=88=90(=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E6=8C=82=E8=BD=BD/trace=20=E4=B8=AD=E9=97=B4=E4=BB=B6?= =?UTF-8?q?/lifespan)=20+=20B7=20=E6=8C=82=E8=B4=A6=E6=94=B6=E5=8F=A3(?= =?UTF-8?q?=E5=BC=95=E6=93=8E=E5=B7=A5=E5=8E=82/=E7=BB=9F=E4=B8=80?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E4=BD=93/=E9=94=81=E5=85=AC=E5=85=B1?= =?UTF-8?q?=E5=8C=96/L3=20=E7=BC=93=E5=AD=98=20DEL/=E6=AD=BB=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E6=B8=85=E7=90=86/=E5=A4=84=E7=BD=AE=E5=8E=9F?= =?UTF-8?q?=E5=AD=90=E4=BA=8B=E5=8A=A1/=E5=90=AF=E5=8A=A8=E6=9C=9F=20debug?= =?UTF-8?q?=20=E6=A0=A1=E9=AA=8C,=20B7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/deps.py | 29 ++++---- app/api/risk.py | 19 ++--- app/api/simulate.py | 15 ++-- app/gateway/gateway_repository.py | 15 +--- app/main.py | 66 +++++++++++++++-- app/repository/core_ro.py | 37 +--------- app/repository/risk_repository.py | 80 +++++++++++++++------ app/service/risk/alert_service.py | 101 +++++++------------------- app/service/risk/locks.py | 40 +++++++++++ app/service/risk/profile_l3.py | 15 ++-- app/service/risk/redis_gateway.py | 68 ++++++++++++++++++ app/utils/db.py | 33 +++++++++ app/utils/exceptions.py | 15 ++++ app/utils/response.py | 34 ++++++++- app/utils/trace.py | 12 +++- docs/memory/FLOW.md | 4 +- docs/memory/FRAMEWORK.md | 27 +++---- docs/memory/MEMORY.md | 30 +++++--- docs/memory/REQUIREMENTS.md | 22 +++--- docs/memory/TODO.md | 2 +- tests/test_alert_service.py | 4 ++ tests/test_aml_service.py | 4 ++ tests/test_main.py | 114 ++++++++++++++++++++++++++++++ tests/test_profile_l3.py | 15 ++++ tests/test_risk_api.py | 26 ++++--- tests/test_risk_engine.py | 4 ++ tests/test_trade_gateway.py | 11 ++- 27 files changed, 608 insertions(+), 234 deletions(-) create mode 100644 app/service/risk/locks.py create mode 100644 app/service/risk/redis_gateway.py create mode 100644 app/utils/db.py create mode 100644 tests/test_main.py diff --git a/app/api/deps.py b/app/api/deps.py index 92caa1d..370193d 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -2,33 +2,39 @@ AuthContext 模型在 A4 冻结（开发计划 v1.1）；B6 落地 `get_auth_context()`： dev（app_env=development）从 `X-Debug-Role` / `X-Debug-Actor` 请求头构造 -（customer 角色 customer_id=actor_id），非 dev 直接拒绝（启动期全局检查归 -B7 lifespan）；T-01 就绪后仅替换工厂内部为 JWT 解析，签名与调用方零改动。 +（customer 角色 customer_id=actor_id），非 dev 请求时拒绝（启动期全局检查 +归 B7 lifespan：非 dev 且 AUTH_FACTORY_IS_DEBUG 拒绝启动，挂账⑤）； +T-01 就绪后替换工厂内部为 JWT 解析并置 AUTH_FACTORY_IS_DEBUG=False， +签名与调用方零改动。 归属断言 `assert_customer_access` 对齐 JWT 手册 §6.1/§6.2 与 PRD G-01： customer 仅本人（AUTH_403_NOT_OWNER）、advisor 经 customer_advisor_rel （AUTH_403_NOT_ASSIGNED）、risk_officer 全量；越权 403 + audit 留痕（A-9）。 所有 403/401 一律经 `deny`/`unauthenticated` 审计（手册 P-05 全局铁律， -B6 评审 P1-1）；多角色按 fail-closed 口径固化（customer 分支优先，命中 deny -即拒，不再并集放宽——评审 P3-1②，T-01 引入 token_type 后收紧）。 +B6 评审 P1-1）；响应体为统一错误结构（utils/response，手册 §10，B7 挂账④）； +多角色按 fail-closed 口径固化（customer 分支优先，命中 deny 即拒，不再并集 +放宽——评审 P3-1②，T-01 引入 token_type 后收紧）。 """ from __future__ import annotations -from fastapi import HTTPException, Request -from fastapi.responses import JSONResponse +from fastapi import Request from pydantic import BaseModel, Field from app.config.settings import settings from app.repository.core_ro import CoreReadOnlyRepository from app.repository.risk_repository import RiskRepository -from app.utils.exceptions import PermissionDenied +from app.utils.exceptions import ApiError, PermissionDenied from app.utils.trace import current_trace, new_trace STAFF_FULL_ACCESS_ROLES = ("risk_officer",) DEBUG_ROLE_HEADER = "X-Debug-Role" DEBUG_ACTOR_HEADER = "X-Debug-Actor" +# B7 挂账⑤：debug 头工厂是 T-01 过渡实现；main.lifespan 据此在非 dev 环境 +# 拒绝启动。T-01 接入 JWT 工厂后置 False（或改为按注册工厂判定）。 +AUTH_FACTORY_IS_DEBUG = True + class AuthContext(BaseModel): """统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。""" @@ -44,11 +50,6 @@ class AuthContext(BaseModel): return "customer" in self.roles -def permission_denied_handler(request: Request, exc: PermissionDenied) -> JSONResponse: - """FastAPI 异常 handler：PermissionDenied → 403 + 错误码（B7 注册进 main）。""" - return JSONResponse(status_code=403, content={"detail": str(exc), "code": exc.code}) - - def _authz_audit( risk_repo: RiskRepository, auth: AuthContext | None, @@ -119,7 +120,9 @@ def get_auth_context(request: Request) -> AuthContext: "handler_comment": None, } ) - raise HTTPException(status_code=401, detail="AUTH_401_MISSING_DEBUG_HEADERS") + raise ApiError( + 401, "AUTH_401_MISSING_DEBUG_HEADERS", "missing X-Debug-Role/X-Debug-Actor headers" + ) return AuthContext( actor_id=actor_id, roles=roles, diff --git a/app/api/risk.py b/app/api/risk.py index 0b8ec77..54e3ae6 100644 --- a/app/api/risk.py +++ b/app/api/risk.py @@ -8,8 +8,9 @@ handle 仅 risk_officer；suitability/check 走 G-01；aml/scan 仅 risk_officer handler_result 枚举由请求模型 Literal 校验（repo 不校验，开发计划 B6 备注）。 直调 suitability/check 每次落 audit（request_ref='api:suitability_check'， 评审 P2-3/P3-9）；aml/scan 幂等防护挂账 B9b 前（评审 P3-6）。 -挂载：B7 集成 main.py；统一响应外壳（含错误体结构对齐手册 §10、预警类响应 -附 disclaimer）挂账 B7（评审 P3-2/P3-7）。 +挂载：main.py include（B7）；错误体统一 ApiError → 手册 §10 结构 +（utils/response.register_error_handlers，挂账④；预警类响应固定 disclaimer +挂账 B9b 核查单①）。 """ from __future__ import annotations @@ -18,7 +19,7 @@ from dataclasses import asdict from datetime import datetime from typing import Literal -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, Field from app.api.deps import AuthContext, assert_customer_access, deny, get_auth_context @@ -27,7 +28,7 @@ from app.repository.risk_repository import RiskRepository from app.service.risk.alert_service import handle_alert from app.service.risk.aml_service import scan_all from app.service.suitability import suitability_check -from app.utils.exceptions import StateConflict +from app.utils.exceptions import ApiError, StateConflict from app.utils.trace import current_trace, new_trace router = APIRouter(prefix="/api/risk", tags=["risk"]) @@ -87,10 +88,10 @@ def handle_alert_api(alert_id: str, req: HandleRequest, auth: AuthContext = Depe return handle_alert( alert_id, req.handler_result, auth.actor_id, req.handler_comment, risk_repo=_repo() ) - except LookupError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + except LookupError as exc: # NotFoundError 子类（B6 评审 P3-5 收敛） + raise ApiError(404, "NOT_FOUND", str(exc)) from exc except StateConflict as exc: - raise HTTPException(status_code=409, detail=str(exc)) from exc + raise ApiError(409, "STATE_CONFLICT", str(exc)) from exc @router.post("/suitability/check") @@ -108,8 +109,8 @@ def suitability_check_api(req: SuitabilityCheckRequest, auth: AuthContext = Depe req.customer_id, req.product_id, core_ro=core, risk_repo=repo, request_ref="api:suitability_check", ) - except LookupError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + except LookupError as exc: # customer/product not found + raise ApiError(404, "NOT_FOUND", str(exc)) from exc repo.insert_audit_log( { "trace_id": current_trace() or new_trace(), diff --git a/app/api/simulate.py b/app/api/simulate.py index 28f66bf..fb27d9c 100644 --- a/app/api/simulate.py +++ b/app/api/simulate.py @@ -3,22 +3,23 @@ 鉴权：`Depends(get_auth_context)`（B6 回挂，评审 P2-2）——一期接受 risk_demo 演示账号或客户本人（auth.customer_id == 请求 customer_id， PRD FR-1 §鉴权）；越权经 deps.deny 审计后 403。T-01 后工厂内部换 JWT。 -trace：B7 中间件贯通；B7 前由 service 层 ensure_trace 兜底。 -挂载：B7 集成 main.py（当前仅 TestClient 独立挂 router 验证）。 -统一响应外壳：utils/response.py 为占位（P1 任务），落地点挂账 B7（届时 -simulate/risk 一并包裹，本路由返回体不变）。 +trace：main 中间件贯通（B7），响应头 X-Trace-Id 回写；service 层 ensure_trace +仍兜底脚本/测试直调场景。 +挂载：main.py include（B7）。错误体统一 ApiError → 手册 §10 结构（挂账④）， +convert 400 / 资源 404 不变 HTTP 语义。 """ from __future__ import annotations from decimal import Decimal -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from pydantic import BaseModel, Field from app.api.deps import AuthContext, deny, get_auth_context from app.gateway.trade_gateway import UnsupportedTradeType, submit_trade from app.repository.risk_repository import RiskRepository +from app.utils.exceptions import ApiError router = APIRouter(prefix="/api/simulate", tags=["simulate"]) @@ -47,6 +48,6 @@ def submit_trade_api(req: TradeRequest, auth: AuthContext = Depends(get_auth_con try: return submit_trade(req.model_dump()) except UnsupportedTradeType as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc + raise ApiError(400, "BAD_REQUEST", str(exc)) from exc except LookupError as exc: # NotFoundError 亦为其子类；已统一（B6 评审 P3-5） - raise HTTPException(status_code=404, detail=str(exc)) from exc + raise ApiError(404, "NOT_FOUND", str(exc)) from exc diff --git a/app/gateway/gateway_repository.py b/app/gateway/gateway_repository.py index fb63d2d..2343025 100644 --- a/app/gateway/gateway_repository.py +++ b/app/gateway/gateway_repository.py @@ -10,27 +10,18 @@ from datetime import datetime from decimal import Decimal from typing import Any -from sqlalchemy import create_engine, text +from sqlalchemy import text from sqlalchemy.engine import Engine from app.config.settings import settings +from app.utils.db import get_engine class GatewayRepository: """core_trade 唯一写入口；仅 INSERT，不改不删（模拟网关语义）。""" def __init__(self, engine: Engine | None = None) -> None: - self._engine = engine or self._default_engine() - - @staticmethod - def _default_engine() -> Engine: - pwd = settings.mysql_password - auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user - url = ( - f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}" - f"/{settings.mysql_core_database}?charset=utf8mb4" - ) - return create_engine(url, pool_pre_ping=True) + self._engine = engine or get_engine(settings.mysql_core_database) def insert_trade( self, diff --git a/app/main.py b/app/main.py index 12fbe63..7be5186 100644 --- a/app/main.py +++ b/app/main.py @@ -1,15 +1,69 @@ -"""FastAPI 入口：挂载路由、中间件（JWT/RBAC）、生命周期。""" +"""FastAPI 入口（B7 集成）：路由挂载、trace 中间件、lifespan、统一错误体。 -from fastapi import FastAPI +lifespan（B7 挂账⑤⑥）： +- 启动期校验：非 development 环境且鉴权工厂仍为 debug 头实现 → 拒绝启动 + （T-01 接入 JWT 后置 deps.AUTH_FACTORY_IS_DEBUG=False 放行）； +- Redis 网关单例注册（惰性连接，publish/DEL 失败降级不阻塞业务）； +- shutdown 统一 dispose 数据库引擎（utils/db 工厂，B6 复审 P3 泄漏收口）。 +trace：X-Trace-Id 透传/生成 + 响应头回写（trace.py 约束：call_next 前 set， +同步路由线程池由 anyio 传播，一致性 B8 断言兜底）。 +""" +from __future__ import annotations + +import re +from contextlib import asynccontextmanager + +from fastapi import FastAPI, Request + +from app.api import deps +from app.api.risk import router as risk_router +from app.api.simulate import router as simulate_router from app.config.settings import settings +from app.service.risk import redis_gateway +from app.utils.db import dispose_engines +from app.utils.response import register_error_handlers +from app.utils.trace import new_trace, reset_trace, set_trace -app = FastAPI(title="JinRong Agent Platform", version="0.1.0") +# 透传外部 X-Trace-Id 的格式白名单（防响应头注入；不合规一律新生成） +_TRACE_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$") + + +@asynccontextmanager +async def lifespan(_: FastAPI): + if settings.app_env != "development" and deps.AUTH_FACTORY_IS_DEBUG: + raise RuntimeError( + "debug auth factory is wired but app_env is not 'development'; " + "deploy T-01 JWT auth first or set app_env=development" + ) + redis_gateway.set_gateway(redis_gateway.RedisGateway()) + try: + yield + finally: + redis_gateway.set_gateway(None) + dispose_engines() + + +app = FastAPI(title="JinRong Agent Platform", version="0.2.0", lifespan=lifespan) +register_error_handlers(app) +app.include_router(risk_router) +app.include_router(simulate_router) + + +@app.middleware("http") +async def trace_middleware(request: Request, call_next): + """trace_id 贯通：透传合法 X-Trace-Id，否则生成；响应头回写。""" + incoming = request.headers.get("X-Trace-Id", "") + trace_id = incoming if _TRACE_ID_PATTERN.fullmatch(incoming) else new_trace() + token = set_trace(trace_id) + try: + response = await call_next(request) + finally: + reset_trace(token) + response.headers["X-Trace-Id"] = trace_id + return response @app.get("/health") def health(): return {"status": "ok", "env": settings.app_env} - - -# TODO: 挂载 app.api 路由；接入 Auth SDK 中间件 diff --git a/app/repository/core_ro.py b/app/repository/core_ro.py index 2736803..7897806 100644 --- a/app/repository/core_ro.py +++ b/app/repository/core_ro.py @@ -6,27 +6,18 @@ from datetime import date, datetime, time, timedelta from decimal import Decimal from typing import Any -from sqlalchemy import create_engine, text +from sqlalchemy import text from sqlalchemy.engine import Engine from app.config.settings import settings +from app.utils.db import get_engine class CoreReadOnlyRepository: """仅 SELECT jinrong_core；禁止写操作。""" def __init__(self, engine: Engine | None = None) -> None: - self._engine = engine or self._default_engine() - - @staticmethod - def _default_engine() -> Engine: - pwd = settings.mysql_password - auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user - url = ( - f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}" - f"/{settings.mysql_core_database}?charset=utf8mb4" - ) - return create_engine(url, pool_pre_ping=True) + self._engine = engine or get_engine(settings.mysql_core_database) def get_customer_l0(self, customer_id: str) -> dict[str, Any] | None: sql = text( @@ -55,28 +46,6 @@ class CoreReadOnlyRepository: with self._engine.connect() as conn: return [dict(r) for r in conn.execute(sql, {"cid": customer_id}).mappings()] - def list_trades( - self, customer_id: str, since: date | None = None, limit: int = 50 - ) -> list[dict[str, Any]]: - sql = text( - """ - SELECT t.*, p.product_name - FROM core_trade t - JOIN core_product p ON p.product_id = t.product_id - WHERE t.customer_id = :cid - AND (:since IS NULL OR t.traded_at >= :since) - ORDER BY t.traded_at DESC - LIMIT :lim - """ - ) - with self._engine.connect() as conn: - return [ - dict(r) - for r in conn.execute( - sql, {"cid": customer_id, "since": since, "lim": limit} - ).mappings() - ] - def list_trades_range( self, customer_id: str, start: datetime, end: datetime, limit: int = 10000 ) -> list[dict[str, Any]]: diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py index 0dafb83..27625a6 100644 --- a/app/repository/risk_repository.py +++ b/app/repository/risk_repository.py @@ -11,29 +11,30 @@ from datetime import datetime from decimal import Decimal from typing import Any -from sqlalchemy import create_engine, text +from sqlalchemy import text from sqlalchemy.engine import Engine from app.config.settings import settings +from app.utils.db import get_engine EVENT_ALERT_TYPES = ("large_amount", "freq_trade", "pattern") +_AUDIT_SQL = text( + """ + INSERT INTO audit_log + (trace_id, event_type, agent_type, actor_id, customer_id, rule_id, + input_summary, decision, risk_score, handler_id, handler_result, handler_comment) + VALUES (:trace_id, :event_type, :agent_type, :actor_id, :customer_id, :rule_id, + :input_summary, :decision, :risk_score, :handler_id, :handler_result, :handler_comment) + """ +) + class RiskRepository: """风控产出表读写；不修改表结构（PRD 冻结约束）。""" def __init__(self, engine: Engine | None = None) -> None: - self._engine = engine or self._default_engine() - - @staticmethod - def _default_engine() -> Engine: - pwd = settings.mysql_password - auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user - url = ( - f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}" - f"/{settings.mysql_database}?charset=utf8mb4" - ) - return create_engine(url, pool_pre_ping=True) + self._engine = engine or get_engine(settings.mysql_database) # ---------- risk_alert ---------- @@ -206,6 +207,44 @@ class RiskRepository: ) return res.rowcount == 1 + def handle_alert_with_audit( + self, + alert_id: str, + handler_result: str, + handler_id: str, + handler_comment: str | None, + audit_entry: dict[str, Any], + ) -> bool: + """状态机变更 + 处置审计**同事务**（B7 挂账⑦）。 + + UPDATE ... WHERE status='pending_review'（防并发抢先）命中后才同事务 + INSERT audit_log——审计失败整体回滚，不产生无痕状态变更（P-05）；返回 + False 表示已被并发处置，service 层转 StateConflict。 + """ + update_sql = text( + """ + UPDATE risk_alert + SET status = :result, handler_id = :hid, handler_result = :result, + handler_comment = :comment, handled_at = :handled_at + WHERE alert_id = :aid AND status = 'pending_review' + """ + ) + with self._engine.begin() as conn: + res = conn.execute( + update_sql, + { + "result": handler_result, + "hid": handler_id, + "comment": handler_comment, + "handled_at": datetime.now(), + "aid": alert_id, + }, + ) + if res.rowcount != 1: + return False + conn.execute(_AUDIT_SQL, self._dump_audit(audit_entry)) + return True + @staticmethod def _dump_alert(alert: dict[str, Any]) -> dict[str, Any]: out = dict(alert) @@ -222,21 +261,16 @@ class RiskRepository: # ---------- audit_log（风控判定审计 · 只 INSERT，PRD §7.3）---------- def insert_audit_log(self, entry: dict[str, Any]) -> None: - sql = text( - """ - INSERT INTO audit_log - (trace_id, event_type, agent_type, actor_id, customer_id, rule_id, - input_summary, decision, risk_score, handler_id, handler_result, handler_comment) - VALUES (:trace_id, :event_type, :agent_type, :actor_id, :customer_id, :rule_id, - :input_summary, :decision, :risk_score, :handler_id, :handler_result, :handler_comment) - """ - ) + with self._engine.begin() as conn: + conn.execute(_AUDIT_SQL, self._dump_audit(entry)) + + @staticmethod + def _dump_audit(entry: dict[str, Any]) -> dict[str, Any]: params = dict(entry) params["input_summary"] = json.dumps( entry.get("input_summary") or {}, ensure_ascii=False, default=str ) - with self._engine.begin() as conn: - conn.execute(sql, params) + return params # ---------- risk_suitability_log ---------- diff --git a/app/service/risk/alert_service.py b/app/service/risk/alert_service.py index e607534..f2a3c42 100644 --- a/app/service/risk/alert_service.py +++ b/app/service/risk/alert_service.py @@ -1,86 +1,32 @@ """预警单聚合与通知（B2 · 架构 §5.2 / PRD FR-4）。 职责：单事件命中规则的**聚合决策与编排**（merge 原语 append_alert_event 在 repo，A3 评审口径）； -进程内锁防并发首单双单（多进程部署时换 Redis SET NX，接口不变）；审计落库；Pub/Sub 通知广播。 +聚合锁原语公共化至 locks（B7 挂账①）；审计落库；Pub/Sub 通知广播经 +redis_gateway 单例（B7：连接由 lifespan 管理，publish 失败降级不阻塞落库）。 """ from __future__ import annotations -import json -import logging -import threading from datetime import date, datetime, time -from typing import Any, Callable +from typing import Any from uuid import uuid4 -from app.config.settings import settings from app.repository.risk_repository import EVENT_ALERT_TYPES, RiskRepository +from app.service.risk import redis_gateway +from app.service.risk.locks import run_locked from app.service.risk.rules import RuleHit from app.utils.exceptions import NotFoundError, StateConflict from app.utils.trace import current_trace, new_trace -logger = logging.getLogger(__name__) - -LOCK_TIMEOUT_SECONDS = 2.0 TIER_SCORE = {"large_amount": 70, "freq_trade": 50, "pattern": 80, "suitability": 90, "aml": 95} -class RedisAlertPublisher: - """同步 Redis Pub/Sub 发布器（事件线与同步调用链一致；B7 lifespan 管理连接）。""" - - def __init__(self, url: str | None = None) -> None: - self._url = url or settings.redis_url - self._client = None - - def publish(self, channel: str, payload: dict[str, Any]) -> None: - if self._client is None: - import redis - - self._client = redis.Redis.from_url(self._url, decode_responses=True) - self._client.publish(channel, json.dumps(payload, ensure_ascii=False)) +def set_publisher(publisher: Any | None) -> None: + """测试/集成注入点（旧名兼容，B7 起转发 redis_gateway.set_gateway）。""" + redis_gateway.set_gateway(publisher) -_publisher: RedisAlertPublisher | None = None - - -def set_publisher(publisher: RedisAlertPublisher | None) -> None: - """测试/集成注入点（B7 lifespan 初始化，测试注入 fake）。""" - global _publisher - _publisher = publisher - - -def _publish(payload: dict[str, Any]) -> None: - try: - (_publisher or RedisAlertPublisher()).publish("risk:pub:alert", payload) - except Exception: # 通知失败不阻塞预警落库（DB 为权威） - logger.exception("publish risk:pub:alert failed") - - -# ---------- 进程内聚合锁（架构 §5.2；拿不到锁降级独立出单，宁多勿漏） ---------- - -_locks: dict[str, threading.Lock] = {} -_locks_guard = threading.Lock() - - -def _lock_for(key: str) -> threading.Lock: - with _locks_guard: - lock = _locks.get(key) - if lock is None: - lock = threading.Lock() - _locks[key] = lock - return lock - - -def _run_locked(key: str, fn: Callable[[bool], Any]) -> Any: - """锁内执行 fn(locked=True)；获取超时降级 fn(locked=False)。""" - lock = _lock_for(key) - if not lock.acquire(timeout=LOCK_TIMEOUT_SECONDS): - logger.warning("agg lock timeout, degrade to standalone alert: %s", key) - return fn(locked=False) - try: - return fn(locked=True) - finally: - lock.release() +# ---------- 进程内聚合锁：公共原语见 locks.py（B7 挂账①） ---------- def _day_start(day: date | None = None) -> datetime: @@ -121,7 +67,8 @@ def _audit( def _publish_alert(alert: dict[str, Any]) -> None: - _publish( + redis_gateway.publish( + "risk:pub:alert", { "alert_id": alert["alert_id"], "alert_type": alert["alert_type"], @@ -129,7 +76,7 @@ def _publish_alert(alert: dict[str, Any]) -> None: "risk_score": alert["risk_score"], "trace_id": alert["trace_id"], "notify_role": ["risk_officer"] + (["compliance"] if alert["alert_type"] == "aml" else []), - } + }, ) @@ -197,7 +144,7 @@ def record_trade_alerts( _publish_alert(alert) return alert - return _run_locked(f"agg:event:{trade['customer_id']}:{date.today()}", _agg) + return run_locked(f"agg:event:{trade['customer_id']}:{date.today()}", _agg) def record_suitability_alert( @@ -256,7 +203,7 @@ def record_suitability_alert( _publish_alert(alert) return alert - return _run_locked( + return run_locked( f"agg:suitability:{trade_request['customer_id']}:{trade_request['product_id']}:{date.today()}", _agg, ) @@ -271,8 +218,9 @@ def handle_alert( ) -> dict[str, Any]: """人工处置（FR-4；角色鉴权在依赖层，本层管状态机与审计）。 - 仅 pending_review 可处置（禁止跳改已处置单，A-7）；处置动作全量写 - audit_log（agent_type='risk'，handler_* 三字段落审计行）。 + 仅 pending_review 可处置（禁止跳改已处置单，A-7）；状态变更与处置审计 + **同事务提交**（B7 挂账⑦：原两事务非原子，审计失败会留下无痕的状态变更， + 违反 P-05 全量留痕）；处置动作全量写 audit_log（agent_type='risk'）。 """ repo = risk_repo or RiskRepository() alert = repo.get_alert(alert_id) @@ -280,11 +228,12 @@ def handle_alert( raise NotFoundError(f"alert not found: {alert_id}") if alert["status"] != "pending_review": raise StateConflict(f"alert {alert_id} already handled (status={alert['status']})") - ok = repo.update_alert_status(alert_id, handler_result, handler_id, handler_comment) - if not ok: # 读后并发窗口：他人已抢先处置 - raise StateConflict(f"alert {alert_id} state changed concurrently") - repo.insert_audit_log( - { + ok = repo.handle_alert_with_audit( + alert_id, + handler_result, + handler_id, + handler_comment, + audit_entry={ "trace_id": current_trace() or new_trace(), "event_type": "alert_handle", "agent_type": "risk", @@ -297,8 +246,10 @@ def handle_alert( "handler_id": handler_id, "handler_result": handler_result, "handler_comment": handler_comment, - } + }, ) + if not ok: # 读后并发窗口：他人已抢先处置 + raise StateConflict(f"alert {alert_id} state changed concurrently") return repo.get_alert(alert_id) diff --git a/app/service/risk/locks.py b/app/service/risk/locks.py new file mode 100644 index 0000000..116583d --- /dev/null +++ b/app/service/risk/locks.py @@ -0,0 +1,40 @@ +"""进程内聚合锁原语（B7 · 挂账①：从 alert_service 公共化，profile_l3 同用）。 + +单进程内按 key 串行减少并发首单冲突（多进程部署换 Redis SET NX，接口不变）； +拿锁超时降级独立执行，冲突安全由调用方兜底（预警聚合：同日单条锚点查询在 +锁内重查；L3：乐观锁重试），宁多勿漏。 +""" + +from __future__ import annotations + +import logging +import threading +from typing import Any, Callable + +logger = logging.getLogger(__name__) + +LOCK_TIMEOUT_SECONDS = 2.0 + +_locks: dict[str, threading.Lock] = {} +_locks_guard = threading.Lock() + + +def lock_for(key: str) -> threading.Lock: + with _locks_guard: + lock = _locks.get(key) + if lock is None: + lock = threading.Lock() + _locks[key] = lock + return lock + + +def run_locked(key: str, fn: Callable[[bool], Any]) -> Any: + """锁内执行 fn(locked=True)；获取超时降级 fn(locked=False)。""" + lock = lock_for(key) + if not lock.acquire(timeout=LOCK_TIMEOUT_SECONDS): + logger.warning("agg lock timeout, degrade to standalone alert: %s", key) + return fn(locked=False) + try: + return fn(locked=True) + finally: + lock.release() diff --git a/app/service/risk/profile_l3.py b/app/service/risk/profile_l3.py index 761bbc4..bfc587a 100644 --- a/app/service/risk/profile_l3.py +++ b/app/service/risk/profile_l3.py @@ -26,11 +26,13 @@ from typing import Any from sqlalchemy.exc import IntegrityError from app.repository.risk_repository import RiskRepository -from app.service.risk.alert_service import _run_locked # 同包复用聚合锁；B7 收敛至公共原语 +from app.service.risk import redis_gateway +from app.service.risk.locks import run_locked logger = logging.getLogger(__name__) TIER_ORDER = ("normal", "watch", "high") +L3_CACHE_KEY = "profile:l3:{customer_id}" # redis-keys 手册：String(JSON) · 5m · l3 UPDATE 时 DEL ALERT_TYPE_TIER = { "aml": "high", "pattern": "watch", @@ -102,8 +104,9 @@ def upsert_profile_l3( ) -> dict[str, Any]: """预警事件 → L3 upsert（aml 自动追加待复核标签）。 - 返回合并后的 L3 行（含 customer_id）。B7 接 Redis 缓存时须在此处加 - 写侧 DEL 钩子（PRD §5.1：MySQL 更新时 DEL profile:l3:{customer_id}）。 + 返回合并后的 L3 行（含 customer_id）。MySQL 写成功后 DEL Redis 读缓存 + `profile:l3:{customer_id}`（B7 挂账②落地，PRD §5.1；DEL 失败降级 TTL 过期， + 不阻塞业务）。 """ repo = risk_repo or RiskRepository() mapped_tier = tier_of(alert_type) @@ -144,9 +147,11 @@ def upsert_profile_l3( logger.warning("L3 update lost race, retrying (attempt %d): %s", attempt, customer_id) raise RuntimeError(f"L3 upsert conflicted after {_MAX_RETRIES} retries: {customer_id}") - return _run_locked(f"l3:{customer_id}", _write) + merged = run_locked(f"l3:{customer_id}", _write) + redis_gateway.cache_delete(L3_CACHE_KEY.format(customer_id=customer_id)) + return merged def get_profile_l3(customer_id: str, risk_repo: RiskRepository | None = None) -> dict[str, Any] | None: - """L3 只读薄封装（对话线/引擎复用；Redis 热缓存与写侧 DEL 钩子待 B7 lifespan 一并接入）。""" + """L3 只读薄封装（对话线/引擎复用；Redis 热读缓存归对话线接入时实现）。""" return (risk_repo or RiskRepository()).get_l3(customer_id) diff --git a/app/service/risk/redis_gateway.py b/app/service/risk/redis_gateway.py new file mode 100644 index 0000000..4dfaad0 --- /dev/null +++ b/app/service/risk/redis_gateway.py @@ -0,0 +1,68 @@ +"""Redis 单例网关（B7 lifespan 管理连接）：预警广播 publish + L3 缓存失效 DEL。 + +惰性连接（首次 publish/DEL 才建）；连接/执行失败一律降级日志，不阻塞业务 +（DB 为权威，PRD FR-4 通知语义）。测试注入 fake：monkeypatch 本模块 +`_gateway`，实现 publish(channel, payload) / delete(*keys) 即可。 +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from app.config.settings import settings + +logger = logging.getLogger(__name__) + + +class RedisGateway: + def __init__(self, url: str | None = None) -> None: + self._url = url or settings.redis_url + self._client: Any = None + + def _ensure(self) -> Any: + if self._client is None: + import redis + + self._client = redis.Redis.from_url(self._url, decode_responses=True) + return self._client + + def publish(self, channel: str, payload: dict[str, Any]) -> None: + self._ensure().publish(channel, json.dumps(payload, ensure_ascii=False)) + + def delete(self, *keys: str) -> None: + self._ensure().delete(*keys) + + +_gateway: RedisGateway | Any | None = None + + +def set_gateway(gateway: RedisGateway | Any | None) -> None: + """单例注入点（B7 lifespan 注册；测试注入 fake）。""" + global _gateway + _gateway = gateway + + +def get_gateway() -> RedisGateway: + """取网关单例；未注册时惰性创建（脚本直调场景）。""" + global _gateway + if _gateway is None: + _gateway = RedisGateway() + return _gateway + + +def publish(channel: str, payload: dict[str, Any]) -> None: + """广播；失败降级（不阻塞预警落库）。""" + try: + get_gateway().publish(channel, payload) + except Exception: + logger.exception("publish %s failed", channel) + + +def cache_delete(*keys: str) -> None: + """写侧缓存失效（PRD §5.1：MySQL 更新时 DEL）；失败降级（TTL 兜底过期）。""" + try: + get_gateway().delete(*keys) + except Exception: + logger.warning("cache DEL failed (degrade to TTL): %s", keys, exc_info=True) diff --git a/app/utils/db.py b/app/utils/db.py new file mode 100644 index 0000000..2468886 --- /dev/null +++ b/app/utils/db.py @@ -0,0 +1,33 @@ +"""SQLAlchemy 引擎工厂（B7 · 开发计划挂账⑥）。 + +按库名缓存单例 Engine：deps / risk / simulate 路由每请求实例化 Repository 时 +复用同一连接池，不再每次 create_engine（B6 复审 P3：实例化点泄漏）；应用 +shutdown 经 dispose_engines 统一释放。测试直传 engine= 参数的用法不受影响； +monkeypatch settings 后须先 dispose_engines() 清缓存。 +""" + +from __future__ import annotations + +from functools import lru_cache + +from sqlalchemy import create_engine +from sqlalchemy.engine import Engine + +from app.config.settings import settings + + +@lru_cache(maxsize=None) +def get_engine(database: str) -> Engine: + """取指定库的单例 Engine（进程内共享连接池）。""" + pwd = settings.mysql_password + auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user + url = ( + f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}" + f"/{database}?charset=utf8mb4" + ) + return create_engine(url, pool_pre_ping=True) + + +def dispose_engines() -> None: + """释放全部缓存 Engine（lifespan shutdown / 测试隔离）。""" + get_engine.cache_clear() diff --git a/app/utils/exceptions.py b/app/utils/exceptions.py index 0a82bd0..5b3c009 100644 --- a/app/utils/exceptions.py +++ b/app/utils/exceptions.py @@ -22,3 +22,18 @@ class PermissionDenied(PermissionError): class StateConflict(ValueError): """状态机冲突（HTTP 409；如处置非 pending_review 的预警单）。""" + + +class ApiError(Exception): + """业务 HTTP 错误（统一错误体出口 · 手册 §10，B7 挂账④）。 + + error_code 取手册 §10 AUTH 系列（AUTH_401_*）或通用码 + （NOT_FOUND / STATE_CONFLICT / BAD_REQUEST）；register_error_handlers + 转为 {error_code, message, trace_id, request_id} 响应。 + """ + + def __init__(self, status_code: int, error_code: str, message: str) -> None: + super().__init__(message) + self.status_code = status_code + self.error_code = error_code + self.message = message diff --git a/app/utils/response.py b/app/utils/response.py index a868c47..064b12c 100644 --- a/app/utils/response.py +++ b/app/utils/response.py @@ -1 +1,33 @@ -"""统一 API 响应格式。""" +"""统一 API 响应外壳（B7 · 挂账④）。 + +成功响应：业务字段平铺不变（B5 评审 P2-2 口径「路由返回体不变」），trace_id +经 X-Trace-Id 响应头贯通（main 中间件）；错误响应统一 JWT 手册 §10 结构 +{error_code, message, trace_id, request_id}（request_id 沿用 trace_id， +独立请求级标识尚未引入）。main 与测试 app 共用 register_error_handlers。 +""" + +from __future__ import annotations + +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse + +from app.utils.exceptions import ApiError, PermissionDenied +from app.utils.trace import current_trace, new_trace + + +def error_body(error_code: str, message: str) -> dict[str, str]: + """手册 §10 错误体（trace 缺失时兜底生成，保证响应可归因）。""" + tid = current_trace() or new_trace() + return {"error_code": error_code, "message": message, "trace_id": tid, "request_id": tid} + + +def register_error_handlers(app: FastAPI) -> None: + """注册统一错误体 handler（替代 FastAPI 默认 detail 结构）。""" + + @app.exception_handler(ApiError) + async def _api_error_handler(request: Request, exc: ApiError) -> JSONResponse: + return JSONResponse(status_code=exc.status_code, content=error_body(exc.error_code, exc.message)) + + @app.exception_handler(PermissionDenied) + async def _permission_denied_handler(request: Request, exc: PermissionDenied) -> JSONResponse: + return JSONResponse(status_code=403, content=error_body(exc.code, str(exc))) diff --git a/app/utils/trace.py b/app/utils/trace.py index bde6235..ace7460 100644 --- a/app/utils/trace.py +++ b/app/utils/trace.py @@ -8,7 +8,7 @@ from __future__ import annotations -from contextvars import ContextVar +from contextvars import ContextVar, Token from uuid import uuid4 _trace_id: ContextVar[str] = ContextVar("trace_id", default="") @@ -21,6 +21,16 @@ def new_trace(trace_id: str | None = None) -> str: return tid +def set_trace(trace_id: str) -> Token: + """绑定外部 trace_id（中间件入口），返回 token 供请求结束复位。""" + return _trace_id.set(trace_id) + + +def reset_trace(token: Token) -> None: + """复位到中间件之前的上下文（防异步任务间串号）。""" + _trace_id.reset(token) + + def current_trace() -> str: """读取当前 trace_id；未初始化时返回空串（调用方应兜底生成）。""" return _trace_id.get() diff --git a/docs/memory/FLOW.md b/docs/memory/FLOW.md index 7326fb0..5d1fde8 100644 --- a/docs/memory/FLOW.md +++ b/docs/memory/FLOW.md @@ -38,7 +38,7 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md ``` -**尚未自动化：** agent 专用表 SQL、Milvus 建 Collection、JWT 中间件 — 见 `TODO.md`。 +**尚未自动化：** Milvus 建 Collection、JWT 中间件（现 `api/deps.py` debug 头过渡，T-01 就绪换工厂内部）— 见 `TODO.md`；agent 专用表已有 `02-mysql-agent专用.sql`（手工执行，含 risk_aml_list）。 ------ @@ -48,7 +48,7 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md Client → Gateway(JWT/RBAC) → api/chat → agent_service(LangGraph) → Tools → 存储 → 响应 + audit_log ``` -当前：`main.py` 仅 health；全链路待 Wave 0 起逐步挂载。 +当前：**风控事件线已集成**（交易网关 → 规则引擎 → 预警/AML/L3 → 4 API，`main.py` 路由挂载 + trace 中间件 + lifespan 完成，B1~B7，180 测试绿）；对话链路待 T-01/T-07。 ------ diff --git a/docs/memory/FRAMEWORK.md b/docs/memory/FRAMEWORK.md index ae989e9..b5f23ab 100644 --- a/docs/memory/FRAMEWORK.md +++ b/docs/memory/FRAMEWORK.md @@ -34,13 +34,13 @@ | 模块 | 职责 | 依赖 | 代码状态 | | --- | --- | --- | --- | -| Agent Gateway / Auth SDK | JWT、RBAC、归属校验 | Redis、MySQL customer_advisor_rel | **未做**（见 JWT 手册） | +| Agent Gateway / Auth SDK | JWT、RBAC、归属校验 | Redis、MySQL customer_advisor_rel | **过渡版已实现**（`api/deps.py` debug 头 AuthContext + 归属断言 + 越权审计，B6；JWT 解析归 T-01，仅换工厂内部） | | 客户财富 Agent | L1 画像、事实查询、阈值提醒 | Core RO、Milvus 产品库 | 空壳 service | | 代理人助手 Agent | L2 画像、RAG、草稿 | L1 只读、Milvus | 空壳 service | | 数据分析 Agent | NL→SQL→解读 | Core RO、画像只读 | 空壳 service | -| 风控监测 Agent | 预警、L3、R-02 适当性 | 交易事件、AML 名单 | 空壳 service | -| Core 只读层 | L0 事实查询 | `jinrong_core` | **core_ro.py 已实现** | -| 共用底座 | 会话、审计、输入防护 | MySQL 11 表 + Redis | SQL 已定；代码未接 | +| 风控监测 Agent | 预警、L3、R-02 适当性 | 交易事件、AML 名单 | **已实现**（B1~B7：rules/alert_service/profile_l3/aml_service/engine/scoring 占位 + 4 API + suitability + main 集成与挂账①~⑦） | +| Core 只读层 | L0 事实查询 | `jinrong_core` | **core_ro.py 已实现**（含风控扩展查询） | +| 共用底座 | 会话、审计、输入防护 | MySQL 11 表 + Redis | **部分接入**（风控审计落库 / Redis 推送已接；会话、全量审计中间件、输入防护待 T-02/T-03） | | 同步脚本 | 归属、Neo4j | Core → agent / 图库 | **sync_*.py 已实现** | ------ @@ -48,15 +48,16 @@ ## 3. 后端分层（app/） ```text -api/ → 路由（chat、knowledge、admin）；薄，不含业务 【空壳】 -service/ → agent_service、rag_service、memory_service、risk/* 【空壳】 -tool/ → document_parser、embedding_tool、milvus_tool 【空壳】 -repository/ → core_ro（Core 只读）；后续 agent 库 Repository 【core_ro 已实现】 -gateway/ → 模拟交易网关（外部 Core 交易系统替身，非 Agent 分层）【PRD v1.0 新增】 -model/ → schemas（Pydantic）、entities（ORM） 【占位】 -config/ → settings、database 【settings 已实现】 -utils/ → response、exceptions、logger 【占位】 -main.py → FastAPI 入口；当前仅 /health 【部分】 +api/ → 风控 risk / simulate / deps 已实现（main 已挂载）；chat、knowledge、admin 空壳【部分】 +service/ → risk/*（rules/alert/aml/engine/l3/scoring/locks/redis_gateway）+ suitability 已实现； + agent_service、rag_service、memory_service 空壳 【部分】 +tool/ → document_parser、embedding_tool、milvus_tool 【空壳】 +repository/ → core_ro（Core 只读）+ risk_repository（风控四表） 【已实现】 +gateway/ → 模拟交易网关（trade_gateway + gateway_repository 仅 INSERT core_trade） 【已实现 B5】 +model/ → schemas（Pydantic）、entities（ORM） 【占位】（风控暂用 dict，收敛待后续） +config/ → settings（双库 + risk_* 阈值）、database 【settings 已实现】 +utils/ → trace / desensitize / db（引擎工厂）/ response（统一错误体）/ exceptions 【基本就绪；logger 占位】 +main.py → 路由挂载 + trace 中间件 + lifespan（B7） 【已集成】 允许：api → service → tool / repository / model / config 禁止：api 直连 Milvus/MySQL 写复杂逻辑；tool 写业务流程；repository 写 Core diff --git a/docs/memory/MEMORY.md b/docs/memory/MEMORY.md index ebbc586..8de59fd 100644 --- a/docs/memory/MEMORY.md +++ b/docs/memory/MEMORY.md @@ -9,20 +9,28 @@ **项目是什么：** 金融四 Agent（客户财富 / 代理人 / 数据分析 / 风控）共用数据层与合规底座；**不**互调 LLM，跨 Agent 走 L1/L2/L3 画像与预警表。 -**当前进度：** 需求与表设计已定 · 后端 **脚手架 + Core 模拟库脚本** 已落地 · **业务 API / JWT / LangGraph 图尚未实现**（多为空壳模块）。 +**当前进度：** 需求与表设计已定 · **风控模块已落地 B1~B7**（规则/预警聚合/L3/AML/引擎/交易网关/鉴权+4 API/适当性校验/**main 集成+挂账①~⑦**，**180 测试绿**）· 剩 **B8 conftest+集成测试 → B9a 脚本 → B9b 演示走查** · **JWT(T-01) / 审计中间件(T-02) / LangGraph 对话线(T-07) 未做**（chat/knowledge/admin 仍空壳）。 **仓库地图：** | 路径 | 状态 | 说明 | | --- | --- | --- | -| `app/main.py` | 可跑 | 仅 `/health`；路由未挂载 | -| `app/api/*.py` | 空壳 | chat / knowledge / admin 待实现 | -| `app/service/*.py` | 空壳 | agent / rag / memory 待实现 | -| `app/repository/core_ro.py` | **已实现** | Core 只读 SELECT（jinrong_core） | -| `app/config/settings.py` | **已实现** | 双库 `jinrong_agent` + `jinrong_core` | +| `app/main.py` | **已集成** | 路由挂载 + trace 中间件（X-Trace-Id 贯通）+ lifespan（启动期 debug 校验 · Redis 网关注册 · 引擎 dispose，B7） | +| `app/api/risk.py` `simulate.py` `deps.py` | **已实现** | 风控 4 API + 模拟网关路由 + debug 头 AuthContext（T-01 过渡版，JWT 就绪只换工厂内部） | +| `app/api/chat.py` `knowledge.py` `admin.py` | 空壳 | 待 T-06 / T-07 | +| `app/service/risk/*` + `service/suitability.py` | **已实现** | RISK-001~005 规则 / 预警聚合 / L3 写入 / AML / 引擎编排 / SUIT-001~008 / locks+redis_gateway 公共原语（B7） | +| `app/service/agent_service.py` 等 | 空壳 | LangGraph 对话线待 T-07 | +| `app/gateway/` | **已实现** | 模拟交易网关（仅 gateway_repository 可 INSERT core_trade，B5） | +| `app/repository/core_ro.py` | **已实现** | Core 只读 SELECT（含风控扩展 sum_trades_on_date / list_trades_range / list_active_customers） | +| `app/repository/risk_repository.py` | **已实现** | risk_alert / risk_suitability_log / L3 / risk_aml_list 读写 | +| `app/utils/` | **基本就绪** | trace / desensitize / db（引擎工厂）/ response（统一错误体）/ exceptions（含 ApiError）已实现；logger 占位 | +| `app/config/settings.py` | **已实现** | 双库 `jinrong_agent` + `jinrong_core` + risk_* 阈值 | | `scripts/core/*.sql` + `reset.ps1` | **已实现** | Core 模拟库 DDL + 种子 | +| `scripts/agent/` `scripts/demo/` | **已实现** | AML 名单种子 + 风控演示数据 | | `scripts/sync/*.py` | **已实现** | 归属同步 + Neo4j 全图 | +| `tests/` | **已实现** | 13 个测试文件 173 用例（sqlite 隔离） | | `docs/需求拆解/` | 已定 | 场景 P0、矩阵、合规原文 | +| `docs/PRD/PRD-风控监测Agent.md` | **已冻结** | 风控 PRD v1.0 + 规则表附录 | | `docs/项目框架设计/表设计/` | 已定 | Agent 共用 11 表 + agent 专用 SQL | | `docs/项目框架设计/Core模拟底座/` | 已定 | 无真实 Core 时的 L0 方案 | | `web/` | **不存在** | 前端 React 待 init | @@ -39,7 +47,7 @@ 7. uvicorn app.main:app --reload # GET /health ``` -**下一步开发（见 TODO）：** T-01 JWT 中间件 → 挂载 api → LangGraph agent_service → Wave 0 验收。 +**下一步开发（见 TODO）：** 风控 B8 conftest+集成测试（A-1~A-5/A-7~A-9 + trace 一致性 + sqlite DDL 收敛）→ B9a 演示/运维脚本 → B9b 演示走查；随后 Wave 0 T-01 JWT / T-02 审计中间件 / T-06 / T-07 LangGraph。 **禁止（改代码前必记）：** Core 正式 C1~C5 不可被画像覆盖 · 审计表只 INSERT · 代理人草稿不外发 · 仅 R-02 可阻断交易 · 四 Agent 不互调 LLM。 @@ -48,8 +56,8 @@ ## 1. 项目简介 - **名称：** JinRong 金融四 Agent 智能管家 -- **当前阶段：** MVP 脚手架期（设计已定，Wave 0 开发中） -- **当前优先级：** Wave 0 共用底座 → Wave 1 代理人/分析闭环 +- **当前阶段：** 风控模块（Wave 2 提前开发）B7~B9 收尾中；Wave 0 底座被风控借道部分落地 +- **当前优先级：** 风控 B7/B8/B9 → Wave 0（T-01/T-02/T-06/T-07）→ Wave 1 ------ @@ -107,10 +115,12 @@ audit_log 等审计表（只 INSERT） ```text 后端入口：app/main.py · 配置 app/config/settings.py · Core 只读 app/repository/core_ro.py -Agent 库 SQL：docs/项目框架设计/表设计/01-mysql-共用底座.sql +Agent 库 SQL：docs/项目框架设计/表设计/01-mysql-共用底座.sql · 02-mysql-agent专用.sql（含 risk_aml_list） Core 模拟：scripts/core/reset.ps1 · 文档 docs/项目框架设计/Core模拟底座/ +种子：scripts/agent/seed-aml-list.sql（AML 名单）· scripts/demo/prepare_risk_demo.sql（reset 后重跑） 依赖：requirements.txt（LangGraph + langchain-core/openai + FastAPI + SQLAlchemy） 启动：uvicorn app.main:app --reload → GET /health +测试：python -m pytest（173 用例，sqlite 隔离） 配置：.env（见 .env.example） RBAC 联调账号：scripts/dev/rbac-seed-reference.md ``` diff --git a/docs/memory/REQUIREMENTS.md b/docs/memory/REQUIREMENTS.md index ae8cbf3..b74c91a 100644 --- a/docs/memory/REQUIREMENTS.md +++ b/docs/memory/REQUIREMENTS.md @@ -17,14 +17,16 @@ | ID | 需求 | 验收对照 | 状态 | TODO | | --- | --- | --- | --- | --- | -| F-01 | JWT + RBAC + 数据归属 | 越权 403 + audit；JWT 手册 §13 | 未做 | T-01 | -| F-02 | 全量审计留痕 | trace_id 可还原 | 未做 | T-02 | -| F-03 | 输入防护 | input_guard_log | 未做 | T-03 | -| F-04 | Core 只读层 | 不改 Core 账；Repository 只 SELECT | **部分** | T-04 | +| F-01 | JWT + RBAC + 数据归属 | 越权 403 + audit；JWT 手册 §13 | **部分**（deps.py debug 头 AuthContext + 归属断言 + 越权审计已落地，B6；JWT 解析待 T-01） | T-01 | +| F-02 | 全量审计留痕 | trace_id 可还原 | **部分**（utils/trace 贯通 + 风控审计 INSERT 已接；全量审计中间件待 T-02） | T-02 | +| F-03 | 输入防护 | input_guard_log | 未做（双写缺口挂账 B7 接 T-02 时统一补） | T-03 | +| F-04 | Core 只读层 | 不改 Core 账；Repository 只 SELECT | **部分**（core_ro 已扩展风控查询；接入 api/service Tool 与归属校验待 T-04） | T-04 | | R0-DB | MySQL 共用 11 表 + Redis | 01-mysql-共用底座.sql | SQL 已定；灌库待验 | T-05 | | R0-CORE | Core 模拟 + 同步 | reset.ps1 + sync 脚本 | **脚本已落地** | T-05 | -**F-04 部分完成说明：** `app/repository/core_ro.py` + `scripts/core/*` + `settings.mysql_core_database` 已有；尚未接入 api/service Tool 与归属校验。 +**F-01 部分完成说明：** B6 已落地 `app/api/deps.py`（AuthContext Pydantic + `get_auth_context()` 工厂，dev 从 `X-Debug-Role`/`X-Debug-Actor` 头构造；T-01 就绪后仅替换工厂内部为 JWT 解析）。 + +**F-04 部分完成说明：** `app/repository/core_ro.py`（含风控扩展 sum_trades_on_date / list_trades_range / list_active_customers）+ `scripts/core/*` + `settings.mysql_core_database` 已有；尚未接入对话 Tool 与归属校验。 ## Wave 1 · 内部 Agent P0 @@ -39,11 +41,11 @@ | ID | 需求 | 验收对照 | 状态 | TODO | | --- | --- | --- | --- | --- | -| R-01 | 大额预警 | risk_alert pending_review | 未做 | T-30 | -| R-02 | 适当性阻断请求 | 唯一阻断场景 | 未做 | T-31 | -| R-03 | AML 命中通知 | 不自动冻户 | 未做 | T-32 | -| R-04 | 频繁/试探模式预警 | 聚合出单 + 去重 | 未做 | T-30 | -| R-05 | L3 画像写入 | 最小写入（最高档合并） | 未做 | T-30 | +| R-01 | 大额预警 | risk_alert pending_review | **已实现**（B1~B6，A-1/A-2 用例绿） | T-30 | +| R-02 | 适当性阻断请求 | 唯一阻断场景 | **已实现**（suitability SUIT-001~008 + check API 带鉴权 + 网关阻断，B5/B6；MySQL 手工 SQL 对照挂账 B9b） | T-31 | +| R-03 | AML 命中通知 | 不自动冻户 | **已实现**（aml_service + scan API + L3 联动；对话通知线待 T-32 阶段 C） | T-32 | +| R-04 | 频繁/试探模式预警 | 聚合出单 + 去重 | **已实现**（B1 规则 + B2 聚合去重） | T-30 | +| R-05 | L3 画像写入 | 最小写入（最高档合并） | **已实现**（profile_l3 防降级；`risk_score` 一期保持 NULL 归评分模型首写） | T-30 | ## Wave 3 · 客户 P0 diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md index fc26ba1..48b9c5b 100644 --- a/docs/memory/TODO.md +++ b/docs/memory/TODO.md @@ -20,7 +20,7 @@ ### 风控模块（PRD v1.0 已冻结 · `docs/PRD/PRD-风控监测Agent.md`，事件驱动线不依赖 T-07 可先行） -- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关 / B6 鉴权+4 API+处置编排（含独立 AI 复审通过：P0/P1=0，P2 测试隔离已修，agent_type 口径等 P3 已修或挂账 B7/B9b，173 测试绿）均已完成并经独立 AI 评审闭环；剩 B7 main 集成、B8 conftest+集成测试、B9a 脚本、B9b 演示走查** +- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关 / B6 鉴权+4 API+处置编排 / B7 main 集成（路由挂载 + trace 中间件 + lifespan 启动期 debug 校验；挂账①锁公共化 ②L3 缓存 DEL ③死代码 ④统一错误体 ⑤启动校验 ⑥引擎工厂 ⑦处置原子事务均落地；⑧ input_guard_log 双写仍随 T-02）均完成并经独立 AI 评审闭环（180 测试绿）；剩 B8 conftest+集成测试、B9a 脚本、B9b 演示走查** - [ ] T-31 `service/suitability.py` 公共校验（SUIT-001~008）+ `POST /api/risk/suitability/check` + 单测验收 A-8 —— **代码已完成（A1~A4 · 77 测试绿；B6 起该 API 带鉴权+直调审计）；MySQL 手工 SQL 对照挂账至 B9b 执行（阶段 A 评审 P2-8）** - [ ] T-32 预警台账与人工处置 API（`GET /alerts`、`POST /handle`，risk_officer/compliance 权限）+ 对话线（依赖 T-01/T-03/T-07）验收 A-6/A-7 —— **API 部分已由 B6 覆盖并复审通过（A-7/A-9 用例绿）；对话线仍依赖 T-01/T-03/T-07** diff --git a/tests/test_alert_service.py b/tests/test_alert_service.py index e653123..c42e926 100644 --- a/tests/test_alert_service.py +++ b/tests/test_alert_service.py @@ -28,11 +28,15 @@ class FakePublisher: def __init__(self): self.messages = [] + self.deletes = [] def publish(self, channel, payload): assert isinstance(payload, dict) self.messages.append((channel, payload)) + def delete(self, *keys): + self.deletes.append(keys) + @pytest.fixture() def env(): diff --git a/tests/test_aml_service.py b/tests/test_aml_service.py index e222613..b98edcc 100644 --- a/tests/test_aml_service.py +++ b/tests/test_aml_service.py @@ -20,11 +20,15 @@ from app.service.risk.profile_l3 import AML_PENDING_TAG class FakePublisher: def __init__(self): self.messages = [] + self.deletes = [] def publish(self, channel, payload): assert isinstance(payload, dict) self.messages.append((channel, payload)) + def delete(self, *keys): + self.deletes.append(keys) + @pytest.fixture() def env(): diff --git a/tests/test_main.py b/tests/test_main.py new file mode 100644 index 0000000..cc0ccdf --- /dev/null +++ b/tests/test_main.py @@ -0,0 +1,114 @@ +"""main 集成冒烟（B7）：路由挂载 / trace 中间件 / lifespan 启动期校验 / 统一错误体。 + +main app 全路由经 TestClient 走真实 lifespan（dev 环境）；仓储 monkeypatch +注入 sqlite（audit_log 供 401 留痕）；Redis 网关注入 fake（不依赖本机 Redis）。 +全链路 trace 一致性与集成测试归 B8 conftest，此处只验中间件行为本身。 +""" + +import re + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +from app.config.settings import settings +from app.main import app +from app.repository.risk_repository import RiskRepository +from app.service.risk import redis_gateway + +TRACE_HEADER_PATTERN = r"^trc-[0-9a-f]{16}$" + + +class FakeGateway: + def __init__(self): + self.messages = [] + self.deletes = [] + + def publish(self, channel, payload): + self.messages.append((channel, payload)) + + def delete(self, *keys): + self.deletes.append(keys) + + +@pytest.fixture() +def client(monkeypatch): + engine = create_engine( + "sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False} + ) + with engine.begin() as conn: + conn.execute( + text( + """CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY, trace_id VARCHAR(64), event_type VARCHAR(64), + agent_type VARCHAR(16), actor_id VARCHAR(64), customer_id VARCHAR(64), + rule_id VARCHAR(64), input_summary TEXT, decision VARCHAR(64), risk_score INTEGER, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""" + ) + ) + repo = RiskRepository(engine=engine) + from app.api import deps as deps_mod + from app.api import risk as risk_api + from app.api import simulate as simulate_mod + + monkeypatch.setattr(risk_api, "_repo", lambda: repo) + monkeypatch.setattr(simulate_mod, "_repo", lambda: repo) + monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) + with TestClient(app) as c: # lifespan：dev 放行；Redis 惰性连接不触网 + monkeypatch.setattr(redis_gateway, "_gateway", FakeGateway()) + yield c + engine.dispose() + + +def test_health(client): + r = client.get("/health") + assert r.status_code == 200 and r.json()["status"] == "ok" + + +def test_all_routers_mounted(client): + paths = client.get("/openapi.json").json()["paths"] + assert set(paths) == { + "/health", + "/api/risk/alerts", + "/api/risk/alerts/{alert_id}/handle", + "/api/risk/suitability/check", + "/api/risk/aml/scan", + "/api/simulate/trade", + } + + +def test_trace_header_generated(client): + r = client.get("/health") + assert r.headers["X-Trace-Id"] and re.fullmatch(TRACE_HEADER_PATTERN, r.headers["X-Trace-Id"]) + + +def test_trace_header_passthrough(client): + tid = "trc-abc123def45678" + assert client.get("/health", headers={"X-Trace-Id": tid}).headers["X-Trace-Id"] == tid + + +def test_trace_header_invalid_regenerated(client): + bad = "bad id!" + tid = client.get("/health", headers={"X-Trace-Id": bad}).headers["X-Trace-Id"] + assert tid != bad and re.fullmatch(TRACE_HEADER_PATTERN, tid) + + +def test_unified_error_body_401_with_trace(client): + """手册 §10 错误体四键 + trace_id 与响应头一致（中间件贯通初验）。""" + r = client.get("/api/risk/alerts") # 无 debug 头 + assert r.status_code == 401 + body = r.json() + assert body["error_code"] == "AUTH_401_MISSING_DEBUG_HEADERS" + assert body["message"] + assert set(body) == {"error_code", "message", "trace_id", "request_id"} + assert body["trace_id"] == body["request_id"] == r.headers["X-Trace-Id"] + + +def test_lifespan_rejects_debug_factory_in_non_dev(monkeypatch): + """挂账⑤：非 dev 环境且 debug 头工厂在场 → 拒绝启动（T-01 前防误部署）。""" + monkeypatch.setattr(settings, "app_env", "production") + with pytest.raises(RuntimeError, match="debug auth factory is wired"): + with TestClient(app): + pass diff --git a/tests/test_profile_l3.py b/tests/test_profile_l3.py index 778b4b2..b93fba0 100644 --- a/tests/test_profile_l3.py +++ b/tests/test_profile_l3.py @@ -12,6 +12,7 @@ from sqlalchemy import create_engine, text from sqlalchemy.pool import StaticPool from app.repository.risk_repository import RiskRepository +from app.service.risk import redis_gateway from app.service.risk.profile_l3 import ( AML_PENDING_TAG, ALERT_TYPE_TIER, @@ -23,6 +24,20 @@ from app.service.risk.profile_l3 import ( ) +@pytest.fixture(autouse=True) +def _fake_redis(monkeypatch): + """L3 写侧 DEL 钩子隔离（B7 挂账②）：upsert 后失效缓存不触真 Redis。""" + + class _FakeGateway: + def publish(self, channel, payload): + pass + + def delete(self, *keys): + pass + + monkeypatch.setattr(redis_gateway, "_gateway", _FakeGateway()) + + @pytest.fixture() def env(): engine = create_engine( diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py index eab4ff1..7d06a4e 100644 --- a/tests/test_risk_api.py +++ b/tests/test_risk_api.py @@ -12,12 +12,13 @@ from sqlalchemy import create_engine, text from sqlalchemy.pool import StaticPool from app.api import risk as risk_api -from app.api.deps import assert_customer_access, permission_denied_handler +from app.api.deps import assert_customer_access from app.api.risk import router as risk_router from app.repository.core_ro import CoreReadOnlyRepository from app.repository.risk_repository import RiskRepository -from app.service.risk import alert_service +from app.service.risk import redis_gateway from app.utils.exceptions import PermissionDenied +from app.utils.response import register_error_handlers class FakePublisher: @@ -25,10 +26,14 @@ class FakePublisher: def __init__(self): self.messages = [] + self.deletes = [] def publish(self, channel, payload): self.messages.append((channel, payload)) + def delete(self, *keys): + self.deletes.append(keys) + def _alert(alert_id, customer, atype, status="pending_review", score=70): return { @@ -140,11 +145,11 @@ def client(env, monkeypatch): from app.api import deps as deps_mod monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) - # aml 命中会广播 Pub/Sub:注入 fake,不触真 Redis(复审 P2) - monkeypatch.setattr(alert_service, "_publisher", FakePublisher()) + # aml 命中广播 Pub/Sub + L3 缓存 DEL:注入 fake,不触真 Redis(复审 P2) + monkeypatch.setattr(redis_gateway, "_gateway", FakePublisher()) app = FastAPI() app.include_router(risk_router) - app.add_exception_handler(PermissionDenied, permission_denied_handler) + register_error_handlers(app) # 统一错误体(手册 §10,与 main 同一 handler 集) with TestClient(app) as c: yield c @@ -305,7 +310,7 @@ def test_a9_customer_other_403_with_audit(client, env): headers=CUST_1001, # 查他人 ) assert r.status_code == 403 - assert "AUTH_403_NOT_OWNER" in r.json()["detail"] + assert r.json()["error_code"] == "AUTH_403_NOT_OWNER" assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden'") == 1 @@ -317,7 +322,7 @@ def test_a9_advisor_not_assigned_403_with_audit(client, env): headers=ADV_01, # 非名下 ) assert r.status_code == 403 - assert "AUTH_403_NOT_ASSIGNED" in r.json()["detail"] + assert r.json()["error_code"] == "AUTH_403_NOT_ASSIGNED" assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden'") == 1 @@ -327,7 +332,7 @@ def test_compliance_suitability_check_403(client): json={"customer_id": "CUST-3001", "product_id": "PROD-510300"}, headers=COMPLIANCE, # 审计角色不在客户业务数据白名单 ) - assert r.status_code == 403 and "AUTH_403_SCOPE" in r.json()["detail"] + assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SCOPE" # ---------- POST /aml/scan ---------- @@ -342,6 +347,9 @@ def test_aml_scan_officer_only_with_audit(client, env): assert len(body["alerts"]) == 1 l3 = repo.get_l3("CUST-3001") assert l3["monitor_tier"] == "high" # scan 命中标记 L3 + # L3 写侧缓存 DEL 钩子（B7 挂账②）：命中客户 upsert 后失效读缓存 + fake = redis_gateway.get_gateway() + assert ("profile:l3:CUST-3001",) in fake.deletes assert _counts(engine, "audit_log", "event_type='aml_scan' AND decision='scan_completed'") == 1 for headers in (COMPLIANCE, CUST_1001): assert client.post("/api/risk/aml/scan", headers=headers).status_code == 403 @@ -398,4 +406,4 @@ def test_multi_role_combinations_fixed_behavior(client): json={"customer_id": "CUST-1001", "product_id": "PROD-161725"}, headers=_h("customer,advisor", "CUST-3001"), ) - assert r.status_code == 403 and "AUTH_403_NOT_OWNER" in r.json()["detail"] + assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_OWNER" diff --git a/tests/test_risk_engine.py b/tests/test_risk_engine.py index 132e0c7..2431b4b 100644 --- a/tests/test_risk_engine.py +++ b/tests/test_risk_engine.py @@ -22,10 +22,14 @@ from app.service.risk.scoring import recompute_customer_score class FakePublisher: def __init__(self): self.messages = [] + self.deletes = [] def publish(self, channel, payload): self.messages.append((channel, payload)) + def delete(self, *keys): + self.deletes.append(keys) + @pytest.fixture() def env(): diff --git a/tests/test_trade_gateway.py b/tests/test_trade_gateway.py index 14c407b..b837aca 100644 --- a/tests/test_trade_gateway.py +++ b/tests/test_trade_gateway.py @@ -10,7 +10,6 @@ from decimal import Decimal import pytest from fastapi import FastAPI -from app.api.deps import PermissionDenied, permission_denied_handler from fastapi.testclient import TestClient from sqlalchemy import create_engine, text from sqlalchemy.pool import StaticPool @@ -23,15 +22,20 @@ from app.repository.core_ro import CoreReadOnlyRepository from app.repository.risk_repository import RiskRepository from app.service.risk import alert_service from app.service.risk.profile_l3 import AML_PENDING_TAG +from app.utils.response import register_error_handlers class FakePublisher: def __init__(self): self.messages = [] + self.deletes = [] def publish(self, channel, payload): self.messages.append((channel, payload)) + def delete(self, *keys): + self.deletes.append(keys) + DDL = [ """CREATE TABLE core_customer ( @@ -243,7 +247,7 @@ def client(env, monkeypatch): monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) monkeypatch.setattr(simulate_mod, "_repo", lambda: repo) app = FastAPI() - app.add_exception_handler(PermissionDenied, permission_denied_handler) + register_error_handlers(app) # 统一错误体(手册 §10,与 main 同一 handler 集) app.include_router(simulate_router) with TestClient(app) as c: yield c @@ -255,7 +259,8 @@ DEMO = {"X-Debug-Role": "risk_demo", "X-Debug-Actor": "STAFF-90001"} def test_api_convert_returns_400(client): r = client.post("/api/simulate/trade", json=_req(ttype="convert"), headers=DEMO) assert r.status_code == 400 - assert "转换交易暂不支持" in r.json()["detail"] + assert r.json()["error_code"] == "BAD_REQUEST" + assert "转换交易暂不支持" in r.json()["message"] def test_api_blocked_returns_200_with_blocked_true(client):