pub mod ai;pub mod auth;pub mod block;#[cfg(not(target_family = "wasm"))]pub(crate) mod download;pub mod factory;pub mod harness_support;pub mod integrations;pub mod managed_mcp;pub mod managed_secrets;pub mod object;pub(crate) mod presigned_upload;pub mod referral;pub mod team;#[cfg(feature = "tui")]pub mod tui_onboarding;pub mod workspace; use std::ops::Deref;use std::path::Path;use std::sync::Arc;use std::time::Duration; use ::http::header::CONTENT_LENGTH;use ai::AIClient;use anyhow::{Context, Result, anyhow};use auth::AuthClient;use block::BlockClient;use channel_versions::ChannelVersions;use chrono::{DateTime, FixedOffset};use factory::FactoryClient;use instant::Instant;use managed_mcp::ManagedMcpClient;use managed_secrets::AppManagedSecretsClient;use object::ObjectClient;use parking_lot::Mutex;use referral::ReferralsClient;use reqwest::StatusCode;use serde::{Deserialize, Serialize};use team::TeamClient;#[cfg(feature = "tui")]use tui_onboarding::TuiOnboardingClient;use url::Url;use warp_core::context_flag::ContextFlag;use warp_core::telemetry::TelemetryEvent;use warp_errors::{AnyhowErrorExt, ErrorExt, register_error, report_error};use warp_server_client::HttpStatusError;use warp_server_client::auth::{AuthClientImpl, AuthEvent, EXPERIMENT_ID_HEADER};use warp_server_client::base_client::{    AmbientHeaderPolicy, AuthenticatedGraphqlConfig, BaseClient, GraphqlRoutingConfig,    TEAM_UID_HEADER,};use warp_server_client::iap::{IapManager, IapState};use warp_server_client::network_logging::NetworkLogModel;use warpui::r#async::BoxFuture;use warpui::{Entity, ModelContext, SingletonEntity};use workspace::WorkspaceClient; use super::experiments::{ServerExperiment, ServerExperiments};use crate::ai::ambient_agents::AmbientAgentTaskId;use crate::ai::get_relevant_files::api::{GetRelevantFiles, GetRelevantFilesResponse};use crate::ai::predict::generate_ai_input_suggestions::GenerateAIInputSuggestionsRequest;use crate::ai::predict::generate_am_query_suggestions::GenerateAMQuerySuggestionsRequest;use crate::ai::predict::predict_am_queries::{PredictAMQueriesRequest, PredictAMQueriesResponse};use crate::ai::predict::{generate_ai_input_suggestions, generate_am_query_suggestions};use crate::ai::voice::transcribe::{TranscribeRequest, TranscribeResponse};use crate::auth::auth_manager::AuthManager;use crate::auth::auth_state::AuthState;use crate::server::team_scope::RequestTeamScope;use crate::server::telemetry::TelemetryApi;use crate::settings::PrivacySettingsSnapshot;use crate::{ChannelState, settings_view}; pub const FETCH_CHANNEL_VERSIONS_TIMEOUT: std::time::Duration = Duration::from_secs(60);#[derive(Serialize)]struct AgentTipShownAnalyticsRequest {    tip: String,} /// We use a special error code header `X-Warp-Error-Code` to allow the server to send/// more specific error code information, so that the client can discern between different/// errors with the same error code./// See errors/http_error_codes.go on the server for possible values.const WARP_ERROR_CODE_HEADER: &str = "X-Warp-Error-Code"; /// An error indicating the user is out of credits. The server sends 429s to communicate this/// state, but if Cloud Run is overloaded, it can also send 429s that aren't credit-related./// So we use this to distinguish between the two cases.const WARP_ERROR_CODE_OUT_OF_CREDITS: &str = "OUT_OF_CREDITS"; /// Error code indicating the user has reached their cloud agent concurrency limit.const WARP_ERROR_CODE_AT_CAPACITY: &str = "AT_CLOUD_AGENT_CAPACITY"; /// ResponseType received by Client#[derive(thiserror::Error, Debug, Serialize, Deserialize)]#[error("{error}")]pub struct ClientError {    pub error: String,    // We unconditionally check for GitHub auth errors in any public API response. It'd be much better    // to have the server return error codes that we can parse, but this isn't yet supported.    // See REMOTE-666    #[serde(skip_serializing_if = "Option::is_none")]    pub auth_url: Option<String>,} impl Deref for ServerApi {    type Target = BaseClient;     fn deref(&self) -> &Self::Target {        &self.base_client    }} /// Error when the user is at their cloud agent concurrency limit.#[derive(thiserror::Error, Debug, Clone, Deserialize)]#[error("{error} (running agents: {running_agents})")]pub struct CloudAgentCapacityError {    pub error: String,    pub running_agents: i32,} #[derive(Deserialize, Debug)]struct TimeResponse {    current_time: DateTime<FixedOffset>,} #[derive(Debug, Clone)]pub struct ServerTime {    time_at_fetch: DateTime<FixedOffset>,    fetched_at: Instant,} impl ServerTime {    pub fn current_time(&self) -> DateTime<FixedOffset> {        let elapsed = chrono::Duration::from_std(self.fetched_at.elapsed())            .expect("duration should not be bigger than limit");        self.time_at_fetch + elapsed    }} /// Wrapper for deserialization errors. This covers both:/// * Using `serde` directly/// * Using `reqwest` decoding utilities#[derive(thiserror::Error, Debug)]pub enum DeserializationError {    #[error(transparent)]    Json(#[from] serde_json::Error),    #[error(transparent)]    Transport(reqwest::Error),} #[derive(Deserialize, Debug)]struct OutOfCreditsResponse {    #[serde(default, rename = "userDisplayMessage")]    user_display_message: Option<String>,} #[derive(thiserror::Error, Debug)]pub enum AIApiError {    #[error("Request failed due to lack of AI quota.")]    QuotaLimit {        user_display_message: Option<String>,    },     #[error("Warp is currently overloaded. Please try again later.")]    ServerOverloaded,     #[error("Internal error occurred at transport layer.")]    Transport(#[source] reqwest::Error),     #[error("Failed to deserialize API response.")]    Deserialization(#[source] DeserializationError),     #[error("No context found on context search.")]    NoContextFound,     #[error("Failed with status code {0}: {1}")]    ErrorStatus(http::StatusCode, String),     #[error(transparent)]    Other(#[from] anyhow::Error),     #[error("Got error when streaming {stream_type}: {source:#}")]    Stream {        stream_type: &'static str,        #[source]        source: anyhow::Error,    },     /// Synthesized client-side when a response stream ends without a stream-finished    /// event: the server always sends one, but the transport can truncate the response    /// between chunks, surfacing as a clean EOF.    #[error("Response stream ended unexpectedly before completion.")]    UnexpectedEof,     /// Synthesized client-side when a request that uses the connected Grok    /// subscription can't be sent because its expired OAuth token failed to    /// refresh. Surfaced as a terminal, user-visible error asking the user to    /// reconnect, rather than sending a request that would fail authentication.    #[error(        "Grok subscription token could not be refreshed. Please try reconnecting your subscription."    )]    GrokSubscriptionTokenRefreshFailed,} impl From<http_client::ResponseError> for AIApiError {    fn from(err: http_client::ResponseError) -> Self {        let http_client::ResponseError {            source,            headers,            body,        } = err;        Self::from_response_error(source, &headers, body)    }} impl From<reqwest::Error> for AIApiError {    fn from(err: reqwest::Error) -> Self {        Self::from_transport_error(err)    }} impl From<serde_json::Error> for AIApiError {    fn from(err: serde_json::Error) -> Self {        AIApiError::Deserialization(err.into())    }} impl AIApiError {    /// Converts a reqwest error to an AIApiError, using response headers to distinguish    /// between different types of 429 errors.    fn from_response_error(        err: reqwest::Error,        headers: &::http::HeaderMap,        body: Option<String>,    ) -> Self {        // For HTTP 429 errors, check the X-Warp-Error-Code header to distinguish        // between out-of-credits and server-overload.        if err.status() == Some(http::StatusCode::TOO_MANY_REQUESTS) {            return Self::error_for_429(headers, body);        }         Self::from_transport_error(err)    }     /// Converts a transport-level reqwest error (no HTTP response) to an AIApiError.    fn from_transport_error(err: reqwest::Error) -> Self {        // Unfortunately, `reqwest` reports some non-decoding errors as decoding errors (e.g.        // unexpected disconnects or timeouts while deserializing a response body). Since we        // render deserialization and transport errors differently, we try to detect those cases        // here.        if err.is_timeout() {            return AIApiError::Transport(err);        }        if err.is_decode() {            #[cfg(not(target_family = "wasm"))]            {                use std::error::Error as _;                let mut source = err.source();                while let Some(underlying) = source {                    if underlying.is::<hyper::Error>() {                        return AIApiError::Transport(err);                    }                     source = underlying.source();                }            }             return AIApiError::Deserialization(DeserializationError::Transport(err));        }         AIApiError::Transport(err)    }     /// Returns the appropriate error for a 429 response by checking the X-Warp-Error-Code header.    fn error_for_429(headers: &::http::HeaderMap, body: Option<String>) -> Self {        if headers            .get(WARP_ERROR_CODE_HEADER)            .and_then(|v| v.to_str().ok())            == Some(WARP_ERROR_CODE_OUT_OF_CREDITS)        {            let user_display_message = body                .and_then(|body| serde_json::from_str::<OutOfCreditsResponse>(&body).ok())                .and_then(|r| r.user_display_message);            AIApiError::QuotaLimit {                user_display_message,            }        } else {            AIApiError::ServerOverloaded        }    }     /// Format a stream error into a human-readable error message. This will read the response    /// body if there is one.    pub(crate) async fn from_stream_error(        stream_type: &'static str,        err: reqwest_eventsource::Error,    ) -> Self {        match err {            reqwest_eventsource::Error::InvalidStatusCode(                http::StatusCode::TOO_MANY_REQUESTS,                res,            ) => {                let headers = res.headers().clone();                let body = res.text().await.ok();                Self::error_for_429(&headers, body)            }            reqwest_eventsource::Error::InvalidStatusCode(status, res) => Self::ErrorStatus(                status,                res.text()                    .await                    .unwrap_or_else(|e| format!("(no response body: {e:#})")),            ),            reqwest_eventsource::Error::Transport(err) => Self::from_transport_error(err),            err => AIApiError::Stream {                stream_type,                // On WASM, `reqwest_eventsource::Error` doesn't implement `Into<anyhow::Error>` or                // `Send` because it may contain a `wasm_bindgen` JS value.                #[cfg(target_family = "wasm")]                source: anyhow!("{err:#?}"),                #[cfg(not(target_family = "wasm"))]                source: anyhow!(err),            },        }    }     /// Whether the error is worth an automatic recovery attempt — a fresh request may    /// succeed. Gates both retry (pre-actions) and resume (post-actions).    pub fn is_recoverable(&self) -> bool {        // Don't recover from client errors, except timeouts and rate limits.        fn is_recoverable_status(status: http::StatusCode) -> bool {            !status.is_client_error()                || status == http::StatusCode::REQUEST_TIMEOUT                || status == http::StatusCode::TOO_MANY_REQUESTS        }         match self {            AIApiError::ErrorStatus(status, _) => is_recoverable_status(*status),            AIApiError::Transport(e) => {                if let Some(status) = e.status() {                    return is_recoverable_status(status);                }                true            }            // A failed Grok token refresh is a credential problem the user must            // fix by reconnecting, so retrying or resuming won't help.            AIApiError::GrokSubscriptionTokenRefreshFailed => false,            // By default, attempt recovery on error.            _ => true,        }    }} impl ErrorExt for AIApiError {    fn is_actionable(&self) -> bool {        match self {            AIApiError::Deserialization(error) => match error {                DeserializationError::Json(_) => true,                DeserializationError::Transport(error) => error.is_actionable(),            },            AIApiError::Transport(error) => error.is_actionable(),            AIApiError::Other(error) => error.is_actionable(),            AIApiError::Stream { source, .. } => source.is_actionable(),            AIApiError::ErrorStatus(_, _) => self.is_recoverable(),            AIApiError::UnexpectedEof => true,            AIApiError::QuotaLimit { .. }            | AIApiError::ServerOverloaded            | AIApiError::NoContextFound            | AIApiError::GrokSubscriptionTokenRefreshFailed => false,        }    }}register_error!(AIApiError); #[derive(thiserror::Error, Debug)]pub enum TranscribeError {    #[error("Request failed due to lack of Voice quota.")]    QuotaLimit,     #[error("Warp is currently overloaded. Please try again later.")]    ServerOverloaded,     #[error("Internal error occurred at transport layer.")]    Transport(#[source] reqwest::Error),     #[error("Failed with status code {0}")]    ErrorStatus(http::StatusCode),     #[error("Failed to deserialize JSON.")]    Deserialization(#[source] DeserializationError),     #[error(transparent)]    Other(#[from] anyhow::Error),} impl TranscribeError {    fn from_json_error(err: reqwest::Error) -> Self {        if err.is_decode() {            #[cfg(not(target_family = "wasm"))]            {                use std::error::Error as _;                let mut source = err.source();                while let Some(underlying) = source {                    if underlying.is::<hyper::Error>() {                        return TranscribeError::Transport(err);                    }                    source = underlying.source();                }            }            return TranscribeError::Deserialization(DeserializationError::Transport(err));        }        TranscribeError::Transport(err)    }} impl ErrorExt for TranscribeError {    fn is_actionable(&self) -> bool {        match self {            TranscribeError::Transport(error) => error.is_actionable(),            TranscribeError::ErrorStatus(status) => {                !status.is_server_error() && *status != http::StatusCode::TOO_MANY_REQUESTS            }            TranscribeError::Other(error) => error.is_actionable(),            TranscribeError::Deserialization(error) => match error {                DeserializationError::Json(_) => true,                DeserializationError::Transport(error) => error.is_actionable(),            },            TranscribeError::QuotaLimit | TranscribeError::ServerOverloaded => false,        }    }}register_error!(TranscribeError); /// An API wrapper struct with methods to requests to warp-server.////// Prefer NOT adding new methods directly on this struct; instead, add to one of the existing/// client trait objects, or create your own. This helps keep `ServerApi` from being overloaded/// with disparate types of calls, and allows you to mock methods in tests.pub struct ServerApi {    base_client: Arc<BaseClient>,    // TODO(jeff): Make `TelemetryApi` another type of client, and move it off `ServerApi`.    telemetry_api: TelemetryApi,    last_server_time: Arc<Mutex<Option<ServerTime>>>,} impl ServerApi {    fn new(        auth_state: Arc<AuthState>,        event_sender: async_channel::Sender<AuthEvent>,        agent_source: Option<ai::AgentSource>,        iap_state: Option<Arc<IapState>>,        ctx: &mut ModelContext<ServerApiProvider>,    ) -> Self {        let mut client = http_client::Client::new();        let iap_token_provider = iap_state.map(|state| {            client.set_iap_token_provider(state.clone());            state as Arc<dyn http_client::iap::IapTokenProvider>        });        let mut telemetry_api = TelemetryApi::new();        if ContextFlag::NetworkLogConsole.is_enabled() {            NetworkLogModel::handle(ctx).update(ctx, |model, model_ctx| {                model.install_on_clients([&mut client, &mut telemetry_api.client], model_ctx);            });        }        Self::new_with_parts(            Arc::new(client),            auth_state,            event_sender,            agent_source,            iap_token_provider,            telemetry_api,        )    }     fn new_with_parts(        client: Arc<http_client::Client>,        auth_state: Arc<AuthState>,        event_sender: async_channel::Sender<AuthEvent>,        agent_source: Option<ai::AgentSource>,        iap_token_provider: Option<Arc<dyn http_client::iap::IapTokenProvider>>,        telemetry_api: TelemetryApi,    ) -> Self {        let graphql_routing = GraphqlRoutingConfig {            #[cfg(feature = "agent_mode_evals")]            path_prefix: Some("/agent-mode-evals".to_string()),            #[cfg(not(feature = "agent_mode_evals"))]            path_prefix: None,        };        let authenticated_graphql = AuthenticatedGraphqlConfig::default();        let base_client = Arc::new(BaseClient::new(            client,            auth_state,            event_sender,            agent_source.map(|source| source.as_str().to_string()),            graphql_routing,            authenticated_graphql,            iap_token_provider,        ));         Self {

Lines 1–500 of 1599. Load more