//! Stdio transport for ACP
//!
//! This module provides a transport implementation that communicates with ACP agents
//! via stdin/stdout using line-delimited JSON. This is the standard transport for
//! agents that run as child processes.
//!
//! # Protocol
//!
//! Messages are sent as line-delimited JSON:
//! - Each JSON-RPC message is serialized to a single line
//! - Lines are terminated with `\n`
//! - The process's stdout is read line-by-line to receive messages
//!
//! # Lifecycle
//!
//! 1. Create StdioTransport with command and arguments
//! 2. Call `connect()` to spawn the process
//! 3. Use `send()` and `recv()` for communication
//! 4. Call `close()` to terminate the process gracefully
//!
//! # Thread Safety
//!
//! StdioTransport uses `Arc<Mutex<T>>` for thread-safe access to stdin/stdout handles.
//! This allows multiple tasks to share the transport, though typically only one task
//! should be reading (recv) and one writing (send) at a time.

use super::{AcpTransport, TransportResult};
use crate::connectors::acp::logging::AcpProtocolLogger;
use async_trait::async_trait;
use serde_json::Value;
use std::collections::VecDeque;
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

/// Context captured when a child process crashes.
///
/// Assembled from three data sources that are normally siloed:
/// stderr output, process exit status, and partial stdout data.
#[derive(Debug, Clone)]
pub struct CrashContext {
    /// Recent stderr lines from the child process (last 50 lines)
    pub recent_stderr: Vec<String>,
    /// Exit status if available at crash detection time
    pub exit_status: Option<std::process::ExitStatus>,
    /// Partial stdout data that was being written when the process crashed
    pub partial_stdout: Option<String>,
}

/// Stdio transport implementation
///
/// Spawns a child process and communicates via stdin/stdout using line-delimited JSON.
///
/// # Example
///
/// ```no_run
/// use dirigent_core::connectors::acp::transport::{AcpTransport, StdioTransport};
/// use serde_json::json;
///
/// # async fn example() -> anyhow::Result<()> {
/// // Spawn dirigate process
/// let mut transport = StdioTransport::new(
///     "dirigate",
///     &["serve", "--fixtures", "./fixtures", "--stdio"]
/// );
///
/// // Connect (spawns process)
/// transport.connect().await?;
///
/// // Send initialize request
/// transport.send(json!({
///     "jsonrpc": "2.0",
///     "method": "initialize",
///     "id": 1,
///     "params": {
///         "protocolVersion": 1,
///         "clientCapabilities": {}
///     }
/// })).await?;
///
/// // Receive response
/// if let Some(response) = transport.recv().await? {
///     println!("Initialized: {:?}", response);
/// }
///
/// // Clean up
/// transport.close().await?;
/// # Ok(())
/// # }
/// ```
pub struct StdioTransport {
    /// Command to execute (e.g., "dirigate" or "/path/to/agent")
    command: String,

    /// Arguments to pass to the command
    args: Vec<String>,

    /// Working directory for the process (optional)
    cwd: Option<PathBuf>,

    /// Environment variables to set (key-value pairs)
    env: Vec<(String, String)>,

    /// Child process handle
    ///
    /// Wrapped in Arc<Mutex> for thread-safe access. This allows the
    /// transport to be used across multiple async tasks.
    child: Arc<Mutex<Option<Child>>>,

    /// Stdin handle for sending messages
    ///
    /// Wrapped in Arc<Mutex> for thread-safe writes. Only one task should
    /// write at a time to avoid message interleaving.
    stdin: Arc<Mutex<Option<ChildStdin>>>,

    /// Stdout handle for receiving messages
    ///
    /// Wrapped in BufReader for line-by-line reading, then in Arc<Mutex>
    /// for thread-safe access. Only one task should read at a time.
    stdout: Arc<Mutex<Option<BufReader<ChildStdout>>>>,

    /// Stderr handle for draining logs
    ///
    /// Wrapped in BufReader for line-by-line reading, then in Arc<Mutex>
    /// for thread-safe access. A background task continuously drains this.
    stderr: Arc<Mutex<Option<BufReader<ChildStderr>>>>,

    /// Background task handle for stderr draining
    ///
    /// Continuously reads stderr and logs it to prevent buffer overflow.
    stderr_task: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,

    /// Directory for protocol logging (optional)
    ///
    /// When set, all JSON-RPC messages are logged to JSONL files in this directory.
    log_dir: Option<PathBuf>,

    /// Protocol logger (created in connect() if log_dir is set)
    logger: Option<AcpProtocolLogger>,

    /// Timestamp of the last write (send) operation
    last_io_write: Arc<Mutex<Option<std::time::Instant>>>,

    /// Timestamp of the last read (recv) operation
    last_io_read: Arc<Mutex<Option<std::time::Instant>>>,

    /// Ring buffer of recent stderr lines from the child process.
    /// Capped at 50 lines. Used to assemble crash context.
    recent_stderr: Arc<Mutex<VecDeque<String>>>,

    /// Crash context captured when recv() detects a child process crash.
    /// Set by recv() when it detects a partial read (crash mid-write).
    last_crash_context: Arc<Mutex<Option<CrashContext>>>,

    /// Pending message from a previous multi-message parse.
    ///
    /// When `try_extract_json` finds two JSON values in one buffer
    /// (e.g., a complete multi-line message followed by the start of the next),
    /// the second value is stored here and returned on the next `recv()` call.
    ///
    /// See [`crate::acp::transport::json_reader::try_extract_json`].
    pending_message: Arc<Mutex<Option<Value>>>,

    /// Incomplete remainder from a previous recv that contained a complete JSON
    /// message followed by the start of another message that wasn't yet complete.
    ///
    /// Without this, the partial data would be discarded and the next `read_line()`
    /// would pick up mid-message, causing a parse failure.
    pending_remainder: Arc<Mutex<Option<String>>>,

    /// Optional process lifecycle manager for graceful shutdown.
    ///
    /// When set, `connect()` registers the child PID with the lifecycle manager
    /// and `close()` performs a graceful shutdown (SIGTERM + timeout) before
    /// force-killing the process. Without it, the original hard-kill behavior
    /// is preserved.
    #[cfg(feature = "server")]
    process_lifecycle: Option<Box<dyn dirigent_process::ProcessLifecycle>>,
}

impl StdioTransport {
    /// Create a new StdioTransport
    ///
    /// # Arguments
    ///
    /// * `command` - Command to execute (e.g., "dirigate")
    /// * `args` - Arguments to pass to the command
    ///
    /// # Returns
    ///
    /// A new StdioTransport that is not yet connected. Call `connect()` to
    /// spawn the process.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use dirigent_core::connectors::acp::transport::StdioTransport;
    ///
    /// let transport = StdioTransport::new(
    ///     "dirigate",
    ///     &["serve", "--stdio"]
    /// );
    /// ```
    pub fn new(command: impl Into<String>, args: &[impl AsRef<str>]) -> Self {
        let command = command.into();
        let args = args.iter().map(|s| s.as_ref().to_string()).collect();

        info!(
            command = %command,
            args = ?args,
            "Creating stdio transport"
        );

        Self {
            command,
            args,
            cwd: None,
            env: vec![],
            child: Arc::new(Mutex::new(None)),
            stdin: Arc::new(Mutex::new(None)),
            stdout: Arc::new(Mutex::new(None)),
            stderr: Arc::new(Mutex::new(None)),
            stderr_task: Arc::new(Mutex::new(None)),
            log_dir: None,
            logger: None,
            last_io_write: Arc::new(Mutex::new(None)),
            last_io_read: Arc::new(Mutex::new(None)),
            recent_stderr: Arc::new(Mutex::new(VecDeque::new())),
            last_crash_context: Arc::new(Mutex::new(None)),
            pending_message: Arc::new(Mutex::new(None)),
            pending_remainder: Arc::new(Mutex::new(None)),
            #[cfg(feature = "server")]
            process_lifecycle: None,
        }
    }

    /// Get the PID of the child process, if connected.
    pub async fn pid(&self) -> Option<u32> {
        let child_lock = self.child.lock().await;
        child_lock.as_ref().and_then(|c| c.id())
    }

    /// Get the crash context captured by the last recv() call, if any.
    pub async fn get_crash_context(&self) -> Option<CrashContext> {
        self.last_crash_context.lock().await.clone()
    }

    /// Get recent stderr lines from the child process.
    pub async fn get_recent_stderr(&self) -> Vec<String> {
        self.recent_stderr.lock().await.iter().cloned().collect()
    }

    /// Get the command used to spawn this transport.
    pub fn command(&self) -> &str {
        &self.command
    }

    /// Get the arguments passed to the command.
    pub fn args(&self) -> &[String] {
        &self.args
    }

    /// Get the timestamp of the last write (send) operation.
    pub async fn last_io_write(&self) -> Option<std::time::Instant> {
        *self.last_io_write.lock().await
    }

    /// Get the timestamp of the last read (recv) operation.
    pub async fn last_io_read(&self) -> Option<std::time::Instant> {
        *self.last_io_read.lock().await
    }

    /// Set the working directory for the spawned process
    ///
    /// Must be called before `connect()`.
    pub fn set_cwd(&mut self, cwd: PathBuf) {
        self.cwd = Some(cwd);
    }

    /// Add an environment variable for the spawned process
    ///
    /// Must be called before `connect()`.
    pub fn set_env(&mut self, key: String, value: String) {
        self.env.push((key, value));
    }

    /// Set the directory for protocol logging
    ///
    /// When set, all JSON-RPC messages (stdin/stdout) will be logged to JSONL files
    /// in this directory. Useful for debugging protocol issues.
    ///
    /// Must be called before `connect()`.
    pub fn set_log_dir(&mut self, log_dir: PathBuf) {
        self.log_dir = Some(log_dir);
    }

    /// Set the process lifecycle manager for graceful shutdown.
    ///
    /// When set, the transport will:
    /// - Call `configure_async_command` on the command before spawning
    /// - Register the child PID immediately after spawn
    /// - Perform a graceful shutdown (SIGTERM → wait → SIGKILL) in `close()`
    ///
    /// Must be called before `connect()`.
    #[cfg(feature = "server")]
    pub fn set_process_lifecycle(&mut self, lifecycle: Box<dyn dirigent_process::ProcessLifecycle>) {
        self.process_lifecycle = Some(lifecycle);
    }

    /// Log a parsed message and return it, updating I/O timestamp.
    async fn log_and_return_message(&mut self, message: Value) -> TransportResult<Option<Value>> {
        let msg_type = if message.get("id").is_some() {
            if message.get("method").is_some() {
                "request"
            } else {
                "response"
            }
        } else if message.get("method").is_some() {
            "notification"
        } else {
            "unknown"
        };

        info!(
            msg_type = msg_type,
            method = message.get("method").and_then(|m| m.as_str()).unwrap_or(""),
            id = message
                .get("id")
                .map(|id| id.to_string())
                .unwrap_or_default(),
            "📦 Parsed message type"
        );

        *self.last_io_read.lock().await = Some(std::time::Instant::now());

        Ok(Some(message))
    }

    /// Core recv implementation that reads from stdout with three-tier parsing:
    ///
    /// 1. **Happy path**: `serde_json::from_str` on the complete line — fast,
    ///    no risk of partial extraction.
    /// 2. **Fallback**: `StreamDeserializer` via `try_extract_json` — handles
    ///    multi-line JSON, concatenated messages, and carries over remainders.
    /// 3. **Recovery**: If both fail, log the bad data and skip it instead of
    ///    killing the connection. Only EOF and I/O errors are fatal.
    async fn recv_from_stdout(&mut self) -> TransportResult<Option<Value>> {
        // Buffer for assembling multi-line JSON messages.
        let mut json_buffer = String::new();
        let mut buffered_lines: usize = 0;

        // Safety limit: max lines to buffer before giving up on this message.
        const MAX_BUFFER_LINES: usize = 50_000;
        // Safety limit: max buffer size in bytes (10 MB)
        const MAX_BUFFER_BYTES: usize = 10 * 1024 * 1024;

        // Initialize buffer from pending remainder if available.
        // This carries over incomplete data from a previous recv() where we
        // extracted one complete JSON message but the remainder was an
        // incomplete second message.
        if let Some(remainder) = self.pending_remainder.lock().await.take() {
            warn!(
                remainder_bytes = remainder.len(),
                remainder_preview = %&remainder[..remainder.len().min(120)],
                "Initializing buffer from pending remainder"
            );
            json_buffer = remainder;
            buffered_lines = 1;
        }

        loop {
            let mut stdout_lock = self.stdout.lock().await;
            let stdout = stdout_lock
                .as_mut()
                .ok_or("Transport not connected (no stdout)")?;

            let mut line = String::new();
            let bytes_read = stdout
                .read_line(&mut line)
                .await
                .map_err(|e| format!("Failed to read from stdout: {}", e))?;

            // EOF — this IS fatal (process closed stdout)
            if bytes_read == 0 {
                if !json_buffer.is_empty() {
                    warn!(
                        buffer_lines = buffered_lines,
                        buffer_bytes = json_buffer.len(),
                        "EOF with incomplete multi-line JSON buffer"
                    );
                }
                debug!("Received EOF from child process");
                return Ok(None);
            }

            // Detect partial read (crash mid-write signature) — fatal
            if bytes_read > 0 && !line.ends_with('\n') {
                warn!(
                    bytes_read,
                    partial_data = %line,
                    "Partial read from child process (likely crash mid-write)"
                );
                drop(stdout_lock);

                let exit_status = {
                    let mut child_lock = self.child.lock().await;
                    child_lock.as_mut().and_then(|c| c.try_wait().ok().flatten())
                };
                let recent_stderr = self.recent_stderr.lock().await.iter().cloned().collect();
                let crash_ctx = CrashContext {
                    recent_stderr,
                    exit_status,
                    partial_stdout: Some(line),
                };
                info!(
                    exit_status = ?crash_ctx.exit_status,
                    stderr_lines = crash_ctx.recent_stderr.len(),
                    "Crash context assembled for child process crash"
                );
                *self.last_crash_context.lock().await = Some(crash_ctx);
                return Ok(None);
            }

            // Log every line read for framing diagnostics
            // TODO: downgrade to debug once framing bug is resolved
            warn!(
                bytes_read,
                line_starts_with = %&line[..line.len().min(40)].escape_debug(),
                buffer_lines = buffered_lines,
                buffer_bytes = json_buffer.len(),
                "📥 read_line from child stdout"
            );

            // Skip empty lines when not buffering
            if line.trim().is_empty() && json_buffer.is_empty() {
                warn!("Received empty line from child process, skipping");
                drop(stdout_lock);
                continue;
            }

            // Release stdout lock before doing work
            drop(stdout_lock);

            // ── Tier 1: Happy path — from_str on fresh single line ──
            //
            // If we have no accumulated buffer, try parsing the line directly.
            // This is the fast path for well-formed single-line JSON (99% of messages).
            // Unlike StreamDeserializer, from_str is all-or-nothing: it cannot
            // extract a truncated value from a line with escaping issues.
            if json_buffer.is_empty() {
                if let Ok(message) = serde_json::from_str::<Value>(line.trim()) {
                    info!(
                        method = message.get("method").and_then(|m| m.as_str()).unwrap_or(""),
                        id = message.get("id").map(|i| i.to_string()).unwrap_or_default(),
                        bytes = bytes_read,
                        "📨 Received message (from_str fast path)"
                    );
                    if let Some(ref mut logger) = self.logger {
                        logger.log_incoming_raw(line.trim());
                    }
                    return self.log_and_return_message(message).await;
                }
                // from_str failed — fall through to tier 2 (buffer + StreamDeserializer)
            }

            // ── Tier 2: Fallback — multi-line buffering with StreamDeserializer ──
            //
            // Append to buffer and try StreamDeserializer, which handles:
            // - Multi-line / pretty-printed JSON
            // - Concatenated messages (extracts first, stores remainder)
            // - Incomplete JSON (returns None, we keep buffering)

            // Append to buffer
            if json_buffer.is_empty() {
                json_buffer = line;
                buffered_lines = 1;
            } else {
                json_buffer.push_str(&line);
                buffered_lines += 1;
            }

            if let Some((message, remainder)) = crate::acp::transport::json_reader::try_extract_json(&json_buffer) {
                let method = message.get("method").and_then(|m| m.as_str()).unwrap_or("");
                let msg_id = message.get("id").map(|i| i.to_string()).unwrap_or_default();
                if buffered_lines > 1 {
                    warn!(
                        lines = buffered_lines,
                        bytes = json_buffer.len(),
                        method,
                        id = msg_id,
                        remainder_bytes = remainder.len(),
                        "Recovered multi-line JSON message from child process"
                    );
                } else {
                    info!(method, id = msg_id, "📨 Received message (StreamDeserializer fallback)");
                }

                // Log raw content to protocol log
                if let Some(ref mut logger) = self.logger {
                    let parsed_end = json_buffer.len() - remainder.len();
                    logger.log_incoming_raw(&json_buffer[..parsed_end]);
                }

                // Handle leftover data from concatenated messages
                if !remainder.is_empty() {
                    warn!(
                        remainder_bytes = remainder.len(),
                        remainder_preview = %&remainder[..remainder.len().min(120)],
                        "Buffer has remaining data after complete JSON, checking for second message"
                    );
                    if let Some((second_msg, second_remainder)) = crate::acp::transport::json_reader::try_extract_json(&remainder)
                    {
                        warn!(
                            second_msg_method = second_msg.get("method").and_then(|m| m.as_str()).unwrap_or("?"),
                            second_remainder_bytes = second_remainder.len(),
                            "Extracted second complete message from remainder"
                        );
                        *self.pending_message.lock().await = Some(second_msg);
                        if !second_remainder.is_empty() {
                            warn!(
                                remainder_bytes = second_remainder.len(),
                                remainder_preview = %&second_remainder[..second_remainder.len().min(120)],
                                "Carrying over third message fragment as pending remainder"
                            );
                            *self.pending_remainder.lock().await = Some(second_remainder);
                        }
                    } else {
                        warn!(
                            remainder_bytes = remainder.len(),
                            remainder_preview = %&remainder[..remainder.len().min(120)],
                            "Carrying over incomplete remainder for next recv()"
                        );
                        *self.pending_remainder.lock().await = Some(remainder);
                    }
                }

                return self.log_and_return_message(message).await;
            }

            // ── Tier 3: Recovery — skip unparseable data ──
            //
            // If the buffer doesn't start with a JSON token, it's garbage.
            // Instead of killing the connection, log the bad data and reset.
            // One lost notification is better than a dead session.
            let buffer_trimmed = json_buffer.trim_start();
            if !buffer_trimmed.starts_with('{') && !buffer_trimmed.starts_with('[') {
                warn!(
                    target: "acp_protocol",
                    buffered_lines,
                    buffer_bytes = json_buffer.len(),
                    line_bytes = bytes_read,
                    raw_preview = %&buffer_trimmed[..buffer_trimmed.len().min(300)],
                    "Skipping non-JSON data from child process (connection preserved)"
                );
                if let Some(ref mut logger) = self.logger {
                    logger.log_incoming_raw(buffer_trimmed);
                }
                // Reset and keep reading
                json_buffer.clear();
                buffered_lines = 0;
                continue;
            }

            // Safety limits — these ARE fatal (runaway buffering)
            if buffered_lines >= MAX_BUFFER_LINES {
                error!(
                    target: "acp_protocol",
                    lines = buffered_lines,
                    bytes = json_buffer.len(),
                    "Multi-line JSON buffer exceeded line limit, aborting"
                );
                return Err(format!(
                    "Multi-line JSON message exceeded {} lines without completing",
                    MAX_BUFFER_LINES
                )
                .into());
            }
            if json_buffer.len() > MAX_BUFFER_BYTES {
                error!(
                    target: "acp_protocol",
                    lines = buffered_lines,
                    bytes = json_buffer.len(),
                    "Multi-line JSON buffer exceeded size limit, aborting"
                );
                return Err(format!(
                    "Multi-line JSON message exceeded {} bytes without completing",
                    MAX_BUFFER_BYTES
                )
                .into());
            }

            if buffered_lines % 1000 == 0 {
                debug!(
                    lines = buffered_lines,
                    bytes = json_buffer.len(),
                    "Still buffering multi-line JSON message..."
                );
            }
        }
    }
}

#[async_trait]
impl AcpTransport for StdioTransport {
    async fn connect(&mut self) -> TransportResult<()> {
        let mut child_lock = self.child.lock().await;

        if child_lock.is_some() {
            return Err("Transport already connected".into());
        }

        info!(
            command = %self.command,
            args = ?self.args,
            "Spawning child process"
        );

        // Spawn the child process
        let mut command = Command::new(&self.command);
        command
            .args(&self.args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped()) // Pipe stderr to prevent buffer overflow
            .kill_on_drop(true); // Ensure process is killed when dropped

        // Set working directory if specified
        if let Some(cwd) = &self.cwd {
            command.current_dir(cwd);
        }

        // Set environment variables
        for (key, value) in &self.env {
            command.env(key, value);
        }

        // Allow the process lifecycle manager to configure platform-specific
        // flags (e.g. Windows Job Object, Linux process group) before spawning.
        #[cfg(feature = "server")]
        if let Some(ref lifecycle) = self.process_lifecycle {
            lifecycle.configure_async_command(&mut command);
        }

        let mut child = command
            .spawn()
            .map_err(|e| format!("Failed to spawn process '{}': {}", self.command, e))?;

        // Register the child PID with the lifecycle manager so it is included
        // in the platform-level process group / job object.
        #[cfg(feature = "server")]
        if let Some(ref lifecycle) = self.process_lifecycle {
            if let Some(pid) = child.id() {
                if let Err(e) = lifecycle.register_child(pid) {
                    warn!(error = %e, pid, "Failed to register child with process lifecycle");
                }
            }
        }

        // Take stdin/stdout/stderr handles
        let stdin = child.stdin.take().ok_or("Failed to open child stdin")?;
        let stdout = child.stdout.take().ok_or("Failed to open child stdout")?;
        let stderr = child.stderr.take().ok_or("Failed to open child stderr")?;

        // Store stdin/stdout handles
        *child_lock = Some(child);
        *self.stdin.lock().await = Some(stdin);
        *self.stdout.lock().await = Some(BufReader::new(stdout));

        // Store stderr handle and spawn draining task
        let stderr_reader = BufReader::new(stderr);
        *self.stderr.lock().await = Some(stderr_reader);

        // Spawn background task to drain stderr continuously
        let stderr_arc = Arc::clone(&self.stderr);
        let recent_stderr_arc = Arc::clone(&self.recent_stderr);
        let command_name = self.command.clone();
        let stderr_handle = tokio::spawn(async move {
            debug!(target: "child_stderr", command = %command_name, "Starting stderr draining task");

            loop {
                let mut stderr_lock = stderr_arc.lock().await;
                if let Some(stderr_reader) = stderr_lock.as_mut() {
                    let mut line = String::new();
                    match stderr_reader.read_line(&mut line).await {
                        Ok(0) => {
                            // EOF - process closed stderr
                            debug!(target: "child_stderr", command = %command_name, "Stderr EOF reached, task exiting");
                            break;
                        }
                        Ok(_) => {
                            // Got a line - log it
                            let trimmed = line.trim();
                            if !trimmed.is_empty() {
                                // Non-empty stderr from child process is often indicative of problems
                                // Use warn level so it's visible in console by default
                                warn!(target: "child_stderr", command = %command_name, line = %trimmed, "Child process stderr output");
                                {
                                    let mut buffer = recent_stderr_arc.lock().await;
                                    buffer.push_back(trimmed.to_string());
                                    while buffer.len() > 50 {
                                        buffer.pop_front();
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            // Error reading - log and exit
                            warn!(target: "child_stderr", command = %command_name, error = %e, "Error reading stderr, task exiting");
                            break;
                        }
                    }
                } else {
                    // Stderr handle was taken (during close), exit task
                    debug!(target: "child_stderr", command = %command_name, "Stderr handle removed, task exiting");
                    break;
                }
                // Release lock between reads to allow close() to proceed
                drop(stderr_lock);
            }

            debug!(target: "child_stderr", command = %command_name, "Stderr draining task finished");
        });

        *self.stderr_task.lock().await = Some(stderr_handle);

        // Create protocol logger if log_dir is configured
        if let Some(log_dir) = &self.log_dir {
            self.logger = AcpProtocolLogger::new(log_dir.clone(), &self.command);
        }

        info!("Child process spawned successfully with stderr draining");

        Ok(())
    }

    async fn send(&mut self, message: Value) -> TransportResult<()> {
        let mut stdin_lock = self.stdin.lock().await;
        let stdin = stdin_lock
            .as_mut()
            .ok_or("Transport not connected (no stdin)")?;

        // Serialize to JSON and add newline
        let json_str = serde_json::to_string(&message)
            .map_err(|e| format!("Failed to serialize message: {}", e))?;

        // Determine message type for logging
        let msg_type = if message.get("method").is_some() {
            if message.get("id").is_some() {
                "request"
            } else {
                "notification"
            }
        } else {
            "response"
        };
        let method = message.get("method").and_then(|m| m.as_str()).unwrap_or("");
        let id = message.get("id").map(|i| i.to_string()).unwrap_or_default();

        info!(
            msg_type = msg_type,
            method = method,
            id = id,
            "📤 Sending message to child process"
        );
        let masked_message = dirigent_protocol::log_utils::mask_json_string(&json_str);
        debug!(message = %masked_message, "Full message JSON");

        // Log to protocol log file if enabled
        if let Some(ref mut logger) = self.logger {
            logger.log_outgoing(&message);
        }

        // Write to stdin
        stdin
            .write_all(json_str.as_bytes())
            .await
            .map_err(|e| format!("Failed to write to stdin: {}", e))?;
        stdin
            .write_all(b"\n")
            .await
            .map_err(|e| format!("Failed to write newline to stdin: {}", e))?;
        stdin
            .flush()
            .await
            .map_err(|e| format!("Failed to flush stdin: {}", e))?;

        debug!("Message sent and flushed successfully");

        // Record I/O timestamp
        *self.last_io_write.lock().await = Some(std::time::Instant::now());

        Ok(())
    }

    async fn recv(&mut self) -> TransportResult<Option<Value>> {
        // Check if we have a leftover message from a previous multi-message parse.
        // Clone the Arc to avoid holding an immutable borrow of self across the
        // mutable call to recv_from_stdout/log_and_return_message.
        let pending_arc = Arc::clone(&self.pending_message);
        let pending = pending_arc.lock().await.take();

        if let Some(msg) = pending {
            debug!("Returning pending message from previous recv() parse");
            self.log_and_return_message(msg).await
        } else {
            self.recv_from_stdout().await
        }
    }

    async fn close(&mut self) -> TransportResult<()> {
        info!("Closing stdio transport");

        // Abort stderr draining task first
        if let Some(task) = self.stderr_task.lock().await.take() {
            debug!("Aborting stderr draining task");
            task.abort();
            // Wait briefly for task to finish with timeout
            let _ = tokio::time::timeout(std::time::Duration::from_millis(100), task).await;
        }

        // Drop stdin/stdout/stderr to close pipes (signals process to exit)
        *self.stdin.lock().await = None;
        *self.stdout.lock().await = None;
        *self.stderr.lock().await = None;

        // Terminate the child process
        let mut child_lock = self.child.lock().await;
        if let Some(mut child) = child_lock.take() {
            debug!("Terminating child process");

            #[cfg(feature = "server")]
            let used_lifecycle = {
                if let Some(ref lifecycle) = self.process_lifecycle {
                    if child.id().is_some() {
                        // Graceful shutdown: SIGTERM/CTRL_BREAK → wait → force kill
                        dirigent_process::graceful_shutdown_async(
                            lifecycle.as_ref(),
                            &mut child,
                            std::time::Duration::from_secs(5),
                        )
                        .await;
                        true
                    } else {
                        // Process already exited
                        true
                    }
                } else {
                    false
                }
            };

            #[cfg(not(feature = "server"))]
            let used_lifecycle = false;

            if !used_lifecycle {
                // Fallback: hard kill (original behavior, belt-and-suspenders)
                if let Err(e) = child.start_kill() {
                    warn!(error = %e, "Failed to send kill signal (process may have already exited)");
                }

                // Wait briefly with timeout for process to exit
                let wait_result =
                    tokio::time::timeout(std::time::Duration::from_millis(500), child.wait()).await;

                match wait_result {
                    Ok(Ok(status)) => {
                        info!(status = ?status, "Child process exited");
                    }
                    Ok(Err(e)) => {
                        warn!(error = %e, "Error waiting for child process");
                    }
                    Err(_) => {
                        warn!("Child process did not exit within timeout, leaving to OS cleanup");
                    }
                }
            }
        }

        Ok(())
    }

    async fn get_crash_context(&self) -> Option<CrashContext> {
        self.last_crash_context.lock().await.clone()
    }

    async fn pid(&self) -> Option<u32> {
        let child_lock = self.child.lock().await;
        child_lock.as_ref().and_then(|c| c.id())
    }
}

impl Drop for StdioTransport {
    fn drop(&mut self) {
        // Attempt to kill the child process if it's still running
        // This is a best-effort cleanup - we can't use async in Drop
        if let Ok(mut child_lock) = self.child.try_lock() {
            if let Some(mut child) = child_lock.take() {
                // Try to kill synchronously (blocking)
                let _ = child.start_kill();
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    /// Helper to find the mocker binary
    fn mocker_binary() -> std::path::PathBuf {
        let manifest_dir = env!("CARGO_MANIFEST_DIR");
        std::path::Path::new(manifest_dir)
            .parent()
            .unwrap()
            .parent()
            .unwrap()
            .join("target")
            .join("debug")
            .join(if cfg!(windows) {
                "dirigate.exe"
            } else {
                "dirigate"
            })
    }

    /// Helper to find the fixture file
    fn fixture_path() -> std::path::PathBuf {
        let manifest_dir = env!("CARGO_MANIFEST_DIR");
        std::path::Path::new(manifest_dir)
            .parent()
            .unwrap()
            .join("dirigate")
            .join("examples")
            .join("basic.yaml")
    }

    #[tokio::test]
    async fn test_stdio_transport_basic_flow() {
        let binary = mocker_binary();
        if !binary.exists() {
            eprintln!(
                "Skipping test: mocker binary not found at {:?}. Run 'cargo build' first.",
                binary
            );
            return;
        }

        let fixture = fixture_path();
        assert!(fixture.exists(), "Fixture file not found at {:?}", fixture);

        // Create transport
        let mut transport = StdioTransport::new(
            binary.to_str().unwrap(),
            &["serve", "--fixtures", fixture.to_str().unwrap(), "--stdio"],
        );

        // Connect
        transport.connect().await.expect("Failed to connect");

        // Test 1: Initialize
        let init_request = json!({
            "jsonrpc": "2.0",
            "method": "initialize",
            "id": 1,
            "params": {
                "protocolVersion": 1,
                "clientCapabilities": {}
            }
        });

        transport
            .send(init_request)
            .await
            .expect("Failed to send init request");

        let init_response = transport
            .recv()
            .await
            .expect("Failed to receive init response")
            .expect("Got None instead of response");

        assert_eq!(init_response["jsonrpc"], "2.0");
        assert_eq!(init_response["id"], 1);
        assert!(init_response["result"]["protocolVersion"].is_number());

        // Test 2: Create session
        let session_request = json!({
            "jsonrpc": "2.0",
            "method": "session/new",
            "id": 2,
            "params": {
                "cwd": ".",
                "mcpServers": []
            }
        });

        transport
            .send(session_request)
            .await
            .expect("Failed to send session request");

        let session_response = transport
            .recv()
            .await
            .expect("Failed to receive session response")
            .expect("Got None instead of response");

        assert_eq!(session_response["jsonrpc"], "2.0");
        assert_eq!(session_response["id"], 2);
        let session_id = session_response["result"]["sessionId"]
            .as_str()
            .expect("sessionId should be a string");
        assert!(!session_id.is_empty());

        // Test 3: Send a prompt
        let prompt_request = json!({
            "jsonrpc": "2.0",
            "method": "session/prompt",
            "id": 3,
            "params": {
                "sessionId": session_id,
                "prompt": [
                    {
                        "type": "text",
                        "text": "hello"
                    }
                ]
            }
        });

        transport
            .send(prompt_request)
            .await
            .expect("Failed to send prompt request");

        // Receive response (skip any notifications)
        let mut prompt_response = None;
        for _ in 0..10 {
            // Max 10 messages to avoid infinite loop
            if let Some(msg) = transport.recv().await.expect("Failed to receive") {
                // Check if this is the response we're looking for
                if msg.get("id") == Some(&json!(3)) {
                    prompt_response = Some(msg);
                    break;
                }
                // Skip notifications
            } else {
                break;
            }
        }

        let prompt_response = prompt_response.expect("Did not receive response to prompt request");
        assert_eq!(prompt_response["jsonrpc"], "2.0");
        assert_eq!(prompt_response["id"], 3);
        assert!(prompt_response["result"]["stopReason"].is_string());

        // Close
        transport.close().await.expect("Failed to close");
    }

    #[tokio::test]
    async fn test_stdio_transport_process_exit() {
        let binary = mocker_binary();
        if !binary.exists() {
            eprintln!("Skipping test: mocker binary not found");
            return;
        }

        let fixture = fixture_path();
        if !fixture.exists() {
            eprintln!("Skipping test: fixture not found");
            return;
        }

        let mut transport = StdioTransport::new(
            binary.to_str().unwrap(),
            &["serve", "--fixtures", fixture.to_str().unwrap(), "--stdio"],
        );

        transport.connect().await.expect("Failed to connect");

        // Close the transport (kills the process)
        transport.close().await.expect("Failed to close");

        // Try to send after close - should fail
        let result = transport.send(json!({"test": "message"})).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_stdio_transport_malformed_json() {
        // This test would require a mock process that sends malformed JSON
        // For now, we'll skip it as it requires more infrastructure
        // In a real implementation, you might use a test harness process
    }

    // try_extract_json unit tests are in crate::acp::transport::json_reader::tests

    #[tokio::test]
    async fn test_stdio_transport_concurrent_sends() {
        let binary = mocker_binary();
        if !binary.exists() {
            eprintln!("Skipping test: mocker binary not found");
            return;
        }

        let fixture = fixture_path();
        if !fixture.exists() {
            eprintln!("Skipping test: fixture not found");
            return;
        }

        let mut transport = StdioTransport::new(
            binary.to_str().unwrap(),
            &["serve", "--fixtures", fixture.to_str().unwrap(), "--stdio"],
        );

        transport.connect().await.expect("Failed to connect");

        // Initialize first
        transport
            .send(json!({
                "jsonrpc": "2.0",
                "method": "initialize",
                "id": 1,
                "params": {
                    "protocolVersion": 1,
                    "clientCapabilities": {}
                }
            }))
            .await
            .expect("Failed to send init");

        // Drain the response
        transport.recv().await.ok();

        // Send multiple messages in quick succession
        let transport = Arc::new(Mutex::new(transport));

        let mut handles = vec![];
        for i in 0..5 {
            let t = Arc::clone(&transport);
            let handle = tokio::spawn(async move {
                let mut transport = t.lock().await;
                transport
                    .send(json!({
                        "jsonrpc": "2.0",
                        "method": "session/new",
                        "id": i + 10,
                        "params": {
                            "cwd": ".",
                            "mcpServers": []
                        }
                    }))
                    .await
            });
            handles.push(handle);
        }

        // All sends should succeed
        for handle in handles {
            handle.await.expect("Task panicked").expect("Send failed");
        }

        // Close
        let mut transport = transport.lock().await;
        transport.close().await.expect("Failed to close");
    }
}
