Native: Complete server-client cycle

Author: sudeeptarlekar

application/apps/indexer/gui/application/src/session/service/mod.rs

@@ -8,7 +8,7 @@ use tokio::{
 use uuid::Uuid;
 
 use mcp::{client::McpClient, server::tasks::Tasks, types::Prompt};
-use processor::grabber::LineRange;
+use processor::{grabber::LineRange, search::filter::SearchFilter};
 use session_core::session::Session;
 use stypes::{CallbackEvent, ComputationError, ObserveOptions, ObserveOrigin, Transport};
 
@@ -122,7 +122,38 @@ impl SessionService {
                 },
 
                 Ok(task) = self.mcp_task_rx.recv() => {
-                    println!("🔵 Session Service {} received MCP task: {:?}", self.session_id(), task);
+                    match task {
+                        Tasks::ApplySearchFilter { session_id, filters, task_result_tx } => {
+                            if self.session_id() == session_id {
+                                let cmd = SessionCommand::ApplySearchFilter { operation_id: session_id, filters: filters.clone() };
+                                println!("🔵 Forwarding ApplySearchFilter command to Session Service {}: {:?}", self.session_id(), cmd);
+                                match self.handle_command(cmd, prompt_tx.clone()).await {
+                                    Ok(ControlFlow::Break(())) => {
+                                        println!("✅ Successfully handled ApplySearchFilter command with BREACK for session {}", self.session_id());
+
+                                        if let Err(err) = task_result_tx.send(Ok(())).await {
+                                            log::error!("Failed to send task result for ApplySearchFilter: {err:?}");
+                                        }
+                                        break
+                                    },
+                                    Ok(ControlFlow::Continue(())) => {
+                                        println!("✅ Successfully handled ApplySearchFilter command with CONTINUE for session {}", self.session_id());
+
+                                        if let Err(err) = task_result_tx.send(Ok(())).await {
+                                            log::error!("Failed to send task result for ApplySearchFilter: {err:?}");
+                                        }
+                                    },
+                                    Err(error) => {
+                                        println!("❌ Error while handling ApplySearchFilter command: {error:?}");
+                                        self.send_error(error).await;
+                                    }
+                                }
+                            }
+                        },
+                        _ => {
+                            todo!();
+                        }
+                    }
                 },
 
                 // Callback receiver won't be dropped when session is dropped.
@@ -190,7 +221,13 @@ impl SessionService {
                 operation_id,
                 filters,
             } => {
-                self.session.apply_search_filters(operation_id, filters)?;
+                let filts = vec![SearchFilter::new("time=13".to_string(), false, true, false)];
+                println!(
+                    "☑️ Session::SessionService::handle_command: Applying search filters for session {}: {:?}",
+                    self.session_id(),
+                    filters
+                );
+                self.session.apply_search_filters(operation_id, filts)?;
             }
             SessionCommand::DropSearch { operation_id } => {
                 if let Some(filter_op) = operation_id {

application/apps/indexer/gui/application/src/session/ui/bottom_panel/search/search_bar.rs

@@ -70,6 +70,7 @@ impl SearchBar {
 
         // Apply temp filter on pressing enter.
         if enter_pressed && !self.query.is_empty() {
+            println!("🔍 SearchBar: Applying search with query '{}'", self.query);
             if self.temp_filter.is_some() {
                 self.drop_search(shared, actions);
             }
@@ -86,6 +87,7 @@ impl SearchBar {
                 operation_id,
                 filters: vec![filter.clone()],
             };
+
             if actions.try_send_command(&self.cmd_tx, cmd) {
                 shared.search.set_search_operation(operation_id);
                 self.temp_filter = Some(filter);

application/apps/indexer/mcp/Cargo.toml

@@ -20,6 +20,7 @@ rmcp = { version = "0.11", features = [
 ] }
 anyhow.workspace = true
 log.workspace = true
+processor.workspace = true
 reqwest = { version = "0.12.25", features = ["json"] }
 schemars = "1.1"
 serde_json.workspace = true

application/apps/indexer/mcp/src/agents/ollama.rs

@@ -20,7 +20,7 @@ pub struct Ollama {
 impl Default for Ollama {
     fn default() -> Self {
         Self {
-            model: String::from("llama3.2"),
+            model: String::from("qwen3:8b"),
             url: String::from("http://localhost:11434"),
             api_key: None,
         }

application/apps/indexer/mcp/src/client/mod.rs

@@ -94,11 +94,6 @@ impl McpClient {
                 message: e.to_string(),
             })?;
 
-        warn!(
-            "🟢 MCP client connected to MCP server: {:?}",
-            mcp_service.peer_info()
-        );
-
         // let llm = Llm::from_config(self.llm_config);
 
         tokio::spawn(async move {
@@ -113,8 +108,6 @@ impl McpClient {
                 error!("MCP client event loop ended: {:?}", e);
             }
         });
-        warn!("✅ MCP client started");
-
         Ok(())
     }
 
@@ -131,6 +124,7 @@ impl McpClient {
             select! {
                 Some(prompt) = prompt_rx.recv() => {
                     let tools = mcp_service.list_tools(Default::default()).await?;
+                    let message = format!("User prompt: {}\n\n session_id:\n{}", prompt.message, prompt.id);
 
                     let response = match std::env::var("LLM_AGENT")
                         .ok()
@@ -139,12 +133,12 @@ impl McpClient {
                     {
                         Some("openai") => {
                             agents::open_ai::OpenAI::default()
-                                .send_chat_message(prompt.message.clone(), &mut history, tools)
+                                .send_chat_message(message.clone(), &mut history, tools)
                                 .await
                         }
                         _ => {
                             agents::ollama::Ollama::default()
-                                .send_chat_message(prompt.message.clone(), &mut history, tools)
+                                .send_chat_message(message.clone(), &mut history, tools)
                                 .await
                         }
                     };
@@ -176,8 +170,8 @@ impl McpClient {
                                 }
                             }
                         },
-                        _ => {
-                            error!("🔴 MCP Client failed to get mock prompt response:");
+                        Err(err) => {
+                            error!("🔴 MCP Client failed to get mock prompt response: {err:?}");
                         }
                     }
                     // println!("Chipmunk request {chipmunk_request:?}");

application/apps/indexer/mcp/src/server/mod.rs

@@ -9,13 +9,14 @@ use rmcp::{
     model::{CallToolResult, Content, ErrorCode, ServerCapabilities, ServerInfo},
     tool, tool_handler, tool_router,
     transport::streamable_http_server::{
-        session::local::LocalSessionManager,
+        session::{self, local::LocalSessionManager},
         tower::{StreamableHttpServerConfig, StreamableHttpService},
     },
 };
 use tokio::sync::{broadcast, mpsc};
 
-use crate::{errors::McpError, types::*};
+use crate::errors::McpError;
+use crate::types::*;
 use tasks::{Tasks, Tasks::*};
 
 #[derive(Clone, Debug)]
@@ -78,15 +79,19 @@ impl McpServer {
 
     #[tool(description = r#"Generate SearchFilter objects for filtering logs.
 
-This tool accepts one or more filter specifications and returns a list of SearchFilter objects.
+This tool accepts one or more filter specifications along with session_id and returns a list of SearchFilter objects.
 Each filter can be customized with flags for regex matching, case sensitivity, and word boundaries.
+Session id is present in the prompt and can be used by the LLM Model and sent back in the parameters.
+This will help the MCP server to identify which session has requested for the filter and apply it to the correct session.
+
 
 **Input Parameters:**
 - `filters`: An list of filter objects, where each object contains:
   - `value` (string): The text or pattern to search for
   - `is_regex` (boolean): true if the filter is a regular expression pattern
-  - `ignore_case` (boolean): true for case-insensitive matching
+  - `ignore_case` (boolean): true for case-insensitive matching, mostly it is false.
   - `is_word` (boolean): true to match whole words only (word boundary matching)
+- `session_id` (string): Unique identifier for the session requesting the filter, used to apply the filter to the correct session. Present in Chat message.
 
 **Usage Examples:**
 
@@ -123,10 +128,31 @@ When the user provides natural language instructions, interpret them as follows:
         Parameters(params): Parameters<SearchFilters>,
     ) -> Result<CallToolResult, RmcpError> {
         let (task_result_tx, task_result_rx) = mpsc::channel(1);
+        let session_id = uuid::Uuid::parse_str(params.session_id.as_str()).map_err(|e| {
+            RmcpError::new(
+                ErrorCode(503),
+                format!("Invalid session_id format: {}", e),
+                None,
+            )
+        })?;
+        let filters: Vec<processor::search::filter::SearchFilter> = params
+            .filters
+            .iter()
+            .map(|f| {
+                processor::search::filter::SearchFilter::new(
+                    f.value.clone(),
+                    f.is_regex,
+                    f.ignore_case,
+                    f.is_word,
+                )
+            })
+            .collect();
         let task = ApplySearchFilter {
-            filters: params.filters.clone(),
+            session_id,
             task_result_tx,
+            filters,
         };
+        println!("🔵 Received task from LLM {task:?}");
         // Send task over communication channel in a separate thread,
         // in future, we can skip match over task spawn
         match self.task_tx.send(task) {
@@ -160,9 +186,17 @@ Returns a histogram of matches for the current search within an optional range.
     ) -> Result<CallToolResult, RmcpError> {
         let (task_result_tx, task_result_rx) = mpsc::channel(1);
         let range = params.range.map(|r| (r.start, r.end));
+        let session_id = uuid::Uuid::parse_str(params.session_id.as_str()).map_err(|e| {
+            RmcpError::new(
+                ErrorCode(503),
+                format!("Invalid session_id format: {}", e),
+                None,
+            )
+        })?;
         let task = Tasks::GetChartHistogram {
             dataset_len: params.dataset_len,
             range,
+            session_id,
             task_result_tx,
         };
         self.task_tx.send(task).map_err(|e| {
@@ -193,9 +227,17 @@ Returns point data for line plots based on extracted values within an optional r
     ) -> Result<CallToolResult, RmcpError> {
         let (task_result_tx, task_result_rx) = mpsc::channel(1);
         let range = params.range.map(|r| (r.start, r.end));
+        let session_id = uuid::Uuid::parse_str(params.session_id.as_str()).map_err(|e| {
+            RmcpError::new(
+                ErrorCode(503),
+                format!("Invalid session_id format: {}", e),
+                None,
+            )
+        })?;
         let task = Tasks::GetChartLinePlots {
             dataset_len: params.dataset_len,
             range,
+            session_id,
             task_result_tx,
         };
         self.task_tx.send(task).map_err(|e| {

application/apps/indexer/mcp/src/server/tasks.rs

@@ -1,25 +1,30 @@
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
+use uuid::Uuid;
 
 use crate::errors::McpError;
-use crate::types::SearchFilter;
+use processor::search::filter::SearchFilter;
 
 #[derive(Debug, Clone)]
 pub enum Tasks {
     ApplySearchFilter {
+        session_id: Uuid,
         filters: Vec<SearchFilter>,
         task_result_tx: mpsc::Sender<Result<(), McpError>>,
     },
     GetChartHistogram {
+        session_id: Uuid,
         dataset_len: u16,
         range: Option<(u64, u64)>,
         task_result_tx: mpsc::Sender<Result<(), McpError>>,
     },
     GetChartLinePlots {
+        session_id: Uuid,
         dataset_len: u16,
         range: Option<(u64, u64)>,
         task_result_tx: mpsc::Sender<Result<(), McpError>>,
     },
     GenericTask {
+        session_id: Uuid,
         task_result_tx: mpsc::Sender<Result<(), McpError>>,
     },
 }

application/apps/indexer/mcp/src/types.rs

@@ -31,6 +31,7 @@ pub struct RangeU64 {
 #[derive(Clone, Debug, JsonSchema, Serialize, Deserialize)]
 pub struct SearchFilters {
     pub filters: Vec<SearchFilter>,
+    pub session_id: String,
 }
 
 #[derive(Clone, Debug, JsonSchema, Serialize, Deserialize)]
@@ -42,12 +43,14 @@ pub struct SearchValuesFilters {
 pub struct MapRequest {
     pub dataset_len: u16,
     pub range: Option<RangeU64>,
+    pub session_id: String,
 }
 
 #[derive(Clone, Debug, JsonSchema, Serialize, Deserialize)]
 pub struct ValuesRequest {
     pub dataset_len: u16,
     pub range: Option<RangeU64>,
+    pub session_id: String,
 }
 
 #[derive(Clone, Debug, JsonSchema, Serialize, Deserialize)]