use super::collect_output_until_exit;
use super::combine_spawned_output;
use super::find_python;
use super::wait_for_output_contains;
use crate::TerminalSize;
use crate::spawn_pipe_process_no_stdin;
use crate::spawn_pty_process;
use pretty_assertions::assert_eq;
use std::collections::HashMap;
use std::os::windows::io::AsRawHandle;
use std::os::windows::io::BorrowedHandle;
use std::os::windows::io::FromRawHandle;
use std::os::windows::io::OwnedHandle;
use std::os::windows::process::CommandExt;
use std::path::Path;
use std::process::Stdio;
use std::time::Duration;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::process::Command;
use winapi::um::jobapi::IsProcessInJob;
use winapi::um::processthreadsapi::OpenProcess;
use winapi::um::synchapi::WaitForSingleObject;
use winapi::um::winbase::WAIT_OBJECT_0;
use winapi::um::winnt::PROCESS_QUERY_LIMITED_INFORMATION;
use winapi::um::winnt::SYNCHRONIZE;

const READY_MARKER: &str = "__CODEX_CHILD_READY__";
const VALUE_MARKER: &str = "__CODEX_CHILD_VALUE__";

#[tokio::test]
async fn piped_child_has_no_console_with_or_without_job_containment() -> anyhow::Result<()> {
    // Nextest itself can run without a visible console. Re-execute this test
    // detached, like app-server-daemon, so children cannot inherit its console.
    if std::env::var_os("CODEX_PTY_DETACHED_CONSOLE_PROBE").is_none() {
        let output = std::process::Command::new(std::env::current_exe()?)
            .args([
                "--exact",
                "tests::windows_tests::piped_child_has_no_console_with_or_without_job_containment",
                "--nocapture",
            ])
            .env("CODEX_PTY_DETACHED_CONSOLE_PROBE", "1")
            .creation_flags(winapi::um::winbase::DETACHED_PROCESS)
            .output()?;
        anyhow::ensure!(
            output.status.success(),
            "detached probe failed:\n{}\n{}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr),
        );
        return Ok(());
    }
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping Windows pipe console test");
        return Ok(());
    };
    let job = crate::JobObject::create_without_breakaway()?;
    for containment in [None, Some(&job)] {
        let mut command = crate::Command::new(&python);
        command.envs(std::env::vars()).args([
            "-u",
            "-c",
            "import ctypes,sys; print(ctypes.windll.kernel32.GetConsoleWindow()); print(sys.stdin.readline().strip())",
        ]);
        if let Some(job) = containment {
            command.prepare_suspended_spawn(job);
        }
        let mut child = command.spawn()?;
        if let Some(job) = containment {
            let pid = child
                .id()
                .ok_or_else(|| anyhow::anyhow!("missing child pid"))?;
            anyhow::ensure!(
                job.assign_and_resume_process(pid)?.is_some(),
                "job assignment failed"
            );
        }
        child
            .stdin
            .as_mut()
            .unwrap()
            .write_all(b"stdio works\n")
            .await?;
        let output =
            tokio::time::timeout(Duration::from_secs(10), child.wait_with_output()).await??;
        assert!(
            output.status.success(),
            "{}",
            String::from_utf8_lossy(&output.stderr)
        );
        assert_eq!(
            String::from_utf8(output.stdout)?.replace("\r\n", "\n"),
            "0\nstdio works\n"
        );
    }
    Ok(())
}
struct WindowsShell {
    name: &'static str,
    program: String,
    args: Vec<String>,
    child_command: String,
}

fn find_powershell() -> Option<String> {
    ["pwsh.exe", "powershell.exe"]
        .into_iter()
        .find_map(|candidate| {
            std::process::Command::new(candidate)
                .args(["-NoLogo", "-NoProfile", "-Command", "exit 0"])
                .status()
                .ok()
                .filter(std::process::ExitStatus::success)
                .map(|_| candidate.to_string())
        })
}

fn utf8_hex(value: &str) -> String {
    value
        .as_bytes()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<Vec<_>>()
        .join("")
}

async fn wait_for_path(path: &Path, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    loop {
        if path.exists() {
            return true;
        }
        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
        if remaining.is_zero() {
            return false;
        }
        tokio::time::sleep(remaining.min(Duration::from_millis(25))).await;
    }
}

async fn assert_terminate_kills_descendant(
    backend: &str,
    python: &str,
    env: &HashMap<String, String>,
) -> anyhow::Result<()> {
    let marker = std::env::temp_dir().join(format!(
        "codex-job-descendant-{backend}-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_nanos()
    ));
    let child_code = format!(
        "import pathlib,time; print('{READY_MARKER}',flush=True); time.sleep(1); pathlib.Path(bytes.fromhex('{}').decode()).write_text('survived')",
        utf8_hex(&marker.to_string_lossy())
    );
    // Exercise descendants created after the best-effort pipe assignment,
    // without making the test depend on winning the intentionally accepted race.
    let code = format!(
        "import subprocess,sys,time; time.sleep(0.5); code=bytes.fromhex('{}').decode(); subprocess.Popen([sys.executable,'-u','-c',code]); time.sleep(60)",
        utf8_hex(&child_code)
    );
    let args = vec!["-u".to_string(), "-c".to_string(), code];
    let spawned = if backend == "pipe" {
        spawn_pipe_process_no_stdin(python, &args, Path::new("."), env, /*arg0*/ &None, &[]).await?
    } else {
        spawn_pty_process(
            python,
            &args,
            Path::new("."),
            env,
            /*arg0*/ &None,
            TerminalSize::default(),
            crate::ChildFds::Inherited(&[]),
        )
        .await?
    };
    let (session, mut output_rx, exit_rx) = combine_spawned_output(spawned);
    wait_for_output_contains(&mut output_rx, READY_MARKER, /*timeout_ms*/ 10_000).await?;
    session.request_terminate();
    let (_, exit_code) = collect_output_until_exit(output_rx, exit_rx, /*timeout_ms*/ 10_000).await;
    assert_ne!(
        exit_code, -1,
        "{backend} root did not exit after termination"
    );
    tokio::time::sleep(Duration::from_secs(2)).await;
    let survived = marker.exists();
    if survived {
        std::fs::remove_file(&marker)?;
    }
    assert!(!survived, "{backend} descendant survived termination");
    Ok(())
}

async fn assert_normal_exit_preserves_descendant(
    backend: &str,
    python: &str,
    env: &HashMap<String, String>,
) -> anyhow::Result<()> {
    let marker_base = std::env::temp_dir().join(format!(
        "codex-job-natural-exit-{backend}-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_nanos()
    ));
    let ready_marker = marker_base.with_extension("ready");
    let survival_marker = marker_base.with_extension("survived");
    let child_code = format!(
        "import pathlib,time; pathlib.Path(bytes.fromhex('{}').decode()).write_text('ready'); time.sleep(1); pathlib.Path(bytes.fromhex('{}').decode()).write_text('survived')",
        utf8_hex(&ready_marker.to_string_lossy()),
        utf8_hex(&survival_marker.to_string_lossy())
    );
    let code = format!(
        "import pathlib,subprocess,sys,time; code=bytes.fromhex('{}').decode(); ready=pathlib.Path(bytes.fromhex('{}').decode()); subprocess.Popen([sys.executable,'-u','-c',code],stdin=subprocess.DEVNULL,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,creationflags=subprocess.DETACHED_PROCESS|subprocess.CREATE_NEW_PROCESS_GROUP); deadline=time.time()+10\nwhile not ready.exists() and time.time()<deadline: time.sleep(.05)\nsys.exit(0 if ready.exists() else 2)",
        utf8_hex(&child_code),
        utf8_hex(&ready_marker.to_string_lossy())
    );
    let args = vec!["-u".to_string(), "-c".to_string(), code];
    let spawned = if backend == "pipe" {
        spawn_pipe_process_no_stdin(python, &args, Path::new("."), env, /*arg0*/ &None, &[]).await?
    } else {
        spawn_pty_process(
            python,
            &args,
            Path::new("."),
            env,
            /*arg0*/ &None,
            TerminalSize::default(),
            crate::ChildFds::Inherited(&[]),
        )
        .await?
    };
    let (session, output_rx, exit_rx) = combine_spawned_output(spawned);
    let (_, exit_code) = collect_output_until_exit(output_rx, exit_rx, /*timeout_ms*/ 10_000).await;
    assert_eq!(exit_code, 0, "{backend} root did not exit normally");
    drop(session);

    let survived = wait_for_path(&survival_marker, Duration::from_secs(10)).await;
    let _ = std::fs::remove_file(ready_marker);
    let _ = std::fs::remove_file(survival_marker);
    assert!(survived, "{backend} descendant did not survive normal exit");
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn terminate_kills_descendants_for_best_effort_pipe_and_atomic_conpty() -> anyhow::Result<()>
{
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping Windows process-tree termination test");
        return Ok(());
    };
    let env: HashMap<String, String> = std::env::vars().collect();
    assert_terminate_kills_descendant("pipe", &python, &env).await?;
    assert_terminate_kills_descendant("ConPTY", &python, &env).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn normal_exit_preserves_descendants_for_pipe_and_conpty() -> anyhow::Result<()> {
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping Windows process-tree natural-exit test");
        return Ok(());
    };
    let env: HashMap<String, String> = std::env::vars().collect();
    assert_normal_exit_preserves_descendant("pipe", &python, &env).await?;
    assert_normal_exit_preserves_descendant("ConPTY", &python, &env).await
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn contained_spawn_owns_immediate_descendant() -> anyhow::Result<()> {
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping Windows contained-spawn test");
        return Ok(());
    };

    let mut command = Command::new(&python);
    command
        .args([
            "-u",
            "-c",
            "import subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); print(child.pid,flush=True); child.wait()",
        ])
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::null());

    let job = crate::JobObject::create()?;
    let mut root = job.spawn_contained(&mut command)?;
    let stdout = root
        .stdout
        .take()
        .ok_or_else(|| anyhow::anyhow!("missing contained process stdout"))?;
    let mut stdout = BufReader::new(stdout);
    let mut child_pid = String::new();
    tokio::time::timeout(Duration::from_secs(10), stdout.read_line(&mut child_pid)).await??;
    let child_pid: u32 = child_pid.trim().parse()?;

    let process = unsafe {
        OpenProcess(
            PROCESS_QUERY_LIMITED_INFORMATION | SYNCHRONIZE,
            /*bInheritHandle*/ 0,
            child_pid,
        )
    };
    anyhow::ensure!(!process.is_null(), "failed to open immediate child process");
    let process = unsafe { OwnedHandle::from_raw_handle(process.cast()) };
    let mut in_job = 0;
    let checked = unsafe {
        IsProcessInJob(
            process.as_raw_handle().cast(),
            job.as_raw_handle().cast(),
            &mut in_job,
        )
    };
    anyhow::ensure!(checked != 0, "failed to inspect child Job Object");
    anyhow::ensure!(in_job != 0, "immediate child escaped its Job Object");

    let root_handle = root
        .raw_handle()
        .ok_or_else(|| anyhow::anyhow!("missing contained root handle"))?;
    let root_handle = unsafe { BorrowedHandle::borrow_raw(root_handle) }.try_clone_to_owned()?;
    job.terminate_and_wait(&root_handle, Duration::from_secs(10))?;
    let child_status = unsafe {
        WaitForSingleObject(process.as_raw_handle().cast(), /*dwMilliseconds*/ 0)
    };
    assert_eq!(child_status, WAIT_OBJECT_0);
    tokio::time::timeout(Duration::from_secs(10), root.wait()).await??;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rejected_job_assignment_resumes_existing_job_member() -> anyhow::Result<()> {
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping Windows nested-job fallback test");
        return Ok(());
    };

    let owning_job = crate::JobObject::create()?;
    let rejected_job = crate::JobObject::create_without_breakaway()?;
    let mut occupied_command = Command::new(&python);
    occupied_command
        .args(["-c", "import time; time.sleep(60)"])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    let mut existing_member = rejected_job.spawn_contained(&mut occupied_command)?;

    let mut command = Command::new(&python);
    command
        .args([
            "-u",
            "-c",
            "import time; print('resumed',flush=True); time.sleep(60)",
        ])
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::null());
    rejected_job.prepare_suspended_spawn(&mut command);
    let mut root = command.spawn()?;
    let process_handle = root
        .raw_handle()
        .ok_or_else(|| anyhow::anyhow!("missing suspended process handle"))?;
    owning_job.assign_process(process_handle)?;
    let process_id = root
        .id()
        .ok_or_else(|| anyhow::anyhow!("missing suspended process id"))?;

    assert!(
        rejected_job
            .assign_and_resume_process(process_id)?
            .is_none(),
        "unrelated nested job unexpectedly accepted the process"
    );
    let stdout = root
        .stdout
        .take()
        .ok_or_else(|| anyhow::anyhow!("missing resumed process stdout"))?;
    let mut stdout = BufReader::new(stdout);
    let mut marker = String::new();
    tokio::time::timeout(Duration::from_secs(10), stdout.read_line(&mut marker)).await??;
    assert_eq!(marker.trim(), "resumed");

    let process_handle = crate::JobObject::open_process_handle(process_id)?;
    crate::JobObject::terminate_process_handle(&process_handle)?;
    rejected_job.terminate()?;
    let status = tokio::time::timeout(Duration::from_secs(10), root.wait()).await??;
    assert_eq!(status.code(), Some(1));
    tokio::time::timeout(Duration::from_secs(10), existing_member.wait()).await??;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conpty_delivers_input_to_foreground_children() -> anyhow::Result<()> {
    let Some(python) = find_python() else {
        eprintln!("python not found; skipping ConPTY input test");
        return Ok(());
    };
    let code = format!(
        "print('__CODEX_CHILD_'+'READY__', flush=True); value=input(); print('{VALUE_MARKER}'+value.encode('utf-8').hex(), flush=True)"
    );
    let expected = "cafeé 漢字";
    let expected_marker = format!("{VALUE_MARKER}{}", utf8_hex(expected));
    let mut shells = vec![WindowsShell {
        name: "cmd",
        program: std::env::var("COMSPEC").unwrap_or_else(|_| "cmd.exe".to_string()),
        args: vec!["/D".to_string(), "/Q".to_string()],
        child_command: format!("\"{}\" -u -c \"{code}\"", python.replace('"', "\"\"")),
    }];
    if let Some(program) = find_powershell() {
        shells.push(WindowsShell {
            name: "PowerShell",
            program,
            args: vec!["-NoLogo".to_string(), "-NoProfile".to_string()],
            child_command: format!("& '{}' -u -c \"{code}\"", python.replace('\'', "''")),
        });
    }
    let env: HashMap<String, String> = std::env::vars().collect();

    for shell in shells {
        let spawned = spawn_pty_process(
            &shell.program,
            &shell.args,
            Path::new("."),
            &env,
            /*arg0*/ &None,
            TerminalSize::default(),
            crate::ChildFds::Inherited(&[]),
        )
        .await?;
        let (session, mut output_rx, exit_rx) = combine_spawned_output(spawned);
        let writer = session.writer_sender();
        writer
            .send(format!("{}\n", shell.child_command).into_bytes())
            .await?;
        wait_for_output_contains(&mut output_rx, READY_MARKER, /*timeout_ms*/ 10_000)
            .await
            .map_err(|err| anyhow::anyhow!("{} child did not become ready: {err}", shell.name))?;

        writer
            .send(format!("{expected}X\u{8}\n").into_bytes())
            .await?;
        let mut output =
            wait_for_output_contains(&mut output_rx, &expected_marker, /*timeout_ms*/ 10_000)
                .await
                .map_err(|err| {
                    anyhow::anyhow!("{} child received incorrect input: {err}", shell.name)
                })?;

        writer.send(b"exit 0\n".to_vec()).await?;
        let (remaining, exit_code) =
            collect_output_until_exit(output_rx, exit_rx, /*timeout_ms*/ 10_000).await;
        output.extend_from_slice(&remaining);

        assert_eq!(
            exit_code,
            0,
            "{} did not exit cleanly: {:?}",
            shell.name,
            String::from_utf8_lossy(&output)
        );
    }

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conpty_ctrl_c_interrupts_powershell_foreground_child() -> anyhow::Result<()> {
    let Some(program) = find_powershell() else {
        return Ok(());
    };
    let args = vec!["-NoLogo".to_string(), "-NoProfile".to_string()];
    let env: HashMap<String, String> = std::env::vars().collect();
    let spawned = spawn_pty_process(
        &program,
        &args,
        Path::new("."),
        &env,
        /*arg0*/ &None,
        TerminalSize::default(),
        crate::ChildFds::Inherited(&[]),
    )
    .await?;
    let (session, mut output_rx, exit_rx) = combine_spawned_output(spawned);
    let writer = session.writer_sender();
    // CI may pass its ignored Ctrl-C disposition to children. Reset it in the
    // foreground process itself before checking delivery through ConPTY.
    let python = find_python()
        .ok_or_else(|| anyhow::anyhow!("Python is required for the Ctrl-C fixture"))?;
    let python = python.replace('\'', "''");
    let command = format!(
        r#"& '{python}' -u -c "import ctypes,time; assert ctypes.windll.kernel32.SetConsoleCtrlHandler(None, False); print('__CODEX_CHILD_'+'READY__'); time.sleep(60)""#
    );
    writer.send(format!("{command}\n").into_bytes()).await?;
    wait_for_output_contains(&mut output_rx, READY_MARKER, /*timeout_ms*/ 10_000).await?;

    writer.send(vec![0x03]).await?;
    wait_for_output_contains(&mut output_rx, "PS ", /*timeout_ms*/ 10_000).await?;
    writer.send(b"cmd.exe /D /C ver\n".to_vec()).await?;
    let mut output = wait_for_output_contains(
        &mut output_rx,
        "Microsoft Windows",
        /*timeout_ms*/ 10_000,
    )
    .await?;

    writer.send(b"exit 0\n".to_vec()).await?;
    let (remaining, exit_code) =
        collect_output_until_exit(output_rx, exit_rx, /*timeout_ms*/ 10_000).await;
    output.extend_from_slice(&remaining);
    assert_eq!(
        exit_code,
        0,
        "PowerShell did not resume after Ctrl-C: {:?}",
        String::from_utf8_lossy(&output)
    );
    Ok(())
}