fix(chat): 定时功能审计修复 — 消除重复解析 + ID碰撞 + 输入补全
Some checks failed
CI / Lint & TypeCheck (push) Has been cancelled
CI / Unit Tests (push) Has been cancelled
CI / Build Frontend (push) Has been cancelled
CI / Rust Check (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
CI / E2E Tests (push) Has been cancelled
Some checks failed
CI / Lint & TypeCheck (push) Has been cancelled
CI / Unit Tests (push) Has been cancelled
CI / Build Frontend (push) Has been cancelled
CI / Rust Check (push) Has been cancelled
CI / Security Scan (push) Has been cancelled
CI / E2E Tests (push) Has been cancelled
审计发现修复: - H-01: 存储 ParsedSchedule 避免重复 parse_nl_schedule 调用 - H-03: trigger ID 追加 UUID 片段防止高并发碰撞 - C-02: execute_trigger 验证错误信息明确系统 Hand 必须注册 - M-02: SchedulerService 传递 trigger_name 作为 task_description - M-01: 添加拦截路径跳过 post_hook 的设计注释
This commit is contained in:
@@ -77,7 +77,7 @@ impl SchedulerService {
|
|||||||
kernel_lock: &Arc<Mutex<Option<Kernel>>>,
|
kernel_lock: &Arc<Mutex<Option<Kernel>>>,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
// Collect due triggers under lock
|
// Collect due triggers under lock
|
||||||
let to_execute: Vec<(String, String, String)> = {
|
let to_execute: Vec<(String, String, String, String)> = {
|
||||||
let kernel_guard = kernel_lock.lock().await;
|
let kernel_guard = kernel_lock.lock().await;
|
||||||
let kernel = match kernel_guard.as_ref() {
|
let kernel = match kernel_guard.as_ref() {
|
||||||
Some(k) => k,
|
Some(k) => k,
|
||||||
@@ -103,7 +103,8 @@ impl SchedulerService {
|
|||||||
.filter_map(|t| {
|
.filter_map(|t| {
|
||||||
if let zclaw_hands::TriggerType::Schedule { ref cron } = t.config.trigger_type {
|
if let zclaw_hands::TriggerType::Schedule { ref cron } = t.config.trigger_type {
|
||||||
if Self::should_fire_cron(cron, &now) {
|
if Self::should_fire_cron(cron, &now) {
|
||||||
Some((t.config.id.clone(), t.config.hand_id.clone(), cron.clone()))
|
// (trigger_id, hand_id, cron_expr, trigger_name)
|
||||||
|
Some((t.config.id.clone(), t.config.hand_id.clone(), cron.clone(), t.config.name.clone()))
|
||||||
} else {
|
} else {
|
||||||
None
|
None
|
||||||
}
|
}
|
||||||
@@ -123,7 +124,7 @@ impl SchedulerService {
|
|||||||
// If parallel execution is needed, spawn each execute_hand in a separate task
|
// If parallel execution is needed, spawn each execute_hand in a separate task
|
||||||
// and collect results via JoinSet.
|
// and collect results via JoinSet.
|
||||||
let now = chrono::Utc::now();
|
let now = chrono::Utc::now();
|
||||||
for (trigger_id, hand_id, cron_expr) in to_execute {
|
for (trigger_id, hand_id, cron_expr, trigger_name) in to_execute {
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"[Scheduler] Firing scheduled trigger '{}' → hand '{}' (cron: {})",
|
"[Scheduler] Firing scheduled trigger '{}' → hand '{}' (cron: {})",
|
||||||
trigger_id, hand_id, cron_expr
|
trigger_id, hand_id, cron_expr
|
||||||
@@ -138,6 +139,7 @@ impl SchedulerService {
|
|||||||
let input = serde_json::json!({
|
let input = serde_json::json!({
|
||||||
"trigger_id": trigger_id,
|
"trigger_id": trigger_id,
|
||||||
"trigger_type": "schedule",
|
"trigger_type": "schedule",
|
||||||
|
"task_description": trigger_name,
|
||||||
"cron": cron_expr,
|
"cron": cron_expr,
|
||||||
"fired_at": now.to_rfc3339(),
|
"fired_at": now.to_rfc3339(),
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -305,9 +305,10 @@ impl TriggerManager {
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Get hand (outside of our lock to avoid potential deadlock with hand_registry)
|
// Get hand (outside of our lock to avoid potential deadlock with hand_registry)
|
||||||
|
// System hands (prefixed with '_') must be registered at boot — same rule as create_trigger.
|
||||||
let hand = self.hand_registry.get(&hand_id).await
|
let hand = self.hand_registry.get(&hand_id).await
|
||||||
.ok_or_else(|| zclaw_types::ZclawError::InvalidInput(
|
.ok_or_else(|| zclaw_types::ZclawError::InvalidInput(
|
||||||
format!("Hand '{}' not found", hand_id)
|
format!("Hand '{}' not found (system hands must be registered at boot)", hand_id)
|
||||||
))?;
|
))?;
|
||||||
|
|
||||||
// Update state before execution
|
// Update state before execution
|
||||||
|
|||||||
@@ -221,7 +221,7 @@ pub async fn agent_chat_stream(
|
|||||||
// If the user's message contains a schedule intent (e.g. "每天早上9点提醒我查房"),
|
// If the user's message contains a schedule intent (e.g. "每天早上9点提醒我查房"),
|
||||||
// parse it with NlScheduleParser, create a trigger, and return confirmation
|
// parse it with NlScheduleParser, create a trigger, and return confirmation
|
||||||
// directly without calling the LLM.
|
// directly without calling the LLM.
|
||||||
let mut schedule_intercepted = false;
|
let mut captured_parsed: Option<zclaw_runtime::nl_schedule::ParsedSchedule> = None;
|
||||||
|
|
||||||
if zclaw_runtime::nl_schedule::has_schedule_intent(&message) {
|
if zclaw_runtime::nl_schedule::has_schedule_intent(&message) {
|
||||||
let parse_result = zclaw_runtime::nl_schedule::parse_nl_schedule(&message, &id);
|
let parse_result = zclaw_runtime::nl_schedule::parse_nl_schedule(&message, &id);
|
||||||
@@ -233,7 +233,12 @@ pub async fn agent_chat_stream(
|
|||||||
// Try to create a schedule trigger
|
// Try to create a schedule trigger
|
||||||
let kernel_lock = state.lock().await;
|
let kernel_lock = state.lock().await;
|
||||||
if let Some(kernel) = kernel_lock.as_ref() {
|
if let Some(kernel) = kernel_lock.as_ref() {
|
||||||
let trigger_id = format!("sched_{}", chrono::Utc::now().timestamp_millis());
|
// Use UUID fragment to avoid collision under high concurrency
|
||||||
|
let trigger_id = format!(
|
||||||
|
"sched_{}_{}",
|
||||||
|
chrono::Utc::now().timestamp_millis(),
|
||||||
|
&uuid::Uuid::new_v4().to_string()[..8]
|
||||||
|
);
|
||||||
let trigger_config = zclaw_hands::TriggerConfig {
|
let trigger_config = zclaw_hands::TriggerConfig {
|
||||||
id: trigger_id.clone(),
|
id: trigger_id.clone(),
|
||||||
name: parsed.task_description.clone(),
|
name: parsed.task_description.clone(),
|
||||||
@@ -242,6 +247,7 @@ pub async fn agent_chat_stream(
|
|||||||
cron: parsed.cron_expression.clone(),
|
cron: parsed.cron_expression.clone(),
|
||||||
},
|
},
|
||||||
enabled: true,
|
enabled: true,
|
||||||
|
// 60/hour = once per minute max, reasonable for scheduled tasks
|
||||||
max_executions_per_hour: 60,
|
max_executions_per_hour: 60,
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -251,11 +257,11 @@ pub async fn agent_chat_stream(
|
|||||||
"[agent_chat_stream] Schedule trigger created: {} (cron: {})",
|
"[agent_chat_stream] Schedule trigger created: {} (cron: {})",
|
||||||
trigger_id, parsed.cron_expression
|
trigger_id, parsed.cron_expression
|
||||||
);
|
);
|
||||||
schedule_intercepted = true;
|
captured_parsed = Some(parsed.clone());
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
"[agent_chat_stream] Failed to create schedule trigger: {}",
|
"[agent_chat_stream] Failed to create schedule trigger, falling through to LLM: {}",
|
||||||
e
|
e
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
@@ -272,20 +278,17 @@ pub async fn agent_chat_stream(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Get the streaming receiver while holding the lock, then release it
|
// Get the streaming receiver while holding the lock, then release it
|
||||||
let (mut rx, llm_driver) = if schedule_intercepted {
|
// NOTE: When schedule_intercepted, llm_driver is None so post_conversation_hook
|
||||||
|
// (memory extraction, heartbeat, reflection) is intentionally skipped —
|
||||||
|
// schedule confirmations are system messages, not user conversations.
|
||||||
|
let (mut rx, llm_driver) = if let Some(parsed) = captured_parsed {
|
||||||
// Schedule was intercepted — build confirmation message directly
|
// Schedule was intercepted — build confirmation message directly
|
||||||
let confirm_msg = {
|
let confirm_msg = format!(
|
||||||
let parsed = match zclaw_runtime::nl_schedule::parse_nl_schedule(&message, &id) {
|
|
||||||
zclaw_runtime::nl_schedule::ScheduleParseResult::Exact(p) => p,
|
|
||||||
_ => unreachable!("schedule_intercepted is only true for Exact results"),
|
|
||||||
};
|
|
||||||
format!(
|
|
||||||
"已为您设置定时任务:\n\n- **任务**:{}\n- **时间**:{}\n- **Cron**:`{}`\n\n任务已激活,将在设定时间自动执行。",
|
"已为您设置定时任务:\n\n- **任务**:{}\n- **时间**:{}\n- **Cron**:`{}`\n\n任务已激活,将在设定时间自动执行。",
|
||||||
parsed.task_description,
|
parsed.task_description,
|
||||||
parsed.natural_description,
|
parsed.natural_description,
|
||||||
parsed.cron_expression,
|
parsed.cron_expression,
|
||||||
)
|
);
|
||||||
};
|
|
||||||
|
|
||||||
let (tx, rx) = tokio::sync::mpsc::channel(32);
|
let (tx, rx) = tokio::sync::mpsc::channel(32);
|
||||||
let _ = tx.send(zclaw_runtime::LoopEvent::Delta(confirm_msg)).await;
|
let _ = tx.send(zclaw_runtime::LoopEvent::Delta(confirm_msg)).await;
|
||||||
|
|||||||
Reference in New Issue
Block a user