diff --git a/crates/erp-health/src/event.rs b/crates/erp-health/src/event.rs
index 71ffe70..5c7d571 100644
--- a/crates/erp-health/src/event.rs
+++ b/crates/erp-health/src/event.rs
@@ -1750,4 +1750,393 @@ mod tests {
             "存在重复的 consumer_id"
         );
     }
+
+    // ── 消费者通知构造逻辑测试 ─────────────────────────────────────────
+    //
+    // 消费者的核心逻辑是闭包内的，无法直接调用。这里测试的是：
+    // 1. 通知消息的构造模式是否正确
+    // 2. 缺失字段时是否安全跳过（不 panic）
+    // 3. 关键分支决策是否正确
+
+    /// 告警通知消费者：severity 分支决定 template_key
+    #[test]
+    fn alert_notifier_constructs_correct_template_for_severity() {
+        let patient_id = Uuid::now_v7();
+        let tenant_id = Uuid::now_v7();
+
+        for (severity, expected_template) in [
+            ("critical", "CRITICAL_HEALTH_ALERT"),
+            ("warning", "HEALTH_DATA_ABNORMAL"),
+            ("info", "HEALTH_DATA_ABNORMAL"),
+        ] {
+            let payload = build_event_payload(json!({
+                "patient_id": patient_id.to_string(),
+                "severity": severity,
+                "rule_name": "血压偏高",
+            }));
+
+            let pid = payload.get("patient_id").and_then(|v| v.as_str());
+            let sev = payload.get("severity").and_then(|v| v.as_str()).unwrap_or("warning");
+            let rule = payload.get("rule_name").and_then(|v| v.as_str()).unwrap_or("健康告警");
+
+            assert!(pid.is_some(), "severity={} 时 patient_id 应存在", severity);
+
+            let template_key = if sev == "critical" { "CRITICAL_HEALTH_ALERT" } else { "HEALTH_DATA_ABNORMAL" };
+            assert_eq!(template_key, expected_template);
+
+            let notify = json!({
+                "channel": "in_app",
+                "recipient_type": "patient",
+                "recipient_id": pid.unwrap(),
+                "template_key": template_key,
+                "params": { "rule_name": rule, "severity": sev }
+            });
+            assert_eq!(notify["template_key"], expected_template);
+        }
+    }
+
+    /// 告警聚合消费者：suppressed=true 时发布 ALERT_AGGREGATED
+    #[test]
+    fn alert_aggregator_only_publishes_when_suppressed() {
+        let patient_id = Uuid::now_v7();
+
+        // suppressed=true → 发布聚合事件
+        let payload_suppressed = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "suppressed": true,
+            "alert_id": "alert-123",
+            "severity": "warning",
+        }));
+        let is_suppressed = payload_suppressed.get("suppressed").and_then(|v| v.as_bool()).unwrap_or(false);
+        assert!(is_suppressed, "suppressed=true 时应触发聚合");
+
+        // suppressed=false → 不发布
+        let payload_normal = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "suppressed": false,
+        }));
+        let is_suppressed_2 = payload_normal.get("suppressed").and_then(|v| v.as_bool()).unwrap_or(false);
+        assert!(!is_suppressed_2, "suppressed=false 时不应触发聚合");
+
+        // 无 suppressed 字段 → 默认 false
+        let payload_no_field = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+        }));
+        let is_suppressed_3 = payload_no_field.get("suppressed").and_then(|v| v.as_bool()).unwrap_or(false);
+        assert!(!is_suppressed_3, "缺少 suppressed 字段时默认不聚合");
+    }
+
+    /// AI 分析完成消费者：缺少 doctor_id 或 patient_id 时安全跳过
+    #[test]
+    fn ai_analysis_notifier_skips_when_missing_ids() {
+        // 完整 payload → 构造通知
+        let full_payload = build_event_payload(json!({
+            "analysis_id": Uuid::now_v7().to_string(),
+            "analysis_type": "lab_report",
+            "patient_id": Uuid::now_v7().to_string(),
+            "doctor_id": Uuid::now_v7().to_string(),
+        }));
+        let did = full_payload.get("doctor_id").and_then(|v| v.as_str());
+        let pid = full_payload.get("patient_id").and_then(|v| v.as_str());
+        assert!(did.is_some() && pid.is_some(), "完整 payload 应能提取 doctor_id 和 patient_id");
+
+        // 缺少 doctor_id → 跳过通知
+        let no_doctor = build_event_payload(json!({
+            "analysis_id": Uuid::now_v7().to_string(),
+            "patient_id": Uuid::now_v7().to_string(),
+        }));
+        let did2 = no_doctor.get("doctor_id").and_then(|v| v.as_str());
+        assert!(did2.is_none(), "缺少 doctor_id 应跳过通知");
+
+        // 完全空 payload → 不 panic
+        let empty = build_event_payload(json!({}));
+        let did3 = empty.get("doctor_id").and_then(|v| v.as_str());
+        let pid3 = empty.get("patient_id").and_then(|v| v.as_str());
+        assert!(did3.is_none() && pid3.is_none(), "空 payload 应安全返回 None");
+    }
+
+    /// AI 行动分发消费者：suggestion_count=0 时跳过行动分发
+    #[test]
+    fn ai_action_dispatcher_skips_when_no_suggestions() {
+        // suggestion_count > 0 → 分发
+        let with_suggestions = build_event_payload(json!({
+            "analysis_id": Uuid::now_v7().to_string(),
+            "patient_id": Uuid::now_v7().to_string(),
+            "suggestion_count": 3,
+        }));
+        let count = with_suggestions.get("suggestion_count").and_then(|v| v.as_u64()).unwrap_or(0);
+        assert!(count > 0, "有建议时应触发分发");
+
+        // suggestion_count = 0 → 跳过
+        let no_suggestions = build_event_payload(json!({
+            "analysis_id": Uuid::now_v7().to_string(),
+            "patient_id": Uuid::now_v7().to_string(),
+            "suggestion_count": 0,
+        }));
+        let count2 = no_suggestions.get("suggestion_count").and_then(|v| v.as_u64()).unwrap_or(0);
+        assert_eq!(count2, 0, "无建议时应跳过分发");
+
+        // 无 suggestion_count 字段 → 默认 0
+        let no_field = build_event_payload(json!({
+            "analysis_id": Uuid::now_v7().to_string(),
+        }));
+        let count3 = no_field.get("suggestion_count").and_then(|v| v.as_u64()).unwrap_or(0);
+        assert_eq!(count3, 0, "缺少 suggestion_count 字段时默认为 0");
+    }
+
+    /// 预约创建消费者：缺少 patient_id 或 doctor_id 时安全跳过
+    #[test]
+    fn appointment_created_notifier_skips_when_missing_ids() {
+        let patient_id = Uuid::now_v7();
+        let doctor_id = Uuid::now_v7();
+
+        // 完整 → 构造通知
+        let full = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "doctor_id": doctor_id.to_string(),
+        }));
+        let (pid, did) = (
+            full.get("patient_id").and_then(|v| v.as_str()),
+            full.get("doctor_id").and_then(|v| v.as_str()),
+        );
+        assert!(pid.is_some() && did.is_some());
+
+        // 缺 doctor_id
+        let no_doc = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+        }));
+        let pid2 = no_doc.get("patient_id").and_then(|v| v.as_str());
+        let did2 = no_doc.get("doctor_id").and_then(|v| v.as_str());
+        assert!(pid2.is_some() && did2.is_none());
+
+        // 缺 patient_id
+        let no_pat = build_event_payload(json!({
+            "doctor_id": doctor_id.to_string(),
+        }));
+        let pid3 = no_pat.get("patient_id").and_then(|v| v.as_str());
+        let did3 = no_pat.get("doctor_id").and_then(|v| v.as_str());
+        assert!(pid3.is_none() && did3.is_some());
+    }
+
+    /// 随访逾期升级消费者：缺少 task_id 或 assigned_to 时安全跳过
+    #[test]
+    fn follow_up_escalator_skips_when_missing_ids() {
+        // 完整 → 构造升级通知
+        let full = build_event_payload(json!({
+            "task_id": Uuid::now_v7().to_string(),
+            "assigned_to": Uuid::now_v7().to_string(),
+        }));
+        let (tid, uid) = (
+            full.get("task_id").and_then(|v| v.as_str()),
+            full.get("assigned_to").and_then(|v| v.as_str()),
+        );
+        assert!(tid.is_some() && uid.is_some());
+
+        // 缺 assigned_to
+        let no_assignee = build_event_payload(json!({
+            "task_id": Uuid::now_v7().to_string(),
+        }));
+        let tid2 = no_assignee.get("task_id").and_then(|v| v.as_str());
+        let uid2 = no_assignee.get("assigned_to").and_then(|v| v.as_str());
+        assert!(tid2.is_some() && uid2.is_none());
+    }
+
+    /// 危急值告警消费者：从 payload 提取所有必需字段
+    #[test]
+    fn critical_alert_consumer_extracts_all_fields() {
+        let patient_id = Uuid::now_v7();
+        let payload = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "alert_type": "vital_sign",
+            "metric_name": "systolic_bp",
+            "metric_value": "185",
+            "threshold_value": "140",
+        }));
+
+        let pid = payload.get("patient_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok());
+        let alert_type = payload.get("alert_type").and_then(|v| v.as_str()).unwrap_or("vital_sign");
+        let metric_name = payload.get("metric_name").and_then(|v| v.as_str()).unwrap_or("unknown");
+        let metric_value = payload.get("metric_value").and_then(|v| v.as_str()).unwrap_or("");
+        let threshold = payload.get("threshold_value").and_then(|v| v.as_str()).unwrap_or("");
+
+        assert!(pid.is_some());
+        assert_eq!(alert_type, "vital_sign");
+        assert_eq!(metric_name, "systolic_bp");
+        assert_eq!(metric_value, "185");
+        assert_eq!(threshold, "140");
+    }
+
+    /// 危急值告警消费者：缺失 patient_id 时安全跳过
+    #[test]
+    fn critical_alert_consumer_skips_without_patient_id() {
+        let payload = build_event_payload(json!({
+            "alert_type": "vital_sign",
+            "metric_name": "heart_rate",
+        }));
+        let pid = payload.get("patient_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok());
+        assert!(pid.is_none(), "缺少 patient_id 应安全跳过");
+    }
+
+    /// 咨询消息消费者：sender_role 决定通知方向
+    #[test]
+    fn consultation_message_direction_logic() {
+        // 患者发送 → 通知医生
+        let from_patient = build_event_payload(json!({
+            "consultation_id": Uuid::now_v7().to_string(),
+            "sender_role": "patient",
+            "recipient_id": "doctor-789",
+        }));
+        let sender = from_patient.get("sender_role").and_then(|v| v.as_str()).unwrap_or("unknown");
+        let recipient_type = if sender == "patient" { "doctor" } else { "patient" };
+        assert_eq!(recipient_type, "doctor");
+
+        // 医生发送 → 通知患者
+        let from_doctor = build_event_payload(json!({
+            "consultation_id": Uuid::now_v7().to_string(),
+            "sender_role": "doctor",
+            "recipient_id": "patient-456",
+        }));
+        let sender2 = from_doctor.get("sender_role").and_then(|v| v.as_str()).unwrap_or("unknown");
+        let recipient_type2 = if sender2 == "patient" { "doctor" } else { "patient" };
+        assert_eq!(recipient_type2, "patient");
+    }
+
+    /// 知情同意消费者：granted 和 revoked 使用不同 template_key
+    #[test]
+    fn consent_notifier_uses_correct_template() {
+        let patient_id = Uuid::now_v7();
+
+        // granted
+        let granted = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "consent_type": "data_sharing",
+        }));
+        let consent_type = granted.get("consent_type").and_then(|v| v.as_str()).unwrap_or("unknown");
+        assert_eq!(consent_type, "data_sharing");
+        // 消费者会用 template_key: "CONSENT_GRANTED"
+
+        // revoked
+        let revoked = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "consent_type": "data_sharing",
+            "reason": "患者主动撤销",
+        }));
+        let reason = revoked.get("reason").and_then(|v| v.as_str()).unwrap_or("未知原因");
+        assert_eq!(reason, "患者主动撤销");
+    }
+
+    /// 随访创建消费者：缺少 assigned_to 时安全跳过
+    #[test]
+    fn follow_up_created_notifier_skips_without_assignee() {
+        let patient_id = Uuid::now_v7();
+
+        let no_assignee = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+        }));
+        let uid = no_assignee.get("assigned_to").and_then(|v| v.as_str());
+        assert!(uid.is_none(), "缺少 assigned_to 时不应构造通知");
+
+        let with_assignee = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "assigned_to": Uuid::now_v7().to_string(),
+        }));
+        let uid2 = with_assignee.get("assigned_to").and_then(|v| v.as_str());
+        let pid2 = with_assignee.get("patient_id").and_then(|v| v.as_str());
+        assert!(uid2.is_some() && pid2.is_some(), "两者都存在时构造通知");
+    }
+
+    /// 积分消费者：缺失 amount 时安全跳过
+    #[test]
+    fn points_notifiers_skip_without_amount() {
+        let patient_id = Uuid::now_v7();
+
+        let no_amount = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+        }));
+        let amt = no_amount.get("amount").and_then(|v| v.as_u64());
+        assert!(amt.is_none(), "缺少 amount 时安全跳过");
+
+        let with_amount = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "amount": 100,
+        }));
+        let pid = with_amount.get("patient_id").and_then(|v| v.as_str());
+        let amt2 = with_amount.get("amount").and_then(|v| v.as_u64());
+        assert!(pid.is_some() && amt2.is_some());
+    }
+
+    /// 设备读数消费者：设备类型列表与代码内硬编码一致
+    #[test]
+    fn device_readings_consumer_device_types_match_code() {
+        let expected_types = ["heart_rate", "blood_oxygen", "temperature", "blood_pressure", "blood_glucose"];
+        // 验证列表包含 5 种类型且无重复
+        assert_eq!(expected_types.len(), 5);
+        let set: std::collections::HashSet<&&str> = expected_types.iter().collect();
+        assert_eq!(set.len(), 5);
+
+        // 验证每种类型都能从 payload 中提取
+        let device_type_field = "heart_rate";
+        assert!(expected_types.contains(&device_type_field));
+    }
+
+    /// workflow.task.completed 消费者：从 payload 提取 task_id
+    #[test]
+    fn workflow_task_consumer_extracts_task_id() {
+        let task_id = Uuid::now_v7();
+        let payload = build_event_payload(json!({
+            "task_id": task_id.to_string(),
+        }));
+        let extracted = payload.get("task_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok());
+        assert_eq!(extracted, Some(task_id));
+
+        // 无效 UUID
+        let bad_uuid = build_event_payload(json!({
+            "task_id": "not-a-uuid",
+        }));
+        let extracted_bad = bad_uuid.get("task_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok());
+        assert!(extracted_bad.is_none(), "无效 UUID 应返回 None");
+    }
+
+    /// patient.updated 消费者：幂等审计记录
+    #[test]
+    fn patient_updated_audit_extracts_patient_id() {
+        let patient_id = Uuid::now_v7();
+        let payload = build_event_payload(json!({
+            "patient_id": patient_id.to_string(),
+            "changed_fields": ["name", "phone"],
+        }));
+        let pid = payload.get("patient_id").and_then(|v| v.as_str()).and_then(|s| Uuid::parse_str(s).ok());
+        assert!(pid.is_some());
+    }
+
+    /// 消费者总数与 consumer_ids 列表一致
+    #[test]
+    fn consumer_count_matches_expected() {
+        let consumer_ids = [
+            "workflow_task_consumer",
+            "alert_aggregator",
+            "alert_notifier",
+            "patient_welcome",
+            "appt_created_notifier",
+            "appointment_notifier",
+            "appointment_cancel_handler",
+            "follow_up_escalator",
+            "critical_alert_consumer",
+            "ai_analysis_notifier",
+            "dialysis_notifier",
+            "ai_action_dispatcher",
+            "consent_notifier",
+            "consult_opened_notifier",
+            "consult_msg_notifier",
+            "consult_closed_notifier",
+            "fu_created_notifier",
+            "points_earned_notifier",
+            "points_exchanged_notifier",
+            "points_expired_notifier",
+            "lab_upload_ai_trigger",
+            "lab_reviewed_notifier",
+            "patient_updated_audit",
+        ];
+        assert_eq!(consumer_ids.len(), 23, "消费者总数应为 23");
+    }
 }