refactor(server): split server.rs into concern modules

bt-daemon/src/server/auth.rs

@@ -0,0 +1,209 @@
+//! Per-session authentication: resolving a route to a backend lease through
+//! the host, refreshing it before flushes, and diagnosing failures.
+
+use super::*;
+
+pub(super) fn lease_is_expiring(lease: &AuthLease) -> bool {
+    const REFRESH_WINDOW_MS: i64 = 60_000;
+    let Some(expires_at_ms) = lease.expires_at_ms else {
+        return false;
+    };
+    expires_at_ms <= now_ms().saturating_add(REFRESH_WINDOW_MS)
+}
+
+/// Resolve a route's lease the way event delivery does, including the
+/// organization the route requires.
+pub(crate) async fn resolve_route_auth(
+    provider: &dyn AuthProvider,
+    selection: &AuthSelection,
+    reason: AuthResolveReason,
+    required_org: Option<&str>,
+) -> anyhow::Result<AuthLease> {
+    let lease = provider.resolve(selection, reason).await?;
+    if let Some(expected_org) = required_org {
+        if lease.auth.org_name.as_deref() != Some(expected_org) {
+            anyhow::bail!(
+                "profile {:?} resolved organization {:?}, expected {:?}",
+                lease.selection,
+                lease.auth.org_name,
+                expected_org
+            );
+        }
+    }
+    Ok(lease)
+}
+
+impl Daemon {
+    pub(super) async fn configure_event(&self, env: &mut Envelope) -> anyhow::Result<DeliveryKey> {
+        let Some(provider) = &self.auth_provider else {
+            anyhow::bail!("daemon host has no Braintrust auth provider");
+        };
+        let requested_route = env
+            .route
+            .clone()
+            .ok_or_else(|| anyhow::anyhow!("event is missing its session route"))?;
+        if requested_route.destination.is_none() {
+            anyhow::bail!(
+                "session route is missing its trace destination; select a project or destination during `bt trace setup` or `bt trace run`"
+            );
+        }
+        let requested_key = DeliveryKey::new(&env.source, &env.session_id, &requested_route)?;
+        let key = self
+            .route_aliases
+            .lock()
+            .unwrap()
+            .get(&requested_key)
+            .cloned()
+            .unwrap_or_else(|| requested_key.clone());
+        let (selection, reason, expected_selection) = {
+            let states = self.session_auth.lock().await;
+            match states.get(&key) {
+                Some(state) => {
+                    if !lease_is_expiring(&state.lease) {
+                        env.config = Some(state.route.with_auth(state.lease.auth.clone()));
+                        return Ok(key);
+                    }
+                    (
+                        state.lease.selection.clone(),
+                        AuthResolveReason::Expiring,
+                        Some(state.lease.selection.clone()),
+                    )
+                }
+                None => (
+                    requested_route.auth.clone(),
+                    AuthResolveReason::Initial,
+                    None,
+                ),
+            }
+        };
+
+        let lease = resolve_route_auth(
+            provider.as_ref(),
+            &selection,
+            reason,
+            requested_route.auth.org_name.as_deref(),
+        )
+        .await
+        .map_err(|error| {
+            let message = format!(
+                "could not resolve Braintrust auth for {}: {error}; run `bt login` or select a profile explicitly",
+                env.source
+            );
+            self.record_auth_error(&key, &env.source, message.clone());
+            anyhow::anyhow!(message)
+        })?;
+        if let Some(expected) = expected_selection {
+            if lease.selection != expected {
+                anyhow::bail!(
+                    "credential refresh changed auth selection from {expected:?} to {:?}",
+                    lease.selection
+                );
+            }
+        }
+
+        let mut route = requested_route;
+        if route.auth.effective_source() == crate::wire::AuthSource::SavedProfile
+            && route.auth.profile_id.is_none()
+            && lease.selection.profile_id.is_some()
+        {
+            route.auth = lease.selection.clone();
+            if let Err(error) = crate::settings::migrate_persisted_route(
+                &env.source,
+                env.route.as_ref().expect("route checked above"),
+                &route,
+            ) {
+                tracing::warn!(source = %env.source, "could not migrate legacy profile route: {error}");
+            }
+        }
+        let canonical_key = DeliveryKey::new(&env.source, &env.session_id, &route)?;
+        if canonical_key != requested_key {
+            self.route_aliases
+                .lock()
+                .unwrap()
+                .insert(requested_key, canonical_key.clone());
+        }
+        env.route = Some(route.clone());
+        env.config = Some(route.with_auth(lease.auth.clone()));
+        self.session_auth
+            .lock()
+            .await
+            .insert(canonical_key.clone(), SessionAuthState { route, lease });
+        self.auth_errors.lock().unwrap().remove(&canonical_key);
+        Ok(canonical_key)
+    }
+
+    /// Keep a route's auth failure visible to `status.get` until it succeeds.
+    pub(super) fn record_auth_error(&self, key: &DeliveryKey, source: &str, message: String) {
+        self.auth_errors
+            .lock()
+            .unwrap()
+            .insert(key.clone(), (source.to_string(), message));
+    }
+
+    /// Resolve a route's credentials exactly as event delivery would, so
+    /// diagnostics see this process's credential store and environment rather
+    /// than the caller's.
+    pub(super) async fn diagnose_auth(&self, selection: AuthSelection) -> AuthDiagnoseResult {
+        let result = match &self.auth_provider {
+            Some(provider) => {
+                resolve_route_auth(
+                    provider.as_ref(),
+                    &selection,
+                    AuthResolveReason::Initial,
+                    selection.org_name.as_deref(),
+                )
+                .await
+            }
+            None => Err(anyhow::anyhow!(
+                "daemon host has no Braintrust auth provider"
+            )),
+        };
+        match result {
+            Ok(lease) => AuthDiagnoseResult {
+                selection: Some(lease.selection),
+                org_name: lease.auth.org_name,
+                expires_at_ms: lease.expires_at_ms,
+                error: None,
+            },
+            Err(error) => AuthDiagnoseResult {
+                error: Some(error.to_string()),
+                ..AuthDiagnoseResult::default()
+            },
+        }
+    }
+
+    pub(super) async fn refresh_session_before_flush(
+        &self,
+        key: &DeliveryKey,
+    ) -> anyhow::Result<()> {
+        let Some(provider) = &self.auth_provider else {
+            return Ok(());
+        };
+        let Some(state) = self.session_auth.lock().await.get(key).cloned() else {
+            return Ok(());
+        };
+        if !lease_is_expiring(&state.lease) {
+            return Ok(());
+        }
+        let selection = state.lease.selection.clone();
+        let lease = provider
+            .resolve(&selection, AuthResolveReason::Expiring)
+            .await?;
+        if lease.selection != state.lease.selection {
+            anyhow::bail!("credential refresh changed the session auth selection");
+        }
+        let config = state.route.with_auth(lease.auth.clone());
+        self.session_auth.lock().await.insert(
+            key.clone(),
+            SessionAuthState {
+                route: state.route,
+                lease,
+            },
+        );
+        let session = { self.sessions.lock().unwrap().get(key).cloned() };
+        if let Some(session) = session {
+            session.configure(config).await?;
+        }
+        Ok(())
+    }
+}

bt-daemon/src/server/claude_transcripts.rs

@@ -0,0 +1,148 @@
+//! Claude Code transcript polling. The daemon remembers each Claude session's
+//! transcript and emits synthetic `TranscriptUpdate` events for it, so records
+//! written after the session's last hook are still translated.
+
+use super::*;
+
+#[derive(Clone)]
+pub(super) struct ClaudeTranscriptObservation {
+    /// Synthetic-event provenance only: never retain native prompts or responses.
+    pub(super) template: Envelope,
+    pub(super) through: Option<u64>,
+}
+
+impl Daemon {
+    pub(super) fn remember_claude_transcripts(&self, env: &Envelope) {
+        if env.source != "claude-code" || env.payload.get("_bt_import_through_offset").is_some() {
+            return;
+        }
+        if env.event == "SessionEnd" {
+            self.claude_transcripts
+                .lock()
+                .unwrap()
+                .retain(|key, _| key.source != env.source || key.session_id != env.session_id);
+            return;
+        }
+        let Some(route) = env.route.as_ref() else {
+            return;
+        };
+        let Some(path) = env.payload.get("transcript_path").and_then(Value::as_str) else {
+            return;
+        };
+        let Ok(key) = DeliveryKey::new(&env.source, &env.session_id, route) else {
+            return;
+        };
+        let through = env
+            .payload
+            .get("_bt_claude_transcript_mirrors")
+            .and_then(|mirrors| mirrors.get(path))
+            .or_else(|| {
+                env.payload
+                    .get("_bt_transcript_mirror")
+                    .filter(|mirror| mirror.get("path").and_then(Value::as_str) == Some(path))
+            })
+            .and_then(|mirror| mirror.get("through"))
+            .and_then(Value::as_u64);
+        let mut payload = serde_json::json!({"transcript_path": path});
+        if let Some(cwd) = env.payload.get("cwd").and_then(Value::as_str) {
+            payload["cwd"] = Value::String(cwd.to_string());
+        }
+        let template = Envelope {
+            source: env.source.clone(),
+            source_version: env.source_version.clone(),
+            plugin_version: env.plugin_version.clone(),
+            session_id: env.session_id.clone(),
+            event: "TranscriptUpdate".to_string(),
+            ts_ms: env.ts_ms,
+            managed_run_id: env.managed_run_id.clone(),
+            capture: None,
+            payload,
+            route: env.route.clone(),
+            config: None,
+        };
+        let mut observations = self.claude_transcripts.lock().unwrap();
+        let previous = observations
+            .get(&key)
+            .filter(|observation| {
+                observation
+                    .template
+                    .payload
+                    .get("transcript_path")
+                    .and_then(Value::as_str)
+                    == Some(path)
+            })
+            .and_then(|observation| observation.through);
+        observations.insert(
+            key,
+            ClaudeTranscriptObservation {
+                template,
+                through: through.or(previous),
+            },
+        );
+    }
+
+    /// Poll native paths only on the capture side. Each observed boundary is
+    /// journaled before translation, including observations made during shutdown.
+    pub(super) async fn observe_claude_transcripts(
+        &self,
+        only: Option<&DeliveryKey>,
+        quiesced: bool,
+    ) {
+        // Shutdown already owns the exclusive gate. Ordinary polling must not
+        // append after that final drain begins.
+        let _gate = if quiesced {
+            None
+        } else {
+            Some(self.capture_gate.read().await)
+        };
+        if !quiesced && self.quiescing.load(Ordering::SeqCst) {
+            return;
+        }
+        let keys: Vec<_> = self
+            .claude_transcripts
+            .lock()
+            .unwrap()
+            .keys()
+            .filter(|key| {
+                only.is_none_or(|only| {
+                    key.source == only.source && key.session_id == only.session_id
+                })
+            })
+            .cloned()
+            .collect();
+        for key in keys {
+            // Routes share mirror files, so serialize by source session, not
+            // destination. Never hold this lock across an ingress barrier.
+            let lock = self.session_lock(&key.source, &key.session_id);
+            let _guard = lock.lock().await;
+            let observation = self.claude_transcripts.lock().unwrap().get(&key).cloned();
+            let Some(observation) = observation else {
+                continue;
+            };
+            let path = observation.template.payload["transcript_path"]
+                .as_str()
+                .unwrap();
+            let Ok(metadata) = tokio::fs::metadata(path).await else {
+                continue;
+            };
+            if observation.through == Some(metadata.len()) {
+                continue;
+            }
+            let mut env = observation.template;
+            env.ts_ms = now_ms();
+            hydrate_transcript_reference(&self.data_dir, &mut env).await;
+            let through = env
+                .payload
+                .get("_bt_claude_transcript_mirrors")
+                .and_then(|mirrors| mirrors.get(env.payload["transcript_path"].as_str().unwrap()))
+                .and_then(|mirror| mirror.get("through"))
+                .and_then(Value::as_u64);
+            if through.is_none() || through == observation.through {
+                continue;
+            }
+            if let Err(error) = self.journal_and_enqueue(env).await {
+                tracing::warn!(session_id = %key.session_id, %error, "Claude transcript observation failed");
+            }
+        }
+    }
+}