use crate::product::agent::CodexAuth;
use crate::product::agent::NewThread;
use crate::product::agent::ThreadManager;
use crate::product::agent::parse_turn_item;
use crate::product::agent::protocol::EventMsg;
use crate::product::agent::protocol::Op;
use crate::product::agent::protocol::RolloutItem;
use crate::product::agent::protocol::RolloutLine;
use crate::product::protocol::items::TurnItem;
use crate::product::protocol::user_input::UserInput;
use crate::test_support::core::load_default_config_for_test;
use crate::test_support::core::skip_if_no_network;
use crate::test_support::core::test_codex::test_codex;
use crate::test_support::core::wait_for_event;
use core::time::Duration;
use tempfile::tempdir;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;

/// Build minimal SSE stream with completed marker using the JSON fixture.
fn sse_completed(id: &str) -> String {
    crate::test_support::core::load_sse_fixture_with_id("../fixtures/completed_template.json", id)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn start_thread_notifies_thread_created_subscribers() {
    let lha_home = tempdir().expect("tempdir");
    let config = load_default_config_for_test(&lha_home).await;
    let thread_manager = ThreadManager::with_models_provider(
        CodexAuth::from_api_key("sk-test"),
        config.model_provider.clone(),
    );
    let mut thread_created_rx = thread_manager.subscribe_thread_created();

    let NewThread { thread_id, .. } = thread_manager
        .start_thread(config)
        .await
        .expect("start thread");

    let notified_thread_id = timeout(Duration::from_secs(1), thread_created_rx.recv())
        .await
        .expect("thread created notification")
        .expect("thread created channel open");

    assert_eq!(notified_thread_id, thread_id);
    thread_manager
        .get_thread(notified_thread_id)
        .await
        .expect("notified thread is registered");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn fork_thread_twice_drops_to_first_message() {
    skip_if_no_network!();

    // Start a mock server that completes three turns.
    let server = MockServer::start().await;
    let sse = sse_completed("resp");
    let first = ResponseTemplate::new(200)
        .insert_header("content-type", "text/event-stream")
        .set_body_raw(sse.clone(), "text/event-stream");

    // Expect three calls to /v1/responses – one per user input.
    Mock::given(method("POST"))
        .and(path("/v1/responses"))
        .respond_with(first)
        .expect(3)
        .mount(&server)
        .await;

    let mut builder = test_codex();
    let test = builder.build(&server).await.expect("create conversation");
    let codex = test.codex.clone();
    let thread_manager = test.thread_manager.clone();
    let config_for_fork = test.config.clone();

    // Send three user messages; wait for three completed turns.
    for text in ["first", "second", "third"] {
        codex
            .submit(Op::UserInput {
                items: vec![UserInput::Text {
                    text: text.to_string(),
                    text_elements: Vec::new(),
                }],
                final_output_json_schema: None,
            })
            .await
            .unwrap();
        let _ = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
    }

    // Request history from the base conversation to obtain rollout path.
    let base_path = codex.rollout_path().expect("rollout path");

    // GetHistory flushes before returning the path; no wait needed.

    // Helper: read rollout items (excluding SessionMeta) from a JSONL path.
    let read_items = |p: &std::path::Path| -> Vec<RolloutItem> {
        let text = std::fs::read_to_string(p).expect("read rollout file");
        let mut items: Vec<RolloutItem> = Vec::new();
        for line in text.lines() {
            if line.trim().is_empty() {
                continue;
            }
            let v: serde_json::Value = serde_json::from_str(line).expect("jsonl line");
            let rl: RolloutLine = serde_json::from_value(v).expect("rollout line");
            match rl.item {
                RolloutItem::SessionMeta(_) => {}
                other => items.push(other),
            }
        }
        items
    };

    // Compute expected prefixes after each fork by truncating base rollout
    // strictly before the nth user input (0-based).
    let base_items = read_items(&base_path);
    let find_user_input_positions = |items: &[RolloutItem]| -> Vec<usize> {
        let mut pos = Vec::new();
        for (i, it) in items.iter().enumerate() {
            if let RolloutItem::TranscriptItem(response_item) = it
                && let Some(TurnItem::UserMessage(_)) = parse_turn_item(&response_item.clone())
            {
                // Consider any user message as an input boundary; recorder stores both EventMsg and transcript items.
                // We specifically look for input items, which are represented as ContentItem::InputText.
                pos.push(i);
            }
        }
        pos
    };
    let user_inputs = find_user_input_positions(&base_items);

    // After cutting at nth user input (n=1 → second user message), cut strictly before that input.
    let cut1 = user_inputs.get(1).copied().unwrap_or(0);
    let expected_after_first: Vec<RolloutItem> = base_items[..cut1].to_vec();

    // After dropping again (n=1 on fork1), compute expected relative to fork1's rollout.

    // Fork once with n=1 → drops the last user input and everything after.
    let NewThread {
        thread: codex_fork1,
        ..
    } = thread_manager
        .fork_thread(1, config_for_fork.clone(), base_path.clone())
        .await
        .expect("fork 1");

    let fork1_path = codex_fork1.rollout_path().expect("rollout path");

    // GetHistory on fork1 flushed; the file is ready.
    let fork1_items = read_items(&fork1_path);
    assert!(fork1_items.len() > expected_after_first.len());
    pretty_assertions::assert_eq!(
        serde_json::to_value(&fork1_items[..expected_after_first.len()]).unwrap(),
        serde_json::to_value(&expected_after_first).unwrap()
    );

    // Fork again with n=0 → drops the (new) last user message, leaving only the first.
    let NewThread {
        thread: codex_fork2,
        ..
    } = thread_manager
        .fork_thread(0, config_for_fork.clone(), fork1_path.clone())
        .await
        .expect("fork 2");

    let fork2_path = codex_fork2.rollout_path().expect("rollout path");
    // GetHistory on fork2 flushed; the file is ready.
    let fork1_items = read_items(&fork1_path);
    let fork1_user_inputs = find_user_input_positions(&fork1_items);
    let cut_last_on_fork1 = fork1_user_inputs
        .get(fork1_user_inputs.len().saturating_sub(1))
        .copied()
        .unwrap_or(0);
    let expected_after_second: Vec<RolloutItem> = fork1_items[..cut_last_on_fork1].to_vec();
    let fork2_items = read_items(&fork2_path);
    assert!(fork2_items.len() > expected_after_second.len());
    pretty_assertions::assert_eq!(
        serde_json::to_value(&fork2_items[..expected_after_second.len()]).unwrap(),
        serde_json::to_value(&expected_after_second).unwrap()
    );
}
