forge_agent/

forge_sdk_bridge.rs

//! `ForgeSdkBridge` - in-process driver around a [`forge_sdk::Client`].
//!
//! Drives a [`forge_sdk::Client`] directly - no Node.js subprocess,
//! no NDJSON, no command queue. The bridge holds the spawned
//! `Arc<Client>` and dispatches each method as a direct call
//! (or a `tokio::spawn`'d async task when the method is
//! fire-and-forget). Synthesized events (Connected, `PermissionRequest`,
//! `McpSnapshot`, …) flow back through an `mpsc::UnboundedSender<AgentEvent>`
//! the bridge owns; consumers grab the matching receiver once via
//! [`ForgeSdkBridge::take_events`].
//!
//! ```text
//!     view                    ForgeSdkBridge                  forge_sdk::Client
//!      | trait method            |                                    |
//!      |------------------------>|  client.method().await             |
//!      |                         |----------------------------------->|
//!      |                         |                                    |
//!      |       AgentEvent        |  reader_loop / callbacks           |
//!      |<------ event_tx --------+<-----------------------------------|
//! ```

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;

use forge_sdk::Client;
use parking_lot::Mutex;
use serde_json::Value;
use tokio::sync::{mpsc, oneshot};
use tracing::Instrument;

use crate::client::{AgentEvent, SessionLaunchSettings, SpawnFailureKind};
use crate::forge_sdk_worker;
use crate::forge_sdk_worker::SessionId;
use forge_primitives::{PermissionMode, PermissionOutcome, QuestionOutcome};

/// Name the class of a spawn failure from the SDK error itself, so a
/// consumer never has to infer it from the rendered message.
fn spawn_failure_kind(err: &anyhow::Error) -> SpawnFailureKind {
    match err.downcast_ref::<forge_sdk::Error>() {
        Some(forge_sdk::Error::CwdNotFound { .. }) => SpawnFailureKind::WorkingDirMissing,
        _ => SpawnFailureKind::Unclassified,
    }
}

/// The turn-error event for a failure forge hit itself, classified where
/// the message is built: a reader branching on the class gets it without
/// searching the prose.
fn turn_error(session_id: String, message: String) -> AgentEvent {
    let class = crate::translate::error_handling::classify_turn_error(&message);
    AgentEvent::TurnError { session_id, message, class }
}

/// Sentinel `config_dir` for `ForgeSdkBridge` test stubs that never
/// exercise the path. Production code constructs the bridge with a
/// real account `config_dir`; tests that don't drive a session use
/// this path so the typed field stays non-optional.
#[cfg(any(test, feature = "testing"))]
const TESTING_STUB_CONFIG_DIR: &str = "/tmp/forge-testing-stub";

/// Pending permission responses keyed by `tool_use_id`. The
/// `can_use_tool` callback parks a oneshot here when the CLI asks;
/// dispatch drains it when the matching `permission_response` arrives
/// from a view.
pub(crate) type PendingResponses =
    Arc<Mutex<HashMap<String, oneshot::Sender<forge_primitives::PermissionDecision>>>>;

/// Pending question outcomes keyed by `tool_use_id`. The
/// `AskUserQuestion` driver in the `can_use_tool` callback parks a
/// fresh oneshot per question, emits a `QuestionRequest`, and awaits
/// the matching `question_response`.
pub(crate) type PendingQuestions = Arc<Mutex<HashMap<String, oneshot::Sender<QuestionOutcome>>>>;

/// In-process bridge wrapping a single [`forge_sdk::Client`].
///
/// Single instance per connection. The bridge owns the spawned
/// `forge_sdk::Client`, the `can_use_tool` parking lots, and the
/// outbound `AgentEvent` channel.
#[derive(Clone)]
pub struct ForgeSdkBridge {
    inner: Arc<BridgeInner>,
}

pub(crate) struct BridgeInner {
    /// Set after first `new_session` / `resume_session`; cleared on
    /// session replace or shutdown.
    client: Mutex<Option<Client>>,
    /// Bridge → consumer event emission channel. Cloned freely into the
    /// reader subtask + `can_use_tool` callback closures.
    event_tx: mpsc::UnboundedSender<AgentEvent>,
    /// Single-take receiver handed out via [`ForgeSdkBridge::take_events`].
    events_rx: Mutex<Option<mpsc::UnboundedReceiver<AgentEvent>>>,
    /// Permission round-trip parking lot.
    pub(crate) pending: PendingResponses,
    /// Question round-trip parking lot.
    pub(crate) pending_questions: PendingQuestions,
    /// Current session id, shared with the `can_use_tool` callback so
    /// permission/question events carry the right `session_id`.
    pub(crate) session_id_slot: Arc<Mutex<String>>,
    /// `<config_dir>` this bridge was bound to at construction time
    /// (workspace-driven - typically the picked account's
    /// `config_dir`). Threaded into the spawned `claude`
    /// subprocess as `CLAUDE_CONFIG_DIR` and consulted by every
    /// in-process accessor (oauth, settings, catalog) so they
    /// honour the bound account, not whatever `$CLAUDE_CONFIG_DIR`
    /// the parent shell happened to have.
    config_dir: PathBuf,
    /// Forge-internal account display name from `forge.toml`'s
    /// `[[accounts]]`, set by the workspace picker. `None` when
    /// `Agent::spawn` is called directly (tests, smoke). When
    /// present, read back through the handle's `display_name()` so a
    /// view can render which forge-account the bridge is bound to.
    display_name: Option<String>,
    /// Forge-workspace-supplied in-process MCP servers attached at
    /// every `spawn_session` call. Today this is the per-session
    /// `forge` MCP server, whose tool surface varies by session kind;
    /// future modules (worktree, memory, …) slot in alongside under
    /// the same `forge` server name. Cheap to clone - each entry
    /// is just a name + a few `Arc<dyn Tool>`s.
    extra_mcp_servers: Vec<(String, forge_sdk::mcp::McpServer)>,
    /// The session's resolved forge.toml env - `[env]` merged with
    /// `[accounts.env]` and the spawning project's
    /// env - stamped onto the spawned `claude`
    /// subprocess by `forge_sdk_worker::build_options_with_callback`.
    /// Empty when `Agent::spawn` is called directly without a
    /// workspace (tests, smoke) or when no table declares anything.
    env: HashMap<String, String>,
    /// Session ids whose context-usage probe is out, each against the instant
    /// its mark was taken, added when the ask goes and removed when its
    /// answer, error or timeout lands.
    ///
    /// A probe is answered inline over the CLI's whole transcript, and the
    /// socket asks on every read of a seat that reports no usage, so without
    /// this one unanswered seat parks a probe per read. A probe that times out
    /// emits nothing, so a caller that coalesced only on its own side would
    /// repeat forever.
    ///
    /// The instant bounds the mark the way the probe's own timeout cannot:
    /// every release lives inside the dispatched closure, so a closure that
    /// never runs leaves a mark nothing will ever clear. Any asker past
    /// [`ForgeSdkBridge::PROBE_MARK_DEADLINE`] reclaims it instead.
    context_probes_in_flight: Mutex<HashMap<String, Instant>>,
}

/// Holds a seat's probe mark while its probe is out, and releases it however
/// the probe ends: an answer, an error, a timeout, or a dispatch that never
/// reached a client at all.
///
/// A guard rather than a release at each exit, because the exits multiply - the
/// timeout returns early, and the closure is dropped uncalled when no client is
/// bound.
///
/// It releases only the mark it took: one reclaimed by a later ask belongs to
/// that ask's probe, and a late drop must not clear the live mark.
struct ProbeInFlight {
    inner: Arc<BridgeInner>,
    session_id: String,
    taken_at: Instant,
}

impl Drop for ProbeInFlight {
    fn drop(&mut self) {
        let mut marks = self.inner.context_probes_in_flight.lock();
        if marks.get(&self.session_id) == Some(&self.taken_at) {
            marks.remove(&self.session_id);
        }
    }
}

impl ForgeSdkBridge {
    /// Construct a fresh bridge bound to `config_dir` with an
    /// optional forge-account `display_name`. Both are stored as
    /// typed fields. `config_dir` is consulted by every in-process
    /// accessor (oauth, settings, catalog scans) and exported to
    /// the spawned `claude` subprocess as `CLAUDE_CONFIG_DIR`.
    /// `display_name`, when present, is read back through the
    /// handle's `display_name()`. The internal event channel
    /// is created here; consumers grab the receiver once via
    /// [`ForgeSdkBridge::take_events`].
    pub(crate) fn new(
        config_dir: PathBuf,
        display_name: Option<String>,
        extra_mcp_servers: Vec<(String, forge_sdk::mcp::McpServer)>,
        env: HashMap<String, String>,
    ) -> Self {
        let (event_tx, events_rx) = mpsc::unbounded_channel();
        Self {
            inner: Arc::new(BridgeInner {
                client: Mutex::new(None),
                event_tx,
                events_rx: Mutex::new(Some(events_rx)),
                pending: Arc::new(Mutex::new(HashMap::new())),
                pending_questions: Arc::new(Mutex::new(HashMap::new())),
                session_id_slot: Arc::new(Mutex::new(String::new())),
                config_dir,
                display_name,
                extra_mcp_servers,
                env,
                context_probes_in_flight: Mutex::new(HashMap::new()),
            }),
        }
    }

    /// The resolved forge.toml env to stamp onto every spawned
    /// `claude` subprocess. Cloned per `spawn_session` call, but the
    /// values were read from disk once at forge BOOT - a new bridge
    /// does not re-read them, so a forge.toml edit needs a forge
    /// restart, not a new session.
    pub(crate) fn env(&self) -> HashMap<String, String> {
        self.inner.env.clone()
    }

    /// Forge-workspace-supplied in-process MCP servers to attach to
    /// every spawned `claude` subprocess (e.g. the `forge` server
    /// carrying the peer-coordination tools - #114 v1). Cheap-clone
    /// via the `McpServer`'s `Arc<dyn Tool>` internals; called once
    /// per `spawn_session` invocation.
    pub(crate) fn extra_mcp_servers(&self) -> Vec<(String, forge_sdk::mcp::McpServer)> {
        self.inner.extra_mcp_servers.clone()
    }

    pub(crate) fn event_tx(&self) -> &mpsc::UnboundedSender<AgentEvent> {
        &self.inner.event_tx
    }

    pub(crate) fn inner_pending(&self) -> &PendingResponses {
        &self.inner.pending
    }

    pub(crate) fn inner_pending_questions(&self) -> &PendingQuestions {
        &self.inner.pending_questions
    }

    pub(crate) fn session_id_slot_arc(&self) -> &Arc<Mutex<String>> {
        &self.inner.session_id_slot
    }

    fn client(&self) -> Option<Client> {
        self.inner.client.lock().clone()
    }

    pub(crate) fn set_client(&self, client: Client) {
        *self.inner.client.lock() = Some(client);
    }

    pub(crate) fn clear_client(&self) -> Option<Client> {
        // Drain in-flight permission / question oneshots before the
        // client goes away. The SDK's detached `can_use_tool` /
        // `question` callback tasks are awaiting these receivers; if
        // we drop the client without resolving them they leak the
        // closure (which holds `Arc<event_tx>`, `Arc<PendingResponses>`,
        // …) and the matching workspace forwarders exit on a closed
        // channel without notifying the awaiter.
        for (_, tx) in self.inner.pending.lock().drain() {
            let _ = tx.send(forge_primitives::PermissionDecision::deny("session replaced"));
        }
        for (_, tx) in self.inner.pending_questions.lock().drain() {
            let _ = tx.send(forge_primitives::QuestionOutcome::Cancelled);
        }
        self.inner.client.lock().take()
    }

    /// Send an `AgentEvent::McpOperationError` to the consumer; log a
    /// `BRIDGE_LIFECYCLE` warn if the channel is closed (terminal
    /// event for a user-visible MCP failure - silent-drop on
    /// teardown race would lose the only path to surface).
    /// On send failure we destructure the unsent event back out of
    /// the `SendError` so the warn log carries `session_id`,
    /// `server_name`, and the underlying error text - the most
    /// actionable triage signals.
    fn emit_mcp_error_or_log(
        event_tx: &mpsc::UnboundedSender<AgentEvent>,
        session_id: String,
        operation: &'static str,
        server_name: Option<String>,
        error_msg: String,
    ) {
        let session_id_for_log = session_id.clone();
        let server_name_for_log = server_name.clone();
        let error_msg_for_log = error_msg.clone();
        let event = AgentEvent::McpOperationError {
            session_id,
            error: forge_primitives::McpOperationError {
                operation: operation.to_owned(),
                server_name,
                message: error_msg,
            },
        };
        if event_tx.send(event).is_err() {
            tracing::warn!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                session_id = %session_id_for_log,
                operation,
                server_name = ?server_name_for_log,
                error_msg = %error_msg_for_log,
                "event channel closed; McpOperationError dropped",
            );
        }
    }

    /// Spawn a fire-and-forget client call. Logs and drops on failure.
    fn dispatch<F, Fut>(&self, label: &'static str, f: F) -> anyhow::Result<()>
    where
        F: FnOnce(Client) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        self.dispatch_with_failure(label, f, |_| None)
    }

    /// Spawn a fire-and-forget client call. A failure inside the
    /// spawned future - or the client-less early Err - is emitted as
    /// the caller's typed event (a `None` maps to log-only) - a view
    /// rolls back nothing otherwise: the spinner, the optimistic
    /// chip, the waiting awaiter all unwind only if this event
    /// arrives.
    fn dispatch_with_failure<F, Fut, T>(
        &self,
        label: &'static str,
        f: F,
        on_failure: T,
    ) -> anyhow::Result<()>
    where
        F: FnOnce(Client) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
        T: FnOnce(&anyhow::Error) -> Option<AgentEvent> + Send + 'static,
    {
        let Some(client) = self.client() else {
            let err = anyhow::anyhow!("forge-sdk bridge: {label} called before active session");
            // The caller warn-logs this Err and drops it; the typed
            // event is what unwinds a view's optimistic state.
            if let Some(event) = on_failure(&err)
                && self.inner.event_tx.send(event).is_err()
            {
                tracing::warn!(
                    target: crate::logging::targets::BRIDGE_LIFECYCLE,
                    label,
                    "event channel closed; typed dispatch failure dropped",
                );
            }
            return Err(err);
        };
        let event_tx = self.inner.event_tx.clone();
        let span = tracing::info_span!("bridge_dispatch", label);
        tokio::spawn(
            async move {
                if let Err(err) = f(client).await {
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        label,
                        error = %err,
                        "forge-sdk bridge: dispatch failed",
                    );
                    if let Some(event) = on_failure(&err)
                        && event_tx.send(event).is_err()
                    {
                        tracing::warn!(
                            target: crate::logging::targets::BRIDGE_LIFECYCLE,
                            label,
                            "event channel closed; typed dispatch failure dropped",
                        );
                    }
                }
            }
            .instrument(span),
        );
        Ok(())
    }
}

#[cfg(any(test, feature = "testing"))]
impl Default for ForgeSdkBridge {
    fn default() -> Self {
        Self::new(PathBuf::from(TESTING_STUB_CONFIG_DIR), None, Vec::new(), HashMap::new())
    }
}

// `unused_self` / `needless_pass_by_value` are allowed at the impl
// level: passthrough accessors that delegate to `forge_sdk::*` free
// functions don't logically use `self`, but every method here
// preserves the `&self`-receiver shape so AgentHandle wrappers can
// call them through a single stable seam.
#[allow(clippy::unused_self, clippy::needless_pass_by_value)]
impl ForgeSdkBridge {
    pub(crate) fn take_events(&self) -> Option<mpsc::UnboundedReceiver<AgentEvent>> {
        self.inner.events_rx.lock().take()
    }

    pub(crate) fn prompt_text(
        &self,
        session_id: String,
        text: String,
        uuid: String,
    ) -> anyhow::Result<()> {
        self.prompt_with_images(session_id, text, Vec::new(), uuid)
    }

    pub(crate) fn prompt_with_images(
        &self,
        session_id: String,
        text: String,
        images: Vec<forge_primitives::ImageAttachment>,
        uuid: String,
    ) -> anyhow::Result<()> {
        // No `check_session_id` here - a view commits to its
        // thinking state BEFORE this call, and a silent
        // stale-session drop (returning Ok with no event) would
        // leave the user's spinner running forever with no
        // recovery path. If the prompt does land in the wrong
        // session, normal SDK error surfaces (TurnError, etc.)
        // tear down Thinking via the existing wire path. We still
        // log a breadcrumb on mismatch so the bypass is observable.
        self.trace_session_id_bypass(&session_id, "prompt_with_images");
        let mut chunks: Vec<forge_primitives::PromptChunk> = Vec::with_capacity(1 + images.len());
        for img in images {
            if let Err(reason) = forge_primitives::validate_image(&img.data, &img.mime_type) {
                tracing::warn!(
                    target: crate::logging::targets::APP_INPUT,
                    "forge-sdk bridge: skipping invalid image: {reason}"
                );
                continue;
            }
            chunks.push(forge_primitives::PromptChunk {
                kind: "image".to_owned(),
                value: serde_json::json!({
                    "data": img.data,
                    "mime_type": img.mime_type,
                }),
            });
        }
        chunks.push(forge_primitives::PromptChunk {
            kind: "text".to_owned(),
            value: Value::String(text),
        });
        self.dispatch_with_failure(
            "prompt",
            move |client| async move { forge_sdk_worker::send_prompt(&client, chunks, &uuid).await },
            move |err| Some(turn_error(session_id.clone(), err.to_string())),
        )
    }

    /// Drop one prompt still in the CLI's queue. The answer rides
    /// [`AgentEvent::PromptCancelResolved`] rather than the call's return: the
    /// dispatcher channel is fire-and-forget, and a caller that reads `false`
    /// has to tell the reader the prompt was already taken.
    pub(crate) fn cancel_queued(&self, session_id: String, uuid: String) -> anyhow::Result<()> {
        if !self.check_session_id(&session_id, "cancel_queued") {
            return Ok(());
        }
        let event_tx = self.inner.event_tx.clone();
        let err_session_id = session_id.clone();
        let answered = uuid.clone();
        self.dispatch_with_failure(
            "cancel_queued",
            move |client| async move {
                let cancelled = tokio::time::timeout(
                    Self::CONTROL_RESPONSE_TIMEOUT,
                    client.cancel_queued_message(&uuid),
                )
                .await
                .map_err(|_| {
                    anyhow::anyhow!("cancel_async_message not acknowledged by the CLI")
                })??;
                if event_tx
                    .send(AgentEvent::PromptCancelResolved { uuid: answered, cancelled })
                    .is_err()
                {
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        "event channel closed; PromptCancelResolved dropped",
                    );
                }
                Ok(())
            },
            move |err| Some(turn_error(err_session_id.clone(), err.to_string())),
        )
    }

    pub(crate) fn cancel(&self, session_id: String) -> anyhow::Result<()> {
        if !self.check_session_id(&session_id, "cancel") {
            return Ok(());
        }
        let event_tx = self.inner.event_tx.clone();
        let err_session_id = session_id.clone();
        self.dispatch_with_failure(
            "cancel",
            move |client| async move {
                match tokio::time::timeout(Self::CONTROL_RESPONSE_TIMEOUT, client.interrupt()).await
                {
                    Ok(Ok(())) => Ok(()),
                    Ok(Err(e)) => Err(e.into()),
                    Err(_) => {
                        let message = "interrupt not acknowledged by the CLI".to_owned();
                        if event_tx.send(turn_error(session_id.clone(), message.clone())).is_err() {
                            tracing::warn!(
                                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                                "event channel closed; TurnError dropped",
                            );
                        }
                        Ok(())
                    }
                }
            },
            move |err| {
                Some(turn_error(err_session_id, format!("interrupt not acknowledged: {err}")))
            },
        )
    }

    /// `send_control` parks until the CLI's `control_response`; the
    /// wait is bounded so a silent CLI surfaces as a typed failure
    /// instead of a chip/spinner stuck on an optimistic change that
    /// never took.
    const CONTROL_RESPONSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);

    /// How long a seat's context-usage mark may stand before any asker may
    /// reclaim it. The probe gives up at [`Self::CONTROL_RESPONSE_TIMEOUT`],
    /// counted from its closure's first poll, so a mark past that bound is
    /// either already dead or a probe the timeout is about to drop -
    /// reclaiming costs at worst one overlapping probe, while leaving a dead
    /// mark costs the seat its reading for the life of the process.
    const PROBE_MARK_DEADLINE: std::time::Duration = Self::CONTROL_RESPONSE_TIMEOUT;

    /// The CLI's own refusal text for a failed `set_permission_mode`.
    /// The SDK wraps it twice (`MessageParse` Display + a
    /// `control failed: ` prefix); strip both so the chat line carries
    /// the reason, not the wrapper.
    fn set_mode_rejection_text(err: &forge_sdk::Error) -> String {
        if let forge_sdk::Error::MessageParse { reason, .. } = err
            && let Some(cli_reason) = reason.strip_prefix("control failed: ")
        {
            return cli_reason.to_owned();
        }
        err.to_string()
    }

    pub(crate) fn set_mode(&self, session_id: String, mode: PermissionMode) -> anyhow::Result<()> {
        if !self.check_session_id(&session_id, "set_mode") {
            // A session-swap race dropped the dispatch; surface it on
            // the attempted session so the chip doesn't stay flipped.
            let _ = self.inner.event_tx.send(AgentEvent::SetModeFailed {
                session_id,
                mode,
                message: "the session it was sent to is no longer active".to_owned(),
            });
            return Ok(());
        }
        let event_tx = self.inner.event_tx.clone();
        let err_session_id = session_id.clone();
        self.dispatch_with_failure(
            "set_mode",
            move |client| async move {
                let outcome = tokio::time::timeout(
                    Self::CONTROL_RESPONSE_TIMEOUT,
                    client.set_permission_mode(mode),
                )
                .await;
                let failure = match outcome {
                    Ok(Ok(())) => None,
                    Ok(Err(e)) => {
                        let text = Self::set_mode_rejection_text(&e);
                        Some(if text.trim().is_empty() {
                            "no reason given".to_owned()
                        } else {
                            text
                        })
                    }
                    Err(_) => Some("no response from the CLI".to_owned()),
                };
                if let Some(message) = failure {
                    if event_tx
                        .send(AgentEvent::SetModeFailed {
                            session_id: session_id.clone(),
                            mode,
                            message: message.clone(),
                        })
                        .is_err()
                    {
                        tracing::warn!(
                            target: crate::logging::targets::BRIDGE_LIFECYCLE,
                            error = %message,
                            "event channel closed; SetModeFailed dropped",
                        );
                    } else {
                        tracing::warn!(
                            target: crate::logging::targets::BRIDGE_LIFECYCLE,
                            session_id = %session_id,
                            mode = %mode.as_wire(),
                            error = %message,
                            "set_permission_mode rejected; SetModeFailed emitted",
                        );
                    }
                }
                Ok(())
            },
            move |err| {
                Some(AgentEvent::SetModeFailed {
                    session_id: err_session_id,
                    mode,
                    message: format!("set_mode never reached the CLI: {err}"),
                })
            },
        )
    }

    pub(crate) fn set_model(&self, session_id: String, model: String) -> anyhow::Result<()> {
        if !self.check_session_id(&session_id, "set_model") {
            // A session-swap race dropped the dispatch; surface it on
            // the attempted session so the model chip doesn't stay
            // flipped.
            let _ = self.inner.event_tx.send(AgentEvent::SetModelFailed {
                session_id: session_id.clone(),
                model: model.clone(),
                message: "the session it was sent to is no longer active".to_owned(),
            });
            return Ok(());
        }
        let event_tx = self.inner.event_tx.clone();
        let err_session_id = session_id.clone();
        let err_model = model.clone();
        self.dispatch_with_failure(
            "set_model",
            move |client| async move {
                let outcome = tokio::time::timeout(
                    Self::CONTROL_RESPONSE_TIMEOUT,
                    client.set_model(Some(model.as_str())),
                )
                .await;
                let failure = match outcome {
                    Ok(Ok(())) => None,
                    Ok(Err(e)) => {
                        let text = Self::set_mode_rejection_text(&e);
                        Some(if text.trim().is_empty() {
                            "no reason given".to_owned()
                        } else {
                            text
                        })
                    }
                    Err(_) => Some("no response from the CLI".to_owned()),
                };
                if let Some(message) = failure
                    && event_tx
                        .send(AgentEvent::SetModelFailed {
                            session_id: session_id.clone(),
                            model: model.clone(),
                            message: message.clone(),
                        })
                        .is_err()
                {
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        error = %message,
                        "event channel closed; SetModelFailed dropped",
                    );
                }
                Ok(())
            },
            move |err| {
                Some(AgentEvent::SetModelFailed {
                    session_id: err_session_id,
                    model: err_model,
                    message: format!("set_model never reached the CLI: {err}"),
                })
            },
        )
    }

    /// Verify the requested `session_id` matches the bridge's current
    /// session before dispatching a user-action method. In a session-swap
    /// race a `cancel`/`set_mode`/`set_model` for session A could
    /// otherwise hit session B's `Client`. Logs the mismatch and returns
    /// false; the caller is expected to drop the dispatch with a no-op
    /// `Ok(())`. A request carrying no id at all is refused the same way
    /// and logged at `error`, since there is nothing to match it to.
    ///
    /// Other user-action methods (`prompt_with_images`) intentionally
    /// opt out - see the inline rationale at each call site.
    fn check_session_id(&self, session_id: &str, label: &'static str) -> bool {
        // The slot is written once, where a spawn supplies the id the
        // child will run under, and every production spawn supplies one:
        // `new_session(None)` needs `fresh_session_id_for` to miss, which
        // needs a pooled entry with no `registration`, and the account
        // walk only returns keys from the org pin the pool holds. So a
        // live bridge's slot is never empty here, and a request that does
        // not match is stale rather than unaddressable.
        // An id-less request names no session at all, so there is nothing
        // to match it against, and accepting it would run the dispatch on
        // whichever session the bridge happens to hold. Checked before
        // the equality, since two empty ids would otherwise match.
        if session_id.is_empty() {
            let current = self.inner.session_id_slot.lock().clone();
            tracing::error!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                event_name = "dispatch_without_session_id",
                label,
                current_session_id = %current,
                "dropping a dispatch that carries no session id"
            );
            return false;
        }
        let current = self.inner.session_id_slot.lock().clone();
        if current == session_id {
            return true;
        }
        tracing::warn!(
            target: crate::logging::targets::BRIDGE_LIFECYCLE,
            event_name = "stale_session_dispatch",
            label,
            current_session_id = %current,
            requested_session_id = %session_id,
            "dropping dispatch for stale session id"
        );
        false
    }

    /// Sibling of [`Self::check_session_id`] for methods that intentionally
    /// pass through on mismatch (no silent drop). Logs a breadcrumb at
    /// `trace` so postmortems can correlate "my prompt vanished" with
    /// a session-swap race, without raising the noise floor under
    /// normal operation.
    fn trace_session_id_bypass(&self, session_id: &str, label: &'static str) {
        let current = self.inner.session_id_slot.lock().clone();
        if current == session_id {
            return;
        }
        if session_id.is_empty() {
            tracing::warn!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                event_name = "bypass_without_session_id",
                label,
                current_session_id = %current,
                "passing through a dispatch that carries no session id (intentional bypass)",
            );
            return;
        }
        tracing::trace!(
            target: crate::logging::targets::BRIDGE_LIFECYCLE,
            event_name = "stale_session_bypass",
            label,
            current_session_id = %current,
            requested_session_id = %session_id,
            "passing through dispatch despite stale session id (intentional bypass)",
        );
    }

    pub(crate) fn get_context_usage(&self, session_id: String) -> anyhow::Result<()> {
        let taken_at = Instant::now();
        {
            let mut marks = self.inner.context_probes_in_flight.lock();
            if let Some(out_since) = marks.get(&session_id) {
                let age = taken_at.duration_since(*out_since);
                if age < Self::PROBE_MARK_DEADLINE {
                    tracing::debug!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        event_name = "context_usage_probe_coalesced",
                        session_id = %session_id,
                        "context usage probe already in flight; this ask is coalesced",
                    );
                    return Ok(());
                }
                // The mark outlived its probe's own deadline, so the probe is
                // not coming back for it: this ask takes the seat over.
                tracing::warn!(
                    target: crate::logging::targets::BRIDGE_LIFECYCLE,
                    event_name = "context_usage_probe_mark_reclaimed",
                    session_id = %session_id,
                    age_secs = age.as_secs(),
                    "a context usage probe outlived its deadline; starting a fresh probe",
                );
            }
            marks.insert(session_id.clone(), taken_at);
        }
        let probe = ProbeInFlight {
            inner: Arc::clone(&self.inner),
            session_id: session_id.clone(),
            taken_at,
        };
        let event_tx = self.inner.event_tx.clone();
        self.dispatch("get_context_usage", move |client| async move {
            // Refreshes arrive at turn end, session switch, connect
            // and post-compaction; a wedged CLI must cost one
            // timed-out probe, not a parked task per refresh with the
            // Ctx bar frozen. The expiry below releases the mark; a
            // probe whose closure never runs at all is the park this
            // timeout cannot reach, and the mark's own deadline in
            // `get_context_usage` is what bounds that one.
            let usage = match tokio::time::timeout(
                Self::CONTROL_RESPONSE_TIMEOUT,
                client.get_context_usage(),
            )
            .await
            {
                Ok(Ok(usage)) => usage,
                Ok(Err(e)) => return Err(e.into()),
                Err(_) => {
                    // The mark goes with the timeout's early return: a seat
                    // whose probe timed out is askable again, rather than
                    // wedged for the life of its session.
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        session_id = %session_id,
                        "context usage probe timed out; skipping this poll",
                    );
                    return Ok(());
                }
            };
            // Released before the event is emitted, so an asker woken by the
            // answer finds the seat free rather than racing the drop.
            drop(probe);
            let percentage = forge_sdk_worker::reported_percentage(usage.percentage);
            // `raw_max_tokens` is the model's nominal context-window
            // size; `max_tokens` is the effective cap after autocompact
            // reductions. Forge surfaces the raw size so the panel
            // shows the model's headline capacity (1M / 200K / …)
            // rather than a fluctuating effective number.
            let max_tokens = Some(usage.raw_max_tokens);
            let effective_max_tokens = Some(usage.max_tokens);
            let total_tokens = Some(usage.total_tokens);
            if event_tx
                .send(AgentEvent::ContextUsage {
                    session_id,
                    percentage,
                    max_tokens,
                    total_tokens,
                    effective_max_tokens,
                })
                .is_err()
            {
                tracing::warn!(
                    target: crate::logging::targets::BRIDGE_LIFECYCLE,
                    "event channel closed; ContextUsage dropped",
                );
            }
            Ok(())
        })
    }

    pub(crate) fn get_mcp_snapshot(&self, session_id: String) -> anyhow::Result<()> {
        let event_tx = self.inner.event_tx.clone();
        let err_session_id = session_id.clone();
        self.dispatch_with_failure(
            "get_mcp_snapshot",
            move |client| async move {
                let outcome =
                    tokio::time::timeout(Self::CONTROL_RESPONSE_TIMEOUT, client.mcp_status()).await;
                let response = match outcome {
                    Ok(Err(e)) => return Err(e.into()),
                    Ok(Ok(result)) => result,
                    Err(_) => {
                        Self::emit_mcp_error_or_log(
                            &event_tx,
                            session_id,
                            "status",
                            None,
                            "no response from the CLI".to_owned(),
                        );
                        return Ok(());
                    }
                };
                if event_tx
                    .send(AgentEvent::McpSnapshot {
                        session_id,
                        servers: response.mcp_servers,
                        error: None,
                    })
                    .is_err()
                {
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        "event channel closed; McpSnapshot dropped",
                    );
                }
                Ok(())
            },
            move |err| {
                Some(AgentEvent::McpOperationError {
                    session_id: err_session_id,
                    error: forge_primitives::McpOperationError {
                        operation: "status".to_owned(),
                        server_name: None,
                        message: err.to_string(),
                    },
                })
            },
        )
    }

    /// Decode the launch settings, or report the refusal as the spawn failure
    /// it is (Ved refused the spawn, 2026-10-15).
    ///
    /// **The event is the point of doing it here.** `ConnectionFailed` is what
    /// the seat's own task records a spawn failure by and rolls the row on, so
    /// a payload that does not decode leaves a seat that says why and can be
    /// retried, not one stuck at `Spawning` with nothing to read.
    fn decode_or_report(
        &self,
        session_id: Option<&str>,
        launch_settings: serde_json::Value,
    ) -> anyhow::Result<SessionLaunchSettings> {
        crate::agent::decode_launch_settings(launch_settings).map_err(|error| {
            tracing::error!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                event_name = "launch_settings_decode_failed",
                session_id = session_id.unwrap_or(""),
                "failed to decode launch_settings; the spawn is refused rather than started \
                 with none of them",
            );
            let msg = format!("forge-sdk session spawn refused: {error}");
            let kind = spawn_failure_kind(&error);
            if self
                .event_tx()
                .send(AgentEvent::ConnectionFailed { message: msg.clone(), kind })
                .is_err()
            {
                tracing::warn!(
                    target: crate::logging::targets::BRIDGE_LIFECYCLE,
                    error = %msg,
                    "event channel closed; ConnectionFailed dropped",
                );
            }
            error
        })
    }

    /// Start a session under `session_id`, or under an id the CLI picks
    /// when that is `None`.
    pub(crate) fn new_session(
        &self,
        session_id: Option<String>,
        cwd: String,
        launch_settings: serde_json::Value,
    ) -> anyhow::Result<()> {
        let launch_settings = self.decode_or_report(session_id.as_deref(), launch_settings)?;
        let bridge = self.clone();
        let span = tracing::info_span!("bridge_new_session", cwd = %cwd);
        tokio::spawn(
            async move {
                let id = match session_id.as_deref() {
                    Some(id) => SessionId::Create(id),
                    None => SessionId::Unassigned,
                };
                if let Err(err) =
                    forge_sdk_worker::spawn_session(&bridge, &cwd, id, &launch_settings).await
                {
                    let kind = spawn_failure_kind(&err);
                    let msg = format!("forge-sdk session spawn failed: {err}");
                    if bridge
                        .event_tx()
                        .send(AgentEvent::ConnectionFailed { message: msg.clone(), kind })
                        .is_err()
                    {
                        tracing::warn!(
                            target: crate::logging::targets::BRIDGE_LIFECYCLE,
                            error = %msg,
                            "event channel closed; ConnectionFailed dropped",
                        );
                    }
                }
            }
            .instrument(span),
        );
        Ok(())
    }

    pub(crate) fn resume_session(
        &self,
        session_id: String,
        cwd: String,
        launch_settings: serde_json::Value,
    ) -> anyhow::Result<()> {
        let launch_settings = self.decode_or_report(Some(&session_id), launch_settings)?;
        let bridge = self.clone();
        let span = tracing::info_span!(
            "bridge_resume_session",
            session_id = %session_id,
            cwd = %cwd,
        );
        tokio::spawn(
            async move {
                if let Err(err) = forge_sdk_worker::spawn_session(
                    &bridge,
                    &cwd,
                    SessionId::Resume(&session_id),
                    &launch_settings,
                )
                .await
                {
                    let kind = spawn_failure_kind(&err);
                    let msg = format!("forge-sdk session resume failed: {err}");
                    if bridge
                        .event_tx()
                        .send(AgentEvent::ConnectionFailed { message: msg.clone(), kind })
                        .is_err()
                    {
                        tracing::warn!(
                            target: crate::logging::targets::BRIDGE_LIFECYCLE,
                            error = %msg,
                            "event channel closed; ConnectionFailed dropped",
                        );
                    }
                }
            }
            .instrument(span),
        );
        Ok(())
    }

    /// Re-enter `session_id`, keeping the identity through a refusal.
    /// Three attempts, in order: resume it; when the CLI refuses - the
    /// transcript is gone - start a new session under the SAME id, which
    /// the CLI adopts and which keeps the identity stable; when that is
    /// refused too, start under a fresh id rather than failing. The
    /// third attempt exists because the first two can both be refused
    /// for a reason that is neither of them being absent (an unreadable
    /// transcript), and that is a state the pre-id fallback handled
    /// silently by starting a new session. Only all three failing
    /// surfaces `ConnectionFailed`. Used by project-rooted spawns
    /// (Default / Named) where the catalog's recorded lead may be stale
    /// (e.g. cross-account scan, deleted file, schema drift). See
    /// [`forge_primitives::AgentCommand::ResumeOrNewSession`] for the
    /// motivation.
    ///
    /// `cwd` is also passed to the resume attempt, not just the
    /// fallback. `claude --resume <id>` indexes sessions by the
    /// project key derived from its own working directory, so
    /// inheriting forge's `$PWD` (the default when cwd is empty) makes
    /// every cross-directory resume fail with "No conversation found
    /// with session ID …" even when the `.jsonl` exists.
    pub(crate) fn resume_or_new_session(
        &self,
        session_id: String,
        cwd: String,
        launch_settings: serde_json::Value,
    ) -> anyhow::Result<()> {
        let launch_settings = self.decode_or_report(Some(&session_id), launch_settings)?;
        let bridge = self.clone();
        let span = tracing::info_span!(
            "bridge_resume_or_new_session",
            session_id = %session_id,
            cwd = %cwd,
        );
        tokio::spawn(async move {
            let Err(resume_err) = forge_sdk_worker::spawn_session(
                &bridge,
                &cwd,
                SessionId::Resume(&session_id),
                &launch_settings,
            )
            .await
            else {
                return;
            };
            tracing::warn!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                error = %resume_err,
                session_id = %session_id,
                "session resume failed; starting a new session under the same id",
            );
            let Err(create_err) = forge_sdk_worker::spawn_session(
                &bridge,
                &cwd,
                SessionId::Create(&session_id),
                &launch_settings,
            )
            .await
            else {
                return;
            };
            // Both refused: the id is present on disk but not resumable,
            // so create refuses it as already in use. A fresh id is the
            // only way the session comes up.
            let fresh = uuid::Uuid::new_v4().to_string();
            tracing::warn!(
                target: crate::logging::targets::BRIDGE_LIFECYCLE,
                error = %create_err,
                session_id = %session_id,
                fresh_session_id = %fresh,
                "session could neither be resumed nor created under its id; starting under a \
                 new one",
            );
            if let Err(new_err) = forge_sdk_worker::spawn_session(
                &bridge,
                &cwd,
                SessionId::Create(&fresh),
                &launch_settings,
            )
            .await
            {
                let msg = format!(
                    "forge-sdk session spawn failed after the resume fallbacks (resume err: {resume_err}; same-id err: {create_err}; fresh err: {new_err})",
                );
                if bridge
                    .event_tx()
                    .send(AgentEvent::ConnectionFailed {
                        message: msg.clone(),
                        // Three attempts collapsed into one message, so
                        // there is no single variant to name.
                        kind: SpawnFailureKind::Unclassified,
                    })
                    .is_err()
                {
                    tracing::warn!(
                        target: crate::logging::targets::BRIDGE_LIFECYCLE,
                        error = %msg,
                        "event channel closed; ConnectionFailed dropped",
                    );
                }
            }
        }.instrument(span));
        Ok(())
    }

    // The two response handlers below are intentionally exempt from
    // `check_session_id` - staleness is detected via the pending map
    // (an unknown `tool_call_id` already logs warn in
    // `deliver_*_response`). A late response arriving after a session
    // swap should still be honoured if its tool_call_id is in the
    // shared pending map.

    pub(crate) fn permission_response(
        &self,
        _session_id: String,
        tool_call_id: String,
        outcome: PermissionOutcome,
    ) -> anyhow::Result<()> {
        forge_sdk_worker::deliver_permission_response(&self.inner.pending, &tool_call_id, outcome);
        Ok(())
    }

    pub(crate) fn question_response(
        &self,
        _session_id: String,
        tool_call_id: String,
        outcome: QuestionOutcome,
    ) -> anyhow::Result<()> {
        forge_sdk_worker::deliver_question_response(
            &self.inner.pending_questions,
            &tool_call_id,
            outcome,
        );
        Ok(())
    }

    // ---- Direct-return accessors (delegate to forge_sdk::*) ----

    /// Cheap clone of the bridge's bound `config_dir`. Used by the
    /// session worker to thread `CLAUDE_CONFIG_DIR` into the spawned
    /// `claude` subprocess and by direct-return accessors.
    pub(crate) fn config_dir(&self) -> PathBuf {
        self.inner.config_dir.clone()
    }

    /// Cheap clone of the bridge's bound forge-account
    /// `display_name`, when forge-workspace picked one. Used by the
    /// session worker to attach `forge_account` to the initial
    /// `ForgeAccountIdentity` emit alongside the CLI-side `account`.
    pub(crate) fn display_name(&self) -> Option<String> {
        self.inner.display_name.clone()
    }

    /// OS PID of the spawned `claude` child, when a client is bound
    /// and the transport reported one. Used by
    /// [`crate::env::processes`] to anchor an OS-level walk of the
    /// descendant process tree for the Inspector pane's PROCESSES
    /// section. Returns `None` before the first `new_session` /
    /// `resume_session` lands a client, or after `clear_client`.
    pub(crate) fn claude_pid(&self) -> Option<u32> {
        self.client().and_then(|c| c.claude_pid())
    }

    pub(crate) fn project_memory_path(&self, cwd: &Path) -> PathBuf {
        crate::userdata::memory::project_memory_path(&self.inner.config_dir, cwd)
    }

    pub(crate) fn settings_documents(
        &self,
        cwd: Option<&Path>,
    ) -> crate::userdata::settings::SettingsDocuments {
        crate::userdata::settings::settings_documents(&self.inner.config_dir, cwd)
    }
}

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

    /// The bridge names the SDK variant, so a consumer classifies on it
    /// rather than on rendered prose - the working directory's path runs
    /// through `.claude/worktrees/<label>`, which is the word the
    /// worktree-creation heuristic keys on.
    #[test]
    fn a_missing_working_directory_is_named_on_the_failure() {
        let missing = anyhow::Error::new(forge_sdk::Error::CwdNotFound { path: "/gone".into() });
        assert_eq!(
            spawn_failure_kind(&missing),
            SpawnFailureKind::WorkingDirMissing,
            "the SDK error has to reach the kind through the anyhow wrapper \
             spawn_session hands back",
        );

        let unnamed = anyhow::Error::new(forge_sdk::Error::CliNotFound { binary: "claude".into() });
        assert_eq!(
            spawn_failure_kind(&unnamed),
            SpawnFailureKind::Unclassified,
            "an unnamed failure leaves the message heuristics in charge",
        );
    }

    /// The class rides the event from where the message is built, so a
    /// reader gets the classification instead of searching the prose
    /// again - the failure's text is all a `TurnError` carries.
    #[test]
    fn a_turn_error_carries_the_class_of_its_own_text() {
        let auth = turn_error("s1".to_owned(), "authentication failed: please log in".to_owned());
        let AgentEvent::TurnError { class, .. } = &auth else {
            panic!("a built turn error is a TurnError, got {auth:?}")
        };
        assert_eq!(
            class,
            &forge_primitives::TurnErrorClass::AuthRequired,
            "an auth-shaped failure carries the class, not a value a reader has to re-derive",
        );

        let plain = turn_error("s1".to_owned(), "stdin write failed".to_owned());
        let AgentEvent::TurnError { class, .. } = &plain else {
            panic!("a built turn error is a TurnError, got {plain:?}")
        };
        assert_eq!(
            class,
            &forge_primitives::TurnErrorClass::Other,
            "a failure the classifier does not know is Other rather than unset",
        );
    }

    /// Bind the bridge's id slot. Every production bridge has one from
    /// the start: the slot is written once, where the spawn supplies the
    /// id the child will run under, and nothing writes it per frame. An
    /// unbound slot is therefore a fixture state, and `check_session_id`
    /// refuses a dispatch it cannot match rather than accepting any id.
    fn bind_session_slot(bridge: &ForgeSdkBridge, session_id: &str) {
        bridge.session_id_slot_arc().lock().clone_from(&session_id.to_owned());
    }

    /// The match is by id and only by id: a stale id is refused so a
    /// `cancel`/`set_mode`/`set_model` meant for one session cannot land
    /// on another, and an id-less request is refused rather than being
    /// run against whichever session the bridge happens to hold.
    #[test]
    fn check_session_id_matches_by_id_only() {
        let bridge = test_bridge();
        bind_session_slot(&bridge, "session-1");

        assert!(bridge.check_session_id("session-1", "cancel"), "the session's own id passes");
        assert!(!bridge.check_session_id("session-2", "cancel"), "another session's id is refused");
        assert!(
            !bridge.check_session_id("", "cancel"),
            "a request carrying no id is refused, not matched to anything",
        );
    }

    /// An unbound slot refuses rather than matching everything, which is
    /// what the old `current.is_empty()` wildcard did. That arm is safe
    /// to refuse because nothing writes the slot per frame and every
    /// production spawn supplies an id, so a live bridge is never
    /// unbound; see the invariant on `check_session_id`.
    #[test]
    fn an_unbound_slot_refuses_rather_than_matching_anything() {
        let bridge = test_bridge();

        assert!(
            !bridge.check_session_id("session-1", "cancel"),
            "an unbound slot does not accept a request it cannot check",
        );
        assert!(
            !bridge.check_session_id("", "cancel"),
            "nor an id-less request, which two empty ids would otherwise match",
        );
    }

    fn test_bridge() -> ForgeSdkBridge {
        ForgeSdkBridge::new(
            PathBuf::from(TESTING_STUB_CONFIG_DIR),
            None,
            Vec::new(),
            HashMap::new(),
        )
    }

    #[test]
    fn take_events_returns_some_once_then_none() {
        let bridge = test_bridge();
        assert!(bridge.take_events().is_some());
        assert!(bridge.take_events().is_none());
    }

    #[test]
    fn clear_client_drains_pending_permission_map() {
        let bridge = test_bridge();
        let (tx, rx) = tokio::sync::oneshot::channel::<forge_primitives::PermissionDecision>();
        bridge.inner.pending.lock().insert("tool-id-1".to_owned(), tx);

        bridge.clear_client();

        assert!(bridge.inner.pending.lock().is_empty());
        let decision = rx.blocking_recv().expect("oneshot resolved by clear_client drain");
        assert!(!decision.is_allow(), "drain must resolve with deny");
    }

    #[test]
    fn clear_client_drains_pending_question_map() {
        let bridge = test_bridge();
        let (tx, rx) = tokio::sync::oneshot::channel::<forge_primitives::QuestionOutcome>();
        bridge.inner.pending_questions.lock().insert("tool-id-q1".to_owned(), tx);

        bridge.clear_client();

        assert!(bridge.inner.pending_questions.lock().is_empty());
        let outcome = rx.blocking_recv().expect("oneshot resolved by clear_client drain");
        assert!(matches!(outcome, forge_primitives::QuestionOutcome::Cancelled));
    }

    #[test]
    fn dispatch_without_client_returns_error() {
        let bridge = test_bridge();
        bind_session_slot(&bridge, "session-1");
        let err = bridge.cancel("session-1".to_owned()).unwrap_err();
        assert!(err.to_string().contains("before active session"));
    }

    /// The client-None window (clear_client, then set_client only
    /// after init completes) still routes commands with the old
    /// session id, so the caller's typed event is the only thing that
    /// unwinds a view's optimistic state - a silent Err return loses
    /// the prompt entirely.
    #[tokio::test]
    async fn dispatch_without_client_emits_the_typed_event() {
        let bridge = test_bridge();
        let mut events = bridge.take_events().expect("fresh bridge yields its events receiver");

        let err = bridge
            .dispatch_with_failure(
                "prompt",
                |_client| async { Ok::<(), anyhow::Error>(()) },
                |err| Some(turn_error("s1".to_owned(), err.to_string())),
            )
            .expect_err("no client, dispatch refused");
        assert!(err.to_string().contains("before active session"));

        match tokio::time::timeout(std::time::Duration::from_secs(2), events.recv()).await {
            Ok(Some(AgentEvent::TurnError { message, .. })) => {
                assert!(
                    message.contains("before active session"),
                    "the typed failure carries the early Err: {message}"
                );
            }
            other => panic!("expected a TurnError from the client-less early Err, got {other:?}"),
        }
    }

    /// The client-None window passes `check_session_id` for both
    /// sibling methods (the slot holds the session id the caller
    /// dispatches with), so the typed failure is the only thing that
    /// unflips the optimistic mode chip - a silent Err return leaves it
    /// stuck.
    #[tokio::test]
    async fn set_mode_without_client_emits_set_mode_failed() {
        let bridge = test_bridge();
        bind_session_slot(&bridge, "session-1");
        let mut events = bridge.take_events().expect("fresh bridge yields its events receiver");

        bridge
            .set_mode("session-1".to_owned(), PermissionMode::AcceptEdits)
            .expect_err("no client, dispatch refused");

        match tokio::time::timeout(std::time::Duration::from_secs(2), events.recv()).await {
            Ok(Some(AgentEvent::SetModeFailed { session_id, mode, message })) => {
                assert_eq!(session_id, "session-1");
                assert_eq!(mode, PermissionMode::AcceptEdits);
                assert!(
                    message.contains("set_mode never reached the CLI"),
                    "the typed failure names the op and carries the early Err: {message}"
                );
            }
            other => panic!("expected SetModeFailed from the client-less early Err, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn set_model_without_client_emits_set_model_failed() {
        let bridge = test_bridge();
        bind_session_slot(&bridge, "session-1");
        let mut events = bridge.take_events().expect("fresh bridge yields its events receiver");

        bridge
            .set_model("session-1".to_owned(), "claude-sonnet-5".to_owned())
            .expect_err("no client, dispatch refused");

        match tokio::time::timeout(std::time::Duration::from_secs(2), events.recv()).await {
            Ok(Some(AgentEvent::SetModelFailed { session_id, model, message })) => {
                assert_eq!(session_id, "session-1");
                assert_eq!(model, "claude-sonnet-5");
                assert!(
                    message.contains("set_model never reached the CLI"),
                    "the typed failure names the op and carries the early Err: {message}"
                );
            }
            other => {
                panic!("expected SetModelFailed from the client-less early Err, got {other:?}")
            }
        }
    }

    #[test]
    fn set_mode_rejection_text_strips_the_control_failed_wrapper() {
        let err = forge_sdk::Error::message_parse("control failed: mode not permitted");
        assert_eq!(ForgeSdkBridge::set_mode_rejection_text(&err), "mode not permitted");
        let other = forge_sdk::Error::Connection {
            reason: "subprocess closed before set_permission_mode response".into(),
        };
        assert_eq!(
            ForgeSdkBridge::set_mode_rejection_text(&other),
            other.to_string(),
            "non-control errors pass through as their Display",
        );
    }

    /// forge-sdk's shared dev fixture (same file the forge-sdk
    /// integration tests drive); the bridge only needs a completed
    /// handshake behind `set_client`.
    fn sdk_mock_binary() -> String {
        concat!(env!("CARGO_MANIFEST_DIR"), "/../forge-sdk/tests/fixtures/mock_claude.sh")
            .to_owned()
    }

    async fn bridge_with_mock_client() -> (ForgeSdkBridge, mpsc::UnboundedReceiver<AgentEvent>) {
        bridge_with_mock_client_opts(forge_sdk::OptionsBuilder::new()).await
    }

    async fn bridge_with_mock_client_opts(
        builder: forge_sdk::OptionsBuilder,
    ) -> (ForgeSdkBridge, mpsc::UnboundedReceiver<AgentEvent>) {
        let bridge = ForgeSdkBridge::default();
        let events = bridge.take_events().expect("fresh bridge yields its events receiver");
        let opts = builder.binary(sdk_mock_binary()).build();
        let (client, _client_events) = forge_sdk::Client::spawn(opts).await.expect("mock client");
        bridge.set_client(client);
        (bridge, events)
    }

    /// forge-sdk's control mock, whose `FORGED_MOCK_ECHO_SUBTYPE` hook records
    /// every observed control subtype to `echo`.
    fn control_mock_binary() -> String {
        concat!(env!("CARGO_MANIFEST_DIR"), "/../forge-sdk/tests/fixtures/mock_claude_control.sh")
            .to_owned()
    }

    async fn bridge_with_control_mock(
        echo: &std::path::Path,
    ) -> (ForgeSdkBridge, mpsc::UnboundedReceiver<AgentEvent>) {
        let bridge = ForgeSdkBridge::default();
        let events = bridge.take_events().expect("fresh bridge yields its events receiver");
        let opts = forge_sdk::OptionsBuilder::new()
            .binary(control_mock_binary())
            .env("FORGED_MOCK_ECHO_SUBTYPE", echo.to_string_lossy().as_ref())
            .build();
        let (client, _client_events) = forge_sdk::Client::spawn(opts).await.expect("mock client");
        bridge.set_client(client);
        (bridge, events)
    }

    /// The control subtypes the mock observed, in order.
    fn observed_subtypes(echo: &std::path::Path) -> Vec<String> {
        std::fs::read_to_string(echo)
            .expect("the mock echoes observed subtypes to the file")
            .lines()
            .map(str::to_owned)
            .collect()
    }

    /// A probe already out for a seat is not asked again.
    ///
    /// The socket re-asks on every read of a seat that reports no usage, and a
    /// client reads its seat on a poll, so an unanswered seat would otherwise
    /// park a probe over the CLI's whole transcript per poll - and a probe that
    /// times out emits nothing, so the repeat would never stop. The mark is per
    /// seat, and released when the answer, the error or the timeout lands.
    #[tokio::test]
    async fn a_context_probe_already_in_flight_is_not_asked_again() {
        let dir = tempfile::tempdir().expect("tempdir");
        let echo = dir.path().join("echo");
        let (bridge, mut events) = bridge_with_control_mock(&echo).await;

        bridge.get_context_usage("session-1".to_owned()).expect("the first ask goes");
        bridge
            .get_context_usage("session-1".to_owned())
            .expect("an ask while one is out is answered, not refused");
        bridge.get_context_usage("session-1".to_owned()).expect("and so is a third");
        // The mark itself, so the coalescing is pinned at the ask rather than
        // inferred from the CLI's sight of it.
        assert_eq!(
            bridge.inner.context_probes_in_flight.lock().len(),
            1,
            "the three asks left one seat marked, which is the one probe that went",
        );
        let subtypes = drain_answers(&bridge, &mut events, &echo).await;
        assert_eq!(
            subtypes.iter().filter(|subtype| *subtype == "get_context_usage").count(),
            1,
            "three asks with one probe out reach the CLI once: {subtypes:?}",
        );
        assert!(
            bridge.inner.context_probes_in_flight.lock().is_empty(),
            "and the answer released the seat, so the next ask is not the last",
        );

        // And the seat is free again once its answer lands.
        bridge.get_context_usage("session-1".to_owned()).expect("a later ask goes");
        let subtypes = drain_answers(&bridge, &mut events, &echo).await;
        assert_eq!(
            subtypes.iter().filter(|subtype| *subtype == "get_context_usage").count(),
            2,
            "a further ask reaches the CLI: {subtypes:?}",
        );
    }

    /// A mark left by a probe whose closure never runs is reclaimed by the
    /// next ask past the deadline.
    ///
    /// That closure is the whole of the probe's release machinery: the mark
    /// is taken before the dispatch, and an answer, an error and the timeout
    /// all live inside it. So a dispatch whose task never reaches its first
    /// poll leaves a mark nothing will ever clear, and a seat that coalesces
    /// behind it stays unprobed for the life of the process. The deadline is
    /// the only bound that reaches this park, so the ask past it must start a
    /// fresh probe rather than coalesce.
    #[tokio::test]
    async fn a_wedged_probe_mark_is_reclaimed_by_a_later_ask() {
        let dir = tempfile::tempdir().expect("tempdir");
        let echo = dir.path().join("echo");
        let (bridge, mut events) = bridge_with_control_mock(&echo).await;

        let wedged_at = Instant::now()
            .checked_sub(ForgeSdkBridge::PROBE_MARK_DEADLINE * 4)
            .expect("an instant four deadlines back");
        bridge.inner.context_probes_in_flight.lock().insert("session-1".to_owned(), wedged_at);

        bridge.get_context_usage("session-1".to_owned()).expect("the ask goes");
        // Checked before any wait: a wedged mark either turns over now or it
        // stays the wedged one, and the answered probe's release leaves
        // `None`, which is not the wedged instant either way.
        assert_ne!(
            bridge.inner.context_probes_in_flight.lock().get("session-1").copied(),
            Some(wedged_at),
            "the ask after the deadline must reclaim the wedged mark, not coalesce behind it",
        );

        let subtypes = drain_answers(&bridge, &mut events, &echo).await;
        assert_eq!(
            subtypes.iter().filter(|subtype| *subtype == "get_context_usage").count(),
            1,
            "the reclaimed ask must reach the CLI as one probe: {subtypes:?}",
        );
    }

    /// The guard releases only the mark it took, so a wedged probe's late
    /// release cannot clear the fresh one a reclaim took over.
    #[test]
    fn a_reclaimed_mark_survives_the_wedged_probes_late_release() {
        let bridge = test_bridge();
        let wedged_at = Instant::now()
            .checked_sub(ForgeSdkBridge::PROBE_MARK_DEADLINE * 4)
            .expect("an instant four deadlines back");
        let wedged = ProbeInFlight {
            inner: Arc::clone(&bridge.inner),
            session_id: "session-1".to_owned(),
            taken_at: wedged_at,
        };
        bridge.inner.context_probes_in_flight.lock().insert("session-1".to_owned(), wedged_at);

        let reclaimed_at = Instant::now();
        bridge.inner.context_probes_in_flight.lock().insert("session-1".to_owned(), reclaimed_at);
        drop(wedged);

        assert_eq!(
            bridge.inner.context_probes_in_flight.lock().get("session-1"),
            Some(&reclaimed_at),
            "the wedged probe's late release must leave the reclaimed mark standing",
        );
    }

    /// The subtypes the mock observed, fenced past every ask this call made.
    ///
    /// The read is fenced, not settled. The mock answers strictly in order, so
    /// the answer to a `get_mcp_snapshot` dispatched after the asks proves it
    /// has read every request written before that one: what the count below
    /// reads is what the test caused, at any load. The probe's own answer is
    /// waited for first, which is what keeps the fence last - the asks are
    /// dispatched together, so by the time the probe is back, a wrongly
    /// dispatched ask is long since written.
    ///
    /// Both waits are the answers themselves, never a stopwatch: a round trip
    /// here crosses a subprocess spawning `python3` three times per control
    /// line, and a fixed window expires mid-flight on a loaded machine -
    /// reading a slow mock as an ask that never went, which is the verdict
    /// this test exists to give. The deadline bounds a probe that never
    /// answers.
    async fn drain_answers(
        bridge: &ForgeSdkBridge,
        events: &mut mpsc::UnboundedReceiver<AgentEvent>,
        echo: &std::path::Path,
    ) -> Vec<String> {
        wait_for(events, "the probe's answer", &|event| {
            matches!(event, AgentEvent::ContextUsage { .. })
        })
        .await;
        bridge.get_mcp_snapshot("session-1".to_owned()).expect("the fence ask goes");
        wait_for(events, "the fence's answer", &|event| {
            matches!(event, AgentEvent::McpSnapshot { .. })
        })
        .await;
        observed_subtypes(echo)
    }

    /// The first event `wanted` accepts, inside the deadline.
    async fn wait_for(
        events: &mut mpsc::UnboundedReceiver<AgentEvent>,
        what: &str,
        wanted: &impl Fn(&AgentEvent) -> bool,
    ) -> AgentEvent {
        let deadline = std::time::Duration::from_secs(10);
        let Ok(Some(event)) = tokio::time::timeout(deadline, async {
            loop {
                match events.recv().await {
                    Some(event) if wanted(&event) => return Some(event),
                    Some(_) => {}
                    None => return None,
                }
            }
        })
        .await
        else {
            panic!("{what} arrives inside the deadline");
        };
        event
    }

    /// A failure inside the spawned dispatch future reaches the consumer
    /// as the caller's typed event - the spinner/optimistic state
    /// unwinds only when this arrives, so log-only failure is a
    /// regression.
    #[tokio::test]
    async fn dispatch_failure_emits_the_typed_event() {
        let (bridge, mut events) = bridge_with_mock_client().await;

        bridge
            .dispatch_with_failure(
                "prompt",
                |_client| async { Err::<(), _>(anyhow::anyhow!("stdin write failed")) },
                |err| Some(turn_error("s1".to_owned(), err.to_string())),
            )
            .expect("dispatch accepted");

        match tokio::time::timeout(std::time::Duration::from_secs(2), events.recv()).await {
            Ok(Some(AgentEvent::TurnError { message, .. })) => {
                assert!(
                    message.contains("stdin write failed"),
                    "the typed failure carries the error: {message}"
                );
            }
            other => panic!("expected a TurnError from the dispatch Err arm, got {other:?}"),
        }
    }

    /// A CLI that never answers `set_model` must surface
    /// `SetModelFailed` at the 60s budget (virtual time) instead of
    /// leaving the optimistic model flip unrecoverable. The wedge rides
    /// `Options::env` so the mock subprocess alone skips the subtype.
    /// Time runs REAL for the spawn handshake (a paused clock would
    /// fire the init budget before the mock's real stdout arrives),
    /// then pauses so the 60s dispatch budget elapses virtually.
    #[tokio::test(start_paused = true)]
    async fn set_model_timeout_emits_set_model_failed() {
        tokio::time::resume();
        let (bridge, mut events) = bridge_with_mock_client_opts(
            forge_sdk::OptionsBuilder::new().env("FORGED_MOCK_SKIP_SUBTYPE", "set_model"),
        )
        .await;
        tokio::time::pause();
        bind_session_slot(&bridge, "mock-session-001");

        bridge
            .set_model("mock-session-001".to_owned(), "claude-attempted".to_owned())
            .expect("dispatch");

        // Outer wrapper past the 60s budget: virtual time advances to
        // the SOONEST timer, so the dispatch's 60s fires before this.
        // Either failure text is a legitimate SetModelFailed cause: the
        // timeout's "no response from the CLI", or the connection
        // failure's Display when the mock's stdout ends inside the
        // virtual-time window. The test's job is the mechanism, not
        // which wedge won.
        match tokio::time::timeout(std::time::Duration::from_secs(120), events.recv()).await {
            Ok(Some(AgentEvent::SetModelFailed { model, message, .. })) => {
                assert_eq!(model, "claude-attempted");
                assert!(
                    message == "no response from the CLI"
                        || message.contains("connection to claude subprocess failed"),
                    "a SetModelFailed failure text is required: {message}"
                );
            }
            other => panic!("expected SetModelFailed on the wedged set_model, got {other:?}"),
        }
    }
}