Merge ticket branch t086

src/cli/mod.rs

@@ -71,38 +71,67 @@ pub async fn run() {
 
     let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
 
+    let engine_tx = spawn_engine(&settings).await;
+    load_active_profile(&settings, &settings_path, &engine_tx).await;
+    spawn_ingest(&args, &engine_tx);
+
+    let app = crate::api::router(engine_tx, settings, settings_path.clone());
+    let listener = TcpListener::bind((crate::consts::DEFAULT_BIND_HOST, args.port))
+        .await
+        .expect("failed to bind listener");
+    let bound_port = listener
+        .local_addr()
+        .expect("failed to read back bound address")
+        .port();
+
+    register_session_and_log(&settings_path, &args, bound_port);
+
+    axum::serve(listener, app)
+        .await
+        .expect("server error");
+}
+
+async fn spawn_engine(settings: &Arc<Mutex<Settings>>) -> crate::engine::EngineHandle {
     let engine_tx = crate::engine::spawn(settings.clone(), crate::consts::ENGINE_CHANNEL_CAPACITY);
-    {
-        let max_logs = settings.lock().unwrap().max_logs;
-        let _ = engine_tx.tx.send(crate::engine::EngineMessage::SetMaxLogs(max_logs)).await;
-    }
+    let max_logs = settings.lock().unwrap().max_logs;
+    let _ = engine_tx.tx.send(crate::engine::EngineMessage::SetMaxLogs(max_logs)).await;
+    engine_tx
+}
 
-    {
-        let (active_profile_name, active_profile) = {
-            let mut settings_guard = settings.lock().unwrap();
-            settings_guard
-                .ensure_default_profile(&settings_path)
-                .expect("failed to ensure the Default profile exists");
-            let active_profile_name = settings_guard.active_profile.clone();
-            let active_profile = active_profile_name
-                .as_deref()
-                .and_then(|name| settings_guard.profile(name).cloned());
-            (active_profile_name, active_profile)
-        };
-        match (active_profile_name, active_profile) {
-            (Some(_), Some(active_profile)) => {
-                active_profile.load_into_engine(&engine_tx).await;
-            }
-            (Some(profile_name), None) => {
-                eprintln!(
-                    "warning: active profile '{profile_name}' not found in {}; skipping",
-                    settings_path.display()
-                );
-            }
-            (None, _) => {}
+/// Resolves the active profile (creating the Default profile if none exists yet) and, if
+/// found, replays its declarations into the engine before ingest starts — there's no way to
+/// reprocess earlier lines later, so this must happen before `spawn_ingest`.
+async fn load_active_profile(
+    settings: &Arc<Mutex<Settings>>,
+    settings_path: &std::path::Path,
+    engine_tx: &crate::engine::EngineHandle,
+) {
+    let (active_profile_name, active_profile) = {
+        let mut settings_guard = settings.lock().unwrap();
+        settings_guard
+            .ensure_default_profile(settings_path)
+            .expect("failed to ensure the Default profile exists");
+        let active_profile_name = settings_guard.active_profile.clone();
+        let active_profile = active_profile_name
+            .as_deref()
+            .and_then(|name| settings_guard.profile(name).cloned());
+        (active_profile_name, active_profile)
+    };
+    match (active_profile_name, active_profile) {
+        (Some(_), Some(active_profile)) => {
+            active_profile.load_into_engine(engine_tx).await;
         }
+        (Some(profile_name), None) => {
+            eprintln!(
+                "warning: active profile '{profile_name}' not found in {}; skipping",
+                settings_path.display()
+            );
+        }
+        (None, _) => {}
     }
+}
 
+fn spawn_ingest(args: &Args, engine_tx: &crate::engine::EngineHandle) {
     match args.file.clone() {
         Some(path) => {
             tokio::spawn(crate::engine::ingest::run_file(engine_tx.clone(), path));
@@ -115,22 +144,15 @@ pub async fn run() {
     if let Some(metrics_url) = args.metrics_url.clone() {
         tokio::spawn(crate::engine::ingest::run_metrics_poll(engine_tx.clone(), metrics_url));
     }
+}
 
-    let app = crate::api::router(engine_tx, settings, settings_path.clone());
-    let listener = TcpListener::bind((crate::consts::DEFAULT_BIND_HOST, args.port))
-        .await
-        .expect("failed to bind listener");
-    let bound_port = listener
-        .local_addr()
-        .expect("failed to read back bound address")
-        .port();
-
+fn register_session_and_log(settings_path: &std::path::Path, args: &Args, bound_port: u16) {
     let source = match &args.file {
         Some(path) => format!("tailing {}", path.display()),
         None => "reading logs from stdin".to_string(),
     };
     let session_name = args.name.clone().unwrap_or_else(sessions::generate_name);
-    match sessions::register_session(&settings_path, &session_name, bound_port) {
+    match sessions::register_session(settings_path, &session_name, bound_port) {
         Ok(_) => eprintln!(
             "ottyl serving on http://localhost:{} ({}), session name: {}",
             bound_port, source, session_name
@@ -140,10 +162,6 @@ pub async fn run() {
             bound_port, source, session_name, err
         ),
     }
-
-    axum::serve(listener, app)
-        .await
-        .expect("server error");
 }
 
 async fn run_mcp(settings_path: &std::path::Path, session: Option<String>, port: Option<u16>) {