gemini-cli-sdk 0.1.0

//! Transport trait and `GeminiTransport` production implementation.
//!
//! The [`Transport`] trait abstracts over the subprocess communication channel.
//! [`GeminiTransport`] is the production implementation that spawns
//! `gemini --experimental-acp` as a child process and communicates with it
//! via JSON-RPC 2.0 over NDJSON stdio.
//!
//! # Architecture
//!
//! ```text
//! GeminiTransport
//!   ├── stdin  ──► gemini process ──► stdout  (background reader, T12)
//!   │                                 │
//!   │               JSON-RPC 2.0      ▼
//!   │            pending_requests  message_tx ──► message_rx (stream)
//!   └── reverse requests ◄── ReverseRequestHandler
//! ```
//!
//! The background reader (implemented in T12) feeds `message_tx`. Callers
//! consume parsed JSON values through `read_messages()` or via the higher-level
//! `send_request` / `send_notification` helpers.

use std::collections::HashMap;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;

use async_trait::async_trait;
use dashmap::DashMap;
use futures_core::Stream;
use serde::de::DeserializeOwned;
use serde::Serialize;
use serde_json::Value;
use tokio::io::AsyncWriteExt;
use tokio::process::{Child, ChildStdin};
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio::task::JoinHandle;

use crate::jsonrpc::{JsonRpcId, JsonRpcNotification, JsonRpcRequest, JsonRpcResponse};
use crate::Result;

// ── Transport Trait ──────────────────────────────────────────────────────────

/// Abstract transport for communicating with a Gemini CLI process.
///
/// Implementations are responsible for process lifecycle (spawn / kill),
/// NDJSON framing over stdio, and delivering parsed JSON values to callers.
///
/// # Contract
///
/// - [`connect`] must be called and awaited before any other method.
/// - [`write`] appends a newline after `data` to maintain NDJSON framing.
/// - [`read_messages`] returns a stream that stays open until the process exits.
/// - [`close`] is idempotent: calling it more than once is safe.
///
/// [`connect`]: Transport::connect
/// [`write`]: Transport::write
/// [`read_messages`]: Transport::read_messages
/// [`close`]: Transport::close
#[async_trait]
pub trait Transport: Send + Sync {
    /// Spawn the underlying process and begin reading from its stdout.
    ///
    /// # Errors
    ///
    /// Returns [`Error::SpawnFailed`] when the OS refuses to start the process.
    ///
    /// [`Error::SpawnFailed`]: crate::Error::SpawnFailed
    async fn connect(&self) -> Result<()>;

    /// Write a raw NDJSON line to the process stdin.
    ///
    /// A newline (`\n`) is appended automatically. The data is flushed
    /// immediately so the subprocess receives it without buffering delay.
    ///
    /// # Errors
    ///
    /// - [`Error::NotConnected`] — stdin is not open (connect not called, or
    ///   stdin was already closed via [`end_input`]).
    /// - [`Error::Io`] — the write or flush failed.
    ///
    /// [`Error::NotConnected`]: crate::Error::NotConnected
    /// [`Error::Io`]: crate::Error::Io
    /// [`end_input`]: Transport::end_input
    async fn write(&self, data: &str) -> Result<()>;

    /// Return a stream of parsed JSON values from the process stdout.
    ///
    /// The stream yields one item per NDJSON line. It remains open until the
    /// subprocess exits or the reader task is cancelled.
    ///
    /// The background reader loop (implemented in T12) feeds this stream via
    /// an internal `mpsc` channel.
    fn read_messages(&self) -> Pin<Box<dyn Stream<Item = Result<Value>> + Send>>;

    /// Signal that no more input will be sent; closes stdin.
    ///
    /// Dropping stdin causes the subprocess to receive EOF on its standard
    /// input, which typically causes a clean shutdown of the JSON-RPC server.
    async fn end_input(&self) -> Result<()>;

    /// Send an interrupt signal to the subprocess.
    ///
    /// On Unix, delivers `SIGINT` to the process group. On Windows, sends
    /// `CTRL_C_EVENT` to the console group (implemented in T12/integration).
    async fn interrupt(&self) -> Result<()>;

    /// Returns `true` if the transport has been connected and is ready to use.
    fn is_ready(&self) -> bool;

    /// Shut down the transport cleanly and return the process exit code.
    ///
    /// 1. Cancels the background reader task.
    /// 2. Closes stdin (sending EOF to the subprocess).
    /// 3. Waits for the subprocess to exit.
    /// 4. Returns the exit code, or `None` if the process was already gone.
    async fn close(&self) -> Result<Option<i32>>;
}

// ── ReverseRequestHandler ────────────────────────────────────────────────────

/// Handles reverse requests from the Gemini CLI subprocess.
///
/// The CLI may send JSON-RPC *requests* back to the host process, for example
/// to ask for tool-use permission approval. The background reader (T12)
/// dispatches such incoming requests to this handler. The handler must produce
/// a JSON value to send back as the JSON-RPC response body.
#[async_trait]
pub(crate) trait ReverseRequestHandler: Send + Sync {
    /// Called when the CLI sends a permission-request JSON-RPC call.
    ///
    /// `id` is the JSON-RPC request ID that must be echoed in the response.
    /// `params` is the raw params object from the CLI's request.
    ///
    /// The return value is serialised and sent back as the `result` field of
    /// the JSON-RPC response.
    async fn handle_permission_request(&self, id: JsonRpcId, params: Value) -> Value;
}

// ── GeminiTransport ──────────────────────────────────────────────────────────

/// Production transport that spawns `gemini --experimental-acp` as a subprocess
/// and communicates via JSON-RPC 2.0 over NDJSON stdio.
///
/// # Thread Safety
///
/// `GeminiTransport` is `Send + Sync`. Most mutable state is guarded by
/// `tokio::sync::Mutex` for async-friendly locking. The `process` field uses
/// `std::sync::Mutex` so it can be accessed from the synchronous [`Drop`]
/// implementation without entering an async context.
///
/// # Lifecycle
///
/// 1. Construct via [`GeminiTransport::new`] or [`GeminiTransport::from_config`].
/// 2. Call [`connect`] — this spawns the subprocess and starts the reader task.
/// 3. Use [`send_request`] / [`send_notification`] / [`send_response`] for RPC.
/// 4. Call [`close`] when done. Drop provides best-effort cleanup.
///
/// [`connect`]: Transport::connect
/// [`send_request`]: GeminiTransport::send_request
/// [`send_notification`]: GeminiTransport::send_notification
/// [`send_response`]: GeminiTransport::send_response
/// [`close`]: Transport::close
pub struct GeminiTransport {
    // ── Subprocess configuration ─────────────────────────────────────────────
    pub(crate) cli_path: PathBuf,
    pub(crate) args: Vec<String>,
    pub(crate) cwd: PathBuf,
    pub(crate) env: HashMap<String, String>,

    // ── Process handles ──────────────────────────────────────────────────────
    // `std::sync::Mutex` is used here (not `tokio::sync::Mutex`) so the Drop
    // impl can call `start_kill()` synchronously without needing an async context
    // or a runtime handle.
    pub(crate) process: std::sync::Mutex<Option<Child>>,
    /// Stdin handle. `None` before `connect()` and after `end_input()`.
    /// Wrapped in `Arc` so the background reader can share it for reverse-request responses.
    pub(crate) stdin: Arc<Mutex<Option<ChildStdin>>>,

    // ── JSON-RPC state ───────────────────────────────────────────────────────
    /// Monotonically increasing request ID counter.
    pub(crate) next_request_id: AtomicU64,
    /// In-flight request table: request ID ──► response oneshot sender.
    /// Wrapped in `Arc` so the background reader task can share ownership.
    pub(crate) pending_requests: Arc<DashMap<JsonRpcId, oneshot::Sender<JsonRpcResponse>>>,

    // ── Message routing ──────────────────────────────────────────────────────
    // The background reader task (T12) writes parsed JSON values into
    // `message_tx`. `read_messages()` exposes a stream over `message_rx`.
    pub(crate) message_tx: Mutex<Option<mpsc::Sender<Result<Value>>>>,
    pub(crate) message_rx: Mutex<Option<mpsc::Receiver<Result<Value>>>>,
    pub(crate) reader_handle: Mutex<Option<JoinHandle<()>>>,

    // ── Reverse request handler ──────────────────────────────────────────────
    /// Wrapped in `Arc` so the background reader can clone the inner handler without
    /// holding the lock across await points.
    pub(crate) reverse_handler: Arc<Mutex<Option<Arc<dyn ReverseRequestHandler>>>>,

    // ── State ────────────────────────────────────────────────────────────────
    pub(crate) ready: AtomicBool,
    /// Optional callback receiving every line written to the subprocess stderr.
    pub(crate) stderr_callback: Option<Arc<dyn Fn(String) + Send + Sync>>,
    /// Timeout for `close()` to wait for the subprocess to exit before killing it.
    pub(crate) close_timeout: Option<std::time::Duration>,
}

impl GeminiTransport {
    /// Create a new transport with explicit parameters.
    ///
    /// The transport is not connected after construction. Call [`Transport::connect`]
    /// to spawn the subprocess.
    ///
    /// # Parameters
    ///
    /// - `cli_path` — path to the `gemini` binary.
    /// - `args` — CLI arguments (typically from [`ClientConfig::to_cli_args`]).
    /// - `cwd` — working directory for the subprocess.
    /// - `env` — additional environment variables.
    /// - `stderr_callback` — optional sink for subprocess stderr lines.
    ///
    /// [`ClientConfig::to_cli_args`]: crate::config::ClientConfig::to_cli_args
    pub fn new(
        cli_path: PathBuf,
        args: Vec<String>,
        cwd: PathBuf,
        env: HashMap<String, String>,
        stderr_callback: Option<Arc<dyn Fn(String) + Send + Sync>>,
        close_timeout: Option<std::time::Duration>,
    ) -> Self {
        let (tx, rx) = mpsc::channel(256);
        Self {
            cli_path,
            args,
            cwd,
            env,
            process: std::sync::Mutex::new(None),
            stdin: Arc::new(Mutex::new(None)),
            next_request_id: AtomicU64::new(1),
            pending_requests: Arc::new(DashMap::new()),
            message_tx: Mutex::new(Some(tx)),
            message_rx: Mutex::new(Some(rx)),
            reader_handle: Mutex::new(None),
            reverse_handler: Arc::new(Mutex::new(None)),
            ready: AtomicBool::new(false),
            stderr_callback,
            close_timeout,
        }
    }

    /// Create a transport from a [`ClientConfig`].
    ///
    /// Resolves the CLI binary path via [`discovery::find_cli`] when
    /// `config.cli_path` is `None`, and falls back to the current working
    /// directory when `config.cwd` is `None`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::CliNotFound`] when no `gemini` binary can be located.
    ///
    /// [`ClientConfig`]: crate::config::ClientConfig
    /// [`discovery::find_cli`]: crate::discovery::find_cli
    /// [`Error::CliNotFound`]: crate::Error::CliNotFound
    pub fn from_config(config: &crate::config::ClientConfig) -> crate::Result<Self> {
        let cli_path = if let Some(path) = &config.cli_path {
            path.clone()
        } else {
            crate::discovery::find_cli()?
        };
        let cwd = config
            .cwd
            .clone()
            .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")));
        let args = config.to_cli_args();
        Ok(Self::new(
            cli_path,
            args,
            cwd,
            config.env.clone(),
            config.stderr_callback.clone(),
            config.close_timeout,
        ))
    }

    // ── JSON-RPC helpers ─────────────────────────────────────────────────────

    /// Send a JSON-RPC request and wait for its response.
    ///
    /// Generates a unique request ID, registers a oneshot channel in
    /// `pending_requests`, serialises the request, writes it to stdin, then
    /// awaits the response. The background reader (T12) resolves the oneshot
    /// when the matching response arrives.
    ///
    /// # Errors
    ///
    /// - [`Error::NotConnected`] — stdin is not open.
    /// - [`Error::Transport`] — the response channel was closed before the
    ///   response arrived (process likely exited).
    /// - [`Error::JsonRpcError`] — the server returned a JSON-RPC error object.
    /// - [`Error::Json`] — the response `result` could not be deserialised as `R`.
    ///
    /// [`Error::NotConnected`]: crate::Error::NotConnected
    /// [`Error::Transport`]: crate::Error::Transport
    /// [`Error::JsonRpcError`]: crate::Error::JsonRpcError
    /// [`Error::Json`]: crate::Error::Json
    pub(crate) async fn send_request<P: Serialize, R: DeserializeOwned>(
        &self,
        method: &str,
        params: P,
    ) -> Result<R> {
        let id = JsonRpcId::Number(self.next_request_id.fetch_add(1, Ordering::Relaxed));
        let request =
            JsonRpcRequest::new(id.clone(), method, Some(serde_json::to_value(params)?));

        // Register oneshot before writing so the response cannot race ahead.
        let (tx, rx) = oneshot::channel();
        self.pending_requests.insert(id, tx);

        let line = serde_json::to_string(&request)?;
        self.write(&line).await?;

        let response = rx.await.map_err(|_| {
            crate::Error::Transport(
                "Response channel closed before response received".to_string(),
            )
        })?;

        match response.into_result() {
            Ok(value) => serde_json::from_value(value).map_err(Into::into),
            Err(err) => Err(crate::Error::JsonRpcError {
                code: err.code,
                message: err.message,
                data: err.data,
            }),
        }
    }

    /// Send a JSON-RPC request and return the response receiver **without**
    /// awaiting it.
    ///
    /// This is the non-blocking counterpart of [`send_request`]. The caller
    /// is responsible for awaiting the returned `oneshot::Receiver` (typically
    /// inside a `select!` that also drains a concurrent notification stream).
    ///
    /// [`send_request`]: GeminiTransport::send_request
    pub(crate) async fn send_request_start<P: Serialize>(
        &self,
        method: &str,
        params: P,
    ) -> Result<oneshot::Receiver<JsonRpcResponse>> {
        let id = JsonRpcId::Number(self.next_request_id.fetch_add(1, Ordering::Relaxed));
        let request =
            JsonRpcRequest::new(id.clone(), method, Some(serde_json::to_value(params)?));

        // Register oneshot before writing so the response cannot race ahead.
        let (tx, rx) = oneshot::channel();
        self.pending_requests.insert(id, tx);

        let line = serde_json::to_string(&request)?;
        self.write(&line).await?;

        Ok(rx)
    }

    /// Send a JSON-RPC notification (no response expected or awaited).
    ///
    /// # Errors
    ///
    /// Propagates errors from [`write`].
    ///
    /// [`write`]: Transport::write
    pub(crate) async fn send_notification<P: Serialize>(
        &self,
        method: &str,
        params: P,
    ) -> Result<()> {
        let notif = JsonRpcNotification::new(method, Some(serde_json::to_value(params)?));
        let line = serde_json::to_string(&notif)?;
        self.write(&line).await
    }

    /// Send a JSON-RPC response, answering a reverse request from the agent.
    ///
    /// Called by [`ReverseRequestHandler`] implementations (or the background
    /// reader in T12) after processing an inbound CLI request.
    ///
    /// # Errors
    ///
    /// Propagates errors from [`write`].
    ///
    /// [`write`]: Transport::write
    #[allow(dead_code)]
    pub(crate) async fn send_response(&self, id: JsonRpcId, result: Value) -> Result<()> {
        let response = JsonRpcResponse::success(id, result);
        let line = serde_json::to_string(&response)?;
        self.write(&line).await
    }

    // ── Internal helpers (used by T12 background reader) ─────────────────────

    /// Register the reverse-request handler.
    ///
    /// Must be called before [`connect`] so the handler is available when the
    /// background reader starts.
    ///
    /// [`connect`]: Transport::connect
    pub(crate) async fn set_reverse_handler(&self, handler: Arc<dyn ReverseRequestHandler>) {
        // `reverse_handler` is `Arc<Mutex<...>>` — the Arc is dereffed transparently.
        *self.reverse_handler.lock().await = Some(handler);
    }

    /// Allocate the next request ID without issuing a request.
    ///
    /// Useful when the caller needs to pre-allocate an ID for manual request
    /// construction outside of [`send_request`].
    ///
    /// [`send_request`]: GeminiTransport::send_request
    #[allow(dead_code)]
    pub(crate) fn next_id(&self) -> JsonRpcId {
        JsonRpcId::Number(self.next_request_id.fetch_add(1, Ordering::Relaxed))
    }

    /// Route an incoming JSON-RPC response to the matching pending-request
    /// oneshot channel.
    ///
    /// Called by the background reader (T12) for every response message.
    /// Logs a warning when no matching pending request is found.
    #[allow(dead_code)]
    pub(crate) fn route_response(&self, response: JsonRpcResponse) {
        if let Some((_, tx)) = self.pending_requests.remove(&response.id) {
            // Receiver may have been dropped if the caller gave up; that is fine.
            let _ = tx.send(response);
        } else {
            tracing::warn!(
                id = %response.id,
                "Received JSON-RPC response for unknown request ID — dropping"
            );
        }
    }

    /// Clone the message sender so the background reader (T12) can feed values
    /// into the channel without consuming the sender.
    ///
    /// Returns `None` if `connect()` has not been called yet.
    #[allow(dead_code)]
    pub(crate) async fn take_message_tx(&self) -> Option<mpsc::Sender<Result<Value>>> {
        // Clone rather than take so multiple readers can get a sender if needed.
        self.message_tx.lock().await.clone()
    }

    /// Return a reference-counted handle to the reverse-request handler, if any.
    #[allow(dead_code)]
    pub(crate) async fn get_reverse_handler(&self) -> Option<Arc<dyn ReverseRequestHandler>> {
        self.reverse_handler.lock().await.clone()
    }

    // ── Background reader helpers (T12) ───────────────────────────────────────

    /// Write a line to stdin — used by the background reader task to send
    /// reverse-request responses back to the Gemini CLI process.
    ///
    /// The newline is appended automatically and the write is flushed immediately.
    ///
    /// # Errors
    ///
    /// - [`Error::NotConnected`] — stdin is `None` (not yet connected or already closed).
    /// - [`Error::Io`] — the write or flush failed.
    ///
    /// [`Error::NotConnected`]: crate::Error::NotConnected
    /// [`Error::Io`]: crate::Error::Io
    async fn write_line_to_stdin(stdin: &Mutex<Option<ChildStdin>>, data: &str) -> Result<()> {
        let mut guard = stdin.lock().await;
        let stdin_handle = guard.as_mut().ok_or(crate::Error::NotConnected)?;
        stdin_handle.write_all(data.as_bytes()).await?;
        stdin_handle.write_all(b"\n").await?;
        stdin_handle.flush().await?;
        Ok(())
    }

    /// Spawn the background stdout-reader task.
    ///
    /// This is a static/associated function (not `&self`) because the spawned
    /// task must own its resources independently — it cannot hold a reference to
    /// the `GeminiTransport` struct, which would violate lifetime constraints
    /// across the `tokio::spawn` boundary.
    ///
    /// # Routing logic
    ///
    /// Each NDJSON line is parsed as JSON and classified:
    ///
    /// - **Response** (`id` present, `method` absent): matched to a pending
    ///   request via `pending_requests` and delivered over its oneshot channel.
    /// - **Notification** (`method` present, `id` absent): forwarded to the
    ///   consumer via `message_tx`.
    /// - **Request** (both `id` and `method` present): a reverse request from
    ///   the CLI. Dispatched to `reverse_handler`; the result is written back
    ///   to stdin as a JSON-RPC response.
    pub(crate) fn spawn_reader(
        pending_requests: Arc<DashMap<JsonRpcId, oneshot::Sender<JsonRpcResponse>>>,
        reverse_handler: Arc<Mutex<Option<Arc<dyn ReverseRequestHandler>>>>,
        stdin_writer: Arc<Mutex<Option<ChildStdin>>>,
        message_tx: mpsc::Sender<Result<Value>>,
        stdout: tokio::process::ChildStdout,
    ) -> JoinHandle<()> {
        tokio::spawn(async move {
            use tokio::io::{AsyncBufReadExt, BufReader};

            let reader = BufReader::new(stdout);
            let mut lines = reader.lines();

            while let Ok(Some(line)) = lines.next_line().await {
                let line = line.trim().to_string();
                if line.is_empty() {
                    continue;
                }

                tracing::trace!(line = %line, "reader: received line");

                // Diagnostic: show raw lines for debugging tool-call flows.
                if std::env::var("GEMINI_SDK_DEBUG").is_ok() {
                    let preview = if line.len() > 200 { &line[..200] } else { &line };
                    eprintln!("[SDK reader] raw: {preview}");
                }

                // Parse as JSON; forward a parse error to the consumer stream.
                let value: Value = match serde_json::from_str(&line) {
                    Ok(v) => v,
                    Err(e) => {
                        tracing::warn!(error = %e, line = %line, "reader: JSON parse error");
                        let _ = message_tx
                            .send(Err(crate::Error::ParseError {
                                message: e.to_string(),
                                line: line.clone(),
                            }))
                            .await;
                        continue;
                    }
                };

                // Classify and route the message.
                let kind = crate::jsonrpc::JsonRpcMessage::classify(&value);
                if std::env::var("GEMINI_SDK_DEBUG").is_ok() {
                    let method = value.get("method").and_then(|m| m.as_str()).unwrap_or("(none)");
                    let has_id = value.get("id").is_some();
                    eprintln!("[SDK reader] classify={kind:?} method={method} has_id={has_id}");
                }
                match kind {
                    Ok(crate::jsonrpc::MessageKind::Response) => {
                        // Extract the response ID.
                        let id = match value.get("id") {
                            Some(raw_id) => {
                                match serde_json::from_value::<JsonRpcId>(raw_id.clone()) {
                                    Ok(id) => id,
                                    Err(e) => {
                                        tracing::warn!(
                                            error = %e,
                                            "reader: failed to parse response ID"
                                        );
                                        continue;
                                    }
                                }
                            }
                            None => continue,
                        };

                        // Route to the waiting caller via its oneshot.
                        if let Some((_, tx)) = pending_requests.remove(&id) {
                            match serde_json::from_value::<JsonRpcResponse>(value) {
                                Ok(resp) => {
                                    // Receiver may have been dropped (caller gave up) — ok.
                                    let _ = tx.send(resp);
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        error = %e,
                                        "reader: failed to deserialize response body"
                                    );
                                }
                            }
                        } else {
                            tracing::warn!(
                                id = %id,
                                "reader: response for unknown request ID — dropping"
                            );
                        }
                    }

                    Ok(crate::jsonrpc::MessageKind::Notification) => {
                        // Forward notification to the consumer stream.
                        if message_tx.send(Ok(value)).await.is_err() {
                            tracing::debug!("reader: message channel closed, stopping");
                            break;
                        }
                    }

                    Ok(crate::jsonrpc::MessageKind::Request) => {
                        // Reverse request from the CLI (e.g. permission approval).
                        if std::env::var("GEMINI_SDK_DEBUG").is_ok() {
                            eprintln!("[SDK reader] REVERSE REQUEST detected — dispatching to handler");
                        }
                        let request: JsonRpcRequest = match serde_json::from_value(value) {
                            Ok(r) => r,
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "reader: failed to deserialize reverse request"
                                );
                                continue;
                            }
                        };

                        let request_id = request.id.clone();
                        let method = request.method.clone();
                        let params = request.params.unwrap_or(Value::Null);

                        // Snapshot the handler Arc without holding the lock across await.
                        let maybe_handler: Option<Arc<dyn ReverseRequestHandler>> =
                            reverse_handler.lock().await.clone();

                        let response = if let Some(h) = maybe_handler {
                            let result = h.handle_permission_request(request_id.clone(), params).await;
                            JsonRpcResponse::success(request_id, result)
                        } else {
                            tracing::warn!(
                                method = %method,
                                "reader: no reverse handler registered, responding with error"
                            );
                            JsonRpcResponse::error(
                                request_id,
                                crate::jsonrpc::JsonRpcError {
                                    code: crate::jsonrpc::error_codes::METHOD_NOT_FOUND,
                                    message: format!("No handler registered for '{method}'"),
                                    data: None,
                                },
                            )
                        };
                        let response_line = match serde_json::to_string(&response) {
                            Ok(s) => s,
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "reader: failed to serialize reverse-request response"
                                );
                                continue;
                            }
                        };

                        if std::env::var("GEMINI_SDK_DEBUG").is_ok() {
                            eprintln!("[SDK reader] SENDING reverse response: {response_line}");
                        }

                        if let Err(e) =
                            Self::write_line_to_stdin(&stdin_writer, &response_line).await
                        {
                            tracing::warn!(
                                error = %e,
                                "reader: failed to write reverse-request response to stdin"
                            );
                        } else if std::env::var("GEMINI_SDK_DEBUG").is_ok() {
                            eprintln!("[SDK reader] reverse response SENT OK");
                        }
                    }

                    Err(e) => {
                        tracing::warn!(error = %e, "reader: invalid JSON-RPC message shape");
                    }
                }
            }

            tracing::debug!("reader: stdout closed, reader task exiting");
        })
    }
}

// ── Transport impl ───────────────────────────────────────────────────────────

#[async_trait]
impl Transport for GeminiTransport {
    /// Spawn `gemini --experimental-acp` and start the background reader task.
    ///
    /// This implementation:
    /// 1. Spawns the subprocess with piped stdio.
    /// 2. Optionally spawns a stderr-draining task (feeds `stderr_callback`).
    /// 3. Spawns the stdout reader task via [`GeminiTransport::spawn_reader`],
    ///    which routes JSON-RPC responses, notifications, and reverse requests.
    ///
    /// # Errors
    ///
    /// Returns [`Error::SpawnFailed`] when `tokio::process::Command::spawn`
    /// fails (e.g. binary not found, permission denied).
    /// Returns [`Error::Transport`] when stdio handles cannot be captured.
    ///
    /// [`Error::SpawnFailed`]: crate::Error::SpawnFailed
    /// [`Error::Transport`]: crate::Error::Transport
    async fn connect(&self) -> Result<()> {
        use tokio::process::Command;

        let mut cmd = Command::new(&self.cli_path);
        cmd.args(&self.args)
            .current_dir(&self.cwd)
            .envs(&self.env)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        let mut child = cmd.spawn().map_err(crate::Error::SpawnFailed)?;

        // Take stdio handles before moving `child` into the process lock.
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| crate::Error::Transport("Failed to capture stdout".to_string()))?;
        let child_stdin = child
            .stdin
            .take()
            .ok_or_else(|| crate::Error::Transport("Failed to capture stdin".to_string()))?;
        let stderr = child.stderr.take();

        // Store the process and stdin.
        *self.process.lock().unwrap() = Some(child);
        *self.stdin.lock().await = Some(child_stdin);

        // Spawn optional stderr drainer.
        if let (Some(stderr_stream), Some(cb)) = (stderr, &self.stderr_callback) {
            let cb = Arc::clone(cb);
            tokio::spawn(async move {
                use tokio::io::{AsyncBufReadExt, BufReader};
                let mut lines = BufReader::new(stderr_stream).lines();
                while let Ok(Some(line)) = lines.next_line().await {
                    cb(line);
                }
            });
        }

        // Obtain the message sender (cloned so the field retains its copy).
        let message_tx = self
            .message_tx
            .lock()
            .await
            .clone()
            .ok_or_else(|| crate::Error::Transport("Message channel not initialized".to_string()))?;

        // Spawn the background stdout reader task.
        // Each argument is an Arc clone — the task takes full ownership of these
        // references, allowing it to outlive the `connect()` call safely.
        let handle = Self::spawn_reader(
            Arc::clone(&self.pending_requests),
            Arc::clone(&self.reverse_handler),
            Arc::clone(&self.stdin),
            message_tx,
            stdout,
        );
        *self.reader_handle.lock().await = Some(handle);

        self.ready.store(true, Ordering::Release);
        Ok(())
    }

    async fn write(&self, data: &str) -> Result<()> {
        let mut guard = self.stdin.lock().await;
        let stdin = guard.as_mut().ok_or(crate::Error::NotConnected)?;
        stdin.write_all(data.as_bytes()).await?;
        stdin.write_all(b"\n").await?;
        stdin.flush().await?;
        Ok(())
    }

    /// Return a stream of parsed JSON values from the subprocess stdout.
    ///
    /// The stream is backed by the internal `mpsc` channel fed by the background
    /// reader task spawned in [`connect`]. The receiver is consumed on the first
    /// call; subsequent calls return an empty stream and log a warning.
    ///
    /// # Design
    ///
    /// `read_messages` takes the `mpsc::Receiver` out of `message_rx` using
    /// `try_lock` (non-blocking). This avoids holding the lock across an await
    /// boundary while remaining safe — the receiver is moved into the stream,
    /// which owns it for its lifetime.
    ///
    /// [`connect`]: Transport::connect
    fn read_messages(&self) -> Pin<Box<dyn Stream<Item = Result<Value>> + Send>> {
        // Try to take the receiver without blocking. `try_lock` is safe here
        // because `read_messages` is synchronous (no await points), so we
        // cannot be in the middle of another async operation on this mutex.
        let rx = self
            .message_rx
            .try_lock()
            .ok()
            .and_then(|mut guard| guard.take());

        match rx {
            Some(rx) => Box::pin(async_stream::stream! {
                // Wrap the receiver in a minimal stream that yields until
                // the sender (reader task) is dropped.
                let mut receiver = rx;
                while let Some(item) = receiver.recv().await {
                    yield item;
                }
            }),
            None => {
                tracing::warn!(
                    "read_messages() called after the receiver was already consumed \
                     or the mutex is contended; returning empty stream"
                );
                // Return an immediately-terminating stream by wrapping an empty
                // futures stream. `futures_core::stream::empty()` cannot be used
                // here because its `Item` type requires explicit annotation that
                // conflicts with the `Pin<Box<dyn Stream<...>>>` return; we use
                // `async_stream` with an explicit type-annotated yield that is
                // guarded to never execute.
                let empty: Vec<Result<Value>> = Vec::new();
                Box::pin(async_stream::stream! {
                    for item in empty {
                        yield item;
                    }
                })
            }
        }
    }

    async fn end_input(&self) -> Result<()> {
        // Drop the ChildStdin handle; the subprocess sees EOF on its stdin.
        *self.stdin.lock().await = None;
        Ok(())
    }

    async fn interrupt(&self) -> Result<()> {
        #[cfg(unix)]
        {
            // Lock for the minimum time necessary — just to read the PID.
            let pid = {
                let process = self.process.lock().unwrap();
                process.as_ref().and_then(|c| c.id())
            };
            if let Some(pid) = pid {
                let _ = nix::sys::signal::kill(
                    nix::unistd::Pid::from_raw(pid as i32),
                    nix::sys::signal::Signal::SIGINT,
                );
            }
        }
        #[cfg(windows)]
        {
            // Send CTRL_BREAK_EVENT to the process group so the subprocess can
            // handle a graceful shutdown. Uses GenerateConsoleCtrlEvent from the
            // Win32 Console API (windows-sys dependency already declared).
            let pid = {
                let process = self.process.lock().unwrap();
                process.as_ref().and_then(|c| c.id())
            };
            if let Some(pid) = pid {
                unsafe {
                    let _ = windows_sys::Win32::System::Console::GenerateConsoleCtrlEvent(
                        windows_sys::Win32::System::Console::CTRL_BREAK_EVENT,
                        pid,
                    );
                }
            }
        }
        Ok(())
    }

    fn is_ready(&self) -> bool {
        self.ready.load(Ordering::Relaxed)
    }

    async fn close(&self) -> Result<Option<i32>> {
        // 1. Cancel the background reader task.
        if let Some(handle) = self.reader_handle.lock().await.take() {
            handle.abort();
        }

        // 2. Close stdin (sends EOF to the subprocess).
        self.end_input().await?;

        // 3. Mark as not ready.
        self.ready.store(false, Ordering::Relaxed);

        // 4. Extract the child *before* any await point to avoid holding the
        //    std::sync::MutexGuard across an await (which would be Undefined
        //    Behaviour-adjacent and triggers the "MutexGuard is not Send" lint).
        let child = {
            let mut process = self.process.lock().unwrap();
            process.take()
        }; // Guard is dropped here, before we hit `.await`.

        if let Some(mut child) = child {
            if let Some(duration) = self.close_timeout {
                match tokio::time::timeout(duration, child.wait()).await {
                    Ok(Ok(status)) => return Ok(status.code()),
                    Ok(Err(e)) => return Err(e.into()),
                    Err(_) => {
                        // Timed out waiting for clean exit — kill the process.
                        tracing::warn!(
                            timeout_ms = duration.as_millis() as u64,
                            "close() timed out waiting for subprocess; killing"
                        );
                        child.start_kill().ok();
                        let status = child.wait().await?;
                        return Ok(status.code());
                    }
                }
            } else {
                let status = child.wait().await?;
                return Ok(status.code());
            }
        }

        Ok(None)
    }
}

// ── Drop Safety ──────────────────────────────────────────────────────────────

impl Drop for GeminiTransport {
    /// Best-effort subprocess cleanup on drop.
    ///
    /// Sends `SIGKILL` (Unix) / `TerminateProcess` (Windows) to the subprocess
    /// if it is still running. This is a fire-and-forget best-effort; errors
    /// (e.g. the process already exited) are silently ignored.
    fn drop(&mut self) {
        if let Ok(mut process) = self.process.lock() {
            if let Some(ref mut child) = *process {
                let _ = child.start_kill();
            }
        }
    }
}

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;

    /// Return a minimal transport configured with a nonexistent binary so tests
    /// never accidentally spawn a real process.
    fn make_transport() -> GeminiTransport {
        GeminiTransport::new(
            PathBuf::from("/nonexistent/gemini"),
            vec!["--experimental-acp".to_string()],
            PathBuf::from("/tmp"),
            HashMap::new(),
            None,
            None,
        )
    }

    // ── Construction ─────────────────────────────────────────────────────────

    #[test]
    fn test_transport_new_does_not_panic() {
        // Constructing the transport must not panic regardless of binary path.
        let t = make_transport();
        assert_eq!(t.cli_path, PathBuf::from("/nonexistent/gemini"));
        assert!(!t.args.is_empty());
    }

    // ── is_ready ─────────────────────────────────────────────────────────────

    #[test]
    fn test_is_not_ready_before_connect() {
        let t = make_transport();
        assert!(
            !t.is_ready(),
            "transport must not report ready before connect() is called"
        );
    }

    // ── next_id ──────────────────────────────────────────────────────────────

    #[test]
    fn test_next_id_increments_monotonically() {
        let t = make_transport();
        let id0 = t.next_id();
        let id1 = t.next_id();
        let id2 = t.next_id();

        let n = |id: &JsonRpcId| match id {
            JsonRpcId::Number(n) => *n,
            JsonRpcId::String(_) => panic!("expected Number variant"),
        };

        assert!(
            n(&id0) < n(&id1) && n(&id1) < n(&id2),
            "IDs must be strictly increasing: {id0}, {id1}, {id2}"
        );
    }

    #[test]
    fn test_next_id_starts_at_one() {
        let t = make_transport();
        let id = t.next_id();
        match id {
            JsonRpcId::Number(n) => assert_eq!(n, 1, "first ID must be 1"),
            JsonRpcId::String(_) => panic!("expected Number variant"),
        }
    }

    // ── write (requires stdin) ────────────────────────────────────────────────

    #[tokio::test]
    async fn test_write_returns_not_connected_when_stdin_absent() {
        let t = make_transport();
        // `stdin` is None before connect() — write should return NotConnected.
        let result = t.write("{}").await;
        assert!(
            result.is_err(),
            "write must fail when transport is not connected"
        );
        assert!(
            matches!(result.unwrap_err(), crate::Error::NotConnected),
            "error must be NotConnected"
        );
    }

    // ── end_input ────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_end_input_clears_stdin() {
        let t = make_transport();
        // Simulate a connected state by manually inserting a fake stdin.
        // We use a pipe so ChildStdin would hold the write end; we mimic this
        // by checking that None is set after end_input regardless of initial state.
        // Since we cannot construct a ChildStdin directly in a unit test, we
        // verify the None-path: calling end_input on an already-None stdin is a no-op.
        t.end_input().await.expect("end_input must not fail on None stdin");
        let guard = t.stdin.lock().await;
        assert!(guard.is_none(), "stdin must be None after end_input");
    }

    // ── route_response ───────────────────────────────────────────────────────

    #[test]
    fn test_route_response_resolves_pending_request() {
        let t = make_transport();
        let id = JsonRpcId::Number(42);
        let (tx, mut rx) = oneshot::channel();
        t.pending_requests.insert(id.clone(), tx);

        let response = JsonRpcResponse::success(id, serde_json::json!({"ok": true}));
        t.route_response(response);

        // The oneshot should now be resolved.
        let received = rx.try_recv().expect("response must be available immediately");
        assert!(received.result.is_some());
        assert_eq!(received.result.unwrap(), serde_json::json!({"ok": true}));
    }

    #[test]
    fn test_route_response_unknown_id_does_not_panic() {
        let t = make_transport();
        // Routing a response for an ID nobody registered must not panic.
        let id = JsonRpcId::Number(9999);
        let response = JsonRpcResponse::success(id, serde_json::Value::Null);
        t.route_response(response); // Must not panic.
    }

    // ── read_messages ────────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_read_messages_yields_sent_items() {
        use tokio_stream::StreamExt;

        let t = make_transport();

        // Grab the sender and send a value before consuming the stream.
        let tx = t.take_message_tx().await.expect("sender must be Some");
        tx.send(Ok(serde_json::json!({"hello": "world"})))
            .await
            .expect("send must succeed");
        // Drop the sender so the stream terminates naturally.
        drop(tx);
        // Also drop the field copy so no senders remain.
        *t.message_tx.lock().await = None;

        let mut stream = t.read_messages();
        let item = stream.next().await;
        assert!(item.is_some(), "stream must yield the sent item");
        let value = item.unwrap().expect("item must be Ok");
        assert_eq!(value, serde_json::json!({"hello": "world"}));

        // After the only sender is dropped the stream must terminate.
        let end = stream.next().await;
        assert!(end.is_none(), "stream must terminate after all senders are dropped");
    }

    #[tokio::test]
    async fn test_read_messages_second_call_returns_empty_stream() {
        use tokio_stream::StreamExt;

        let t = make_transport();
        // First call consumes the receiver.
        let _stream1 = t.read_messages();
        // Second call must not panic and must return an empty stream.
        let mut stream2 = t.read_messages();
        let item = stream2.next().await;
        assert!(
            item.is_none(),
            "second read_messages() call must return an empty stream"
        );
    }

    // ── take_message_tx ──────────────────────────────────────────────────────

    #[tokio::test]
    async fn test_take_message_tx_returns_sender() {
        let t = make_transport();
        let sender = t.take_message_tx().await;
        assert!(sender.is_some(), "message_tx must be Some after construction");
    }

    #[tokio::test]
    async fn test_take_message_tx_is_reentrant() {
        // Multiple calls must return a clone each time (not consume the sender).
        let t = make_transport();
        let s1 = t.take_message_tx().await;
        let s2 = t.take_message_tx().await;
        assert!(s1.is_some() && s2.is_some(), "sender must be clonable");
    }
}