diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py
index b30457e..b85481b 100644
--- a/app/repository/risk_repository.py
+++ b/app/repository/risk_repository.py
@@ -219,6 +219,25 @@ class RiskRepository:
         row["payload"] = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
         return row
 
+    # ---------- 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)
+            """
+        )
+        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)
+
     # ---------- risk_suitability_log ----------
 
     def insert_suitability_log(self, log: dict[str, Any]) -> None:
diff --git a/app/service/risk/alert_service.py b/app/service/risk/alert_service.py
new file mode 100644
index 0000000..27c418e
--- /dev/null
+++ b/app/service/risk/alert_service.py
@@ -0,0 +1,286 @@
+"""预警单聚合与通知（B2 · 架构 §5.2 / PRD FR-4）。
+
+职责：单事件命中规则的**聚合决策与编排**（merge 原语 append_alert_event 在 repo，A3 评审口径）；
+进程内锁防并发首单双单（多进程部署时换 Redis SET NX，接口不变）；审计落库；Pub/Sub 通知广播。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import threading
+from datetime import date, datetime, time
+from typing import Any, Callable
+from uuid import uuid4
+
+from app.config.settings import settings
+from app.repository.risk_repository import EVENT_ALERT_TYPES, RiskRepository
+from app.service.risk.rules import RuleHit
+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))
+
+
+_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()
+
+
+def _day_start(day: date | None = None) -> datetime:
+    return datetime.combine(day or date.today(), time.min)
+
+
+def _new_alert_id() -> str:
+    return f"ALT-{datetime.now():%Y%m%d}-{uuid4().hex[:8].upper()}"
+
+
+def _audit(
+    risk_repo: RiskRepository,
+    *,
+    customer_id: str | None,
+    rule_id: str | None,
+    decision: str,
+    risk_score: int | None,
+    input_summary: dict[str, Any],
+    event_type: str = "risk_judgement",
+    actor_id: str = "SYSTEM",
+) -> None:
+    risk_repo.insert_audit_log(
+        {
+            "trace_id": current_trace() or new_trace(),
+            "event_type": event_type,
+            "agent_type": "risk",
+            "actor_id": actor_id,
+            "customer_id": customer_id,
+            "rule_id": rule_id,
+            "input_summary": input_summary,
+            "decision": decision,
+            "risk_score": risk_score,
+            "handler_id": None,
+            "handler_result": None,
+            "handler_comment": None,
+        }
+    )
+
+
+def _publish_alert(alert: dict[str, Any]) -> None:
+    _publish(
+        {
+            "alert_id": alert["alert_id"],
+            "alert_type": alert["alert_type"],
+            "customer_id_mask": alert["customer_id"][:6] + "**",
+            "risk_score": alert["risk_score"],
+            "trace_id": alert["trace_id"],
+            "notify_role": ["risk_officer"] + (["compliance"] if alert["alert_type"] == "aml" else []),
+        }
+    )
+
+
+def record_trade_alerts(
+    trade: dict[str, Any],
+    hits: list[RuleHit],
+    risk_repo: RiskRepository | None = None,
+    customer_context: dict[str, Any] | None = None,
+) -> dict[str, Any] | None:
+    """交易事件预警入口（引擎 B4 调用）。
+
+    hits 为空 → 审计 pass；否则事件类命中聚合为一张单（PRD FR-4：同客户同日仅一张事件类
+    pending 单，alert_type 随最高分规则动态更新）。
+    """
+    risk_repo = risk_repo or RiskRepository()
+    if not hits:
+        _audit(risk_repo, customer_id=trade["customer_id"], rule_id=None, decision="pass",
+               risk_score=None, input_summary={"trade_id": trade["trade_id"]})
+        return None
+
+    best = max(hits, key=lambda h: h.risk_score)
+    triggered_rules = sorted({h.rule_id for h in hits})
+    risk_score = max(h.risk_score for h in hits)
+    event = {
+        "trade_id": trade["trade_id"],
+        "product_id": trade["product_id"],
+        "trade_type": trade["trade_type"],
+        "amount": str(trade["amount"]),
+        "traded_at": str(trade["traded_at"]),
+        "matched_rules": [h.rule_id for h in hits],
+        "details": {h.rule_id: h.detail for h in hits},
+    }
+
+    def _agg(locked: bool) -> dict[str, Any]:
+        if locked:
+            pending = risk_repo.find_pending_event_alert(trade["customer_id"], _day_start())
+            if pending:
+                risk_repo.append_alert_event(
+                    pending["alert_id"], event, triggered_rules, risk_score, best.alert_type
+                )
+                updated = risk_repo.get_alert(pending["alert_id"])
+                _audit(risk_repo, customer_id=trade["customer_id"], rule_id=",".join(triggered_rules),
+                       decision="alert_appended", risk_score=updated["risk_score"],
+                       input_summary=event)
+                _publish_alert(updated)
+                return updated
+        alert = {
+            "alert_id": _new_alert_id(),
+            "trace_id": current_trace() or new_trace(),
+            "customer_id": trade["customer_id"],
+            "trade_id": trade["trade_id"],
+            "alert_type": best.alert_type,
+            "triggered_rules": triggered_rules,
+            "risk_score": risk_score,
+            "status": "pending_review",
+            "payload": {
+                "product_id": trade["product_id"],
+                "events": [event],
+                "customer_context": customer_context or {},
+            },
+        }
+        risk_repo.insert_alert(alert)
+        _audit(risk_repo, customer_id=trade["customer_id"], rule_id=",".join(triggered_rules),
+               decision="alert_created", risk_score=risk_score, input_summary=event)
+        _publish_alert(alert)
+        return alert
+
+    return _run_locked(f"agg:event:{trade['customer_id']}:{date.today()}", _agg)
+
+
+def record_suitability_alert(
+    trade_request: dict[str, Any],
+    rule_id: str,
+    block_reason: str,
+    risk_repo: RiskRepository | None = None,
+) -> dict[str, Any]:
+    """R-02 阻断预警（网关 FR-1 调用）：同客户+产品+日仅一张 pending 单。"""
+    risk_repo = risk_repo or RiskRepository()
+    event = {
+        "trade_id": trade_request["trade_id"],
+        "product_id": trade_request["product_id"],
+        "trade_type": trade_request["trade_type"],
+        "amount": str(trade_request["amount"]),
+        "traded_at": str(trade_request["traded_at"]),
+        "matched_rules": [rule_id],
+        "details": {rule_id: block_reason},
+    }
+    best_type, score = "suitability", TIER_SCORE["suitability"]
+
+    def _agg(locked: bool) -> dict[str, Any]:
+        if locked:
+            pending = risk_repo.find_pending_suitability_alert(
+                trade_request["customer_id"], trade_request["product_id"], _day_start()
+            )
+            if pending:
+                risk_repo.append_alert_event(
+                    pending["alert_id"], event, [rule_id], score, best_type
+                )
+                updated = risk_repo.get_alert(pending["alert_id"])
+                _audit(risk_repo, customer_id=trade_request["customer_id"], rule_id=rule_id,
+                       decision="suitability_alert_appended", risk_score=score,
+                       input_summary=event, event_type="suitability_block")
+                _publish_alert(updated)
+                return updated
+        alert = {
+            "alert_id": _new_alert_id(),
+            "trace_id": current_trace() or new_trace(),
+            "customer_id": trade_request["customer_id"],
+            "trade_id": trade_request["trade_id"],
+            "alert_type": best_type,
+            "triggered_rules": [rule_id],
+            "risk_score": score,
+            "status": "pending_review",
+            "payload": {
+                "product_id": trade_request["product_id"],
+                "events": [event],
+                "customer_context": {},
+            },
+        }
+        risk_repo.insert_alert(alert)
+        _audit(risk_repo, customer_id=trade_request["customer_id"], rule_id=rule_id,
+               decision="suitability_alert_created", risk_score=score,
+               input_summary=event, event_type="suitability_block")
+        _publish_alert(alert)
+        return alert
+
+    return _run_locked(
+        f"agg:suitability:{trade_request['customer_id']}:{trade_request['product_id']}:{date.today()}",
+        _agg,
+    )
+
+
+def record_aml_alert(
+    customer_id: str,
+    aml_detail: dict[str, Any],
+    risk_repo: RiskRepository | None = None,
+) -> dict[str, Any]:
+    """R-03 AML 命中：独立出单不聚合（PRD FR-4），单事件单张。"""
+    risk_repo = risk_repo or RiskRepository()
+    alert = {
+        "alert_id": _new_alert_id(),
+        "trace_id": current_trace() or new_trace(),
+        "customer_id": customer_id,
+        "trade_id": aml_detail.get("trade_id"),
+        "alert_type": "aml",
+        "triggered_rules": ["AML-001"],
+        "risk_score": TIER_SCORE["aml"],
+        "status": "pending_review",
+        "payload": {"product_id": aml_detail.get("product_id"), "events": [aml_detail], "customer_context": {}},
+    }
+    risk_repo.insert_alert(alert)
+    _audit(risk_repo, customer_id=customer_id, rule_id="AML-001", decision="aml_alert_created",
+           risk_score=TIER_SCORE["aml"], input_summary=aml_detail, event_type="aml_hit")
+    _publish_alert(alert)
+    return alert
diff --git a/tests/test_alert_service.py b/tests/test_alert_service.py
new file mode 100644
index 0000000..6af25a0
--- /dev/null
+++ b/tests/test_alert_service.py
@@ -0,0 +1,195 @@
+"""alert_service 单测（B2 · 聚合/去重/审计/推送 + 并发冒烟）。
+
+sqlite StaticPool 单连接共享内存库，跨线程可用；publisher 注入 fake。
+"""
+
+import json
+from datetime import datetime
+from decimal import Decimal
+from threading import Thread
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.pool import StaticPool
+
+from app.repository.risk_repository import RiskRepository
+from app.service.risk import alert_service
+from app.service.risk.alert_service import (
+    record_aml_alert,
+    record_suitability_alert,
+    record_trade_alerts,
+    set_publisher,
+)
+from app.service.risk.rules import RULE_ALERT_TYPES, RuleHit
+
+
+class FakePublisher:
+    def __init__(self):
+        self.messages = []
+
+    def publish(self, channel, payload):
+        self.messages.append((channel, json.loads(payload)))
+
+
+@pytest.fixture()
+def env():
+    engine = create_engine(
+        "sqlite://",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    with engine.begin() as conn:
+        conn.execute(
+            text(
+                """
+                CREATE TABLE risk_alert (
+                    alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64),
+                    customer_id VARCHAR(64), trade_id VARCHAR(64), alert_type VARCHAR(16),
+                    triggered_rules TEXT, risk_score INTEGER,
+                    status VARCHAR(24) DEFAULT 'pending_review', payload TEXT,
+                    handler_id VARCHAR(64), handler_result VARCHAR(64),
+                    handler_comment VARCHAR(512),
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, handled_at TIMESTAMP
+                )
+                """
+            )
+        )
+        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)
+    pub = FakePublisher()
+    set_publisher(pub)
+    yield repo, pub, engine
+    set_publisher(None)
+    engine.dispose()
+
+
+def _trade(trade_id, amount="600000", customer="C1"):
+    return {
+        "trade_id": trade_id,
+        "customer_id": customer,
+        "product_id": "P1",
+        "trade_type": "subscribe",
+        "amount": Decimal(amount),
+        "traded_at": datetime(2026, 9, 6, 14, 0, 0),
+    }
+
+
+def _hit(rule_id, score):
+    return RuleHit(rule_id, RULE_ALERT_TYPES[rule_id], score, "detail")
+
+
+def _counts(engine, table, where="1=1", params=None):
+    with engine.connect() as conn:
+        return conn.execute(text(f"SELECT COUNT(*) FROM {table} WHERE {where}"), params or {}).scalar_one()
+
+
+def test_no_hits_records_pass_audit(env):
+    repo, _, engine = env
+    result = record_trade_alerts(_trade("T1"), [], risk_repo=repo)
+    assert result is None
+    assert _counts(engine, "risk_alert") == 0
+    assert _counts(engine, "audit_log", "decision = 'pass'") == 1
+
+
+def test_first_large_trade_creates_agg_alert(env):
+    repo, pub, engine = env
+    hits = [_hit("RISK-001", 70), _hit("RISK-002", 70)]
+    alert = record_trade_alerts(_trade("T1"), hits, risk_repo=repo)
+    assert alert["alert_type"] == "large_amount"
+    assert alert["triggered_rules"] == ["RISK-001", "RISK-002"]
+    assert alert["risk_score"] == 70
+    assert alert["status"] == "pending_review"
+    assert len(alert["payload"]["events"]) == 1
+    assert alert["trace_id"].startswith("trc-")
+    assert _counts(engine, "risk_alert") == 1
+    assert _counts(engine, "audit_log", "decision = 'alert_created'") == 1
+    (channel, payload), = pub.messages
+    assert channel == "risk:pub:alert"
+    assert payload["alert_id"] == alert["alert_id"]
+    assert payload["customer_id_mask"].endswith("**")
+
+
+def test_same_day_second_trade_appends_not_duplicates(env):
+    repo, pub, engine = env
+    record_trade_alerts(_trade("T1"), [_hit("RISK-001", 70), _hit("RISK-002", 70)], risk_repo=repo)
+    alert = record_trade_alerts(_trade("T2"), [_hit("RISK-001", 70)], risk_repo=repo)
+    assert _counts(engine, "risk_alert") == 1  # 同日仅一张事件类单
+    got = repo.get_alert(alert["alert_id"])
+    assert len(got["payload"]["events"]) == 2
+    assert got["triggered_rules"] == ["RISK-001", "RISK-002"]
+    assert _counts(engine, "audit_log", "decision = 'alert_appended'") == 1
+    assert len(pub.messages) == 2  # 每次事件都广播，消费端按 alert_id 聚合
+
+
+def test_alert_type_follows_highest_score_rule(env):
+    """类型随最高分规则动态更新（PRD FR-4；评审 P1-3 口径）。"""
+    repo, _, _ = env
+    first = record_trade_alerts(_trade("T1"), [_hit("RISK-001", 70)], risk_repo=repo)
+    assert first["alert_type"] == "large_amount"
+    updated = record_trade_alerts(_trade("T2", "800000"), [_hit("RISK-005", 80)], risk_repo=repo)
+    assert updated["alert_id"] == first["alert_id"]
+    got = repo.get_alert(first["alert_id"])
+    assert got["alert_type"] == "pattern" and got["risk_score"] == 80
+    assert set(got["triggered_rules"]) == {"RISK-001", "RISK-005"}
+
+
+def test_suitability_dedup_per_product(env):
+    repo, _, engine = env
+    req1 = _trade("T1", "100000")
+    req2 = _trade("T2", "100000")
+    a1 = record_suitability_alert(req1, "SUIT-001", "C1 不可购 R4", risk_repo=repo)
+    a2 = record_suitability_alert(req2, "SUIT-001", "C1 不可购 R4", risk_repo=repo)
+    assert a1["alert_id"] == a2["alert_id"]  # 同客户+产品+日去重
+    got = repo.get_alert(a1["alert_id"])
+    assert got["alert_type"] == "suitability" and len(got["payload"]["events"]) == 2
+    # 不同产品独立出单
+    req3 = _trade("T3", "100000")
+    req3["product_id"] = "P2"
+    a3 = record_suitability_alert(req3, "SUIT-001", "x", risk_repo=repo)
+    assert a3["alert_id"] != a1["alert_id"]
+    assert _counts(engine, "risk_alert") == 2
+
+
+def test_aml_standalone_alert_with_compliance_notify(env):
+    repo, pub, _ = env
+    alert = record_aml_alert("C1", {"list_type": "sanction", "match": 0.93}, risk_repo=repo)
+    assert alert["alert_type"] == "aml" and alert["risk_score"] == 95
+    (_, payload), = pub.messages
+    assert "compliance" in payload["notify_role"]
+
+
+def test_concurrent_first_alerts_single_row(env):
+    """并发冒烟（B2 验收）：两线程同客户同日首单 → 预警单 1 张、events 含两笔。"""
+    repo, _, engine = env
+    errors = []
+
+    def worker(trade_id):
+        try:
+            record_trade_alerts(
+                _trade(trade_id), [_hit("RISK-001", 70), _hit("RISK-002", 70)], risk_repo=repo
+            )
+        except Exception as exc:  # pragma: no cover
+            errors.append(exc)
+
+    threads = [Thread(target=worker, args=(f"T{i}",)) for i in (1, 2)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    assert not errors, errors
+    assert _counts(engine, "risk_alert") == 1
+    with engine.connect() as conn:
+        events = conn.execute(text("SELECT payload FROM risk_alert")).scalar_one()
+    assert len(json.loads(events)["events"]) == 2