diff --git a/app/api/chat.py b/app/api/chat.py
index 65f71ad..52407fd 100644
--- a/app/api/chat.py
+++ b/app/api/chat.py
@@ -101,6 +101,11 @@ def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get
     if agent_type not in AGENT_TYPES:
         raise ApiError(400, "BAD_REQUEST", f"invalid X-Agent-Type: {agent_type}")
     assert_agent_access(auth, agent_type, risk_repo=_repo())
+    # C5 前置（PRD 4A.1）：对话线不放行 risk_manager——HTTP 台账才放行，保住
+    # FR-6 冻结口径。矩阵放行解决 HTTP 通道，chat 层显式拒绝兜底（manager 根本
+    # 进不了对话线，Tool 层 assert_tool_access 天然 fail-closed）。
+    if agent_type == "risk" and "risk_manager" in auth.roles:
+        deny(auth, "AUTH_403_ROLE", _repo(), message="对话线仅限 risk_officer，请走 HTTP 台账")
 
     message = req.message.strip()
     if not message:
diff --git a/app/api/deps.py b/app/api/deps.py
index 1ec8c43..8bd20a5 100644
--- a/app/api/deps.py
+++ b/app/api/deps.py
@@ -47,7 +47,13 @@ AGENT_ACCESS_MATRIX: dict[str, dict[str, tuple[str, ...]]] = {
     "customer": {"token_types": ("customer",), "roles": ("customer",)},
     "advisor": {"token_types": ("staff",), "roles": ("advisor", "compliance", "ops")},
     "analyst": {"token_types": ("staff",), "roles": ("analyst", "compliance")},
-    "risk": {"token_types": ("staff", "service"), "roles": ("risk_officer", "service_risk")},
+    # C5（PRD 4A.1）：增补 risk_manager——HTTP 台账请求经此交叉校验，
+    # 不加则 manager 连 GET /api/risk/alerts 都会被 AUTH_403_AGENT_MISMATCH 挡掉。
+    # 对话线不放行（FR-6 冻结口径），由 chat.py 显式 deny 兜底。
+    "risk": {
+        "token_types": ("staff", "service"),
+        "roles": ("risk_officer", "risk_manager", "service_risk"),
+    },
 }
 
 # T-01 完成：JWT 已接入，debug 头降级为 dev 兜底（main.lifespan 据此放行
diff --git a/app/api/risk.py b/app/api/risk.py
index 5f0afac..949013a 100644
--- a/app/api/risk.py
+++ b/app/api/risk.py
@@ -64,8 +64,13 @@ def list_alerts_api(
     page_size: int = Query(20, ge=1, le=100),
 ) -> dict:
     """预警台账分页（FR-4）。compliance 强制 aml 过滤（A-7：仅返回 aml 单）。"""
+    # C5 前置（PRD 4A.1）：risk_manager 经 AGENT_ACCESS_MATRIX 放行 HTTP 通道，
+    # 此处同 risk_officer 全量只读路径（无处置入口）；handle/aml/scan 端点角色
+    # 校验保持 risk_officer only，manager 触碰即 403，零代码改动。
     if auth.has_role("risk_officer"):
         pass
+    elif auth.has_role("risk_manager"):
+        pass
     elif auth.has_role("compliance"):
         alert_type = "aml"
     else:
diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py
index 48cf712..5382744 100644
--- a/app/repository/risk_repository.py
+++ b/app/repository/risk_repository.py
@@ -344,6 +344,53 @@ class RiskRepository:
             conn.execute(_AUDIT_SQL, self._dump_audit(audit_entry))
         return True
 
+    def list_pending_alerts_all(self, page_size: int = 1000) -> list[dict]:
+        """全部 pending_review 单（C5 时效升级扫描输入；演示规模一次取回，Python 端判级）。"""
+        with self._engine.connect() as conn:
+            rows = conn.execute(
+                text(
+                    """
+                    SELECT * FROM risk_alert
+                    WHERE status = 'pending_review'
+                    ORDER BY created_at ASC
+                    LIMIT :lim
+                    """
+                ),
+                {"lim": page_size},
+            ).mappings()
+            return [self._parse_alert(dict(r)) for r in rows]
+
+    def update_alert_escalation(
+        self, alert_id: str, escalation_level: int, escalated_at: datetime, trace_id: str
+    ) -> bool:
+        """payload 升级标记独占写入（RISK-007；status/handler_* 列一律不碰）。
+
+        统一读改写（评审 P1-1 定案）：SELECT payload → 合并 escalation_level/
+        escalated_at/escalation_trace_id → UPDATE payload。与 append_alert_event
+        同模式（跨 MySQL/sqlite 已验证），不引入 JSON_SET 方言分支；并发窗口由
+        「单发定时脚本 + 幂等闸门（仅升不降）」兜底。WHERE alert_id=:aid，返回
+        rowcount==1。
+        """
+        with self._engine.begin() as conn:
+            row = conn.execute(
+                text("SELECT payload FROM risk_alert WHERE alert_id = :aid"),
+                {"aid": alert_id},
+            ).mappings().first()
+            if row is None:
+                return False
+            payload = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
+            payload["escalation_level"] = escalation_level
+            payload["escalated_at"] = escalated_at.isoformat()
+            payload["escalation_trace_id"] = trace_id
+            res = conn.execute(
+                text("UPDATE risk_alert SET payload = :payload WHERE alert_id = :aid"),
+                {
+                    "payload": json.dumps(payload, ensure_ascii=False, default=str),
+                    "aid": alert_id,
+                },
+            )
+            return res.rowcount == 1
+
     @staticmethod
     def _dump_alert(alert: dict[str, Any]) -> dict[str, Any]:
         out = dict(alert)
@@ -355,6 +402,18 @@ class RiskRepository:
     def _parse_alert(row: dict[str, Any]) -> dict[str, Any]:
         row["triggered_rules"] = json.loads(row["triggered_rules"])
         row["payload"] = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
+        # sqlite 原生 DDL 无类型信息，created_at/handled_at 读回为字符串；MySQL 端
+        # DATETIME 已是 datetime。统一在解析层转回，避免调用方按 datetime 计算超期/判级。
+        if isinstance(row.get("created_at"), str):
+            try:
+                row["created_at"] = datetime.fromisoformat(row["created_at"])
+            except ValueError:
+                pass
+        if isinstance(row.get("handled_at"), str):
+            try:
+                row["handled_at"] = datetime.fromisoformat(row["handled_at"])
+            except ValueError:
+                pass
         return row
 
     # ---------- audit_log（风控判定审计 · 只 INSERT，PRD §7.3）----------
diff --git a/app/service/risk/chat_tools.py b/app/service/risk/chat_tools.py
index bb5b3fd..666f077 100644
--- a/app/service/risk/chat_tools.py
+++ b/app/service/risk/chat_tools.py
@@ -25,6 +25,7 @@ from __future__ import annotations
 import datetime as _dt
 from typing import Any, Callable
 
+from app.config.settings import settings
 from app.repository.core_ro import CoreReadOnlyRepository
 from app.repository.risk_repository import RiskRepository
 from app.service import suitability
@@ -215,10 +216,60 @@ def aml_lookup(customer_id: str, core_ro: CoreReadOnlyRepository | None = None,
     )
 
 
+def query_overdue_alerts(customer_id: str, core_ro: CoreReadOnlyRepository | None = None,
+                          risk_repo: RiskRepository | None = None, **params: Any) -> dict[str, Any]:
+    """超期 pending 单列表（只读；FR-9 时效升级查询入口）。
+
+    hours = params.get("hours")：缺省取 settings.risk_escalation_l1_hours（普通单
+    L1 阈值），过滤 overdue_hours >= hours 的单；按超期时长降序（对话线查询置顶语义）。
+    跨客户全量（risk_officer 无绑定客户亦可查），customer_id 仅回显 scope。
+    """
+    repo = risk_repo or RiskRepository()
+    now = _dt.datetime.now()
+    hours = float(params.get("hours") or settings.risk_escalation_l1_hours)
+    rows = repo.list_pending_alerts_all()
+    out: list[dict[str, Any]] = []
+    for a in rows:
+        created = a.get("created_at")
+        if not isinstance(created, _dt.datetime):
+            continue
+        overdue_h = (now - created).total_seconds() / 3600.0
+        if overdue_h < hours:
+            continue
+        payload = a.get("payload") or {}
+        out.append(
+            {
+                "alert_id": a.get("alert_id"),
+                "customer_id": a.get("customer_id"),
+                "alert_type": a.get("alert_type"),
+                "created_at": created.isoformat(),
+                "escalation_level": int(payload.get("escalation_level") or 0),
+                "overdue_hours": round(overdue_h, 1),
+                "risk_score": a.get("risk_score"),
+            }
+        )
+    out.sort(key=lambda x: x["overdue_hours"], reverse=True)
+    return _jsonable(
+        {
+            "scope": customer_id or None,
+            "hours_threshold": hours,
+            "overdue_count": len(out),
+            "items": out,
+        }
+    )
+
+
 # ---------- 注册表（供 tool_service.get_registered_tool 分发） ----------
 
 
 RISK_TOOL_REGISTRY: dict[str, RiskToolSpec] = {
+    "query_overdue_alerts": RiskToolSpec(
+        func=query_overdue_alerts,
+        description="查询超期未处置预警（FR-9 时效升级；按超期时长降序，可带 hours 阈值）",
+        requires_customer=False,
+        param_whitelist=("hours",),
+        int_bounds={"hours": (1, 720)},
+    ),
     "alert_query": RiskToolSpec(
         func=alert_query,
         description="查询风控预警（客户维度或全量待审；risk_officer 可不带客户查全量待审）",
diff --git a/app/service/risk/escalation_service.py b/app/service/risk/escalation_service.py
new file mode 100644
index 0000000..19f7c30
--- /dev/null
+++ b/app/service/risk/escalation_service.py
@@ -0,0 +1,205 @@
+"""预警处置时效升级（C5 · PRD FR-9 / 规则表 v1.1 RISK-007 补充约束）。
+
+RISK-007：pending_review 单超时（普通 4h/24h，AML 1h/4h）→ payload 写升级标记
++ 推送升级通知，**不改 status**（人工处置后经 handle_api 自然退出 pending 范围）。
+
+设计要点（实现方案 §3.3）：
+- 升级标记只由本定时任务写（update_alert_escalation 仅 cron 链路调用）；
+  handle_api 走 handle_alert_with_audit，写集不相交，互不覆盖（A-11 断言锚点）。
+- 幂等闸门：computed_level > 已记录 escalation_level 才动作；同级/降级一律跳过，
+  每级别至多推送一次。
+- 先持久化再推送（PRD 拍板语义）：逐单 update_alert_escalation → 逐单审计 →
+  按 (customer_id, level) 降噪合并为一次推送。
+- 通知链累积（评审 P1-4）：L1 → [risk_officer, risk_manager]；
+  L2 → [risk_officer, risk_manager, compliance]（升级到 L2 不把 manager 移出）。
+- 审计只 INSERT（event_type='alert_escalation'，VARCHAR 可直接扩，未改表）。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from app.config.settings import settings
+from app.repository.risk_repository import RiskRepository
+from app.service.risk import redis_gateway
+from app.utils.trace import current_trace, new_trace
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class EscalationThresholds:
+    """升级时效阈值（冻结 dataclass，from_settings 读配置）。"""
+
+    l1_hours: int = 4
+    l2_hours: int = 24
+    aml_l1_hours: int = 1
+    aml_l2_hours: int = 4
+
+    @classmethod
+    def from_settings(cls) -> "EscalationThresholds":
+        return cls(
+            l1_hours=settings.risk_escalation_l1_hours,
+            l2_hours=settings.risk_escalation_l2_hours,
+            aml_l1_hours=settings.risk_escalation_aml_l1_hours,
+            aml_l2_hours=settings.risk_escalation_aml_l2_hours,
+        )
+
+
+def _overdue_hours(alert: dict[str, Any], now: datetime) -> float:
+    created_at = alert.get("created_at")
+    if not isinstance(created_at, datetime):
+        return 0.0
+    return (now - created_at).total_seconds() / 3600.0
+
+
+def compute_level(alert: dict[str, Any], now: datetime, th: EscalationThresholds) -> int:
+    """超时级别：普通单 4h→L1 / 24h→L2；AML 走短通道 1h/4h。"""
+    is_aml = alert.get("alert_type") == "aml"
+    l1 = th.aml_l1_hours if is_aml else th.l1_hours
+    l2 = th.aml_l2_hours if is_aml else th.l2_hours
+    overdue = _overdue_hours(alert, now)
+    if overdue >= l2:
+        return 2
+    if overdue >= l1:
+        return 1
+    return 0
+
+
+def _current_level(alert: dict[str, Any]) -> int:
+    payload = alert.get("payload") or {}
+    return int(payload.get("escalation_level") or 0)
+
+
+def _notify_role(level: int) -> list[str]:
+    """通知角色链（评审 P1-4 累积口径，manager 不因升级到 L2 而移出）。"""
+    if level >= 2:
+        return ["risk_officer", "risk_manager", "compliance"]
+    return ["risk_officer", "risk_manager"]
+
+
+def scan_and_escalate(
+    now: datetime | None = None,
+    risk_repo: RiskRepository | None = None,
+    thresholds: EscalationThresholds | None = None,
+) -> dict[str, Any]:
+    """单次扫描入口（定时脚本 / 测试 / 手动演示共用）。
+
+    返回 {"scanned", "escalated": [...], "merged_notices", "skipped"}。
+    escalated 为逐单升级记录（含 alert_id / customer_id / level）。
+    """
+    repo = risk_repo or RiskRepository()
+    th = thresholds or EscalationThresholds.from_settings()
+    now = now or datetime.now()
+    trace_id = current_trace() or new_trace()
+
+    rows = repo.list_pending_alerts_all()
+    scanned = len(rows)
+    escalated: list[dict[str, Any]] = []
+    skipped = 0
+    merged_notices = 0
+
+    # 降噪分组：同 (customer_id, level) 合并一次推送
+    grouped: dict[tuple, list[dict[str, Any]]] = {}
+    for alert in rows:
+        level = compute_level(alert, now, th)
+        # 幂等闸门：仅升不降；未到期（level=0）或已达该级一并跳过
+        if level <= _current_level(alert):
+            skipped += 1
+            continue
+        grouped.setdefault((alert.get("customer_id"), level), []).append(alert)
+
+    for (customer_id, level), group in grouped.items():
+        notify_role = _notify_role(level)
+        alert_ids: list[str] = []
+        for alert in group:
+            alert_id = alert["alert_id"]
+            # ① 先持久化（独占写 payload，不碰 status/handler_*）
+            ok = repo.update_alert_escalation(alert_id, level, now, trace_id)
+            # ② 逐单审计（不合并）
+            repo.insert_audit_log(
+                {
+                    "trace_id": trace_id,
+                    "event_type": "alert_escalation",
+                    "agent_type": "risk",
+                    "actor_id": "SYSTEM",
+                    "customer_id": customer_id,
+                    "rule_id": None,
+                    "input_summary": {
+                        "alert_id": alert_id,
+                        "reached_level": level,
+                        "escalation_reason": f"pending_review 超期达 LEVEL_{level}",
+                    },
+                    "decision": "escalated",
+                    "risk_score": alert.get("risk_score"),
+                    "handler_id": None,
+                    "handler_result": None,
+                    "handler_comment": None,
+                }
+            )
+            if ok:
+                escalated.append(
+                    {
+                        "alert_id": alert_id,
+                        "customer_id": customer_id,
+                        "alert_type": alert.get("alert_type"),
+                        "level": level,
+                    }
+                )
+                alert_ids.append(alert_id)
+
+        # ③ 降噪：同组多单合并为一次推送（审计仍逐单落）
+        if alert_ids:
+            merged_notices += 1
+            redis_gateway.publish(
+                "risk:pub:alert",
+                {
+                    "alert_id": alert_ids[0] if len(alert_ids) == 1 else f"group:{len(alert_ids)}",
+                    "alert_type": group[0].get("alert_type"),
+                    "customer_id_mask": (customer_id or "unknown")[:6] + "**",
+                    "risk_score": max((int(a.get("risk_score") or 0)) for a in group),
+                    "trace_id": trace_id,
+                    "notify_role": notify_role,
+                    "escalation_level": level,
+                    "alert_ids": alert_ids,
+                    "merged_count": len(alert_ids),
+                },
+            )
+
+    # 任务级审计一条（审计写库失败降级，不阻塞下次扫描，红线 5 口径）
+    try:
+        repo.insert_audit_log(
+            {
+                "trace_id": trace_id,
+                "event_type": "alert_escalation",
+                "agent_type": "risk",
+                "actor_id": "SYSTEM",
+                "customer_id": None,
+                "rule_id": None,
+                "input_summary": {
+                    "decision": "scan_completed",
+                    "scanned": scanned,
+                    "escalated": len(escalated),
+                    "merged_notices": merged_notices,
+                    "skipped": skipped,
+                },
+                "decision": "scan_completed",
+                "risk_score": None,
+                "handler_id": None,
+                "handler_result": None,
+                "handler_comment": None,
+            }
+        )
+    except Exception:
+        logger.exception("escalation scan task audit failed (degraded)")
+
+    # merged_notices 以实际发生合并推送的组数计（len(grouped) 即组数）
+    return {
+        "scanned": scanned,
+        "escalated": escalated,
+        "merged_notices": len(grouped),
+        "skipped": skipped,
+    }
diff --git a/app/service/tool_service.py b/app/service/tool_service.py
index 7099372..ff0370d 100644
--- a/app/service/tool_service.py
+++ b/app/service/tool_service.py
@@ -84,6 +84,8 @@ _INTENT_KEYWORDS: dict[str, list[tuple[str, tuple[str, ...]]]] = {
         _KB_INTENT,
     ],
     "risk": [
+        # FR-9 时效升级查询：置于 alert_query 之前（"超时/超期预警"话术不得误命中台账）
+        ("query_overdue_alerts", ("超期", "超时", "逾期", "多久没处理", "处置时效")),
         ("alert_query", ("预警", "待审", "预警台账", "待审预警")),
         ("customer_context", ("客户上下文", "监测信息", "风险画像", "客户监测")),
         ("suitability_check", ("适当性", "能不能买", "可购", "适合买", "购买资格")),
@@ -367,6 +369,19 @@ def summarize(record: dict[str, Any]) -> str:
             f"申赎合计 {data.get('sum_amount', 0)} 元）"
         )
     # ---- C1 风控四只读 Tool 摘要（降级回复/LLM 上下文共用） ----
+    if name == "query_overdue_alerts":
+        items = data.get("items") or []
+        hours = data.get("hours_threshold")
+        if not items:
+            return f"（超期预警查询：阈值 {hours} 小时内无超期待审预警）"
+        lines = [f"（超期预警：共 {len(items)} 条待审超期（阈值 {hours} 小时），按超期时长降序）"]
+        for it in items[:5]:
+            lvl = it.get("escalation_level")
+            lvl_txt = f"，已升级至 L{lvl}" if lvl else ""
+            lines.append(
+                f"- {it.get('alert_id')}（{it.get('alert_type')}）：已超期 {it.get('overdue_hours')} 小时{lvl_txt}"
+            )
+        return "\n".join(lines)
     if name == "alert_query":
         scope = data.get("scope")
         pending = data.get("pending_count", 0)
diff --git a/docs/项目框架设计/技术选型和版本/02-JWT-RBAC鉴权手册.md b/docs/项目框架设计/技术选型和版本/02-JWT-RBAC鉴权手册.md
index 7968ed1..7b4b384 100644
--- a/docs/项目框架设计/技术选型和版本/02-JWT-RBAC鉴权手册.md
+++ b/docs/项目框架设计/技术选型和版本/02-JWT-RBAC鉴权手册.md
@@ -227,6 +227,7 @@
 | `advisor` | `agent:advisor:chat`, `profile:l1:read`, `profile:l2:read`, `profile:l2:write`, `profile:l3:read`, `core:*:read:assigned` |
 | `analyst` | `agent:analyst:chat`, `profile:l1:read`, `profile:l2:read`, `profile:l3:read`, `core:*:read:scoped`, `sql:execute:readonly`, `risk:alert:read` |
 | `risk_officer` | `agent:risk:chat`, `profile:l1:read`, `profile:l2:read`, `profile:l3:read`, `profile:l3:write`, `risk:alert:write`, `risk:suitability:write`, `core:*:read:all` |
+| `risk_manager` | `risk:alert:read`, `core:*:read:all` | 风控经理：HTTP 台账全量只读，**无处置权、对话线经 chat 层显式拒绝**（PRD 4A.1 / C5 前置） |
 | `compliance` | `audit:read:all`, `agent:advisor:audit`, `compliance:hit:read` |
 | `ops` | `agent:advisor:stats`, `audit:read:aggregated` |
 | `service_risk` | `agent:risk:suitability_check`, `risk:suitability:write`, `profile:l1:read`, `profile:l2:read`, `audit:write` |
@@ -240,7 +241,7 @@
 | `customer` | `customer` | `customer` |
 | `advisor` | `staff` | `advisor`, `compliance`, `ops` |
 | `analyst` | `staff` | `analyst`, `compliance` |
-| `risk` | `staff`, `service` | `risk_officer`, `service_risk` |
+| `risk` | `staff`, `service` | `risk_officer`, `service_risk`, `risk_manager` |
 
 ---
 
@@ -256,6 +257,7 @@ JWT 通过后，**任何涉及 `customer_id` 的读写** 必须执行归属校
 | `advisor` | 客户画像 / 持仓 / 会话 | `customer_advisor_rel` 存在 `advisor_id=sub AND customer_id=? AND rel_status=active` | `AUTH_403_NOT_ASSIGNED` |
 | `analyst` | 客户级明细 | 走 **数据权限目录**；无明细权限则 **仅允许聚合 SQL**（`COUNT`/`GROUP BY`，结果 ≥ k-匿名阈值） | `AUTH_403_SCOPE` |
 | `risk_officer` | 全量客户 | 允许读全部；写仅限 L3 / 预警 / 适当性 | — |
+| `risk_manager` | 全量客户 | **仅读**预警台账（`risk:alert:read`，无 `risk:alert:write`）；对话线（chat）经 `app/api/chat.py` 显式 `deny(AUTH_403_ROLE)` 拒绝，强制走 HTTP 台账 | — |
 | `compliance` | 审计 / 会话 | 可跨客户读 **审计类表**；读 `agent_message` 需 `audit:read:content` 额外权限 | `AUTH_403_AUDIT_SCOPE` |
 | 跨 Agent 读会话 | `agent_session` | 角色只能读 **同 agent_type** 会话；合规除外 | `AUTH_403_SESSION_AGENT` |
 
diff --git a/scripts/core/02-seed-base.sql b/scripts/core/02-seed-base.sql
index 386cfe4..e1ef94c 100644
--- a/scripts/core/02-seed-base.sql
+++ b/scripts/core/02-seed-base.sql
@@ -56,6 +56,12 @@ INSERT INTO core_staff (staff_id, display_name, staff_type, roles) VALUES
 INSERT INTO core_staff (staff_id, display_name, staff_type, roles) VALUES
 ('STAFF-90001', '风控演示账号', 'risk_officer', '["risk_officer", "risk_demo"]');
 
+-- 2 风控经理（C5 前置 · PRD 4A.1：risk_manager 可看台账全量，无处置权）
+-- staff_type 仍为 risk_officer（同岗位序列），roles 标 risk_manager 用于权限区分
+INSERT INTO core_staff (staff_id, display_name, staff_type, roles) VALUES
+('STAFF-31001', '风控经理甲', 'risk_officer', '["risk_manager"]'),
+('STAFF-31002', '风控经理乙', 'risk_officer', '["risk_manager"]');
+
 -- ========== 产品 14 只（含起购/期限 · C-11） ==========
 INSERT INTO core_product (
     product_id, product_name, product_type, min_risk_code,
diff --git a/scripts/cron/escalation_scan.py b/scripts/cron/escalation_scan.py
new file mode 100644
index 0000000..d43899f
--- /dev/null
+++ b/scripts/cron/escalation_scan.py
@@ -0,0 +1,33 @@
+"""RISK-007 处置时效升级扫描（建议 15min 周期；RISK_ESCALATION_SCAN_MINUTES 可配）。
+
+初期独立脚本 + 系统 cron；内嵌 lifespan 归 M4 评估（开发计划挂账 #2）。
+结构对齐 scripts/demo/rebuild_alerts.py：sys.path 引导 → new_trace → 扫描 → 打印
+JSON 摘要（供 cron 日志 / 演示走查）。审计写库失败由 service 内降级留底，不阻塞
+下次扫描（红线 5 口径）。
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+# ① sys.path 引导项目根（rebuild_alerts 先例）
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from app.service.risk.escalation_service import scan_and_escalate  # noqa: E402
+from app.utils.trace import new_trace  # noqa: E402
+
+
+def main() -> int:
+    # ② 显式生成 trace（无 HTTP 上下文，service 内 current_trace 复用）
+    new_trace()
+    # ③ 扫描 → 打印 JSON 摘要
+    summary = scan_and_escalate()
+    print(json.dumps(summary, ensure_ascii=False, default=str, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/conftest.py b/tests/conftest.py
index 778db47..51671d5 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -11,7 +11,7 @@
 
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import pytest
@@ -53,6 +53,61 @@ def sqlite_engine():
     engine.dispose()
 
 
+@pytest.fixture()
+def backdated_alert(sqlite_engine):
+    """注入 created_at 回拨的 pending 单（C5/C6 免等待基建）；teardown 按 alert_id 精确删除。
+
+    用法：
+        aid = backdated_alert(alert_id="ALT-TEST-X", customer_id="C1", hours_ago=5,
+                               payload={...}, alert_type="pattern", risk_score=60)
+    注意：注入与读取必须共用同一个 sqlite_engine（StaticPool 单连接共享）——测试内
+    一律 `RiskRepository(engine=sqlite_engine)`；回拨行 teardown 精确按 alert_id 删，
+    不污染正常 trace 空间（评审必查点，_cleanup_test_rows 的时间窗清不掉回拨行）。
+    """
+    from sqlalchemy import text
+
+    import json
+
+    created_ids: list[str] = []
+
+    def _make(
+        alert_id: str,
+        customer_id: str,
+        hours_ago: float,
+        payload: dict | None = None,
+        alert_type: str = "pattern",
+        risk_score: int = 60,
+    ) -> str:
+        created_at = datetime.now() - timedelta(hours=hours_ago)
+        with sqlite_engine.begin() as conn:
+            conn.execute(
+                text(
+                    "INSERT INTO risk_alert (alert_id, trace_id, customer_id, trade_id,"
+                    " alert_type, triggered_rules, risk_score, status, payload, created_at)"
+                    " VALUES (:aid, 'TRACE-TEST', :cid, 'TRD-TEST-0', :atype, :rules,"
+                    " :score, 'pending_review', :payload, :created_at)"
+                ),
+                {
+                    "aid": alert_id,
+                    "cid": customer_id,
+                    "atype": alert_type,
+                    "rules": json.dumps(["RISK-007"]),
+                    "score": risk_score,
+                    "payload": json.dumps(payload or {}, ensure_ascii=False),
+                    "created_at": created_at,
+                },
+            )
+        created_ids.append(alert_id)
+        return alert_id
+
+    yield _make
+    with sqlite_engine.begin() as conn:
+        for aid in created_ids:
+            conn.execute(
+                text("DELETE FROM risk_alert WHERE alert_id = :aid"), {"aid": aid}
+            )
+
+
 # ---------- 真库集成环境（B8 集成测试专用） ----------
 
 
diff --git a/tests/test_chat.py b/tests/test_chat.py
index 2fd692d..95a275a 100644
--- a/tests/test_chat.py
+++ b/tests/test_chat.py
@@ -288,6 +288,19 @@ def test_chat_tool_risk_agent_no_intent_rule(env):
     assert _rows(env["engine"], "SELECT 1 FROM agent_tool_call") == []
 
 
+def test_chat_risk_manager_entry_denied(env):
+    """C5 前置（PRD 4A.1）：对话线不放行 risk_manager——矩阵放行 HTTP 通道，
+    chat 层显式 deny 保住 FR-6 冻结口径；返回 403 + AUTH_403_ROLE，且不得建会话。"""
+    r = env["client"].post(
+        "/api/chat",
+        json={"message": "查一下预警台账", "customer_id": "CUST-9527"},
+        headers={"X-Debug-Role": "risk_manager", "X-Debug-Actor": "STAFF-31001", "X-Agent-Type": "risk"},
+    )
+    assert r.status_code == 403
+    assert r.json()["error_code"] == "AUTH_403_ROLE"
+    assert _rows(env["engine"], "SELECT 1 FROM agent_session") == []
+
+
 # ---------- JWT 通道（生产主链路 · 评审 P2-3） ----------
 
 
diff --git a/tests/test_chat_tools.py b/tests/test_chat_tools.py
index deefd47..36f0f2f 100644
--- a/tests/test_chat_tools.py
+++ b/tests/test_chat_tools.py
@@ -607,3 +607,44 @@ def test_summarize_tool_rejection_not_authz_wording():
     record = {"tool_name": "query_holdings", "status": "blocked", "error_code": "TOOL_BAD_PARAM"}
     text = tool_service.summarize(record)
     assert "无权访问" not in text and "TOOL_BAD_PARAM" in text
+
+
+# ---------- C5 · FR-9 query_overdue_alerts Tool（直接调函数；需真实 RiskRepository） ----------
+
+
+def test_query_overdue_alerts_filters_by_hours(sqlite_engine, backdated_alert):
+    from app.repository.risk_repository import RiskRepository
+    from app.service.risk.chat_tools import query_overdue_alerts
+
+    backdated_alert("ALT-OV-1", "C1", hours_ago=5)  # 超 4h → 命中
+    backdated_alert("ALT-OV-2", "C1", hours_ago=2)  # 未达 4h → 不命中
+    repo = RiskRepository(engine=sqlite_engine)
+
+    # 缺省 hours = settings.risk_escalation_l1_hours（4）→ 仅 1 张
+    data = query_overdue_alerts(None, core_ro=None, risk_repo=repo)
+    assert data["overdue_count"] == 1
+    assert data["items"][0]["alert_id"] == "ALT-OV-1"
+    assert data["items"][0]["overdue_hours"] >= 4
+
+    # 显式 hours=1 → 两张都超期
+    data2 = query_overdue_alerts(None, core_ro=None, risk_repo=repo, hours=1)
+    assert data2["overdue_count"] == 2
+
+    # 按超期时长降序
+    assert data2["items"][0]["overdue_hours"] >= data2["items"][1]["overdue_hours"]
+
+
+def test_query_overdue_alerts_intent_match_and_summarize(sqlite_engine, backdated_alert):
+    """意图词命中 query_overdue_alerts（置于 alert_query 之前）+ summarize 摘要。"""
+    from app.repository.risk_repository import RiskRepository
+    from app.service.risk.chat_tools import query_overdue_alerts
+
+    backdated_alert("ALT-OV-3", "C1", hours_ago=6)
+    repo = RiskRepository(engine=sqlite_engine)
+    data = query_overdue_alerts(None, core_ro=None, risk_repo=repo, hours=1)
+    assert tool_service.match_intent("risk", "这些预警超时多久没处理了") == "query_overdue_alerts"
+
+    record = {"tool_name": "query_overdue_alerts", "status": "success", "data": data}
+    text = tool_service.summarize(record)
+    assert "超期预警" in text
+    assert "ALT-OV-3" in text
diff --git a/tests/test_escalation_service.py b/tests/test_escalation_service.py
new file mode 100644
index 0000000..199bef5
--- /dev/null
+++ b/tests/test_escalation_service.py
@@ -0,0 +1,186 @@
+"""C5 / FR-9 · RISK-007 处置时效升级：扫描判级 + 幂等 + 降噪合并 + 通知链 + payload 隔离。
+
+覆盖实现方案 §6.2 C5 相关用例（普通 4h/24h、AML 1h/4h 短通道、重复扫描幂等、
+同客户多单合并一次推送、处置后退出扫描、status 全程 pending_review、
+notify_role 链 L1 含 risk_manager / L2 含 compliance、update_alert_escalation
+不碰 handler 列）。
+
+环境：复用 conftest 的 sqlite_engine（StaticPool 单连接共享）+ backdated_alert
+回拨注入；RiskRepository 一律 `engine=sqlite_engine` 与注入同源。
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import text
+
+from app.repository.risk_repository import RiskRepository
+from app.service.risk import redis_gateway
+from app.service.risk.escalation_service import (
+    EscalationThresholds,
+    compute_level,
+    scan_and_escalate,
+)
+
+
+class FakePublisher:
+    def __init__(self):
+        self.messages: list = []
+
+    def publish(self, channel, payload):
+        self.messages.append((channel, payload))
+
+    def delete(self, *keys):
+        pass
+
+
+@pytest.fixture()
+def env(sqlite_engine, monkeypatch):
+    """注入 fake 发布器（升级通知不依赖真实 Redis）。"""
+    repo = RiskRepository(engine=sqlite_engine)
+    pub = FakePublisher()
+    redis_gateway.set_gateway(pub)
+    yield repo, pub, sqlite_engine
+    redis_gateway.set_gateway(None)
+
+
+def _now_after(alert_created_at: datetime, hours: float) -> datetime:
+    """扫描时刻相对回拨注入时刻再后移（避免边界竞态）。"""
+    return alert_created_at + timedelta(hours=hours)
+
+
+# ---------- 判级单测（compute_level） ----------
+
+
+def test_compute_level_boundary():
+    th = EscalationThresholds()
+    base = datetime(2026, 9, 6, 12, 0, 0)
+
+    def mk(hours_ago):
+        return {"alert_type": "pattern", "created_at": base - timedelta(hours=hours_ago)}
+
+    assert compute_level(mk(3.9), base, th) == 0
+    assert compute_level(mk(4.0), base, th) == 1
+    assert compute_level(mk(23.9), base, th) == 1
+    assert compute_level(mk(24.0), base, th) == 2
+
+
+def test_compute_level_aml_short_channel():
+    th = EscalationThresholds()
+    base = datetime(2026, 9, 6, 12, 0, 0)
+
+    def mk(hours_ago):
+        return {"alert_type": "aml", "created_at": base - timedelta(hours=hours_ago)}
+
+    assert compute_level(mk(0.9), base, th) == 0
+    assert compute_level(mk(1.0), base, th) == 1
+    assert compute_level(mk(3.9), base, th) == 1
+    assert compute_level(mk(4.0), base, th) == 2
+
+
+# ---------- 扫描集成 ----------
+
+
+def test_no_escalation_below_l1(env, backdated_alert):
+    repo, pub, engine = env
+    backdated_alert("ALT-E-1", "C1", hours_ago=3.9)
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert result["escalated"] == []
+    assert result["skipped"] >= 1
+    alert = repo.get_alert("ALT-E-1")
+    assert alert["payload"].get("escalation_level", 0) == 0
+    assert alert["status"] == "pending_review"
+
+
+def test_escalate_to_l1_at_4h(env, backdated_alert):
+    repo, pub, engine = env
+    backdated_alert("ALT-E-2", "C1", hours_ago=4.1)
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert len(result["escalated"]) == 1
+    assert result["escalated"][0]["level"] == 1
+    alert = repo.get_alert("ALT-E-2")
+    assert alert["payload"]["escalation_level"] == 1
+    assert alert["status"] == "pending_review"
+    # 通知链 L1 含 risk_manager（不含 compliance）
+    assert len(pub.messages) == 1
+    notify = pub.messages[0][1]["notify_role"]
+    assert "risk_officer" in notify and "risk_manager" in notify
+    assert "compliance" not in notify
+
+
+def test_escalate_to_l2_at_24h(env, backdated_alert):
+    repo, pub, engine = env
+    backdated_alert("ALT-E-3", "C1", hours_ago=24.1)
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert result["escalated"][0]["level"] == 2
+    alert = repo.get_alert("ALT-E-3")
+    assert alert["payload"]["escalation_level"] == 2
+    notify = pub.messages[0][1]["notify_role"]
+    assert "compliance" in notify  # L2 升级到合规
+
+
+def test_aml_escalate_at_1h(env, backdated_alert):
+    repo, pub, engine = env
+    backdated_alert("ALT-E-AML", "C1", hours_ago=1.1, alert_type="aml", risk_score=95)
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert result["escalated"][0]["level"] == 1
+    assert repo.get_alert("ALT-E-AML")["payload"]["escalation_level"] == 1
+
+
+def test_idempotent_repeat_scan(env, backdated_alert):
+    """同单重复扫描不重复推送（幂等闸门：仅升不降）。"""
+    repo, pub, engine = env
+    backdated_alert("ALT-E-IDEM", "C1", hours_ago=5)
+    scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert len(pub.messages) == 1
+    # 第二次：已达 L1，computed_level 不大于 current_level → 跳过，不再推送
+    result2 = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert result2["escalated"] == []
+    assert len(pub.messages) == 1
+    # 第三次：即便继续超时（now 后移），仍只升到 L2 一次，不再重复 L1 推送
+    scan_and_escalate(
+        now=datetime.now() + timedelta(hours=30),
+        risk_repo=repo,
+        thresholds=EscalationThresholds(),
+    )
+    # 仅 L1 一次 + L2 一次 = 2 条推送
+    assert len(pub.messages) == 2
+
+
+def test_merge_same_customer_level(env, backdated_alert):
+    """同客户多单同级别合并一次推送（降噪）。"""
+    repo, pub, engine = env
+    backdated_alert("ALT-E-M1", "C1", hours_ago=5)
+    backdated_alert("ALT-E-M2", "C1", hours_ago=6)
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert len(result["escalated"]) == 2  # 两张单都升级
+    assert result["merged_notices"] == 1  # 但只合并推送一次
+    assert len(pub.messages) == 1
+    assert pub.messages[0][1]["merged_count"] == 2
+
+
+def test_handled_exits_scan(env, backdated_alert):
+    """人工处置后 status 不再是 pending_review → 退出扫描范围。"""
+    repo, pub, engine = env
+    backdated_alert("ALT-E-H", "C1", hours_ago=10)
+    # 模拟人工处置（状态机变更）
+    repo.update_alert_status("ALT-E-H", "confirmed_normal", "STAFF-90001", "已核实")
+    result = scan_and_escalate(now=datetime.now(), risk_repo=repo, thresholds=EscalationThresholds())
+    assert result["scanned"] == 0
+    assert result["escalated"] == []
+
+
+def test_update_alert_escalation_keeps_handler_columns(env, backdated_alert):
+    """升级写入只动 payload，不碰 status / handler_id / handler_result / handled_at。"""
+    repo, pub, engine = env
+    backdated_alert("ALT-E-HC", "C1", hours_ago=5)
+    ok = repo.update_alert_escalation("ALT-E-HC", 1, datetime.now(), "TRACE-X")
+    assert ok is True
+    alert = repo.get_alert("ALT-E-HC")
+    assert alert["status"] == "pending_review"
+    assert alert["handler_id"] is None
+    assert alert["handler_result"] is None
+    assert alert["handled_at"] is None
+    assert alert["payload"]["escalation_level"] == 1
diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py
index f649a7a..9a86d2a 100644
--- a/tests/test_risk_api.py
+++ b/tests/test_risk_api.py
@@ -118,6 +118,7 @@ def _h(role="", actor=""):
 
 OFFICER = _h("risk_officer", "STAFF-90001")
 COMPLIANCE = _h("compliance", "STAFF-40001")
+MANAGER = _h("risk_manager", "STAFF-31001")  # C5 前置（PRD 4A.1）：HTTP 台账全量只读，无处置权
 CUST_1001 = _h("customer", "CUST-1001")
 ADV_01 = _h("advisor", "ADV-01")
 
@@ -148,6 +149,40 @@ def test_other_roles_cannot_list_alerts(client):
         assert r.status_code == 403
 
 
+def test_manager_sees_all_alerts_readonly(client):
+    """C5 前置（PRD 4A.1）：risk_manager 经矩阵放行 HTTP 通道，GET 全量 200（同 officer 查询路径）。"""
+    r = client.get("/api/risk/alerts", headers=MANAGER)
+    assert r.status_code == 200
+    assert r.json()["total"] == 3  # 全量（含已处置 + aml），无处置过滤
+
+
+def test_manager_handle_403(client, env):
+    """manager 无 risk_officer 角色 → handle 端点白名单自然 403（零代码改动）。"""
+    repo, _ = env
+    r = client.post(
+        "/api/risk/alerts/ALT-E1/handle",
+        json={"handler_result": "confirmed_suspicious"},
+        headers=MANAGER,
+    )
+    assert r.status_code == 403
+    assert repo.get_alert("ALT-E1")["status"] == "pending_review"  # 未改动
+
+
+def test_manager_suitability_check_403(client):
+    """评审 P2-5 回归：新增端点仅依赖矩阵放行时，manager 仍须被 G-01 挡在客户业务数据外。"""
+    r = client.post(
+        "/api/risk/suitability/check",
+        json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
+        headers=MANAGER,
+    )
+    assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SCOPE"
+
+
+def test_manager_aml_scan_403(client):
+    """评审 P2-5 回归：aml/scan 仅 risk_officer，manager 触碰即 403。"""
+    assert client.post("/api/risk/aml/scan", headers=MANAGER).status_code == 403
+
+
 def test_alerts_date_filter_and_pagination(client):
     """复审 P3:start_date/end_date 过滤与分页边界回归保护。