In-app auth secret creation flow. (#10420)

## Description
<!-- Please remember to add your design buddy onto the PR for review, if
it contains any UI changes! -->

Completes REMOTE-1613. 

## Screenshots / Videos
<!-- Attach screenshots or a short video demonstrating the change, where
appropriate. Remove this section if it is not relevant to your PR. -->

See looms:
- https://www.loom.com/share/034b88e4b4a940a4895793ad188accdc
- https://www.loom.com/share/5e73fb892deb488abaffaededc4d5d90

## Testing
<!--
How did you test this change? What automated tests did you add? If you
didn't add any new tests, what's your justification for not adding any?
-->

Tested locally that things WAI!

## Agent Mode
- [x] Warp Agent Mode - This PR was created via Warp's AI Agent Mode
duhhhh.

Author: abhishek-pandya

app/src/ai/auth_secret_types.rs

@@ -0,0 +1,115 @@
+use anyhow::{anyhow, Result};
+use warp_cli::agent::Harness;
+use warp_graphql::managed_secrets::ManagedSecretType;
+use warp_managed_secrets::ManagedSecretValue;
+
+pub struct AuthSecretTypeField {
+    pub label: &'static str,
+    pub optional: bool,
+}
+
+pub struct AuthSecretTypeInfo {
+    pub display_name: &'static str,
+    pub secret_type: ManagedSecretType,
+    pub fields: &'static [AuthSecretTypeField],
+}
+
+pub fn auth_secret_types_for_harness(harness: Harness) -> &'static [AuthSecretTypeInfo] {
+    match harness {
+        Harness::Claude => &CLAUDE_AUTH_SECRET_TYPES,
+        _ => &[],
+    }
+}
+
+pub fn build_managed_secret_value(
+    info: &AuthSecretTypeInfo,
+    field_values: &[String],
+) -> Result<ManagedSecretValue> {
+    if field_values.len() != info.fields.len() {
+        return Err(anyhow!(
+            "Expected {} field values, got {}",
+            info.fields.len(),
+            field_values.len()
+        ));
+    }
+    for (field, value) in info.fields.iter().zip(field_values.iter()) {
+        if !field.optional && value.trim().is_empty() {
+            return Err(anyhow!("Field '{}' is required", field.label));
+        }
+    }
+    match info.secret_type {
+        ManagedSecretType::AnthropicApiKey => Ok(ManagedSecretValue::anthropic_api_key(
+            field_values[0].clone(),
+        )),
+        ManagedSecretType::AnthropicBedrockApiKey => {
+            Ok(ManagedSecretValue::anthropic_bedrock_api_key(
+                field_values[0].clone(),
+                field_values[1].clone(),
+            ))
+        }
+        ManagedSecretType::AnthropicBedrockAccessKey => {
+            let session_token = if field_values[2].trim().is_empty() {
+                None
+            } else {
+                Some(field_values[2].clone())
+            };
+            Ok(ManagedSecretValue::anthropic_bedrock_access_key(
+                field_values[0].clone(),
+                field_values[1].clone(),
+                session_token,
+                field_values[3].clone(),
+            ))
+        }
+        ManagedSecretType::RawValue | ManagedSecretType::Dotenvx => Err(anyhow!(
+            "Auth secret type {:?} is not supported via the harness FTUX flow",
+            info.secret_type
+        )),
+    }
+}
+
+static CLAUDE_AUTH_SECRET_TYPES: [AuthSecretTypeInfo; 3] = [
+    AuthSecretTypeInfo {
+        display_name: "Anthropic API Key",
+        secret_type: ManagedSecretType::AnthropicApiKey,
+        fields: &[AuthSecretTypeField {
+            label: "ANTHROPIC_API_KEY",
+            optional: false,
+        }],
+    },
+    AuthSecretTypeInfo {
+        display_name: "Bedrock API Key",
+        secret_type: ManagedSecretType::AnthropicBedrockApiKey,
+        fields: &[
+            AuthSecretTypeField {
+                label: "AWS_BEARER_TOKEN_BEDROCK",
+                optional: false,
+            },
+            AuthSecretTypeField {
+                label: "AWS_REGION",
+                optional: false,
+            },
+        ],
+    },
+    AuthSecretTypeInfo {
+        display_name: "Bedrock Access Key",
+        secret_type: ManagedSecretType::AnthropicBedrockAccessKey,
+        fields: &[
+            AuthSecretTypeField {
+                label: "AWS_ACCESS_KEY_ID",
+                optional: false,
+            },
+            AuthSecretTypeField {
+                label: "AWS_SECRET_ACCESS_KEY",
+                optional: false,
+            },
+            AuthSecretTypeField {
+                label: "AWS_SESSION_TOKEN",
+                optional: true,
+            },
+            AuthSecretTypeField {
+                label: "AWS_REGION",
+                optional: false,
+            },
+        ],
+    },
+];

app/src/ai/cloud_agent_settings.rs

@@ -3,7 +3,10 @@
 //! This module contains user-specific settings for cloud agent features,
 //! such as remembering the last selected environment.
 
-use settings::{macros::define_settings_group, SupportedPlatforms, SyncToCloud};
+use std::collections::HashMap;
+
+use settings::{macros::define_settings_group, Setting as _, SupportedPlatforms, SyncToCloud};
+use warp_cli::agent::Harness;
 
 use crate::server::ids::SyncId;
 
@@ -14,5 +17,32 @@ define_settings_group!(CloudAgentSettings, settings: [
         supported_platforms: SupportedPlatforms::ALL,
         sync_to_cloud: SyncToCloud::Never,
         private: true,
+    },
+    harness_auth_ftux_completed: HarnessAuthFtuxCompleted {
+        type: HashMap<String, bool>,
+        default: HashMap::new(),
+        supported_platforms: SupportedPlatforms::ALL,
+        sync_to_cloud: SyncToCloud::Never,
+        private: true,
     }
 ]);
+
+impl CloudAgentSettings {
+    pub fn is_harness_auth_ftux_completed(&self, harness: Harness) -> bool {
+        self.harness_auth_ftux_completed
+            .value()
+            .get(harness.config_name())
+            .copied()
+            .unwrap_or(false)
+    }
+
+    pub fn mark_harness_auth_ftux_completed(
+        &mut self,
+        harness: Harness,
+        ctx: &mut warpui::ModelContext<Self>,
+    ) {
+        let mut map = self.harness_auth_ftux_completed.value().clone();
+        map.insert(harness.config_name().to_string(), true);
+        let _ = self.harness_auth_ftux_completed.set_value(map, ctx);
+    }
+}

app/src/ai/harness_availability.rs

@@ -1,7 +1,10 @@
+use std::collections::HashMap;
+
 use serde::{Deserialize, Serialize};
 use warp_cli::agent::Harness;
 use warp_core::features::FeatureFlag;
 use warp_core::user_preferences::GetUserPreferences;
+use warp_managed_secrets::{client::SecretOwner, ManagedSecretManager, ManagedSecretValue};
 use warpui::{Entity, ModelContext, SingletonEntity};
 
 use crate::ai::harness_display;
@@ -41,12 +44,29 @@ fn default_harnesses() -> Vec<HarnessAvailability> {
     }]
 }
 
+#[derive(Debug, Clone)]
+pub enum AuthSecretFetchState {
+    NotFetched,
+    Loading,
+    Loaded(Vec<AuthSecretEntry>),
+    Failed(#[allow(dead_code)] String),
+}
+
+#[derive(Debug, Clone)]
+pub struct AuthSecretEntry {
+    pub name: String,
+}
+
 pub enum HarnessAvailabilityEvent {
     Changed,
+    AuthSecretsLoaded,
+    AuthSecretCreated { harness: Harness, name: String },
+    AuthSecretCreationFailed { error: String },
 }
 
 pub struct HarnessAvailabilityModel {
     harnesses: Vec<HarnessAvailability>,
+    auth_secrets: HashMap<Harness, AuthSecretFetchState>,
 }
 
 impl HarnessAvailabilityModel {
@@ -64,6 +84,10 @@ impl HarnessAvailabilityModel {
 
         ctx.subscribe_to_model(&AuthManager::handle(ctx), |me, event, ctx| {
             if let AuthManagerEvent::AuthComplete = event {
+                let cached_harnesses: Vec<Harness> = me.auth_secrets.keys().copied().collect();
+                for harness in cached_harnesses {
+                    me.invalidate_auth_secrets(harness);
+                }
                 me.refresh(ctx);
             }
         });
@@ -74,7 +98,10 @@ impl HarnessAvailabilityModel {
             }
         });
 
-        let me = Self { harnesses };
+        let me = Self {
+            harnesses,
+            auth_secrets: HashMap::new(),
+        };
         me.refresh(ctx);
         me
     }
@@ -116,6 +143,101 @@ impl HarnessAvailabilityModel {
             .filter(|m| !m.is_empty())
     }
 
+    pub fn auth_secrets_for(&self, harness: Harness) -> &AuthSecretFetchState {
+        self.auth_secrets
+            .get(&harness)
+            .unwrap_or(&AuthSecretFetchState::NotFetched)
+    }
+
+    pub fn ensure_auth_secrets_fetched(&mut self, harness: Harness, ctx: &mut ModelContext<Self>) {
+        if matches!(
+            self.auth_secrets_for(harness),
+            AuthSecretFetchState::NotFetched | AuthSecretFetchState::Failed(_)
+        ) {
+            self.fetch_auth_secrets(harness, ctx);
+        }
+    }
+
+    fn fetch_auth_secrets(&mut self, harness: Harness, ctx: &mut ModelContext<Self>) {
+        let Some(agent_harness) = harness_to_graphql_harness(harness) else {
+            return;
+        };
+
+        if !AuthStateProvider::as_ref(ctx).get().is_logged_in() {
+            return;
+        }
+
+        self.auth_secrets
+            .insert(harness, AuthSecretFetchState::Loading);
+
+        let api = ServerApiProvider::as_ref(ctx).get_managed_secrets_client();
+        ctx.spawn(
+            async move { api.list_harness_auth_secrets(agent_harness).await },
+            move |me, result: Result<Vec<warp_graphql::managed_secrets::ManagedSecret>, _>, ctx| {
+                match result {
+                    Ok(secrets) => {
+                        let entries = secrets
+                            .into_iter()
+                            .map(|s| AuthSecretEntry { name: s.name })
+                            .collect();
+                        me.auth_secrets
+                            .insert(harness, AuthSecretFetchState::Loaded(entries));
+                        ctx.emit(HarnessAvailabilityEvent::AuthSecretsLoaded);
+                    }
+                    Err(e) => {
+                        let msg = e.to_string();
+                        report_error!(e.context("Failed to fetch harness auth secrets"));
+                        me.auth_secrets
+                            .insert(harness, AuthSecretFetchState::Failed(msg));
+                    }
+                }
+            },
+        );
+    }
+
+    pub fn invalidate_auth_secrets(&mut self, harness: Harness) {
+        self.auth_secrets.remove(&harness);
+    }
+
+    pub fn create_auth_secret(
+        &mut self,
+        harness: Harness,
+        name: String,
+        value: ManagedSecretValue,
+        ctx: &mut ModelContext<Self>,
+    ) {
+        let manager = ManagedSecretManager::handle(ctx);
+        let create_future =
+            manager
+                .as_ref(ctx)
+                .create_secret(SecretOwner::CurrentUser, name, value, None);
+        ctx.spawn(create_future, move |me, result, ctx| match result {
+            Ok(secret) => {
+                let entry = AuthSecretEntry {
+                    name: secret.name.clone(),
+                };
+                match me.auth_secrets.get_mut(&harness) {
+                    Some(AuthSecretFetchState::Loaded(entries)) => {
+                        entries.push(entry);
+                    }
+                    _ => {
+                        me.auth_secrets
+                            .insert(harness, AuthSecretFetchState::Loaded(vec![entry]));
+                    }
+                }
+                ctx.emit(HarnessAvailabilityEvent::AuthSecretCreated {
+                    harness,
+                    name: secret.name,
+                });
+            }
+            Err(e) => {
+                let msg = e.to_string();
+                report_error!(e.context("Failed to create harness auth secret"));
+                ctx.emit(HarnessAvailabilityEvent::AuthSecretCreationFailed { error: msg });
+            }
+        });
+    }
+
     pub fn refresh(&self, ctx: &mut ModelContext<Self>) {
         // The endpoint queries `user`, which requires auth.
         if !AuthStateProvider::as_ref(ctx).get().is_logged_in() {
@@ -130,6 +252,11 @@ impl HarnessAvailabilityModel {
                     if new_harnesses != me.harnesses {
                         me.harnesses = new_harnesses;
                         me.cache(ctx);
+                        // Invalidate cached auth secrets so the next menu open refetches.
+                        let stale: Vec<Harness> = me.auth_secrets.keys().copied().collect();
+                        for harness in stale {
+                            me.invalidate_auth_secrets(harness);
+                        }
                         ctx.emit(HarnessAvailabilityEvent::Changed);
                     }
                 }
@@ -160,6 +287,16 @@ fn get_cached(ctx: &ModelContext<HarnessAvailabilityModel>) -> Option<Vec<Harnes
     serde_json::from_str::<Vec<HarnessAvailability>>(&raw).ok()
 }
 
+fn harness_to_graphql_harness(harness: Harness) -> Option<warp_graphql::ai::AgentHarness> {
+    match harness {
+        Harness::Oz => Some(warp_graphql::ai::AgentHarness::Oz),
+        Harness::Claude => Some(warp_graphql::ai::AgentHarness::ClaudeCode),
+        Harness::Gemini => Some(warp_graphql::ai::AgentHarness::Gemini),
+        Harness::Codex => Some(warp_graphql::ai::AgentHarness::Codex),
+        Harness::OpenCode | Harness::Unknown => None,
+    }
+}
+
 impl Entity for HarnessAvailabilityModel {
     type Event = HarnessAvailabilityEvent;
 }

app/src/ai/mod.rs

@@ -13,6 +13,7 @@ pub mod ambient_agents;
 pub(crate) mod artifact_download;
 pub mod artifacts;
 pub(crate) mod attachment_utils;
+pub mod auth_secret_types;
 #[cfg(not(target_family = "wasm"))]
 pub mod aws_credentials;
 pub(crate) mod block_context;