feat: 拆分 worker 架构并增强 assistant 安全保护

- 新增 daemon/channel/assistant/function/scheduler/logging worker 架构和 JetStream 消息协议 - 修复邮件通知误入 assistant、迁移 checksum、日志 subject 和工具调用无限循环风险 - 新增 assistant 工具轮次熔断、工具结果截断和提示词软收敛规则 - 补充邮件推送关键字、可信发件人配置、升级日志和验证测试
2026-07-10 00:41:36 +08:00
parent 7deae286ca
commit af6cfdaa83
70 changed files with 5953 additions and 1608 deletions
@@ -0,0 +1,256 @@
+use ias_common::mq::{
+    ConsumerSpec, DeadLetterPayload, InboundPayload, JetStreamBus, JetStreamConfig,
+    MessageEnvelope, StreamKind, StreamNames, inbound_subject,
+};
+
+#[tokio::test]
+#[ignore = "requires a running NATS JetStream server from docker-compose.nats.yml"]
+async fn jetstream_publish_and_ack_roundtrip() -> anyhow::Result<()> {
+    let config = JetStreamConfig::from_env();
+    let bus = JetStreamBus::connect(config.clone()).await?;
+    bus.ensure_streams().await?;
+    let subject = inbound_subject("wechat", "integration");
+    let payload = InboundPayload {
+        channel: "wechat".to_string(),
+        account_id: "integration".to_string(),
+        from_user_id: "tester".to_string(),
+        content: "ping".to_string(),
+        context: None,
+        external_message_id: Some("integration-msg-1".to_string()),
+        session_id: None,
+        group_id: None,
+        create_time_ms: None,
+    };
+    let envelope = MessageEnvelope::new(
+        subject.as_str(),
+        payload.idempotency_key(),
+        "trace-integration",
+        payload,
+    );
+    bus.publish_envelope(subject.as_str(), &envelope).await?;
+
+    let names = StreamNames::new(config.stream_prefix.clone());
+    let spec = ConsumerSpec {
+        stream: names.name(StreamKind::Inbound),
+        durable_name: "integration-ack-test".to_string(),
+        filter_subject: subject.to_string(),
+        ack_wait: config.ack_wait(),
+        max_deliver: config.max_deliver,
+    };
+    let (tx, mut rx) = tokio::sync::mpsc::channel(1);
+    let consumer_bus = bus.clone();
+    let handle = tokio::spawn(async move {
+        consumer_bus
+            .consume_envelopes::<InboundPayload, _, _>(spec, move |message| {
+                let tx = tx.clone();
+                async move {
+                    tx.send(message.message_id).await?;
+                    Ok(())
+                }
+            })
+            .await
+    });
+
+    let message_id = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv())
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("consumer closed before receiving test message"))?;
+    handle.abort();
+
+    assert_eq!(message_id, "wechat/integration/integration-msg-1");
+    Ok(())
+}
+
+#[tokio::test]
+#[ignore = "requires a running NATS JetStream server from docker-compose.nats.yml"]
+async fn jetstream_redelivery_reaches_dlq_after_max_deliver() -> anyhow::Result<()> {
+    let mut config = JetStreamConfig::from_env();
+    config.ack_wait_secs = 1;
+    config.max_deliver = 2;
+    let bus = JetStreamBus::connect(config.clone()).await?;
+    bus.ensure_streams().await?;
+    let names = StreamNames::new(config.stream_prefix.clone());
+    let unique = uuid::Uuid::new_v4().to_string();
+    let account_id = format!("dlq-{unique}");
+    let subject = inbound_subject("wechat", &account_id);
+    let payload = InboundPayload {
+        channel: "wechat".to_string(),
+        account_id: account_id.clone(),
+        from_user_id: "tester".to_string(),
+        content: "force retry".to_string(),
+        context: None,
+        external_message_id: Some(unique.clone()),
+        session_id: None,
+        group_id: None,
+        create_time_ms: None,
+    };
+    let message_id = format!("wechat/{account_id}/{unique}");
+    let envelope = MessageEnvelope::new(
+        subject.as_str(),
+        message_id.clone(),
+        format!("trace-{unique}"),
+        payload,
+    );
+
+    let (dlq_tx, mut dlq_rx) = tokio::sync::mpsc::channel(1);
+    let dlq_bus = bus.clone();
+    let dlq_stream = names.name(StreamKind::Dlq);
+    let dlq_ack_wait = config.ack_wait();
+    let dlq_max_deliver = config.max_deliver;
+    let dlq_handle = tokio::spawn(async move {
+        dlq_bus
+            .consume_envelopes::<DeadLetterPayload, _, _>(
+                ConsumerSpec {
+                    stream: dlq_stream,
+                    durable_name: format!("integration-dlq-{unique}"),
+                    filter_subject: "dlq.>".to_string(),
+                    ack_wait: dlq_ack_wait,
+                    max_deliver: dlq_max_deliver,
+                },
+                move |message| {
+                    let dlq_tx = dlq_tx.clone();
+                    let expected = message_id.clone();
+                    async move {
+                        if message.payload.original_message_id.as_deref() == Some(&expected) {
+                            dlq_tx.send(message.payload.original_message_id).await?;
+                        }
+                        Ok(())
+                    }
+                },
+            )
+            .await
+    });
+
+    let retry_bus = bus.clone();
+    let retry_spec = ConsumerSpec {
+        stream: StreamNames::new(config.stream_prefix.clone()).name(StreamKind::Inbound),
+        durable_name: format!("integration-redelivery-{account_id}"),
+        filter_subject: subject.to_string(),
+        ack_wait: config.ack_wait(),
+        max_deliver: config.max_deliver,
+    };
+    let retry_handle = tokio::spawn(async move {
+        retry_bus
+            .consume_envelopes::<InboundPayload, _, _>(retry_spec, |_message| async move {
+                anyhow::bail!("force redelivery")
+            })
+            .await
+    });
+
+    bus.publish_envelope(subject.as_str(), &envelope).await?;
+    let observed = tokio::time::timeout(std::time::Duration::from_secs(20), dlq_rx.recv())
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("dlq consumer closed before receiving test message"))?;
+    retry_handle.abort();
+    dlq_handle.abort();
+
+    assert_eq!(observed.as_deref(), Some(envelope.message_id.as_str()));
+    Ok(())
+}
+
+#[tokio::test]
+#[ignore = "requires a running NATS JetStream server from docker-compose.nats.yml"]
+async fn jetstream_concurrent_dlq_does_not_lose_messages() -> anyhow::Result<()> {
+    let mut config = JetStreamConfig::from_env();
+    config.ack_wait_secs = 1;
+    config.max_deliver = 2;
+    config.consumer_concurrency = 4;
+    let bus = JetStreamBus::connect(config.clone()).await?;
+    bus.ensure_streams().await?;
+    let names = StreamNames::new(config.stream_prefix.clone());
+    let batch = uuid::Uuid::new_v4().to_string();
+    let count = 8usize;
+    let account_id = format!("cdlq-{batch}");
+
+    // 发布 count 条不同 message_id 的消息
+    let mut message_ids = Vec::new();
+    for index in 0..count {
+        let external = format!("{batch}-{index}");
+        let subject = inbound_subject("wechat", &account_id);
+        let payload = InboundPayload {
+            channel: "wechat".to_string(),
+            account_id: account_id.clone(),
+            from_user_id: "tester".to_string(),
+            content: "force retry".to_string(),
+            context: None,
+            external_message_id: Some(external.clone()),
+            session_id: None,
+            group_id: None,
+            create_time_ms: None,
+        };
+        let message_id = format!("wechat/{account_id}/{external}");
+        message_ids.push(message_id.clone());
+        let envelope = MessageEnvelope::new(
+            subject.as_str(),
+            message_id,
+            format!("trace-{batch}-{index}"),
+            payload,
+        );
+        bus.publish_envelope(subject.as_str(), &envelope).await?;
+    }
+
+    // DLQ consumer 收集 original_message_id
+    let (dlq_tx, mut dlq_rx) = tokio::sync::mpsc::channel(count);
+    let dlq_bus = bus.clone();
+    let dlq_stream = names.name(StreamKind::Dlq);
+    let dlq_ack_wait = config.ack_wait();
+    let dlq_max_deliver = config.max_deliver;
+    let dlq_durable = format!("integration-cdlq-{batch}");
+    let dlq_handle = tokio::spawn(async move {
+        dlq_bus
+            .consume_envelopes::<DeadLetterPayload, _, _>(
+                ConsumerSpec {
+                    stream: dlq_stream,
+                    durable_name: dlq_durable,
+                    filter_subject: "dlq.>".to_string(),
+                    ack_wait: dlq_ack_wait,
+                    max_deliver: dlq_max_deliver,
+                },
+                move |message| {
+                    let dlq_tx = dlq_tx.clone();
+                    async move {
+                        let _ = dlq_tx.send(message.payload.original_message_id).await;
+                        Ok(())
+                    }
+                },
+            )
+            .await
+    });
+
+    // 失败 consumer（总是失败，触发 redelivery -> DLQ）
+    let retry_bus = bus.clone();
+    let retry_spec = ConsumerSpec {
+        stream: names.name(StreamKind::Inbound),
+        durable_name: format!("integration-cdlq-retry-{batch}"),
+        filter_subject: format!("inbound.wechat.{account_id}"),
+        ack_wait: config.ack_wait(),
+        max_deliver: config.max_deliver,
+    };
+    let retry_handle = tokio::spawn(async move {
+        retry_bus
+            .consume_envelopes::<InboundPayload, _, _>(retry_spec, |_| async move {
+                anyhow::bail!("force redelivery")
+            })
+            .await
+    });
+
+    // 收集 count 条 DLQ 消息，验证无碰撞丢失
+    let mut received = Vec::new();
+    for _ in 0..count {
+        let id = tokio::time::timeout(std::time::Duration::from_secs(30), dlq_rx.recv())
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("dlq consumer closed early"))?;
+        if let Some(id) = id {
+            received.push(id);
+        }
+    }
+    retry_handle.abort();
+    dlq_handle.abort();
+
+    received.sort();
+    message_ids.sort();
+    assert_eq!(received.len(), count, "DLQ 丢失了消息");
+    for id in &message_ids {
+        assert!(received.contains(id), "DLQ 缺少消息 {id}");
+    }
+    Ok(())
+}