use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use motosan_agent_loop::{
    AgentLoop, AgentSession, ChatOutput, FileSessionStore, LlmClient, LlmResponse,
    MemorySessionStore, Message, SessionMeta, SessionStore,
};
use motosan_agent_tool::{Tool, ToolContext, ToolDef, ToolResult};

struct SequenceLlm {
    responses: Mutex<Vec<LlmResponse>>,
}

struct SlowTool;

impl Tool for SlowTool {
    fn def(&self) -> ToolDef {
        ToolDef {
            name: "slow".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 + '_>> {
        Box::pin(async move {
            tokio::time::sleep(std::time::Duration::from_millis(80)).await;
            ToolResult::text("slow-ok")
        })
    }
}

impl SequenceLlm {
    fn new(responses: Vec<LlmResponse>) -> Self {
        Self {
            responses: Mutex::new(responses),
        }
    }
}

#[async_trait]
impl LlmClient for SequenceLlm {
    async fn chat(
        &self,
        _messages: &[Message],
        _tools: &[ToolDef],
    ) -> motosan_agent_loop::Result<ChatOutput> {
        let mut responses = self.responses.lock().unwrap();
        if responses.is_empty() {
            panic!("SequenceLlm: no more responses");
        }
        Ok(ChatOutput::new(responses.remove(0)))
    }
}

#[tokio::test]
async fn session_send_updates_history() {
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("hello".into())]));
    let (session, handle) = AgentSession::new(agent, llm, |_| {});

    session.send("hi").await;
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    let history = session.history().await;
    assert!(history.iter().any(|m| m.content == "hi"));
    assert!(history.iter().any(|m| m.content == "hello"));

    drop(session);
    handle.abort();
}

#[tokio::test]
async fn memory_store_roundtrip_and_list() {
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    store.append("s1", &Message::user("u1")).await.unwrap();
    store.append("s1", &Message::assistant("a1")).await.unwrap();
    store.flush("s1").await.unwrap();

    let loaded = store.load("s1").await.unwrap();
    assert_eq!(loaded.len(), 2);
    assert_eq!(loaded[0].content, "u1");
    assert_eq!(loaded[1].content, "a1");

    let listed = store.list().await.unwrap();
    assert_eq!(listed, vec!["s1".to_string()]);

    store.delete("s1").await.unwrap();
    let listed_after = store.list().await.unwrap();
    assert!(listed_after.is_empty());
}

#[tokio::test]
async fn session_resume_uses_store_history() {
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    store
        .append("resume-1", &Message::user("from-store"))
        .await
        .unwrap();

    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));

    let (session, handle) = AgentSession::resume("resume-1", store, agent, llm, |_| {})
        .await
        .unwrap();

    let history = session.history().await;
    assert!(history.iter().any(|m| m.content == "from-store"));

    drop(session);
    handle.abort();
}

#[tokio::test]
async fn file_store_flush_and_resume_roundtrip() {
    let session_id = format!(
        "file-roundtrip-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let store: Arc<dyn SessionStore> = Arc::new(FileSessionStore::new(base_dir.clone()));

    store
        .append(&session_id, &Message::user("u1"))
        .await
        .unwrap();
    store
        .append(&session_id, &Message::assistant("a1"))
        .await
        .unwrap();
    store.flush(&session_id).await.unwrap();

    let loaded = store.load(&session_id).await.unwrap();
    assert_eq!(loaded.len(), 2);
    assert_eq!(loaded[0].content, "u1");
    assert_eq!(loaded[1].content, "a1");

    let listed = store.list().await.unwrap();
    assert!(listed.iter().any(|id| id == &session_id));

    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) = AgentSession::resume(&session_id, store, agent, llm, |_| {})
        .await
        .unwrap();
    let history = session.history().await;
    assert!(history.iter().any(|m| m.content == "u1"));
    assert!(history.iter().any(|m| m.content == "a1"));

    drop(session);
    handle.abort();
    let _ = tokio::fs::remove_dir_all(base_dir).await;
}

#[tokio::test]
async fn file_store_session_meta_header_and_mismatch_is_non_fatal() {
    let session_id = format!(
        "meta-mismatch-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let session_file = base_dir.join(format!("{session_id}.jsonl"));

    let store = FileSessionStore::new(base_dir.clone());
    store
        .append(&session_id, &Message::user("from-file"))
        .await
        .unwrap();
    store.flush(&session_id).await.unwrap();

    let content = tokio::fs::read_to_string(&session_file).await.unwrap();
    let mut lines: Vec<String> = content.lines().map(ToString::to_string).collect();
    let header: SessionMeta = serde_json::from_str(&lines[0]).unwrap();
    assert_eq!(header.session_id, session_id);
    assert!(header.message_count >= 1);

    let mut modified = header.clone();
    modified.agent_loop_version = "0.0.0-mismatch".to_string();
    lines[0] = serde_json::to_string(&modified).unwrap();
    tokio::fs::write(&session_file, format!("{}\n", lines.join("\n")))
        .await
        .unwrap();

    let store_arc: Arc<dyn SessionStore> = Arc::new(FileSessionStore::new(base_dir.clone()));
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) = AgentSession::resume(&session_id, store_arc, agent, llm, |_| {})
        .await
        .unwrap();
    let history = session.history().await;
    assert!(history.iter().any(|m| m.content == "from-file"));

    drop(session);
    handle.abort();
    let _ = tokio::fs::remove_dir_all(base_dir).await;
}

#[test]
fn file_store_without_runtime_periodic_flush_needs_manual_flush() {
    let session_id = format!(
        "manual-flush-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let session_file = base_dir.join(format!("{session_id}.jsonl"));

    let store = FileSessionStore::new(base_dir.clone());
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        store
            .append(&session_id, &Message::user("pending"))
            .await
            .unwrap();
    });

    drop(store);
    assert!(
        !session_file.exists(),
        "without periodic task or explicit flush, data should remain buffered"
    );

    let store = FileSessionStore::new(base_dir.clone());
    rt.block_on(async {
        store
            .append(&session_id, &Message::user("persisted"))
            .await
            .unwrap();
        store.flush(&session_id).await.unwrap();
    });
    assert!(
        session_file.exists(),
        "manual flush must persist buffered content"
    );

    drop(rt);
    let _ = std::fs::remove_dir_all(base_dir);
}

#[test]
fn file_session_store_new_without_runtime_does_not_panic() {
    let session_id = format!(
        "no-rt-store-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let _store = FileSessionStore::new(base_dir.clone());
    let _ = std::fs::remove_dir_all(base_dir);
}

#[test]
fn agent_session_drop_without_runtime_does_not_panic() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();

    let session = rt.block_on(async {
        let agent = AgentLoop::builder().build();
        let llm: Arc<dyn LlmClient> =
            Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
        let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
        let (session, handle) =
            AgentSession::new_with_store("no-rt-drop", store, agent, llm, |_| {});
        handle.abort();
        session
    });
    drop(rt);

    drop(session);
}

#[tokio::test]
async fn agent_session_flush_persists_store_explicitly() {
    let session_id = format!(
        "session-flush-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let store: Arc<dyn SessionStore> = Arc::new(FileSessionStore::new(base_dir.clone()));

    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("hello").await;
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    session.flush().await.unwrap();

    let loaded = store.load(&session_id).await.unwrap();
    assert!(loaded.iter().any(|m| m.content == "hello"));
    assert!(loaded.iter().any(|m| m.content == "ok"));

    drop(session);
    handle.abort();
    let _ = tokio::fs::remove_dir_all(base_dir).await;
}

#[tokio::test]
async fn agent_session_close_persists_store_explicitly() {
    let session_id = format!(
        "session-close-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let base_dir = std::env::temp_dir().join(format!("motosan-agent-loop-{session_id}"));
    let store: Arc<dyn SessionStore> = Arc::new(FileSessionStore::new(base_dir.clone()));

    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("hello-close").await;
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    session.close().await.unwrap();
    handle.abort();

    let loaded = store.load(&session_id).await.unwrap();
    assert!(loaded.iter().any(|m| m.content == "hello-close"));
    assert!(loaded.iter().any(|m| m.content == "ok"));

    let _ = tokio::fs::remove_dir_all(base_dir).await;
}

#[tokio::test]
async fn session_store_does_not_duplicate_user_message_per_turn() {
    let session_id = format!(
        "no-dup-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("hello").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;
    session.flush().await.unwrap();

    let loaded = store.load(&session_id).await.unwrap();
    let hello_count = loaded.iter().filter(|m| m.content == "hello").count();
    assert_eq!(hello_count, 1, "user message should be persisted once");

    drop(session);
    handle.abort();
}

#[tokio::test]
async fn reset_does_not_get_overwritten_by_inflight_turn_result() {
    let session_id = format!(
        "reset-race-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder()
        .tool(Arc::new(SlowTool))
        .max_iterations(3)
        .build();
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::single_tool_call("call_1".into(), "slow".into(), serde_json::json!({})),
        LlmResponse::Message("done".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("will-reset").await;
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    session.reset().await;
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;

    let history = session.history().await;
    assert!(
        history.is_empty(),
        "history should remain empty after reset"
    );
    let loaded = store.load(&session_id).await.unwrap();
    assert!(loaded.is_empty(), "store should remain empty after reset");

    drop(session);
    handle.abort();
}

// ---------------------------------------------------------------------------
// Regression tests for lifecycle race hardening (#74)
// ---------------------------------------------------------------------------

/// After `close()`, new `send()` calls must be silently dropped and not
/// trigger any further turn processing or persistence.
#[tokio::test]
async fn close_prevents_further_sends() {
    let session_id = format!(
        "close-no-send-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::Message("first-reply".into()),
        // If a second turn were to run, this would be consumed and the test
        // would fail because the store would contain "after-close".
        LlmResponse::Message("should-not-appear".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    // Process one turn normally.
    session.send("before-close").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;

    // Clone before close so we can attempt a send on a surviving handle.
    let session_clone = session.clone();

    // Close the session.
    session.close().await.unwrap();

    // Attempt to send on the clone — should be silently dropped because
    // the shared `closed` flag is set.
    session_clone.send("after-close").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;

    // Verify persisted state via the store.
    let loaded = store.load(&session_id).await.unwrap();
    assert!(
        loaded.iter().any(|m| m.content == "before-close"),
        "message sent before close should be persisted"
    );
    assert!(
        !loaded.iter().any(|m| m.content == "after-close"),
        "no message sent after close should appear"
    );
    assert!(
        !loaded.iter().any(|m| m.content == "should-not-appear"),
        "second LLM response should never be triggered"
    );

    handle.abort();
}

/// `is_closed()` should reflect the closed state accurately.
#[tokio::test]
async fn is_closed_reflects_state() {
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> =
        Arc::new(SequenceLlm::new(vec![LlmResponse::Message("ok".into())]));
    let (session, handle) = AgentSession::new(agent, llm, |_| {});

    assert!(
        !session.is_closed(),
        "session should not be closed initially"
    );

    // Clone before close to check state after.
    let session_clone = session.clone();
    session.close().await.unwrap();

    assert!(
        session_clone.is_closed(),
        "cloned handle should observe closed state"
    );

    handle.abort();
}

/// Consecutive resets during in-flight slow tool execution must each
/// discard the stale turn results.  After the final reset (with no new
/// messages sent), both history and store must be empty.
#[tokio::test]
async fn consecutive_resets_discard_inflight_turns() {
    let session_id = format!(
        "consec-reset-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder()
        .tool(Arc::new(SlowTool))
        .max_iterations(3)
        .build();
    // The slow tool takes 80ms; we reset while it is executing.
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::single_tool_call("c1".into(), "slow".into(), serde_json::json!({})),
        LlmResponse::Message("done-1".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    // Send a message, wait for the turn to enter the slow tool, then reset.
    session.send("will-be-reset").await;
    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    session.reset().await;

    // Allow the interrupted turn to complete and be discarded.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    let history = session.history().await;
    assert!(
        history.is_empty(),
        "history must be empty after reset, got {} messages",
        history.len()
    );

    let loaded = store.load(&session_id).await.unwrap();
    assert!(
        loaded.is_empty(),
        "store must be empty after reset, got {} messages",
        loaded.len()
    );

    drop(session);
    handle.abort();
}

/// Two sequential turns must each persist the user message exactly once,
/// verifying the dedup logic works across multiple turns.
#[tokio::test]
async fn no_duplicate_user_message_across_two_turns() {
    let session_id = format!(
        "no-dup-2turn-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::Message("reply-1".into()),
        LlmResponse::Message("reply-2".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("turn-1").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;

    session.send("turn-2").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;

    session.flush().await.unwrap();

    let loaded = store.load(&session_id).await.unwrap();
    let turn1_count = loaded.iter().filter(|m| m.content == "turn-1").count();
    let turn2_count = loaded.iter().filter(|m| m.content == "turn-2").count();
    assert_eq!(
        turn1_count, 1,
        "first user message should appear exactly once in store"
    );
    assert_eq!(
        turn2_count, 1,
        "second user message should appear exactly once in store"
    );

    drop(session);
    handle.abort();
}

/// Interrupt during a slow tool execution must not leave partial data
/// in the store beyond what was persisted before the turn started.
#[tokio::test]
async fn interrupt_during_tool_does_not_corrupt_store() {
    let session_id = format!(
        "interrupt-tool-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder()
        .tool(Arc::new(SlowTool))
        .max_iterations(3)
        .build();
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::single_tool_call("c1".into(), "slow".into(), serde_json::json!({})),
        LlmResponse::Message("after-slow".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    session.send("will-interrupt").await;
    // Let the turn start and hit the slow tool.
    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    session.interrupt().await;
    // Allow the interrupted turn to complete.
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    session.flush().await.unwrap();

    let loaded = store.load(&session_id).await.unwrap();
    // The user message should be persisted (it was appended before the turn).
    let user_count = loaded
        .iter()
        .filter(|m| m.content == "will-interrupt")
        .count();
    assert!(
        user_count <= 1,
        "user message should appear at most once, got {}",
        user_count
    );
    // No duplicate assistant messages.
    let assistant_msgs: Vec<_> = loaded
        .iter()
        .filter(|m| m.role == motosan_agent_loop::Role::Assistant)
        .collect();
    let unique_contents: std::collections::HashSet<_> =
        assistant_msgs.iter().map(|m| &m.content).collect();
    assert_eq!(
        assistant_msgs.len(),
        unique_contents.len(),
        "no duplicate assistant messages should exist in store"
    );

    drop(session);
    handle.abort();
}

/// Reset followed by a new turn must only contain the new turn's data.
#[tokio::test]
async fn reset_then_new_turn_only_has_new_data() {
    let session_id = format!(
        "reset-new-turn-{}",
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let store: Arc<dyn SessionStore> = Arc::new(MemorySessionStore::new());
    let agent = AgentLoop::builder().build();
    let llm: Arc<dyn LlmClient> = Arc::new(SequenceLlm::new(vec![
        LlmResponse::Message("old-reply".into()),
        LlmResponse::Message("new-reply".into()),
    ]));
    let (session, handle) =
        AgentSession::new_with_store(session_id.clone(), store.clone(), agent, llm, |_| {});

    // First turn.
    session.send("old-msg").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;

    // Reset clears everything.
    session.reset().await;
    tokio::time::sleep(std::time::Duration::from_millis(20)).await;

    // Second turn after reset.
    session.send("new-msg").await;
    tokio::time::sleep(std::time::Duration::from_millis(60)).await;
    session.flush().await.unwrap();

    let history = session.history().await;
    assert!(
        !history.iter().any(|m| m.content == "old-msg"),
        "old message should not appear in history after reset"
    );
    assert!(
        !history.iter().any(|m| m.content == "old-reply"),
        "old reply should not appear in history after reset"
    );
    assert!(
        history.iter().any(|m| m.content == "new-msg"),
        "new message should appear in history"
    );
    assert!(
        history.iter().any(|m| m.content == "new-reply"),
        "new reply should appear in history"
    );

    let loaded = store.load(&session_id).await.unwrap();
    assert!(
        !loaded.iter().any(|m| m.content == "old-msg"),
        "old message should not appear in store after reset"
    );
    assert!(
        loaded.iter().any(|m| m.content == "new-msg"),
        "new message should appear in store"
    );

    drop(session);
    handle.abort();
}
