Codex

#[cfg(any(windows, test))]
use serde_json::json;
use serde_json::Value;

#[cfg(any(windows, test))]
const MAX_FRAME: usize = 8 * 1024 * 1024;
#[cfg(any(windows, test))]
fn frame(value: &Value) -> Result<Vec<u8>, String> {
    let body = serde_json::to_vec(value).map_err(|e| e.to_string())?;
    if body.len() > MAX_FRAME {
        return Err("app_tools_frame_too_large".into());
    }
    let mut frame = (body.len() as u32).to_le_bytes().to_vec();
    frame.extend(body);
    Ok(frame)
}

#[cfg(not(windows))]
pub fn request(_method: &str, _params: Value, _endpoint: Option<&str>) -> Result<Value, String> {
    Err("desktop_tools_require_windows".into())
}

#[cfg(windows)]
pub fn request(method: &str, params: Value, endpoint: Option<&str>) -> Result<Value, String> {
    use std::{
        fs::OpenOptions,
        io::{Read, Write},
        os::windows::io::AsRawHandle,
        time::{Duration, Instant},
    };
    use windows_sys::Win32::System::Pipes::{GetNamedPipeServerProcessId, PeekNamedPipe};
    let inv = crate::desktop::inventory()?;
    let owners: Vec<u32> = inv["windows"]
        .as_array()
        .unwrap_or(&vec![])
        .iter()
        .filter_map(|w| w["pid"].as_u64().map(|p| p as u32))
        .collect();
    if owners.is_empty() {
        return Err("desktop_window_not_running".into());
    }
    let listing = crate::process::powershell(
        r#"@(Get-ChildItem \\.\pipe\ | Where-Object {$_.Name -like 'codex-browser-use-*'} | ForEach-Object {$_.Name}) | ConvertTo-Json -Compress"#,
        Duration::from_secs(10),
    )?;
    let value: Value = serde_json::from_str(listing["stdout"].as_str().unwrap_or("").trim())
        .map_err(|_| "desktop_tools_pipe_not_found")?;
    let names = if value.is_array() {
        value.as_array().unwrap().clone()
    } else {
        vec![value]
    };
    let mut verified = Vec::new();
    for name in names {
        let Some(name) = name.as_str() else {
            continue;
        };
        if !name.starts_with("codex-browser-use-")
            || name.contains(['\\', '/'])
            || endpoint.is_some_and(|e| e != name)
        {
            continue;
        }
        let path = format!(r"\\.\pipe\{name}");
        if let Ok(pipe) = OpenOptions::new().read(true).write(true).open(&path) {
            let mut pid = 0;
            if unsafe { GetNamedPipeServerProcessId(pipe.as_raw_handle(), &mut pid) } != 0
                && owners.contains(&pid)
            {
                verified.push((pipe, pid, name.to_string()));
            }
        }
    }
    if verified.len() != 1 {
        return Err(format!(
            "desktop_tools_pipe_ambiguous_or_missing: specify desktopPipe from verified endpoints {:?}",
            verified.iter().map(|(_,pid,name)|(pid,name)).collect::<Vec<_>>()
        ));
    }
    let (mut pipe, pid, endpoint) = verified.pop().unwrap();
    let id = uuid::Uuid::new_v4().to_string();
    pipe.write_all(&frame(
        &json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}),
    )?)
    .map_err(|_| "desktop_tools_write_failed")?;
    let deadline = Instant::now() + Duration::from_secs(120);
    fn read_until(
        pipe: &mut std::fs::File,
        buf: &mut [u8],
        deadline: Instant,
    ) -> Result<(), String> {
        let mut offset = 0;
        while offset < buf.len() {
            let mut available = 0;
            if unsafe {
                PeekNamedPipe(
                    pipe.as_raw_handle(),
                    std::ptr::null_mut(),
                    0,
                    std::ptr::null_mut(),
                    &mut available,
                    std::ptr::null_mut(),
                )
            } == 0
            {
                return Err("desktop_tools_disconnected".into());
            }
            if available == 0 {
                if Instant::now() >= deadline {
                    return Err("desktop_tools_timeout: native request may still need a user response; inspect desktop state".into());
                }
                std::thread::sleep(Duration::from_millis(20));
                continue;
            }
            let end = buf.len().min(offset + available as usize);
            let n = pipe
                .read(&mut buf[offset..end])
                .map_err(|_| "desktop_tools_read_failed")?;
            if n == 0 {
                return Err("desktop_tools_disconnected".into());
            }
            offset += n;
        }
        Ok(())
    }
    let mut size = [0; 4];
    read_until(&mut pipe, &mut size, deadline)?;
    let size = u32::from_le_bytes(size) as usize;
    if size > MAX_FRAME {
        return Err("desktop_tools_frame_too_large".into());
    }
    let mut body = vec![0; size];
    read_until(&mut pipe, &mut body, deadline)?;
    let response: Value =
        serde_json::from_slice(&body).map_err(|_| "desktop_tools_invalid_response")?;
    if response["id"] != id {
        return Err("desktop_tools_response_id_mismatch".into());
    }
    if let Some(error) = response.get("error") {
        return Err(format!("desktop_tools_error: {error}"));
    }
    Ok(
        json!({"result":response["result"],"desktopPid":pid,"desktopPipe":endpoint,"transport":"desktop-native-pipe"}),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn framing_matches_native_little_endian_protocol() {
        let v = json!({"method":"tools/list","params":{"threadStartKind":"all"}});
        let f = frame(&v).unwrap();
        assert_eq!(
            u32::from_le_bytes(f[..4].try_into().unwrap()) as usize,
            f.len() - 4
        );
        assert_eq!(serde_json::from_slice::<Value>(&f[4..]).unwrap(), v);
    }
}