OPS-248 Port critical upstream security and relay fixes

crates/buzz-relay/src/handlers/ingest.rs

@@ -196,6 +196,26 @@ fn emit_product_feedback_success(
     );
 }
 
+fn reaction_write_action(event: &Event, channel_id: Option<Uuid>, inserted: bool) -> TraceAction {
+    let claimed_community = claimed_community_from_event(event);
+    match (channel_id, inserted) {
+        (Some(channel), true) => TraceAction::WriteInsert {
+            msg_id: msg_id_label(event.id.as_bytes()),
+            channel: channel_label(channel),
+            claimed_community,
+        },
+        (Some(channel), false) => TraceAction::WriteDuplicate {
+            msg_id: msg_id_label(event.id.as_bytes()),
+            channel: channel_label(channel),
+            claimed_community,
+        },
+        (None, _) => TraceAction::WriteInsertGlobal {
+            msg_id: msg_id_label(event.id.as_bytes()),
+            claimed_community,
+        },
+    }
+}
+
 /// Increment the rejection counter with a bounded reason and transport label.
 ///
 /// Shared by the WS `EVENT` handler and the HTTP `POST /events` handler so
@@ -2824,6 +2844,8 @@ async fn ingest_event_inner(
                 ));
             }
             buzz_db::ReactionEventInsertOutcome::Duplicate => {
+                let action = reaction_write_action(&event, channel_id, false);
+                emit(tracer, action, state_for_request(tenant, auth.pubkey()));
                 return Ok(IngestResult {
                     event_id: event_id_hex,
                     accepted: false,
@@ -2837,25 +2859,14 @@ async fn ingest_event_inner(
         };
 
         let pubkey_hex = auth.pubkey().to_hex();
-        // Spec WriteInsert (line 514) / WriteDuplicate (line 606): emit
-        // the abstract write action. The persist API returns
-        // `was_inserted` (true → Insert, false → Duplicate). This branch
-        // is the reaction path; channel_id is always Some here, so
-        // WriteInsertGlobal does not apply.
-        let claimed = claimed_community_from_event(&event);
-        let action = if was_inserted {
-            TraceAction::WriteInsert {
-                msg_id: msg_id_label(event.id.as_bytes()),
-                channel: channel_label(channel_id.expect("reaction path has channel")),
-                claimed_community: claimed,
-            }
-        } else {
-            TraceAction::WriteDuplicate {
-                msg_id: msg_id_label(event.id.as_bytes()),
-                channel: channel_label(channel_id.expect("reaction path has channel")),
-                claimed_community: claimed,
-            }
-        };
+        // Spec WriteInsert (line 514) / WriteDuplicate (line 606) /
+        // WriteInsertGlobal (line 559): emit the abstract write action. The
+        // persist API returns `was_inserted` (true → Insert/Global, false →
+        // Duplicate). Reactions on project events (issue/PR roots and their
+        // comments) carry no `h` tag, so `channel_id` can be `None` here —
+        // mirror the message write's three-way split instead of asserting a
+        // channel, which panicked the ingest worker on those events.
+        let action = reaction_write_action(&event, channel_id, was_inserted);
         emit(tracer, action, state_for_request(tenant, auth.pubkey()));
         dispatch_persistent_event(
             tenant,
@@ -3280,6 +3291,24 @@ mod tests {
         assert!(!requires_h_channel_scope(KIND_REACTION));
     }
 
+    #[test]
+    fn reaction_duplicate_trace_covers_channel_and_global_targets() {
+        let keys = nostr::Keys::generate();
+        let event = EventBuilder::new(Kind::Custom(KIND_REACTION as u16), "+")
+            .sign_with_keys(&keys)
+            .expect("sign reaction");
+        let channel = Uuid::new_v4();
+
+        assert!(matches!(
+            reaction_write_action(&event, Some(channel), false),
+            TraceAction::WriteDuplicate { .. }
+        ));
+        assert!(matches!(
+            reaction_write_action(&event, None, false),
+            TraceAction::WriteInsertGlobal { .. }
+        ));
+    }
+
     #[test]
     fn long_form_is_in_scope_allowlist() {
         let dummy = make_dummy_event();

desktop/src-tauri/src/commands/agent_models.rs

@@ -5,6 +5,7 @@ use serde::Deserialize;
 use tauri::{AppHandle, State};
 
 use super::agent_model_process::run_agent_models_command;
+use super::managed_agent_definition::apply_model_provider_prompt_update;
 // The map-only lookup is reached solely from the base-URL helpers that exist for
 // their unit tests; discovery itself always goes through the process-env variant.
 #[cfg(test)]
@@ -696,35 +697,6 @@ use databricks::{
 };
 use databricks::{discover_databricks_models, DatabricksAuthIntent};
 
-/// Apply an `UpdateManagedAgentRequest`'s model/provider/system_prompt patch
-/// to `record`, enforcing the linked-instance write guard: a definition-linked
-/// record's model/provider/prompt are definition-authoritative (see
-/// `effective_config::resolve_linked`), so writes to these three fields are
-/// silently dropped for a linked instance rather than persisting a byte the
-/// resolver will never read. Definition-less instances accept the patch
-/// as-is. Extracted so the guard is exercised by both `update_managed_agent`
-/// and its regression tests — a test that reimplements this check instead of
-/// calling it can go green after the real guard is deleted.
-fn apply_model_provider_prompt_update(
-    record: &mut crate::managed_agents::ManagedAgentRecord,
-    model: Option<Option<String>>,
-    provider: Option<Option<String>>,
-    system_prompt: Option<Option<String>>,
-) {
-    if record.persona_id.is_some() {
-        return;
-    }
-    if let Some(model_update) = model {
-        record.model = model_update;
-    }
-    if let Some(provider_update) = provider {
-        record.provider = provider_update;
-    }
-    if let Some(prompt_update) = system_prompt {
-        record.system_prompt = prompt_update;
-    }
-}
-
 /// Update mutable fields on an existing managed agent record.
 ///
 /// Does NOT auto-restart the agent. Runtime config changes (system prompt,
@@ -769,7 +741,7 @@ pub async fn update_managed_agent(
             input.model,
             input.provider,
             input.system_prompt,
-        );
+        )?;
         if let Some(parallelism) = input.parallelism {
             record.parallelism = parallelism;
         }

desktop/src-tauri/src/commands/agent_models_databricks.rs

@@ -1,7 +1,8 @@
 //! Databricks v1/v2 model discovery and interactive reauthentication.
 
-use std::collections::BTreeMap;
-use std::sync::LazyLock;
+use std::collections::{BTreeMap, HashMap};
+use std::sync::{LazyLock, Mutex, MutexGuard};
+use std::time::{Duration, Instant};
 
 use crate::commands::agent_models_env::{
     env_or_process_value, redaction_env_with_value, DiscoveryProvider,
@@ -13,6 +14,77 @@ use crate::managed_agents::AgentModelsResponse;
 // callback listener/browser flow for the process-wide OAuth cache.
 static AUTH_GATE: LazyLock<tokio::sync::Mutex<()>> = LazyLock::new(|| tokio::sync::Mutex::new(()));
 
+// Hard cap on the interactive browser flow launched from a discovery surface.
+// An abandoned SSO tab must fail discovery cleanly rather than wedge the
+// dropdown forever. (`authenticate_databricks` has its own 60s callback wait;
+// this outer bound also covers endpoint discovery and token exchange.)
+const AUTH_FLOW_TIMEOUT: Duration = Duration::from_secs(150);
+
+// How long a failed/cancelled interactive sign-in suppresses re-launching the
+// browser from passive surfaces.
+pub(super) const AUTH_COOLDOWN: Duration = Duration::from_secs(5 * 60);
+
+/// Per-host record of a recently failed, cancelled, or timed-out interactive
+/// sign-in.
+///
+/// Passive discovery surfaces fire on every form-state change, so without this
+/// a cancelled SSO page would re-pop the browser on the very next keystroke.
+/// Entries expire so a genuine later retry still launches; the saved-model
+/// picker bypasses the cooldown and a success clears it.
+#[derive(Default)]
+pub(super) struct AuthCooldown {
+    until: Mutex<HashMap<String, Instant>>,
+}
+
+impl AuthCooldown {
+    fn map(&self) -> MutexGuard<'_, HashMap<String, Instant>> {
+        // The critical sections below are panic-free map ops, so recover from a
+        // poisoned lock rather than wedge every future sign-in on one panic.
+        self.until
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub(super) fn is_active(&self, host: &str, now: Instant) -> bool {
+        let mut map = self.map();
+        match map.get(host) {
+            Some(&expiry) if now < expiry => true,
+            Some(_) => {
+                map.remove(host);
+                false
+            }
+            None => false,
+        }
+    }
+
+    pub(super) fn record(&self, host: &str, now: Instant) {
+        self.map().insert(host.to_string(), now + AUTH_COOLDOWN);
+    }
+
+    pub(super) fn clear(&self, host: &str) {
+        self.map().remove(host);
+    }
+
+    /// Whether the interactive browser flow may launch now under `auth_intent`.
+    /// Passive surfaces are suppressed while a per-host cooldown is active; the
+    /// explicit picker path always launches and clears any stale suppression.
+    pub(super) fn permits_launch(
+        &self,
+        auth_intent: DatabricksAuthIntent,
+        host: &str,
+        now: Instant,
+    ) -> bool {
+        if auth_intent.respects_cooldown() {
+            !self.is_active(host, now)
+        } else {
+            self.clear(host);
+            true
+        }
+    }
+}
+
+static AUTH_COOLDOWNS: LazyLock<AuthCooldown> = LazyLock::new(AuthCooldown::default);
+
 pub(super) fn is_databricks_provider(provider: Option<&str>) -> bool {
     matches!(
         provider
@@ -50,8 +122,14 @@ pub(super) enum DatabricksAuthIntent {
 }
 
 impl DatabricksAuthIntent {
-    fn allows_interactive_auth(self) -> bool {
-        matches!(self, Self::InteractiveModelPicker)
+    /// Passive draft discovery honors (and, on failure, writes) the per-host
+    /// cooldown so a cancelled SSO page does not re-pop on the next form
+    /// keystroke. The saved-model picker is an explicit user action, so it
+    /// bypasses the cooldown and clears it before launching. Both surfaces
+    /// launch the browser flow (Phase 2 goose-parity); this predicate is the
+    /// only behavioral difference between them.
+    fn respects_cooldown(self) -> bool {
+        matches!(self, Self::PassiveDraftDiscovery)
     }
 }
 
@@ -60,11 +138,16 @@ pub(super) fn databricks_sign_in_required_error() -> String {
         .to_string()
 }
 
-pub(super) fn should_start_interactive_auth(
-    api_key: &str,
-    auth_intent: DatabricksAuthIntent,
-) -> bool {
-    api_key.is_empty() && auth_intent.allows_interactive_auth()
+pub(super) fn databricks_sign_in_timed_out_error() -> String {
+    "Databricks sign-in timed out; open the model picker to retry, or run `buzz-agent auth databricks`"
+        .to_string()
+}
+
+pub(super) fn should_start_interactive_auth(api_key: &str) -> bool {
+    // Phase 2: both discovery surfaces launch the browser flow when no static
+    // token is configured. Which surface is allowed to actually pop the browser
+    // (vs. respect a cooldown) is decided via `AuthCooldown::permits_launch`.
+    api_key.is_empty()
 }
 
 pub(super) async fn discover_databricks_models(
@@ -93,22 +176,26 @@ pub(super) async fn discover_databricks_models(
 
     let entries = match buzz_agent_pkg::discover_databricks_models(&config).await {
         Ok(entries) => entries,
-        Err(buzz_agent_pkg::AgentError::LlmAuth(_))
-            if should_start_interactive_auth(&api_key, auth_intent) =>
-        {
+        Err(buzz_agent_pkg::AgentError::LlmAuth(_)) if should_start_interactive_auth(&api_key) => {
             let _auth = AUTH_GATE.lock().await;
             match buzz_agent_pkg::discover_databricks_models(&config).await {
+                // A peer sign-in under the gate already succeeded.
                 Ok(entries) => entries,
                 Err(buzz_agent_pkg::AgentError::LlmAuth(_)) => {
-                    buzz_agent_pkg::authenticate_databricks(&host)
-                        .await
-                        .map_err(|error| {
-                            format_redacted_error(
-                                "Databricks sign-in failed",
-                                &error,
-                                &redaction_env,
-                            )
-                        })?;
+                    // Passive surfaces suppress the browser while a recent
+                    // failure/cancel is cooling down; the explicit picker path
+                    // always launches (and clears any stale cooldown).
+                    if !AUTH_COOLDOWNS.permits_launch(auth_intent, &host, Instant::now()) {
+                        return Err(databricks_sign_in_required_error());
+                    }
+                    run_interactive_databricks_auth(
+                        buzz_agent_pkg::authenticate_databricks(&host),
+                        AUTH_FLOW_TIMEOUT,
+                        &AUTH_COOLDOWNS,
+                        &host,
+                        &redaction_env,
+                    )
+                    .await?;
                     buzz_agent_pkg::discover_databricks_models(&config)
                         .await
                         .map_err(|error| {
@@ -172,3 +259,43 @@ fn format_redacted_error(
     let message = crate::managed_agents::redact_env_values_in(&error.to_string(), redaction_env);
     format!("{context}: {message}")
 }
+
+/// Run the interactive browser OAuth flow under a hard timeout and maintain the
+/// per-host cooldown. Success clears the cooldown; a failure, cancel, or
+/// timeout records it so passive surfaces stop re-launching the browser on the
+/// next form keystroke. `timeout` is injected (production passes
+/// [`AUTH_FLOW_TIMEOUT`]) so the timeout/cooldown policy is unit-testable
+/// without a live browser.
+pub(super) async fn run_interactive_databricks_auth<Fut>(
+    auth: Fut,
+    timeout: Duration,
+    cooldowns: &AuthCooldown,
+    host: &str,
+    redaction_env: &BTreeMap<String, String>,
+) -> Result<(), String>
+where
+    Fut: std::future::Future<Output = Result<(), buzz_agent_pkg::AgentError>>,
+{
+    match tokio::time::timeout(timeout, auth).await {
+        Ok(Ok(())) => {
+            cooldowns.clear(host);
+            Ok(())
+        }
+        Ok(Err(error)) => {
+            cooldowns.record(host, Instant::now());
+            Err(format_redacted_error(
+                "Databricks sign-in failed",
+                &error,
+                redaction_env,
+            ))
+        }
+        Err(_elapsed) => {
+            cooldowns.record(host, Instant::now());
+            Err(databricks_sign_in_timed_out_error())
+        }
+    }
+}
+
+#[cfg(test)]
+#[path = "agent_models_databricks_tests.rs"]
+mod tests;