use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use motosan_agent_loop::{
    AgentEvent, AgentLoop, AgentOp, ChatOutput, LlmClient, LlmResponse, Message, ToolCallItem,
};
use motosan_agent_tool::{Tool, ToolContext, ToolDef, ToolResult};

struct MockLlm {
    responses: Mutex<Vec<LlmResponse>>,
    calls: Arc<std::sync::atomic::AtomicUsize>,
    captured: Arc<Mutex<Vec<Vec<Message>>>>,
}

struct SlowTool {
    name: &'static str,
    delay_ms: u64,
}

impl Tool for SlowTool {
    fn def(&self) -> ToolDef {
        ToolDef {
            name: self.name.to_string(),
            description: "slow tool".to_string(),
            input_schema: serde_json::json!({
                "type": "object",
                "properties": {},
                "required": []
            }),
        }
    }

    fn call(
        &self,
        _args: serde_json::Value,
        _ctx: &ToolContext,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ToolResult> + Send + '_>> {
        let delay = self.delay_ms;
        Box::pin(async move {
            tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
            ToolResult::text("ok")
        })
    }
}

struct StartSignalTool {
    name: &'static str,
    started: Arc<AtomicBool>,
    delay_ms: u64,
}

impl Tool for StartSignalTool {
    fn def(&self) -> ToolDef {
        ToolDef {
            name: self.name.to_string(),
            description: "signal tool".to_string(),
            input_schema: serde_json::json!({
                "type": "object",
                "properties": {},
                "required": []
            }),
        }
    }

    fn call(
        &self,
        _args: serde_json::Value,
        _ctx: &ToolContext,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ToolResult> + Send + '_>> {
        let started = Arc::clone(&self.started);
        let delay = self.delay_ms;
        Box::pin(async move {
            started.store(true, Ordering::SeqCst);
            tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
            ToolResult::text("ok")
        })
    }
}

impl MockLlm {
    fn new(
        responses: Vec<LlmResponse>,
    ) -> (
        Self,
        Arc<std::sync::atomic::AtomicUsize>,
        Arc<Mutex<Vec<Vec<Message>>>>,
    ) {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let captured = Arc::new(Mutex::new(Vec::new()));
        (
            Self {
                responses: Mutex::new(responses),
                calls: calls.clone(),
                captured: captured.clone(),
            },
            calls,
            captured,
        )
    }
}

#[async_trait]
impl LlmClient for MockLlm {
    async fn chat(
        &self,
        messages: &[Message],
        _tools: &[ToolDef],
    ) -> motosan_agent_loop::Result<ChatOutput> {
        self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        self.captured.lock().unwrap().push(messages.to_vec());

        let mut responses = self.responses.lock().unwrap();
        if responses.is_empty() {
            panic!("MockLlm: no more responses");
        }
        Ok(ChatOutput::new(responses.remove(0)))
    }
}

#[tokio::test]
async fn run_with_ops_none_matches_run() {
    let agent = AgentLoop::builder().build();
    let (llm1, _, _) = MockLlm::new(vec![LlmResponse::Message("ok".to_string())]);
    let (llm2, _, _) = MockLlm::new(vec![LlmResponse::Message("ok".to_string())]);

    let baseline = agent
        .run(&llm1, vec![Message::user("hello")], |_| {})
        .await
        .unwrap();
    let with_ops = agent
        .run_with_ops(&llm2, vec![Message::user("hello")], None, |_| {})
        .await
        .unwrap();

    assert_eq!(with_ops.answer, baseline.answer);
    assert_eq!(with_ops.iterations, baseline.iterations);
    assert_eq!(with_ops.tool_calls, baseline.tool_calls);
}

#[tokio::test]
async fn interrupt_stops_before_next_llm_call() {
    let (llm, llm_calls, _) = MockLlm::new(vec![
        LlmResponse::single_tool_call(
            "call_1".to_string(),
            "ask_user".to_string(),
            serde_json::json!({"question": "pick one"}),
        ),
        LlmResponse::Message("should-not-happen".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user_timeout(std::time::Duration::from_millis(50))
        .max_iterations(5)
        .build();

    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    ops_tx.send(AgentOp::Interrupt).await.unwrap();

    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let events_clone = events.clone();
    let result = agent
        .run_with_ops(
            &llm,
            vec![Message::user("loop")],
            Some(ops_rx),
            move |event| {
                if let AgentEvent::Interrupted = event {
                    events_clone.lock().unwrap().push("interrupted".to_string());
                }
            },
        )
        .await
        .unwrap();

    assert_eq!(result.answer, "(interrupted)");
    assert_eq!(llm_calls.load(std::sync::atomic::Ordering::SeqCst), 0);
    assert_eq!(events.lock().unwrap().as_slice(), ["interrupted"]);
}

#[tokio::test]
async fn inject_ops_are_drained_in_order_before_next_llm_call() {
    let (llm, _, captured) = MockLlm::new(vec![LlmResponse::Message("done".to_string())]);
    let agent = AgentLoop::builder().max_iterations(2).build();
    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    ops_tx
        .send(AgentOp::InjectUserMessage("first".to_string()))
        .await
        .unwrap();
    ops_tx
        .send(AgentOp::InjectHint("second".to_string()))
        .await
        .unwrap();

    let _ = agent
        .run_with_ops(&llm, vec![Message::user("base")], Some(ops_rx), |_| {})
        .await
        .unwrap();

    let calls = captured.lock().unwrap();
    let messages = &calls[0];
    assert_eq!(messages[0].content, "base");
    assert_eq!(messages[1].content, "first");
    assert_eq!(messages[2].content, "[Note: second]");
}

#[tokio::test]
async fn ask_user_emits_event_and_consumes_answer() {
    let (llm, _, captured) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![ToolCallItem {
            id: "call_1".to_string(),
            name: "ask_user".to_string(),
            args: serde_json::json!({
                "question": "Which one?",
                "options": ["A", "B"]
            }),
        }]),
        LlmResponse::Message("thanks".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user()
        .max_iterations(3)
        .build();
    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);

    let ask_event_seen = Arc::new(Mutex::new(false));
    let ask_event_seen_clone = ask_event_seen.clone();
    let ops_tx_clone = ops_tx.clone();

    let result = agent
        .run_with_ops(
            &llm,
            vec![Message::user("start")],
            Some(ops_rx),
            move |event| {
                if let AgentEvent::AskUser {
                    call_id,
                    question,
                    options,
                } = event
                {
                    assert_eq!(question, "Which one?");
                    assert_eq!(options, vec!["A".to_string(), "B".to_string()]);
                    *ask_event_seen_clone.lock().unwrap() = true;
                    let tx = ops_tx_clone.clone();
                    tokio::spawn(async move {
                        tx.send(AgentOp::AskUserAnswer {
                            call_id: Some(call_id),
                            answer: "B".to_string(),
                        })
                        .await
                        .unwrap();
                    });
                }
            },
        )
        .await
        .unwrap();

    assert_eq!(result.answer, "thanks");
    assert!(*ask_event_seen.lock().unwrap());

    let calls = captured.lock().unwrap();
    assert_eq!(calls.len(), 2);
    let second_call_messages = &calls[1];
    let tool_msg = second_call_messages
        .iter()
        .find(|m| m.role == motosan_agent_loop::Role::Tool)
        .expect("tool result message must exist");
    assert_eq!(tool_msg.content, "B");
}

#[tokio::test]
async fn ask_user_accepts_answer_with_none_call_id() {
    let (llm, _, captured) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![ToolCallItem {
            id: "call_1".to_string(),
            name: "ask_user".to_string(),
            args: serde_json::json!({
                "question": "Pick?",
                "options": ["X", "Y"]
            }),
        }]),
        LlmResponse::Message("done".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user()
        .max_iterations(3)
        .build();
    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let tx = ops_tx.clone();

    let _ = agent
        .run_with_ops(
            &llm,
            vec![Message::user("start")],
            Some(ops_rx),
            move |event| {
                if let AgentEvent::AskUser { .. } = event {
                    let tx_clone = tx.clone();
                    tokio::spawn(async move {
                        tx_clone
                            .send(AgentOp::AskUserAnswer {
                                call_id: None,
                                answer: "Y".to_string(),
                            })
                            .await
                            .unwrap();
                    });
                }
            },
        )
        .await
        .unwrap();

    let calls = captured.lock().unwrap();
    let second_call_messages = &calls[1];
    let tool_msg = second_call_messages
        .iter()
        .find(|m| m.role == motosan_agent_loop::Role::Tool)
        .expect("tool result message must exist");
    assert_eq!(tool_msg.content, "Y");
}

#[tokio::test]
async fn interrupt_while_waiting_for_ask_user_exits_without_timeout() {
    let (llm, llm_calls, _) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![ToolCallItem {
            id: "call_1".to_string(),
            name: "ask_user".to_string(),
            args: serde_json::json!({
                "question": "continue?",
                "options": ["yes", "no"]
            }),
        }]),
        LlmResponse::Message("should-not-happen".to_string()),
    ]);
    let agent = AgentLoop::builder()
        .with_ask_user_timeout(std::time::Duration::from_secs(3))
        .max_iterations(5)
        .build();

    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let events_clone = events.clone();

    let result = agent
        .run_with_ops(
            &llm,
            vec![Message::user("start")],
            Some(ops_rx),
            move |event| match event {
                AgentEvent::AskUser { .. } => {
                    let tx = ops_tx.clone();
                    tokio::spawn(async move {
                        tx.send(AgentOp::Interrupt).await.unwrap();
                    });
                }
                AgentEvent::Interrupted => events_clone.lock().unwrap().push("interrupted".into()),
                AgentEvent::AskUserTimeout { .. } => {
                    events_clone.lock().unwrap().push("timeout".into())
                }
                _ => {}
            },
        )
        .await
        .unwrap();

    assert_eq!(result.answer, "(interrupted)");
    assert_eq!(llm_calls.load(std::sync::atomic::Ordering::SeqCst), 1);
    let seen = events.lock().unwrap();
    assert!(seen.iter().any(|v| v == "interrupted"));
    assert!(!seen.iter().any(|v| v == "timeout"));
}

#[tokio::test]
async fn run_with_ops_preserves_parallel_tool_execution_without_ask_user() {
    let (llm, _, _) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![
            ToolCallItem {
                id: "call_1".to_string(),
                name: "slow_a".to_string(),
                args: serde_json::json!({}),
            },
            ToolCallItem {
                id: "call_2".to_string(),
                name: "slow_b".to_string(),
                args: serde_json::json!({}),
            },
        ]),
        LlmResponse::Message("done".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .tool(Arc::new(SlowTool {
            name: "slow_a",
            delay_ms: 120,
        }))
        .tool(Arc::new(SlowTool {
            name: "slow_b",
            delay_ms: 120,
        }))
        .max_iterations(3)
        .build();

    let started = std::time::Instant::now();
    let result = agent
        .run_with_ops(&llm, vec![Message::user("go")], None, |_| {})
        .await
        .unwrap();
    let elapsed = started.elapsed();

    assert_eq!(result.answer, "done");
    assert!(
        elapsed < std::time::Duration::from_millis(200),
        "expected parallel tool execution, elapsed={elapsed:?}"
    );
}

#[tokio::test]
async fn mixed_ask_user_and_regular_tool_calls_complete_without_deadlock() {
    let (llm, _, captured) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![
            ToolCallItem {
                id: "call_ask".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "continue?",
                    "options": ["yes", "no"]
                }),
            },
            ToolCallItem {
                id: "call_slow".to_string(),
                name: "slow_a".to_string(),
                args: serde_json::json!({}),
            },
        ]),
        LlmResponse::Message("done".to_string()),
    ]);

    let tool_started = Arc::new(AtomicBool::new(false));
    let overlap_observed = Arc::new(AtomicBool::new(false));

    let agent = AgentLoop::builder()
        .with_ask_user()
        .tool(Arc::new(StartSignalTool {
            name: "slow_a",
            started: Arc::clone(&tool_started),
            delay_ms: 120,
        }))
        .max_iterations(3)
        .build();
    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let overlap_seen_for_cb = Arc::clone(&overlap_observed);

    let result = agent
        .run_with_ops(
            &llm,
            vec![Message::user("start")],
            Some(ops_rx),
            move |event| {
                if let AgentEvent::AskUser { .. } = event {
                    let tx = ops_tx.clone();
                    let started_flag = Arc::clone(&tool_started);
                    let overlap = Arc::clone(&overlap_seen_for_cb);
                    tokio::spawn(async move {
                        for _ in 0..30 {
                            if started_flag.load(Ordering::SeqCst) {
                                overlap.store(true, Ordering::SeqCst);
                                break;
                            }
                            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
                        }
                        tx.send(AgentOp::AskUserAnswer {
                            call_id: None,
                            answer: "yes".to_string(),
                        })
                        .await
                        .unwrap();
                    });
                }
            },
        )
        .await
        .unwrap();

    assert_eq!(result.answer, "done");
    assert_eq!(result.tool_calls.len(), 2);
    assert!(result.tool_calls.iter().any(|(name, _)| name == "ask_user"));
    assert!(result.tool_calls.iter().any(|(name, _)| name == "slow_a"));

    let calls = captured.lock().unwrap();
    assert_eq!(calls.len(), 2);
    let second_call = &calls[1];
    assert!(second_call
        .iter()
        .any(|m| m.role == motosan_agent_loop::Role::Tool && m.content == "yes"));
    assert!(
        overlap_observed.load(Ordering::SeqCst),
        "expected non-ask tool to start while ask_user was waiting for answer"
    );
}

#[tokio::test]
async fn mixed_tool_calls_emit_stable_event_ordering() {
    let (llm, _, _) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![
            ToolCallItem {
                id: "call_ask".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "continue?",
                    "options": ["yes", "no"]
                }),
            },
            ToolCallItem {
                id: "call_slow".to_string(),
                name: "slow_a".to_string(),
                args: serde_json::json!({}),
            },
        ]),
        LlmResponse::Message("done".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user()
        .with_ask_user_timeout(std::time::Duration::from_millis(500))
        .tool(Arc::new(SlowTool {
            name: "slow_a",
            delay_ms: 120,
        }))
        .max_iterations(3)
        .build();

    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let events_for_cb = Arc::clone(&events);
    let saw_slow_started = Arc::new(AtomicBool::new(false));
    let pending_ask = Arc::new(AtomicBool::new(false));

    let result = agent
        .run_with_ops(&llm, vec![Message::user("start")], Some(ops_rx), {
            let saw_slow_started_cb = Arc::clone(&saw_slow_started);
            let pending_ask_cb = Arc::clone(&pending_ask);
            move |event| match event {
                AgentEvent::ToolStarted { name } => {
                    if name == "slow_a" {
                        saw_slow_started_cb.store(true, Ordering::SeqCst);
                        if pending_ask_cb.swap(false, Ordering::SeqCst) {
                            ops_tx
                                .try_send(AgentOp::AskUserAnswer {
                                    call_id: None,
                                    answer: "yes".to_string(),
                                })
                                .expect("failed to send ask_user answer after slow_a start");
                        }
                    }
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("started:{name}"));
                }
                AgentEvent::AskUser { question, .. } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("ask_user:{question}"));
                    if saw_slow_started_cb.load(Ordering::SeqCst) {
                        ops_tx
                            .try_send(AgentOp::AskUserAnswer {
                                call_id: None,
                                answer: "yes".to_string(),
                            })
                            .expect("failed to send ask_user answer after ask_user event");
                    } else {
                        pending_ask_cb.store(true, Ordering::SeqCst);
                    }
                }
                AgentEvent::ToolCompleted { name, .. } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("completed:{name}"));
                }
                AgentEvent::AskUserTimeout { call_id, question } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("timeout:{call_id}:{question}"));
                }
                _ => {}
            }
        })
        .await
        .unwrap();

    assert_eq!(result.answer, "done");
    let captured = events.lock().unwrap();
    let idx = |name: &str| {
        captured
            .iter()
            .position(|e| e == name)
            .unwrap_or_else(|| panic!("missing event: {name}, got: {:?}", *captured))
    };

    let ask_started = idx("started:ask_user");
    let ask_event = idx("ask_user:continue?");
    let ask_completed = idx("completed:ask_user");
    let slow_started = idx("started:slow_a");
    let slow_completed = idx("completed:slow_a");

    assert!(
        !captured.iter().any(|e| e.starts_with("timeout:")),
        "unexpected ask_user timeout events: {:?}",
        *captured
    );
    assert!(ask_started < ask_event);
    assert!(ask_event < ask_completed);
    assert!(slow_started < slow_completed);
    assert!(ask_started < ask_completed);
    assert!(slow_started < ask_completed);
}

#[tokio::test]
async fn multiple_ask_user_and_regular_tools_complete_in_single_round() {
    let (llm, _, captured) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![
            ToolCallItem {
                id: "call_ask_1".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "pick-a?",
                    "options": ["A", "B"]
                }),
            },
            ToolCallItem {
                id: "call_slow_1".to_string(),
                name: "slow_a".to_string(),
                args: serde_json::json!({}),
            },
            ToolCallItem {
                id: "call_ask_2".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "pick-b?",
                    "options": ["C", "D"]
                }),
            },
            ToolCallItem {
                id: "call_slow_2".to_string(),
                name: "slow_b".to_string(),
                args: serde_json::json!({}),
            },
        ]),
        LlmResponse::Message("done".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user()
        .with_ask_user_timeout(std::time::Duration::from_millis(500))
        .tool(Arc::new(SlowTool {
            name: "slow_a",
            delay_ms: 80,
        }))
        .tool(Arc::new(SlowTool {
            name: "slow_b",
            delay_ms: 80,
        }))
        .max_iterations(3)
        .build();

    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let ask_idx = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let events_for_cb = Arc::clone(&events);

    let result = agent
        .run_with_ops(&llm, vec![Message::user("start")], Some(ops_rx), {
            let ask_idx_cb = Arc::clone(&ask_idx);
            move |event| match event {
                AgentEvent::ToolStarted { name } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("started:{name}"));
                }
                AgentEvent::AskUser { question, .. } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("ask_user:{question}"));
                    let i = ask_idx_cb.fetch_add(1, Ordering::SeqCst);
                    let answer = if i == 0 { "A" } else { "D" };
                    ops_tx
                        .try_send(AgentOp::AskUserAnswer {
                            call_id: None,
                            answer: answer.to_string(),
                        })
                        .expect("failed to send ask_user answer in multi-ask test");
                }
                AgentEvent::ToolCompleted { name, .. } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("completed:{name}"));
                }
                AgentEvent::AskUserTimeout { call_id, question } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("timeout:{call_id}:{question}"));
                }
                _ => {}
            }
        })
        .await
        .unwrap();

    assert_eq!(result.answer, "done");
    assert_eq!(result.tool_calls.len(), 4);
    assert_eq!(ask_idx.load(Ordering::SeqCst), 2);
    let asks = result
        .tool_calls
        .iter()
        .filter(|(name, _)| name == "ask_user")
        .count();
    assert_eq!(asks, 2);

    let seen = events.lock().unwrap();
    assert!(seen.iter().any(|e| e == "started:slow_a"));
    assert!(seen.iter().any(|e| e == "started:slow_b"));
    assert!(seen.iter().any(|e| e == "ask_user:pick-a?"));
    assert!(seen.iter().any(|e| e == "ask_user:pick-b?"));
    assert!(
        !seen.iter().any(|e| e.starts_with("timeout:")),
        "unexpected timeout events in multi-ask test: {:?}",
        *seen
    );

    let llm_calls = captured.lock().unwrap();
    assert_eq!(llm_calls.len(), 2);
    let second_call = &llm_calls[1];
    let tool_messages: Vec<&Message> = second_call
        .iter()
        .filter(|m| m.role == motosan_agent_loop::Role::Tool)
        .collect();
    assert_eq!(tool_messages.len(), 4);
    assert!(tool_messages.iter().any(|m| m.content == "A"));
    assert!(tool_messages.iter().any(|m| m.content == "D"));
}

#[tokio::test]
async fn multiple_ask_user_answers_match_explicit_call_ids() {
    let (llm, _, captured) = MockLlm::new(vec![
        LlmResponse::ToolCalls(vec![
            ToolCallItem {
                id: "call_ask_1".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "pick-a?",
                    "options": ["A", "B"]
                }),
            },
            ToolCallItem {
                id: "call_slow_1".to_string(),
                name: "slow_a".to_string(),
                args: serde_json::json!({}),
            },
            ToolCallItem {
                id: "call_ask_2".to_string(),
                name: "ask_user".to_string(),
                args: serde_json::json!({
                    "question": "pick-b?",
                    "options": ["C", "D"]
                }),
            },
            ToolCallItem {
                id: "call_slow_2".to_string(),
                name: "slow_b".to_string(),
                args: serde_json::json!({}),
            },
        ]),
        LlmResponse::Message("done".to_string()),
    ]);

    let agent = AgentLoop::builder()
        .with_ask_user()
        .with_ask_user_timeout(std::time::Duration::from_millis(500))
        .tool(Arc::new(SlowTool {
            name: "slow_a",
            delay_ms: 80,
        }))
        .tool(Arc::new(SlowTool {
            name: "slow_b",
            delay_ms: 80,
        }))
        .max_iterations(3)
        .build();

    let (ops_tx, ops_rx) = tokio::sync::mpsc::channel(8);
    let ask_seen = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
    let events_for_cb = Arc::clone(&events);

    let result = agent
        .run_with_ops(&llm, vec![Message::user("start")], Some(ops_rx), {
            let ask_seen_cb = Arc::clone(&ask_seen);
            move |event| match event {
                AgentEvent::AskUser {
                    call_id, question, ..
                } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("ask_user:{call_id}:{question}"));
                    let n = ask_seen_cb.fetch_add(1, Ordering::SeqCst);
                    if n == 0 {
                        // Intentionally send the second answer first to verify call_id matching.
                        ops_tx
                            .try_send(AgentOp::AskUserAnswer {
                                call_id: Some("call_ask_2".to_string()),
                                answer: "D".to_string(),
                            })
                            .expect("failed to queue call_ask_2 answer");
                        ops_tx
                            .try_send(AgentOp::AskUserAnswer {
                                call_id: Some("call_ask_1".to_string()),
                                answer: "A".to_string(),
                            })
                            .expect("failed to queue call_ask_1 answer");
                    }
                }
                AgentEvent::AskUserTimeout { call_id, question } => {
                    events_for_cb
                        .lock()
                        .unwrap()
                        .push(format!("timeout:{call_id}:{question}"));
                }
                _ => {}
            }
        })
        .await
        .unwrap();

    assert_eq!(result.answer, "done");
    assert_eq!(ask_seen.load(Ordering::SeqCst), 2);
    let seen = events.lock().unwrap();
    assert!(
        !seen.iter().any(|e| e.starts_with("timeout:")),
        "unexpected timeout events in explicit-call-id test: {:?}",
        *seen
    );

    let llm_calls = captured.lock().unwrap();
    assert_eq!(llm_calls.len(), 2);
    let second_call = &llm_calls[1];
    let tool_messages: Vec<&Message> = second_call
        .iter()
        .filter(|m| m.role == motosan_agent_loop::Role::Tool)
        .collect();
    assert_eq!(tool_messages.len(), 4);

    let ask1 = tool_messages
        .iter()
        .find(|m| m.tool_call_id.as_deref() == Some("call_ask_1"))
        .expect("missing tool message for call_ask_1");
    let ask2 = tool_messages
        .iter()
        .find(|m| m.tool_call_id.as_deref() == Some("call_ask_2"))
        .expect("missing tool message for call_ask_2");
    assert_eq!(ask1.content, "A");
    assert_eq!(ask2.content, "D");
}
