forge_sdk/

client.rs

//! Public [`Client`] - the entry point consumers hold.
//!
//! `Client` is a cheap-clone handle (`Arc`-backed) over a background
//! reader task that owns the `claude` subprocess. All methods take
//! `&self`, so the same `Client` can be passed by reference into
//! multiple tasks. The events stream is returned alongside the Client
//! at spawn time as a [`ClientEvents`] receiver - single-consumer,
//! independent of the writer-side handle.
//!
//! Internally:
//! - The reader task pulls lines from the subprocess, decodes them,
//!   pushes regular [`Message`]s onto an mpsc that backs the returned
//!   [`ClientEvents`], dispatches inbound `control_request`s on
//!   detached tasks, and routes outbound `control_response`s to
//!   per-request oneshots so [`Client::send_control`] callers can
//!   `await` their typed reply.
//! - The writer half is a clonable
//!   [`SharedWriter`](crate::transport::process::SharedWriter) cloned
//!   from [`Subprocess`]; outbound
//!   writes go through it without contending on `&mut self`.

pub(crate) mod control_dispatch;
mod control_send;
pub(crate) mod runtime;

pub(crate) use control_dispatch::ControlDispatchHandle;

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tokio::sync::{Mutex, mpsc, oneshot};
use tokio::task::JoinHandle;
use tracing::debug;

use crate::Error;
use crate::client::runtime::{PendingControls, SharedSessionId, excerpt, spawn_reader_task};
use crate::mcp::orchestration::McpHosts;
use crate::options::Options;
use crate::transport::codec::{DecodedLine, decode_dispatch};
use crate::transport::process::Subprocess;
use forge_primitives::Message;

/// Stream of regular [`Message`]s produced by the reader task.
/// Returned alongside [`Client`] from [`Client::spawn`]. Single-consumer;
/// the writer-side [`Client`] is `Clone` and can move freely between
/// tasks while one task owns the receiver.
pub type ClientEvents = mpsc::UnboundedReceiver<Result<Message, Error>>;

/// Removes a `pending_controls` entry when the awaiting future is
/// dropped without a response having been routed - the drop guard
/// `send_control` parks in its future.
struct PendingGuard {
    pending: PendingControls,
    request_id: String,
    defused: bool,
}
impl Drop for PendingGuard {
    fn drop(&mut self) {
        if self.defused {
            return;
        }
        match self.pending.try_lock() {
            Ok(mut pending) => {
                pending.remove(&self.request_id);
            }
            // The mutex is momentarily held (the reader routing a
            // response); the removal must still land, so finish it on a
            // task once the holder releases.
            Err(_) if tokio::runtime::Handle::try_current().is_ok() => {
                let pending = Arc::clone(&self.pending);
                let request_id = self.request_id.clone();
                tokio::spawn(async move {
                    pending.lock().await.remove(&request_id);
                });
            }
            Err(_) => {
                tracing::debug!(
                    target: crate::logging::targets::BRIDGE_LIFECYCLE,
                    request_id = %self.request_id,
                    "pending_controls mutex contended outside a runtime; entry removal skipped",
                );
            }
        }
    }
}

/// Budget for the `initialize` handshake inside [`Client::spawn`]. A
/// wedged CLI (alive, stdout open, never answering) would otherwise
/// park the caller forever with the child leaked.
///
/// Published because a caller that waits on a spawn is waiting on this
/// budget: a wait shorter than it fails a machine that is merely slow,
/// so a test derives its own wait from this number rather than
/// hand-picking one under it.
pub const INIT_HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);

/// An active `claude` binary subprocess.
///
/// Construct via [`spawn`](Self::spawn). The init handshake
/// (`initialize` `control_request` + drained response) runs inside
/// `spawn`; the `system/init` frame is consumed and its session id
/// cached on [`session_id()`](Self::session_id). The returned
/// [`ClientEvents`] receiver yields the clean conversational stream
/// starting with hook/assistant/user frames.
///
/// `Client` is `Clone`. All clones share the same underlying
/// subprocess, reader task, and pending-control map. Cloning is
/// cheap (an `Arc` increment); pass `Client` by value or reference
/// freely.
#[derive(Clone)]
pub struct Client {
    inner: Arc<ClientInner>,
}

struct ClientInner {
    /// Cloned writer (mpsc-backed via the transport's writer task).
    writer: std::sync::Arc<crate::transport::process::SharedWriter>,
    /// PID of the spawned `claude` child captured at spawn time.
    /// Stable for the lifetime of the subprocess; consumers (e.g. the
    /// Inspector pane's OS-level process walker) cache snapshots
    /// keyed off this value. `None` when the transport couldn't
    /// report a PID (in practice never for a live subprocess, but
    /// the optional makes the test stubs and post-disconnect path
    /// total).
    claude_pid: Option<u32>,
    /// Cached response from the `initialize` `control_request` -
    /// populated during spawn and never mutated afterwards.
    initialization_result: Option<serde_json::Value>,
    /// Captured `system/init` payload (`model`, `tools`, `mcp_servers`,
    /// `slash_commands`, …). The CLI emits this once after init and the
    /// SDK strips it from the user-visible `Message` stream; cache it
    /// here so `forge-agent` can answer footer / slash queries without
    /// re-running the handshake.
    cached_init_data: Option<serde_json::Value>,
    /// Captured session id. The reader task updates it as messages
    /// arrive; consumers read the current value via
    /// [`Client::session_id`].
    session_id: SharedSessionId,
    /// In-flight outbound `control_request`s waiting on responses.
    /// The reader task routes incoming `control_response`s here.
    pending_controls: PendingControls,
    /// Reader task handle. Awaited on [`Client::disconnect`].
    reader_task: Mutex<Option<JoinHandle<()>>>,
    /// Shutdown signal for the reader task. `take()`'d on the first
    /// [`Client::disconnect`] call; subsequent calls are no-ops.
    shutdown_tx: Mutex<Option<oneshot::Sender<()>>>,
    /// Set by [`Client::disconnect`] before the
    /// shutdown signal, so the stream's end can be read as "told to
    /// stop" rather than pipe death.
    disconnect_requested: AtomicBool,
}

impl std::fmt::Debug for Client {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Client")
            .field("session_id", &"<shared>")
            .field(
                "initialization_result",
                &self.inner.initialization_result.as_ref().map(|_| "<cached>"),
            )
            .finish_non_exhaustive()
    }
}

impl Client {
    /// Spawn `claude` with the given options and run the init handshake.
    ///
    /// Wire-recording for conformance baselines is configured via
    /// [`Options::tee_inbound`](crate::Options) /
    /// [`Options::tee_outbound`](crate::Options) callbacks - the
    /// `forge-test-harness` crate drives this.
    ///
    /// # Errors
    ///
    /// Any [`Error`] variant; see field docs.
    pub async fn spawn(options: Options) -> Result<(Self, ClientEvents), Error> {
        let mut sub = Subprocess::spawn(&options).await?;
        let writer = sub.clone_writer();
        // Capture the PID before the reader task consumes `sub`. It
        // doesn't change for the subprocess's lifetime so caching
        // once is correct.
        let claude_pid = sub.child_pid();
        // **The transport's own container, not a second one**: the writer
        // task names the session in a stall's record, and `/new` swaps the
        // occupant of this same handle, so both sides read the current id.
        let session_id = sub.session_id();
        let mcp_hosts =
            McpHosts::new(options.mcp_servers.clone(), options.external_mcp_servers.clone());
        let hook_registry = options.hooks.mint_registry();
        let hook_payload = hook_registry.to_initialize_payload();
        let hook_callbacks = hook_registry.by_id;

        // Build the dispatch handle now so the init loop can route
        // interleaved `control_request`s through it.
        let dispatch = ControlDispatchHandle::new(
            writer.clone(),
            options.can_use_tool.clone(),
            options.auto_approve_tool.clone(),
            mcp_hosts,
            hook_callbacks,
            session_id.clone(),
        );

        // Build the initialize control_request body. the CLI keeps
        // the same field-inclusion rules:
        // `hooks` always present (null when empty), `excludeDynamicSections`
        // only when explicitly set.
        let exclude_dynamic_sections = match &options.system_prompt {
            Some(crate::options::SystemPromptKind::Preset {
                exclude_dynamic_sections: Some(v),
                ..
            }) => Some(*v),
            _ => None,
        };
        let hooks_field = if hook_payload.as_object().is_some_and(serde_json::Map::is_empty) {
            serde_json::Value::Null
        } else {
            hook_payload
        };
        let mut init_body = serde_json::Map::new();
        init_body.insert("subtype".into(), serde_json::Value::String("initialize".into()));
        init_body.insert("hooks".into(), hooks_field);
        if let Some(flag) = exclude_dynamic_sections {
            init_body.insert("excludeDynamicSections".into(), serde_json::Value::Bool(flag));
        }
        let init_request_id = crate::request_id::next();
        let init_envelope = serde_json::json!({
            "type": "control_request",
            "request_id": init_request_id,
            "request": serde_json::Value::Object(init_body),
        });
        let mut init_line = serde_json::to_string(&init_envelope)
            .map_err(|e| Error::encode("initialize body", e))?;
        init_line.push('\n');

        // **Send initialize FIRST.** The CLI in stream-json interactive
        // mode is driven entirely by stdin: it will not emit `system/init`
        // until BOTH (a) an `initialize` control_request is received AND
        // (b) a user message arrives. It DOES emit a `control_response` to
        // the initialize right after hooks complete - that gives us the
        // server-info payload (commands, agents, models, account, pid)
        // without needing to drive a conversation yet.
        sub.write_line(&init_line).await?;

        let mut pre_init_messages: VecDeque<Message> = VecDeque::new();
        let mut cached_init_data: Option<serde_json::Value> = None;
        let mut line_number: u64 = 0;
        let initialization_result = tokio::time::timeout(INIT_HANDSHAKE_TIMEOUT, async {
            let response: Option<serde_json::Value> = loop {
                line_number += 1;
                let line = sub.read_line().await?.ok_or_else(|| Error::Connection {
                    reason: "transport closed stdout before initialize control_response".into(),
                })?;
                let value: serde_json::Value = serde_json::from_str(&line)
                    .map_err(|source| Error::JsonDecode { line: line_number, source })?;
                let ty = value.get("type").and_then(serde_json::Value::as_str).unwrap_or("");
                match ty {
                    "control_response" => {
                        let resp_request_id = value
                            .pointer("/response/request_id")
                            .and_then(serde_json::Value::as_str);
                        if resp_request_id != Some(&init_request_id) {
                            return Err(Error::message_parse(format!(
                                "init: control_response request_id mismatch \
                                     (expected {init_request_id:?}, got {resp_request_id:?}); \
                                     raw line: {}",
                                line.trim_end()
                            )));
                        }
                        let resp_subtype =
                            value.pointer("/response/subtype").and_then(serde_json::Value::as_str);
                        if resp_subtype == Some("success") {
                            let body = value
                                .pointer("/response/response")
                                .cloned()
                                .unwrap_or(serde_json::Value::Null);
                            break match body {
                                serde_json::Value::Null => None,
                                v => Some(v),
                            };
                        }
                        let err_msg = value
                            .pointer("/response/error")
                            .and_then(serde_json::Value::as_str)
                            .map_or_else(
                                || {
                                    format!(
                                        "no `error` string field; full response: {}",
                                        value.pointer("/response").map_or_else(
                                            || "<missing>".to_string(),
                                            ToString::to_string,
                                        )
                                    )
                                },
                                ToString::to_string,
                            );
                        return Err(Error::message_parse(format!("initialize failed: {err_msg}")));
                    }
                    "control_request" => {
                        // Interleaved CLI → SDK request during init -
                        // most commonly an MCP `mcp_message` bootstrap.
                        // Dispatch synchronously through the dispatch
                        // handle (the reader task isn't running yet, so
                        // there's no concurrent reader to worry about).
                        let req: crate::control::ControlRequest = serde_json::from_value(value)
                            .map_err(|e| {
                                Error::message_parse(format!(
                                    "line {line_number}: control_request decode: {e}"
                                ))
                            })?;
                        dispatch.dispatch(req).await?;
                    }
                    "control_cancel_request" => {
                        tracing::debug!(
                            line_number,
                            "control_cancel_request during init; nothing live to cancel"
                        );
                    }
                    _ => match decode_dispatch(&line, line_number) {
                        DecodedLine::Message(msg) => {
                            debug!(line_number, "buffering pre-init frame for caller");
                            // Capture session id off pre-init messages so
                            // callers reading session_id() right after
                            // spawn see a real value when the CLI happens
                            // to emit init early.
                            if let Some(id) = msg.session_id()
                                && !id.is_empty()
                            {
                                let mut current = session_id.write();
                                if current.is_empty() {
                                    *current = id.to_string();
                                }
                            }
                            // Drop `system/init` from the pre-init buffer -
                            // the CLI consumes it inside `query._fetch_init`
                            // and never surfaces it to callers; we mirror.
                            // Cache its `data` so `forge-agent` can read
                            // model / mcp / slash-command info off it.
                            if let Message::System { ref subtype, ref data, .. } = *msg
                                && subtype == "init"
                            {
                                cached_init_data = Some(data.clone());
                            } else {
                                pre_init_messages.push_back(*msg);
                            }
                        }
                        DecodedLine::Unknown { type_str, raw } => {
                            tracing::warn!(
                                type = %type_str,
                                raw = %raw,
                                line = line_number,
                                "unknown top-level type during init - buffering as Message::Unknown"
                            );
                            pre_init_messages.push_back(Message::Unknown { type_str, raw });
                        }
                        DecodedLine::Malformed { line: line_no, reason } => {
                            // Pre-init frames stay strict: unparsable
                            // output before initialize completes is a
                            // broken spawn, not a degradable stream.
                            return Err(Error::message_parse(format!(
                                "line {line_no}: {reason}; raw line: {}",
                                excerpt(&line, 160)
                            )));
                        }
                        other => {
                            debug!(
                                line_number,
                                ?other,
                                "unexpected DecodedLine during init fallthrough; ignoring"
                            );
                        }
                    },
                }
            };
            Ok(response)
        })
        .await;
        let initialization_result = match initialization_result {
            Ok(result) => result?,
            Err(_) => {
                return Err(Error::Connection {
                    reason: format!(
                        "initialize handshake timed out after {}s; the CLI never answered the \
                         initialize control_request",
                        INIT_HANDSHAKE_TIMEOUT.as_secs()
                    ),
                });
            }
        };
        debug!("client init handshake complete");

        let pending_controls: PendingControls = Arc::new(Mutex::new(HashMap::new()));
        let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel();
        let (shutdown_tx, shutdown_rx) = oneshot::channel();

        let reader_task = spawn_reader_task(
            sub,
            dispatch,
            pending_controls.clone(),
            events_tx,
            pre_init_messages.into_iter().collect(),
            shutdown_rx,
        );

        let inner = Arc::new(ClientInner {
            writer,
            claude_pid,
            initialization_result,
            cached_init_data,
            session_id,
            pending_controls,
            reader_task: Mutex::new(Some(reader_task)),
            shutdown_tx: Mutex::new(Some(shutdown_tx)),
            disconnect_requested: AtomicBool::new(false),
        });

        Ok((Self { inner }, events_rx))
    }

    /// Cached response from the `initialize` `control_request` - holds
    /// the CLI's server capabilities, available commands, and output
    /// styles. Returns `None` when the CLI didn't attach a body to its
    /// initialize response.
    pub fn get_server_info(&self) -> Option<&serde_json::Value> {
        self.inner.initialization_result.as_ref()
    }

    /// OS-level PID of the spawned `claude` child, captured at spawn
    /// time. Stable for the lifetime of this `Client`. Used by
    /// downstream consumers (e.g. `forge_agent::env::processes`) to
    /// walk the child process tree for the Inspector pane's
    /// PROCESSES section.
    ///
    /// Returns `None` only when the transport couldn't report a PID
    /// at spawn - never happens for a real subprocess on supported
    /// platforms, but the `Option` keeps the test-stub path
    /// expressible.
    pub fn claude_pid(&self) -> Option<u32> {
        self.inner.claude_pid
    }

    /// SIGKILL the child's process group now, without the reader-task drain
    /// [`disconnect`](Self::disconnect) performs.
    ///
    /// For a session that has no task to run its exit path: dropping the
    /// handle alone kills only the direct child and leaves whatever the CLI
    /// already started behind.
    pub fn kill_process_group(&self) {
        let Some(pgid) = self.inner.claude_pid else { return };
        if let Some(error) = crate::transport::process::kill_group(pgid) {
            tracing::warn!(
                target: "forge_sdk",
                event_name = "subprocess_group_signal_failed",
                session = %self.session_id(),
                pgid,
                error = %error,
                "the session child's process group could not be signalled; \
                 a descendant may outlive the session"
            );
        }
    }

    /// Captured `system/init` payload - the CLI's first session-scoped
    /// frame, carrying `model`, `tools`, `mcp_servers`, `slash_commands`,
    /// `agents`, `skills`, etc. Stripped from the user-visible `Message`
    /// stream during init; cached here for callers that need the
    /// initial session context (e.g. forge-agent's status snapshot +
    /// slash autocomplete).
    pub fn initial_session_data(&self) -> Option<&serde_json::Value> {
        self.inner.cached_init_data.as_ref()
    }

    /// The session id captured from the init message. Returns an empty
    /// string until the first session-scoped frame arrives.
    pub fn session_id(&self) -> String {
        self.inner.session_id.read().clone()
    }

    /// Read the bare `apiKeySource` field from the cached
    /// `system/init` payload, returning a partial
    /// [`forge_primitives::AccountInfo`] (auth source only - no
    /// email/org/subscription). Returns `None` when the payload is
    /// absent (init not yet arrived), the field is missing, or the
    /// value is empty / `"none"`.
    pub fn account_info_from_init(&self) -> Option<forge_primitives::AccountInfo> {
        let data = self.inner.cached_init_data.as_ref()?;
        let api_key_source = data
            .get("apiKeySource")
            .and_then(serde_json::Value::as_str)
            .map(str::to_owned)
            .filter(|s| !s.is_empty() && s != "none")?;
        Some(forge_primitives::AccountInfo {
            api_key_source: Some(api_key_source.clone()),
            token_source: Some(api_key_source),
            ..forge_primitives::AccountInfo::default()
        })
    }

    /// Send a user prompt as a stream-json user turn.
    ///
    /// Mints the prompt's own id and returns it: the id is stamped on the
    /// frame, and the CLI echoes it on that prompt's `command_lifecycle`
    /// frames, so a caller that keeps it can key its own row to them.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on pipe write failure.
    pub async fn send_user_message(&self, prompt: &str) -> Result<String, Error> {
        let uuid = crate::request_id::next_prompt_id();
        self.send_user_message_under(prompt, &uuid).await?;
        Ok(uuid)
    }

    /// Send a user prompt under an id the caller minted.
    ///
    /// Use this when the id must be known before the write - a dispatcher
    /// that draws the prompt's row and reconciles against the lifecycle
    /// frames by id. Mint with
    /// [`crate::request_id::next_prompt_id`] and never reuse one: the CLI
    /// drops a repeated id's prompt silently.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on pipe write failure;
    /// [`Error::MessageParse`] on JSON serialization failure.
    pub async fn send_user_message_under(&self, prompt: &str, uuid: &str) -> Result<(), Error> {
        let session_id = self.session_id();
        let line = crate::transport::codec::encode_user_prompt(prompt, &session_id, uuid)?;
        self.inner.writer.write_line(&line).await
    }

    /// Send a user prompt with structured content blocks (text +
    /// image, etc.) as a stream-json user turn. Use this when the
    /// prompt is multi-modal; for plain text use
    /// [`send_user_message`](Self::send_user_message).
    ///
    /// `content` is forwarded verbatim as the message body's
    /// `content` field - callers must build CLI-compatible block
    /// objects (e.g. `{"type":"text","text":"..."}`,
    /// `{"type":"image","source":{"type":"base64","media_type":"image/png","data":"..."}}`).
    ///
    /// Mints and returns the prompt's id, as [`Self::send_user_message`] does.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on pipe write failure;
    /// [`Error::MessageParse`] on JSON serialization failure.
    pub async fn send_user_message_with_content(
        &self,
        content: &[serde_json::Value],
    ) -> Result<String, Error> {
        let uuid = crate::request_id::next_prompt_id();
        self.send_user_message_with_content_under(content, &uuid).await?;
        Ok(uuid)
    }

    /// Send structured content under an id the caller minted, as
    /// [`Self::send_user_message_under`] does.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on pipe write failure;
    /// [`Error::MessageParse`] on JSON serialization failure.
    pub async fn send_user_message_with_content_under(
        &self,
        content: &[serde_json::Value],
        uuid: &str,
    ) -> Result<(), Error> {
        let session_id = self.session_id();
        let line =
            crate::transport::codec::encode_user_prompt_with_content(content, &session_id, uuid)?;
        self.inner.writer.write_line(&line).await
    }

    /// Close the stdin side of the transport while keeping the client
    /// alive for reading. Signals the CLI that no more user messages
    /// are coming; the CLI drains in-flight turns, emits any final
    /// frames, and closes stdout. The events receiver returns `None`
    /// on EOF afterwards.
    ///
    /// # Errors
    ///
    /// [`Error::Io`] on write failure.
    pub async fn end_input(&self) -> Result<(), Error> {
        self.inner.writer.end_input().await
    }

    /// Graceful shutdown. Signals the reader task to stop, waits for
    /// it to drain and close the subprocess. Idempotent - subsequent
    /// calls (on this clone or any other) are no-ops.
    ///
    /// # Errors
    ///
    /// None today; the inner reader-task close is best-effort. Kept
    /// `Result` for forward-compat.
    pub async fn disconnect(&self) -> Result<(), Error> {
        // The flag goes up BEFORE the signal: a consumer watching the
        // event stream end must be able to tell "told to stop" (a
        // session swap or teardown, expected) from pipe death (a real
        // failure) without any new event field.
        self.inner.disconnect_requested.store(true, Ordering::Release);
        if let Some(tx) = self.inner.shutdown_tx.lock().await.take() {
            let _ = tx.send(());
        }
        if let Some(handle) = self.inner.reader_task.lock().await.take()
            && let Err(join_err) = handle.await
        {
            tracing::error!(
                target: "forge_sdk",
                error = %join_err,
                "reader task panicked during disconnect"
            );
        }
        Ok(())
    }

    /// Whether [`disconnect`](Self::disconnect) has been called on any
    /// clone of this client. Lets a consumer distinguish an expected
    /// stream end (session swap, teardown) from the subprocess dying
    /// on its own.
    pub fn disconnect_requested(&self) -> bool {
        self.inner.disconnect_requested.load(Ordering::Acquire)
    }

    /// Internal: outbound `control_request` issuer. Inserts a oneshot
    /// into `pending_controls`, writes the envelope via the cloned
    /// writer, and awaits the reader task's routed response.
    ///
    /// If the awaiting future is dropped - a caller's timeout fired, or
    /// the task died - the oneshot sender would otherwise sit in
    /// `pending_controls` forever, one leak per dropped call (the
    /// footer's context-usage poll being the unbounded one). The
    /// `PendingGuard` returned in the future removes the entry whenever
    /// the future is dropped without a response having been routed; a
    /// late response then finds no entry, which the reader task already
    /// handles.
    pub(crate) async fn send_control(
        &self,
        subtype: &str,
        extra: serde_json::Value,
    ) -> Result<serde_json::Value, Error> {
        let request_id = crate::request_id::next();
        let mut request_body = serde_json::Map::new();
        request_body.insert("subtype".into(), serde_json::Value::String(subtype.to_string()));
        if let serde_json::Value::Object(extra_map) = extra {
            for (k, v) in extra_map {
                request_body.insert(k, v);
            }
        }
        let envelope = serde_json::json!({
            "type": "control_request",
            "request_id": request_id,
            "request": serde_json::Value::Object(request_body),
        });
        let mut line = serde_json::to_string(&envelope)
            .map_err(|e| Error::encode("control_request envelope", e))?;
        line.push('\n');

        let (resp_tx, resp_rx) = oneshot::channel();
        {
            let mut pending = self.inner.pending_controls.lock().await;
            pending.insert(request_id.clone(), resp_tx);
        }
        let mut guard = PendingGuard {
            pending: Arc::clone(&self.inner.pending_controls),
            request_id: request_id.clone(),
            defused: false,
        };
        if let Err(e) = self.inner.writer.write_line(&line).await {
            // Remove the pending entry on write failure so the EOF
            // drain doesn't fire on a request that never went out.
            let mut pending = self.inner.pending_controls.lock().await;
            pending.remove(&request_id);
            guard.defused = true;
            return Err(e);
        }
        let outcome = match resp_rx.await {
            Ok(outcome) => outcome,
            Err(_) => Err(Error::Connection {
                reason: format!("subprocess closed before {subtype} response"),
            }),
        };
        guard.defused = true;
        outcome
    }
}

#[cfg(test)]
mod tests {
    use super::{Client, PendingGuard};
    use crate::client::runtime::PendingControls;
    use crate::options::OptionsBuilder;
    use std::collections::HashMap;
    use std::sync::Arc;
    use tokio::sync::Mutex;

    fn control_mock_binary() -> String {
        concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fixtures/mock_claude_control.sh").to_owned()
    }

    /// A control request whose awaiting future is dropped (the bridge's
    /// 60s timeout firing) must not leave its `pending_controls` entry
    /// behind - a wedged CLI would otherwise accumulate one entry per
    /// context-usage poll, unbounded.
    #[tokio::test]
    async fn dropped_control_response_future_drains_the_pending_entry() {
        let opts = OptionsBuilder::new()
            .binary(control_mock_binary())
            .env("FORGED_MOCK_SKIP_SUBTYPE", "get_context_usage")
            .build();
        let (client, _events) = Client::spawn(opts).await.expect("spawn");

        let _ =
            tokio::time::timeout(std::time::Duration::from_millis(200), client.get_context_usage())
                .await;

        assert!(
            client.inner.pending_controls.lock().await.is_empty(),
            "the dropped request's pending entry must be removed"
        );
        client.disconnect().await.expect("disconnect");
    }

    /// The reader task can hold `pending_controls` at the instant a
    /// dropped future's guard runs; the removal must still land (via a
    /// spawned retry) rather than silently stranding a dead sender.
    #[tokio::test]
    async fn pending_guard_removal_survives_a_contended_mutex() {
        let pending: PendingControls = Arc::new(Mutex::new(HashMap::new()));
        let (tx, _rx) = tokio::sync::oneshot::channel::<crate::client::runtime::ControlOutcome>();
        pending.lock().await.insert("req-1".to_owned(), tx);

        let guard = PendingGuard {
            pending: Arc::clone(&pending),
            request_id: "req-1".to_owned(),
            defused: false,
        };
        // Hold the map across the drop so the guard's synchronous
        // try_lock loses the race.
        let held = pending.lock().await;
        drop(guard);
        drop(held);

        tokio::time::timeout(std::time::Duration::from_secs(1), async {
            while !pending.lock().await.is_empty() {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("removal retried after the lost race");
    }
}