use anyhow::Result;
use tokio::sync::mpsc;
use tracing::info;
use super::errors::ActionExecutionError;
use super::types::{AppEvent, UserCommand};
use crate::cli::server_startup;
use crate::llm::actions::{get_all_tool_actions, get_user_input_common_actions};
use crate::llm::format_indented_dimmed_lines;
use crate::llm::OllamaClient;
use crate::llm::{execute_actions, CommonAction, Server};
use crate::state::app_state::{AppState, Mode};
use crate::ui::App;
const REQUIRE_DOCS_FOR_OPEN_ACTIONS: bool = false;
fn log_open_server_summary(
status_tx: &mpsc::UnboundedSender<String>,
mac_address: &Option<String>,
interface: &Option<String>,
host: &Option<String>,
port: Option<u16>,
protocol: &str,
send_first: bool,
initial_memory: &Option<String>,
instruction: &str,
startup_params: &Option<serde_json::Value>,
event_handlers: &Option<Vec<serde_json::Value>>,
scheduled_tasks: &Option<Vec<crate::llm::actions::common::ServerTaskDefinition>>,
feedback_instructions: &Option<String>,
) {
let _ = status_tx.send("[INFO] ═══ open_server ═══".to_string());
let _ = status_tx.send(format!("[INFO] Protocol: {}", protocol));
if let Some(p) = port {
let _ = status_tx.send(format!("[INFO] Port: {}", p));
}
if let Some(h) = host {
let _ = status_tx.send(format!("[INFO] Host: {}", h));
}
if let Some(i) = interface {
let _ = status_tx.send(format!("[INFO] Interface: {}", i));
}
if let Some(m) = mac_address {
let _ = status_tx.send(format!("[INFO] MAC Address: {}", m));
}
if send_first {
let _ = status_tx.send("[INFO] Send First: yes".to_string());
}
if !instruction.is_empty() {
let truncated = crate::utils::truncate_for_log(instruction, 100);
let _ = status_tx.send(format!("[INFO] Instruction: {}", truncated));
}
if let Some(mem) = initial_memory {
if !mem.is_empty() {
let truncated = crate::utils::truncate_for_log(mem, 100);
let _ = status_tx.send(format!("[INFO] Initial Memory: {}", truncated));
}
}
if let Some(params) = startup_params {
if !params.is_null() {
let shown = crate::utils::redact::redact_sensitive(params);
let params_str =
serde_json::to_string_pretty(&shown).unwrap_or_else(|_| shown.to_string());
let _ = status_tx.send("[INFO] Startup Params:".to_string());
for line in format_indented_dimmed_lines(¶ms_str, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if let Some(handlers) = event_handlers {
let _ = status_tx.send(format!(
"[INFO] Event Handlers: {} configured",
handlers.len()
));
for handler in handlers {
let event_pattern = handler
.get("event_pattern")
.and_then(|v| v.as_str())
.unwrap_or("*");
let handler_type = handler
.get("handler")
.and_then(|h| h.get("type"))
.and_then(|t| t.as_str())
.unwrap_or("unknown");
let _ = status_tx.send(format!("[INFO] • {} → {}", event_pattern, handler_type));
if handler_type == "script" {
if let Some(code) = handler
.get("handler")
.and_then(|h| h.get("code"))
.and_then(|c| c.as_str())
{
let truncated = crate::utils::truncate_for_log(code, 200);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if handler_type == "static" {
if let Some(actions) = handler.get("handler").and_then(|h| h.get("actions")) {
let actions_str = serde_json::to_string_pretty(actions)
.unwrap_or_else(|_| actions.to_string());
let truncated = crate::utils::truncate_for_log(&actions_str, 300);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
}
}
if let Some(tasks) = scheduled_tasks {
let _ = status_tx.send(format!(
"[INFO] Scheduled Tasks: {} configured",
tasks.len()
));
for task in tasks {
let timing = if task.recurring {
task.interval_secs
.map(|i| format!("every {}s", i))
.unwrap_or_else(|| "recurring".to_string())
} else {
task.delay_secs
.map(|d| format!("after {}s", d))
.unwrap_or_else(|| "one-shot".to_string())
};
let _ = status_tx.send(format!("[INFO] • {} ({})", task.task_id, timing));
let truncated = crate::utils::truncate_for_log(&task.instruction, 100);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if let Some(fb) = feedback_instructions {
let truncated = crate::utils::truncate_for_log(fb, 100);
let _ = status_tx.send(format!("[INFO] Feedback Instructions: {}", truncated));
}
let _ = status_tx.send("[INFO] ══════════════════".to_string());
}
fn log_open_client_summary(
status_tx: &mpsc::UnboundedSender<String>,
protocol: &str,
remote_addr: &str,
instruction: &str,
startup_params: &Option<serde_json::Value>,
initial_memory: &Option<String>,
event_handlers: &Option<Vec<serde_json::Value>>,
scheduled_tasks: &Option<Vec<crate::llm::actions::common::ServerTaskDefinition>>,
feedback_instructions: &Option<String>,
) {
let _ = status_tx.send("[INFO] ═══ open_client ═══".to_string());
let _ = status_tx.send(format!("[INFO] Protocol: {}", protocol));
let _ = status_tx.send(format!("[INFO] Remote: {}", remote_addr));
if !instruction.is_empty() {
let truncated = crate::utils::truncate_for_log(instruction, 100);
let _ = status_tx.send(format!("[INFO] Instruction: {}", truncated));
}
if let Some(mem) = initial_memory {
if !mem.is_empty() {
let truncated = crate::utils::truncate_for_log(mem, 100);
let _ = status_tx.send(format!("[INFO] Initial Memory: {}", truncated));
}
}
if let Some(params) = startup_params {
if !params.is_null() {
let shown = crate::utils::redact::redact_sensitive(params);
let params_str =
serde_json::to_string_pretty(&shown).unwrap_or_else(|_| shown.to_string());
let _ = status_tx.send("[INFO] Startup Params:".to_string());
for line in format_indented_dimmed_lines(¶ms_str, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if let Some(handlers) = event_handlers {
let _ = status_tx.send(format!(
"[INFO] Event Handlers: {} configured",
handlers.len()
));
for handler in handlers {
let event_pattern = handler
.get("event_pattern")
.and_then(|v| v.as_str())
.unwrap_or("*");
let handler_type = handler
.get("handler")
.and_then(|h| h.get("type"))
.and_then(|t| t.as_str())
.unwrap_or("unknown");
let _ = status_tx.send(format!("[INFO] • {} → {}", event_pattern, handler_type));
if handler_type == "script" {
if let Some(code) = handler
.get("handler")
.and_then(|h| h.get("code"))
.and_then(|c| c.as_str())
{
let truncated = crate::utils::truncate_for_log(code, 200);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if handler_type == "static" {
if let Some(actions) = handler.get("handler").and_then(|h| h.get("actions")) {
let actions_str = serde_json::to_string_pretty(actions)
.unwrap_or_else(|_| actions.to_string());
let truncated = crate::utils::truncate_for_log(&actions_str, 300);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
}
}
if let Some(tasks) = scheduled_tasks {
let _ = status_tx.send(format!(
"[INFO] Scheduled Tasks: {} configured",
tasks.len()
));
for task in tasks {
let timing = if task.recurring {
task.interval_secs
.map(|i| format!("every {}s", i))
.unwrap_or_else(|| "recurring".to_string())
} else {
task.delay_secs
.map(|d| format!("after {}s", d))
.unwrap_or_else(|| "one-shot".to_string())
};
let _ = status_tx.send(format!("[INFO] • {} ({})", task.task_id, timing));
let truncated = crate::utils::truncate_for_log(&task.instruction, 100);
for line in format_indented_dimmed_lines(&truncated, 8) {
let _ = status_tx.send(format!("[INFO] {}", line));
}
}
}
if let Some(fb) = feedback_instructions {
let truncated = crate::utils::truncate_for_log(fb, 100);
let _ = status_tx.send(format!("[INFO] Feedback Instructions: {}", truncated));
}
let _ = status_tx.send("[INFO] ══════════════════".to_string());
}
#[derive(Clone)]
pub struct EventHandler {
state: AppState,
llm: OllamaClient,
}
impl EventHandler {
pub fn new(state: AppState, llm: OllamaClient) -> Self {
Self { state, llm }
}
pub async fn list_models(&self) -> Result<Vec<String>> {
self.llm.list_models().await
}
pub fn get_llm_client(&self) -> OllamaClient {
self.llm.clone()
}
pub fn set_llm_client(&mut self, client: OllamaClient) {
self.llm = client;
}
pub async fn handle_event(&mut self, event: AppEvent, ui: &mut App) -> Result<bool> {
match event {
AppEvent::UserCommand(cmd) => self.handle_user_command(cmd, ui).await,
AppEvent::Tick => {
Ok(false)
}
AppEvent::Shutdown => {
info!("Shutdown event received");
Ok(true)
}
}
}
async fn handle_user_command(&mut self, command: UserCommand, ui: &mut App) -> Result<bool> {
match command {
UserCommand::Status => {
self.handle_status(ui).await?;
Ok(false)
}
UserCommand::ShowModel => {
self.handle_show_model(ui).await?;
Ok(false)
}
UserCommand::ChangeModel { model } => {
self.handle_change_model(model, ui).await?;
Ok(false)
}
UserCommand::ShowBackend => {
self.handle_show_backend(ui).await?;
Ok(false)
}
UserCommand::SetBackend { args } => {
self.handle_set_backend(args, ui).await?;
Ok(false)
}
UserCommand::ShowLogLevel => {
ui.add_llm_message(format!("Current log level: {}", ui.log_level.as_str()));
Ok(false)
}
UserCommand::ChangeLogLevel { level } => {
use crate::ui::app::LogLevel;
if let Some(log_level) = LogLevel::parse(&level) {
ui.set_log_level(log_level);
} else {
ui.add_llm_message(format!(
"Invalid log level: {level}. Use: error, warn, info, debug, or trace"
));
}
Ok(false)
}
UserCommand::TestOutput { count } => {
for i in 0..count {
ui.add_llm_message(format!("Test line {} of {}", i + 1, count));
}
Ok(false)
}
UserCommand::TestAsk => {
use crate::llm::actions::tools::{execute_tool, ToolAction};
ui.add_llm_message(
"[INFO] Testing web search approval with DuckDuckGo...".to_string(),
);
let web_search_mode = self.state.get_web_search_mode().await;
let approval_tx = self.state.get_web_approval_channel().await;
let action = ToolAction::WebSearch {
query: "https://duckduckgo.com/?q=test+search+query+with+very+long+parameters&ia=web&category=general&filters=none".to_string(),
};
let result = execute_tool(
&action,
approval_tx.as_ref(),
web_search_mode,
Some(&self.state),
)
.await;
if result.success {
ui.add_llm_message("[INFO] Web search completed successfully".to_string());
ui.add_llm_message(format!("[DEBUG] Result: {}", result.result));
} else {
ui.add_llm_message(format!("[ERROR] Web search failed: {}", result.result));
}
Ok(false)
}
UserCommand::SetFooterStatus { message } => {
if message.is_some() {
ui.add_llm_message(
"Footer status command is only supported in rolling TUI mode".to_string(),
);
}
Ok(false)
}
UserCommand::StopAll => {
self.handle_stop_all(ui).await?;
Ok(false)
}
UserCommand::StopById { id } => {
self.handle_stop_by_id(id, ui).await?;
Ok(false)
}
UserCommand::Save { name, id } => {
self.handle_save(name, id, ui).await?;
Ok(false)
}
UserCommand::Load { name } => {
self.handle_load(name, ui).await?;
Ok(false)
}
#[cfg(feature = "sqlite")]
UserCommand::Sqlite { db_id, query } => {
self.handle_sqlite(db_id, query, ui).await?;
Ok(false)
}
UserCommand::Quit => {
self.handle_quit(ui).await?;
Ok(true) }
UserCommand::UnknownSlashCommand { command } => {
ui.add_llm_message(format!("Unknown command: {command}"));
ui.add_llm_message(
"Available commands: /status, /model [name], /log [level], /docs [protocol], /quit".to_string(),
);
Ok(false)
}
UserCommand::Interpret { input: _ } => {
ui.add_llm_message(
"Internal error: Interpret command should use async path".to_string(),
);
Ok(false)
}
UserCommand::ShowWebSearch => {
ui.add_llm_message(
"Web search command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::SetWebSearch { mode: _ } => {
ui.add_llm_message(
"Web search command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::ShowEventHandler => {
ui.add_llm_message(
"Event handler command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::SetEventHandler { mode: _ } => {
ui.add_llm_message(
"Event handler command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::ShowDocs { protocol } => {
self.handle_show_docs(protocol, ui).await?;
Ok(false)
}
UserCommand::ShowEnvironment => {
self.handle_show_environment(ui).await?;
Ok(false)
}
UserCommand::ShowUsage => {
ui.add_llm_message(
"Usage command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::ShowStability => {
let min = self.state.get_min_stability().await;
for line in crate::protocol::stability_report(min) {
ui.add_llm_message(line);
}
Ok(false)
}
UserCommand::ListSimple => {
ui.add_llm_message(
"Simple protocol command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::StartSimple { protocol: _ } => {
ui.add_llm_message(
"Simple protocol command is only supported in rolling TUI mode".to_string(),
);
Ok(false)
}
UserCommand::Manage | UserCommand::Update { .. } => {
ui.add_llm_message(
"The /manage and /update commands are only supported in rolling TUI mode"
.to_string(),
);
Ok(false)
}
}
}
pub async fn handle_interpret_with_actions(
&mut self,
input: String,
status_tx: mpsc::UnboundedSender<String>,
protocol: Option<Box<dyn Server + Send>>,
) -> Result<()> {
use crate::llm::{ConversationHandler, PromptBuilder};
let protocol_async_actions = if let Some(ref proto) = protocol {
proto.get_async_actions(&self.state)
} else {
Vec::new()
};
let current_model = self.state.get_ollama_model().await;
let model = match crate::llm::ensure_model_selected(
current_model.clone(),
&self.state.get_ollama_url().await,
)
.await
{
Ok(m) => m,
Err(e) => {
return Err(anyhow::anyhow!("Failed to ensure model is selected: {}", e));
}
};
if current_model.is_none() {
let _ = status_tx.send(format!(
"⚠ Auto-selected model: {} (no model was configured)",
model
));
}
let llm_with_status = self.llm.clone().with_status_tx(status_tx.clone());
let web_search_mode = self.state.get_web_search_mode().await;
let approval_tx = self.state.get_web_approval_channel().await;
let conversation_history = self.state.get_user_conversation_history().await;
let is_open_server_enabled = if REQUIRE_DOCS_FOR_OPEN_ACTIONS {
self.state.is_server_docs_read().await
} else {
true
};
let is_open_client_enabled = if REQUIRE_DOCS_FOR_OPEN_ACTIONS {
self.state.is_client_docs_read().await
} else {
true
};
let system_prompt = PromptBuilder::build_user_input_system_prompt_with_docs(
&self.state,
protocol_async_actions.clone(),
conversation_history,
is_open_server_enabled,
is_open_client_enabled,
)
.await;
let selected_mode = self.state.get_selected_scripting_mode().await;
let scripting_env = self.state.get_scripting_env().await;
let mut available_actions = get_user_input_common_actions(
selected_mode,
&scripting_env,
is_open_server_enabled,
is_open_client_enabled,
);
available_actions.extend(get_all_tool_actions(web_search_mode));
available_actions.extend(protocol_async_actions);
let available_actions =
PromptBuilder::advertised_user_input_actions(&self.state, available_actions).await;
let conversation_state = self.state.get_or_create_user_conversation_state().await;
let rate_limiter = self.state.get_rate_limiter().await;
let mut conversation = ConversationHandler::new(
system_prompt,
std::sync::Arc::new(llm_with_status),
model,
rate_limiter,
crate::llm::RequestSource::User, )
.with_native_tools(&available_actions)
.with_status_tx(status_tx.clone())
.with_tracking(
self.state.clone(),
crate::state::app_state::ConversationSource::User,
input.clone(),
)
.with_conversation_state(conversation_state);
self.state
.register_conversation(
conversation.conversation_id().to_string(),
crate::state::app_state::ConversationSource::User,
input.clone(),
)
.await;
conversation.mark_registered();
conversation.add_user_message(input);
const MAX_EXECUTION_RETRIES: usize = 1;
let mut execution_attempts = 0;
loop {
let actions = conversation
.generate_with_tools_and_retry(
approval_tx.clone(),
web_search_mode,
available_actions.clone(),
)
.await;
match actions {
Ok(action_values) => {
let mut should_retry = false;
let mut retry_error: Option<crate::events::ActionExecutionError> = None;
let mut state_changed = false;
for action_value in &action_values {
if let Ok(common_action) = CommonAction::from_json(action_value) {
let modifies_state = matches!(
common_action,
CommonAction::OpenServer { .. }
| CommonAction::CloseServer { .. }
| CommonAction::CloseAllServers
| CommonAction::OpenClient { .. }
| CommonAction::CloseClient { .. }
| CommonAction::CloseAllClients
| CommonAction::CloseConnectionById { .. }
| CommonAction::UpdateServer { .. }
| CommonAction::UpdateClient { .. }
);
match self
.execute_server_management_action(common_action, &status_tx)
.await
{
Ok(_) => {
if modifies_state {
state_changed = true;
}
}
Err(e)
if e.is_retryable()
&& execution_attempts < MAX_EXECUTION_RETRIES =>
{
should_retry = true;
retry_error = Some(e);
break; }
Err(e) => {
let _ = status_tx
.send(format!("[ERROR] Error executing action: {e}"));
}
}
}
}
if state_changed && !should_retry {
conversation.update_current_state(&self.state, None).await;
let _ = status_tx.send(
"[DEBUG] Updated conversation state after server changes".to_string(),
);
}
if should_retry {
if let Some(error) = retry_error {
execution_attempts += 1;
let _ = status_tx.send(format!(
"[INFO] Execution error (attempt {}/{}), retrying with LLM feedback...",
execution_attempts,
MAX_EXECUTION_RETRIES + 1
));
let correction = error.build_correction_message();
conversation.add_user_message(correction);
continue;
}
}
let protocol_ref: Option<&dyn Server> =
protocol.as_ref().map(|p| p.as_ref() as &dyn Server);
match execute_actions(
action_values.clone(),
&self.state,
protocol_ref,
None,
None,
)
.await
{
Ok(result) => {
for msg in result.messages {
let _ = status_tx.send(msg);
}
}
Err(e) => {
let _ =
status_tx.send(format!("[ERROR] Failed to execute actions: {e}"));
}
}
break;
}
Err(e) => {
let _ = status_tx.send(format!("[ERROR] LLM error: {e}"));
conversation.end_tracking().await;
break; }
}
}
Ok(())
}
pub async fn execute_server_management_action(
&mut self,
action: CommonAction,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<(), crate::events::ActionExecutionError> {
match action {
CommonAction::OpenServer {
mac_address,
interface,
host,
port,
protocol,
send_first,
initial_memory,
instruction,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
if REQUIRE_DOCS_FOR_OPEN_ACTIONS && !self.state.is_server_docs_read().await {
use crate::llm::actions::tools::{execute_tool, ToolAction};
let doc_tool = ToolAction::ReadDocumentation {
protocols: vec![protocol.clone()],
protocol: None,
};
let approval_tx = self.state.get_web_approval_channel().await;
let web_search_mode = self.state.get_web_search_mode().await;
let tool_result = execute_tool(
&doc_tool,
approval_tx.as_ref(),
web_search_mode,
Some(&self.state),
)
.await;
if tool_result.success {
self.state
.mark_server_protocols_documented(&[protocol.clone()])
.await;
let original_action = serde_json::json!({
"type": "open_server",
"mac_address": mac_address,
"interface": interface,
"host": host,
"port": port,
"protocol": protocol,
"send_first": send_first,
"initial_memory": initial_memory,
"instruction": instruction,
"startup_params": startup_params,
"event_handlers": event_handlers,
"scheduled_tasks": scheduled_tasks,
"feedback_instructions": feedback_instructions,
});
return Err(ActionExecutionError::DocumentationRequired {
action_type: "open_server".to_string(),
protocol: protocol.clone(),
documentation: tool_result.result,
original_action,
});
} else {
let _ = status_tx.send(format!(
"[WARN] Failed to fetch documentation for {}: {}. Proceeding anyway.",
protocol, tool_result.result
));
}
}
log_open_server_summary(
status_tx,
&mac_address,
&interface,
&host,
port,
&protocol,
send_first,
&initial_memory,
&instruction,
&startup_params,
&event_handlers,
&scheduled_tasks,
&feedback_instructions,
);
let mac_address_clone = mac_address.clone();
let interface_clone = interface.clone();
let host_clone = host.clone();
let initial_memory_clone = initial_memory.clone();
let instruction_clone = instruction.clone();
let startup_params_clone = startup_params.clone();
let event_handlers_clone = event_handlers.clone();
let scheduled_tasks_clone = scheduled_tasks.clone();
let feedback_instructions_clone = feedback_instructions.clone();
match server_startup::start_server_from_action(
&self.state,
mac_address,
interface,
host,
port,
&protocol,
send_first,
initial_memory,
instruction,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
status_tx.clone(),
)
.await
{
Ok(server_id) => {
let _ = status_tx.send(format!(
"[INFO] Server #{} started successfully",
server_id.as_u32()
));
}
Err(e) => {
let error_msg = e.to_string();
let is_port_conflict = error_msg.contains("Address already in use")
|| error_msg.contains("port conflict")
|| error_msg.contains("bind");
let is_validation_error = error_msg
.contains("Invalid event handler configuration")
|| error_msg.contains("Missing 'instruction' field")
|| error_msg.contains("instruction is required");
if is_port_conflict && port.is_some() {
return Err(ActionExecutionError::PortConflict {
port: port.unwrap(),
protocol: protocol.to_string(),
underlying_error: error_msg,
});
} else if is_validation_error {
let original_action = serde_json::json!({
"type": "open_server",
"mac_address": mac_address_clone,
"interface": interface_clone,
"host": host_clone,
"port": port,
"protocol": protocol,
"send_first": send_first,
"initial_memory": initial_memory_clone,
"instruction": instruction_clone,
"startup_params": startup_params_clone,
"event_handlers": event_handlers_clone,
"scheduled_tasks": scheduled_tasks_clone,
"feedback_instructions": feedback_instructions_clone,
});
let parameter_name = event_handler_parameter_name(&error_msg);
return Err(ActionExecutionError::InvalidActionParameters {
action_type: "open_server".to_string(),
parameter_name: parameter_name.to_string(),
error_message: error_msg.clone(),
original_action,
});
} else {
let _ = status_tx
.send(format!("[ERROR] Failed to start server: {}", error_msg));
return Err(ActionExecutionError::Fatal(e));
}
}
}
}
CommonAction::CloseServer { server_id } => {
use crate::state::server::ServerStatus;
let sid = crate::state::ServerId::new(server_id);
self.state
.update_server_status(sid, ServerStatus::Stopped)
.await;
let _ = status_tx.send(format!("[SERVER] Stopped server #{}", sid.as_u32()));
self.state.cleanup_server_tasks(sid).await;
let _ = status_tx.send(format!(
"[TASK] Cleaned up tasks for server #{}",
sid.as_u32()
));
crate::scripting::ResidentScriptManager::shutdown_server(sid.as_u32()).await;
let all_stopped =
self.state.get_all_servers().await.iter().all(|s| {
matches!(s.status, ServerStatus::Stopped | ServerStatus::Error(_))
});
if all_stopped {
self.state.set_mode(Mode::Idle).await;
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::CloseAllServers => {
use crate::state::server::ServerStatus;
let server_ids = self.state.get_all_server_ids().await;
for server_id in server_ids {
self.state
.update_server_status(server_id, ServerStatus::Stopped)
.await;
let _ =
status_tx.send(format!("[SERVER] Stopped server #{}", server_id.as_u32()));
self.state.cleanup_server_tasks(server_id).await;
let _ = status_tx.send(format!(
"[TASK] Cleaned up tasks for server #{}",
server_id.as_u32()
));
crate::scripting::ResidentScriptManager::shutdown_server(server_id.as_u32())
.await;
}
self.state.set_mode(Mode::Idle).await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::UpdateInstruction { instruction } => {
if let Some(server_id) = self.state.get_first_server_id().await {
self.state
.set_instruction(server_id, instruction.clone())
.await;
let _ = status_tx.send(format!(
"[INFO] Server #{} instruction: {}",
server_id.as_u32(),
instruction
));
} else {
let _ =
status_tx.send("[WARN] No server to update instruction for".to_string());
}
}
CommonAction::ChangeModel { model } => {
self.state.set_ollama_model(Some(model.clone())).await;
let _ = status_tx.send(format!("Changed model to: {model}"));
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::SetMemory { value } => {
if let Some(server_id) = self.state.get_first_server_id().await {
self.state.set_memory(server_id, value).await;
}
}
CommonAction::AppendMemory { value } => {
if let Some(server_id) = self.state.get_first_server_id().await {
self.state.append_memory(server_id, value).await;
}
}
CommonAction::AppendToLog { .. } => {
}
CommonAction::ScheduleTask {
task_id,
recurring,
delay_secs,
interval_secs,
max_executions,
server_id,
connection_id,
client_id,
instruction,
context,
script_runtime,
script_language: _,
script_path: _,
script_inline,
script_handles,
} => {
use crate::state::task::{ScheduledTask, TaskScope};
let scope = if let Some(conn_id_str) = connection_id {
if let Some(sid) = server_id {
let server_id_obj = crate::state::ServerId::new(sid);
match crate::server::connection::ConnectionId::from_string(&conn_id_str) {
Some(cid) => {
if let Some(server) = self.state.get_server(server_id_obj).await {
if server.connections.contains_key(&cid) {
TaskScope::Connection(server_id_obj, cid)
} else {
let _ = status_tx.send(format!(
"[ERROR] Connection {} not found on server #{}",
conn_id_str, sid
));
return Ok(());
}
} else {
let _ = status_tx.send(format!(
"[ERROR] Server #{} not found for connection-scoped task",
sid
));
return Ok(());
}
}
None => {
let _ = status_tx.send(format!(
"[ERROR] Invalid connection_id format: {}. Expected 'conn-123' or '123'",
conn_id_str
));
return Ok(());
}
}
} else {
let _ = status_tx.send(
"[ERROR] connection_id requires server_id to be specified".to_string(),
);
return Ok(());
}
} else if let Some(sid) = server_id {
TaskScope::Server(crate::state::ServerId::new(sid))
} else if let Some(cid) = client_id {
let client_id_obj = crate::state::ClientId::new(cid);
if self.state.get_client(client_id_obj).await.is_none() {
let _ = status_tx.send(format!(
"[ERROR] Client #{} not found for client-scoped task",
cid
));
return Ok(());
}
TaskScope::Client(client_id_obj)
} else {
TaskScope::Global
};
let delay = if recurring {
delay_secs.or(interval_secs).unwrap_or(0)
} else {
delay_secs.unwrap_or(0)
};
let mut task = if recurring {
let interval = interval_secs.unwrap_or(delay);
ScheduledTask::new_recurring(
crate::state::TaskId::new(0), task_id.clone(),
scope,
interval,
max_executions,
instruction,
context,
)
} else {
ScheduledTask::new_one_shot(
crate::state::TaskId::new(0), task_id.clone(),
scope,
delay,
instruction,
context,
)
}?;
task.next_execution = crate::state::task::checked_task_deadline(delay)?;
let task_id_num = self.state.add_task(task).await;
let _ = script_runtime; let _ = script_inline; let _ = script_handles;
if recurring {
let interval = interval_secs.unwrap_or(delay);
let max_info = if let Some(max) = max_executions {
format!(" (max {} executions)", max)
} else {
String::new()
};
let _ = status_tx.send(format!(
"[TASK] Scheduled recurring task '{}' (ID: {}) to execute every {}s{}",
task_id, task_id_num, interval, max_info
));
} else {
let _ = status_tx.send(format!(
"[TASK] Scheduled one-shot task '{}' (ID: {}) to execute in {}s",
task_id, task_id_num, delay
));
}
}
CommonAction::CancelTask { task_id } => {
if let Some(task) = self.state.get_task(&task_id).await {
self.state.remove_task(task.id).await;
let _ = status_tx.send(format!("[TASK] Cancelled task '{}'", task_id));
} else {
let _ = status_tx.send(format!("[WARN] Task '{}' not found", task_id));
}
}
CommonAction::OpenClient {
protocol,
remote_addr,
instruction,
startup_params,
initial_memory,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
if REQUIRE_DOCS_FOR_OPEN_ACTIONS && !self.state.is_client_docs_read().await {
use crate::llm::actions::tools::{execute_tool, ToolAction};
let doc_tool = ToolAction::ReadDocumentation {
protocols: vec![protocol.clone()],
protocol: None,
};
let approval_tx = self.state.get_web_approval_channel().await;
let web_search_mode = self.state.get_web_search_mode().await;
let tool_result = execute_tool(
&doc_tool,
approval_tx.as_ref(),
web_search_mode,
Some(&self.state),
)
.await;
if tool_result.success {
self.state
.mark_client_protocols_documented(&[protocol.clone()])
.await;
let original_action = serde_json::json!({
"type": "open_client",
"protocol": protocol,
"remote_addr": remote_addr,
"instruction": instruction,
"startup_params": startup_params,
"initial_memory": initial_memory,
"event_handlers": event_handlers,
"scheduled_tasks": scheduled_tasks,
"feedback_instructions": feedback_instructions,
});
return Err(ActionExecutionError::DocumentationRequired {
action_type: "open_client".to_string(),
protocol: protocol.clone(),
documentation: tool_result.result,
original_action,
});
} else {
let _ = status_tx.send(format!(
"[WARN] Failed to fetch documentation for {}: {}. Proceeding anyway.",
protocol, tool_result.result
));
}
}
let prepared_tasks = crate::state::task::prepare_tasks(
scheduled_tasks.as_deref(),
crate::state::task::TaskScope::Global,
)?;
log_open_client_summary(
status_tx,
&protocol,
&remote_addr,
&instruction,
&startup_params,
&initial_memory,
&event_handlers,
&scheduled_tasks,
&feedback_instructions,
);
use crate::state::client::{ClientInstance, ClientStatus};
let event_handlers_clone = event_handlers.clone();
let feedback_instructions_clone = feedback_instructions.clone();
let initial_memory_clone = initial_memory.clone();
let parsed_event_handlers = match event_handlers {
Some(handlers_json) => match Self::parse_event_handlers(handlers_json) {
Ok(config) => Some(config),
Err(e) => {
let original_action = serde_json::json!({
"type": "open_client",
"protocol": protocol,
"remote_addr": remote_addr,
"instruction": instruction,
"startup_params": startup_params,
"initial_memory": initial_memory_clone,
"event_handlers": event_handlers_clone,
"scheduled_tasks": scheduled_tasks,
"feedback_instructions": feedback_instructions_clone,
});
let error_msg = e.to_string();
let _ = status_tx.send(format!(
"[ERROR] Invalid event handler configuration: {}",
error_msg
));
return Err(ActionExecutionError::InvalidActionParameters {
action_type: "open_client".to_string(),
parameter_name: event_handler_parameter_name(&error_msg)
.to_string(),
error_message: error_msg,
original_action,
});
}
},
None => None,
};
let mut client = ClientInstance::new(
crate::state::ClientId::new(0),
remote_addr.clone(),
protocol.clone(),
instruction.clone(),
);
if let Some(mem) = initial_memory {
client.memory = mem;
}
client.startup_params = startup_params.clone();
client.feedback_instructions = feedback_instructions;
let client_id = self.state.add_client(client).await;
if let Some(config) = parsed_event_handlers {
self.state
.set_client_event_handler_config(client_id, Some(config))
.await;
let _ = status_tx
.send("[INFO] Event handler configuration applied to client".to_string());
}
let _ = status_tx.send(format!(
"[CLIENT] Opening {} client #{} to {}...",
protocol,
client_id.as_u32(),
remote_addr
));
let llm_client = self.llm.clone();
let status_tx_clone = status_tx.clone();
match crate::cli::client_startup::start_client_by_id(
&self.state,
client_id,
&llm_client,
&status_tx_clone,
)
.await
{
Ok(_) => {
let _ = status_tx.send(format!(
"[CLIENT] {} client #{} connected",
protocol,
client_id.as_u32()
));
for mut task in prepared_tasks {
task.scope = crate::state::task::TaskScope::Client(client_id);
let task_name = task.name.clone();
let task_id_num = self.state.add_task(task).await;
let _ = status_tx.send(format!(
"[TASK] Created task '{}' (ID: {}) for client #{}",
task_name,
task_id_num,
client_id.as_u32()
));
}
}
Err(e) => {
self.state
.update_client_status(client_id, ClientStatus::Error(e.to_string()))
.await;
let _ = status_tx.send(format!(
"[ERROR] Failed to connect {} client #{}: {}",
protocol,
client_id.as_u32(),
e
));
return Err(e);
}
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::CloseClient { client_id } => {
use crate::state::client::ClientStatus;
let cid = crate::state::ClientId::new(client_id);
self.state
.update_client_status(cid, ClientStatus::Disconnected)
.await;
let _ = status_tx.send(format!("[CLIENT] Closed client #{}", cid.as_u32()));
self.state.cleanup_client_tasks(cid).await;
let _ = status_tx.send(format!(
"[TASK] Cleaned up tasks for client #{}",
cid.as_u32()
));
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::CloseAllClients => {
use crate::state::client::ClientStatus;
let client_ids = self.state.get_all_client_ids().await;
for client_id in client_ids {
self.state
.update_client_status(client_id, ClientStatus::Disconnected)
.await;
let _ =
status_tx.send(format!("[CLIENT] Closed client #{}", client_id.as_u32()));
self.state.cleanup_client_tasks(client_id).await;
let _ = status_tx.send(format!(
"[TASK] Cleaned up tasks for client #{}",
client_id.as_u32()
));
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::CloseConnectionById { connection_id } => {
use crate::server::connection::ConnectionId;
let conn_id = ConnectionId::new(connection_id);
let all_servers = self.state.get_all_servers().await;
let mut found = false;
for server in all_servers {
if server.connections.contains_key(&conn_id) {
self.state
.close_connection_on_server(server.id, conn_id)
.await;
let _ = status_tx.send(format!(
"[CONNECTION] Closed connection #{} on server #{}",
connection_id,
server.id.as_u32()
));
found = true;
break;
}
}
if !found {
let _ =
status_tx.send(format!("[ERROR] Connection #{} not found", connection_id));
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::ReconnectClient { client_id } => {
let cid = crate::state::ClientId::new(client_id);
let llm_client = self.llm.clone();
let status_tx_clone = status_tx.clone();
let _ =
status_tx.send(format!("[CLIENT] Reconnecting client #{}...", cid.as_u32()));
match crate::cli::client_startup::start_client_by_id(
&self.state,
cid,
&llm_client,
&status_tx_clone,
)
.await
{
Ok(_) => {
let _ = status_tx
.send(format!("[CLIENT] Client #{} reconnected", cid.as_u32()));
}
Err(e) => {
let _ = status_tx.send(format!(
"[ERROR] Failed to reconnect client #{}: {}",
cid.as_u32(),
e
));
return Err(e);
}
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::UpdateClientInstruction {
client_id,
instruction,
} => {
let cid = crate::state::ClientId::new(client_id);
self.state
.set_instruction_for_client(cid, instruction.clone())
.await;
let _ = status_tx.send(format!(
"[CLIENT] Updated instruction for client #{}",
cid.as_u32()
));
let _ = status_tx.send(format!("[CLIENT] New instruction: {}", instruction));
let _ = status_tx.send("__UPDATE_UI__".to_string());
}
CommonAction::UpdateServer {
server_id,
port,
host,
interface,
mac_address,
send_first,
instruction,
initial_memory,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
use crate::cli::management::{self, ServerForm};
let form = ServerForm {
protocol: String::new(), port,
host,
interface,
mac_address,
send_first,
instruction,
initial_memory,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
};
match management::update_server(
&self.state,
crate::state::ServerId::new(server_id),
form,
status_tx.clone(),
)
.await
{
Ok(outcome) => {
let _ = status_tx.send(format!("[INFO] {}", outcome.summary));
}
Err(e) => {
let _ = status_tx.send(format!("[ERROR] Failed to update server: {}", e));
return Err(crate::events::ActionExecutionError::Fatal(e));
}
}
}
CommonAction::UpdateClient {
client_id,
remote_addr,
instruction,
initial_memory,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
use crate::cli::management::{self, ClientForm};
let form = ClientForm {
protocol: String::new(), remote_addr,
instruction,
initial_memory,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
};
match management::update_client(
&self.state,
crate::state::ClientId::new(client_id),
form,
self.llm.clone(),
status_tx.clone(),
)
.await
{
Ok(outcome) => {
let _ = status_tx.send(format!("[INFO] {}", outcome.summary));
}
Err(e) => {
let _ = status_tx.send(format!("[ERROR] Failed to update client: {}", e));
return Err(crate::events::ActionExecutionError::Fatal(e));
}
}
}
#[cfg(feature = "sqlite")]
CommonAction::CreateDatabase {
name,
is_memory,
owner,
schema_ddl,
} => {
use crate::state::DatabaseOwner;
if let Err(e) = crate::state::sqlite::validate_database_name(&name) {
let _ = status_tx.send(format!("[ERROR] {}", e));
return Ok(());
}
let db_path = if is_memory {
crate::state::sqlite::MEMORY_DATABASE_PATH.to_string()
} else {
match crate::state::sqlite::database_file_path(&name) {
Ok(p) => p,
Err(e) => {
let _ = status_tx.send(format!("[ERROR] {}", e));
return Ok(());
}
}
};
let db_owner = if let Some(owner_str) = owner {
if owner_str == "global" {
DatabaseOwner::Global
} else if let Some(id_str) = owner_str.strip_prefix("server-") {
if let Ok(id) = id_str.parse::<u32>() {
DatabaseOwner::Server(crate::state::ServerId::new(id))
} else {
let _ = status_tx
.send(format!("[ERROR] Invalid server ID in owner: {}", owner_str));
return Ok(());
}
} else if let Some(id_str) = owner_str.strip_prefix("client-") {
if let Ok(id) = id_str.parse::<u32>() {
DatabaseOwner::Client(crate::state::ClientId::new(id))
} else {
let _ = status_tx
.send(format!("[ERROR] Invalid client ID in owner: {}", owner_str));
return Ok(());
}
} else {
let _ =
status_tx.send(format!("[ERROR] Invalid owner format: {}", owner_str));
return Ok(());
}
} else {
if let Some(sid) = self.state.get_first_server_id().await {
DatabaseOwner::Server(sid)
} else if let Some(cid) = self.state.get_first_client_id().await {
DatabaseOwner::Client(cid)
} else {
DatabaseOwner::Global
}
};
match self
.state
.create_database(
name.clone(),
db_path.clone(),
db_owner.clone(),
schema_ddl.as_deref(),
)
.await
{
Ok(db_id) => {
let _ = status_tx.send(format!(
"[DB] Created database '{}' ({}) at {} (owner: {})",
name, db_id, db_path, db_owner
));
if let Some(db) = self.state.get_database(db_id).await {
if !db.tables.is_empty() {
let _ =
status_tx.send(format!("[DB] Schema: {}", db.schema_summary()));
}
}
}
Err(e) => {
let _ = status_tx.send(format!("[ERROR] Failed to create database: {}", e));
}
}
}
#[cfg(feature = "sqlite")]
CommonAction::DeleteDatabase { database_id } => {
let db_id = crate::state::DatabaseId::new(database_id);
match self.state.delete_database(db_id).await {
Ok(_) => {
let _ = status_tx.send(format!("[DB] Deleted database {}", db_id));
}
Err(e) => {
let _ = status_tx.send(format!("[ERROR] Failed to delete database: {}", e));
}
}
}
CommonAction::ShowMessage { message } => {
let _ = status_tx.send(format!("[CLIENT] {}", message));
}
CommonAction::ProvideFeedback { .. } => {
}
}
Ok(())
}
pub fn parse_event_handlers(
handlers_json: Vec<serde_json::Value>,
) -> Result<crate::scripting::EventHandlerConfig> {
use crate::scripting::{EventHandler, EventHandlerConfig, EventHandlerType, EventPattern};
let mut config = EventHandlerConfig::new();
for handler_json in handlers_json {
let event_pattern = if let Some(pattern_str) =
handler_json.get("event_pattern").and_then(|v| v.as_str())
{
EventPattern::from(pattern_str)
} else {
EventPattern::wildcard()
};
let handler_type_json = handler_json.get("handler").ok_or_else(|| {
anyhow::anyhow!("Missing 'handler' field in event handler configuration")
})?;
let handler_type_str = handler_type_json
.get("type")
.and_then(|v| v.as_str())
.ok_or_else(|| anyhow::anyhow!("Missing 'type' field in handler configuration"))?;
let handler_type = match handler_type_str {
"llm" => {
let instruction = handler_type_json
.get("instruction")
.and_then(|v| v.as_str())
.ok_or_else(|| {
anyhow::anyhow!("Missing 'instruction' field for llm handler - instruction is required for LLM event handlers")
})?;
EventHandlerType::llm(instruction)
}
"script" => {
let language = handler_type_json
.get("language")
.and_then(|v| v.as_str())
.ok_or_else(|| {
anyhow::anyhow!("Missing 'language' field for script handler")
})?;
let code = handler_type_json
.get("code")
.and_then(|v| v.as_str())
.ok_or_else(|| {
anyhow::anyhow!("Missing 'code' field for script handler")
})?;
let resident = handler_type_json
.get("resident")
.and_then(|v| v.as_bool())
.unwrap_or(false);
if resident {
let scope = handler_type_json
.get("scope")
.and_then(|v| v.as_str())
.map(|s| s.to_string());
EventHandlerType::script_resident(language, code, scope)
} else {
EventHandlerType::script(language, code)
}
}
"static" => {
let actions = handler_type_json
.get("actions")
.and_then(|v| v.as_array())
.ok_or_else(|| {
anyhow::anyhow!("Missing or invalid 'actions' field for static handler")
})?;
EventHandlerType::static_response(actions.clone())
}
"manual" => {
let timeout_secs = match handler_type_json.get("timeout_secs") {
None => crate::scripting::DEFAULT_MANUAL_TIMEOUT_SECS,
Some(v) => v.as_u64().filter(|t| *t > 0).ok_or_else(|| {
anyhow::anyhow!(
"'timeout_secs' for a manual handler must be a positive integer \
number of seconds"
)
})?,
};
EventHandlerType::manual(timeout_secs)
}
_ => anyhow::bail!("Unknown handler type: {}", handler_type_str),
};
handler_type.validate().map_err(|e| {
anyhow::anyhow!(
"Invalid event handler for {}: {}",
describe_pattern(&event_pattern),
e
)
})?;
if let EventHandlerType::Static { ref actions } = handler_type {
validate_static_action_names(&event_pattern, actions)?;
}
config.add_handler(EventHandler::new(event_pattern, handler_type));
}
Ok(config)
}
async fn handle_quit(&mut self, ui: &mut App) -> Result<()> {
ui.add_llm_message("Quitting...".to_string());
Ok(())
}
async fn handle_stop_all(&mut self, ui: &mut App) -> Result<()> {
use crate::state::client::ClientStatus;
use crate::state::server::ServerStatus;
ui.add_llm_message("Stopping all servers, connections, and clients...".to_string());
let server_ids: Vec<_> = self.state.get_all_server_ids().await;
for server_id in server_ids {
self.state
.update_server_status(server_id, ServerStatus::Stopped)
.await;
self.state.cleanup_server_tasks(server_id).await;
crate::scripting::ResidentScriptManager::shutdown_server(server_id.as_u32()).await;
ui.add_llm_message(format!("[SERVER] Stopped server #{}", server_id.as_u32()));
}
let client_ids: Vec<_> = self.state.get_all_client_ids().await;
for client_id in client_ids {
self.state
.update_client_status(client_id, ClientStatus::Disconnected)
.await;
self.state.cleanup_client_tasks(client_id).await;
ui.add_llm_message(format!("[CLIENT] Stopped client #{}", client_id.as_u32()));
}
ui.add_llm_message("All servers and clients stopped.".to_string());
Ok(())
}
async fn handle_stop_by_id(&mut self, id: u32, ui: &mut App) -> Result<()> {
use crate::server::connection::ConnectionId;
use crate::state::client::{ClientId, ClientStatus};
use crate::state::server::{ServerId, ServerStatus};
let mut found = false;
let server_id = ServerId::new(id);
if self.state.get_server(server_id).await.is_some() {
self.state
.update_server_status(server_id, ServerStatus::Stopped)
.await;
self.state.cleanup_server_tasks(server_id).await;
ui.add_llm_message(format!("[SERVER] Stopped server #{}", id));
found = true;
}
let client_id = ClientId::new(id);
if self.state.get_client(client_id).await.is_some() {
self.state
.update_client_status(client_id, ClientStatus::Disconnected)
.await;
self.state.cleanup_client_tasks(client_id).await;
ui.add_llm_message(format!("[CLIENT] Stopped client #{}", id));
found = true;
}
let connection_id = ConnectionId::new(id);
let all_servers = self.state.get_all_servers().await;
for server in all_servers {
if server.connections.contains_key(&connection_id) {
self.state
.close_connection_on_server(server.id, connection_id)
.await;
ui.add_llm_message(format!(
"[CONNECTION] Closed connection #{} on server #{}",
id,
server.id.as_u32()
));
found = true;
break;
}
}
if !found {
ui.add_llm_message(format!(
"No server, client, or connection found with ID #{}",
id
));
}
Ok(())
}
async fn handle_status(&mut self, ui: &mut App) -> Result<()> {
let summary = self.state.get_summary().await;
ui.add_llm_message(format!("Status: {summary}"));
if let Some(server_id) = self.state.get_first_server_id().await {
if let Some(instruction) = self.state.get_instruction(server_id).await {
if !instruction.is_empty() {
ui.add_llm_message(format!(
"Server #{} instruction: {}",
server_id.as_u32(),
instruction
));
}
}
} else {
ui.add_llm_message("No servers running".to_string());
}
Ok(())
}
async fn handle_show_model(&mut self, ui: &mut App) -> Result<()> {
let current_model = self
.state
.get_ollama_model()
.await
.unwrap_or_else(|| "None".to_string());
ui.add_llm_message(format!("Current model: {current_model}"));
ui.add_llm_message("".to_string());
ui.add_llm_message("Fetching available models...".to_string());
match self.llm.list_models().await {
Ok(models) => {
if models.is_empty() {
ui.add_llm_message("No models found. Please pull a model first.".to_string());
ui.add_llm_message("Example: ollama pull llama3.2".to_string());
} else {
ui.add_llm_message(format!("Available models ({}):", models.len()));
for model in &models {
if model == ¤t_model {
ui.add_llm_message(format!(" * {model} (current)"));
} else {
ui.add_llm_message(format!(" {model}"));
}
}
ui.add_llm_message("".to_string());
ui.add_llm_message("To change model, use: /model <name>".to_string());
}
}
Err(e) => {
ui.add_llm_message(format!("Failed to fetch models: {e}"));
ui.add_llm_message("Make sure Ollama is running.".to_string());
}
}
Ok(())
}
async fn handle_change_model(&mut self, model: String, ui: &mut App) -> Result<()> {
match self.llm.list_models().await {
Ok(models) => {
if models.contains(&model) {
self.state.set_ollama_model(Some(model.clone())).await;
ui.add_llm_message(format!("✓ Changed model to: {model}"));
} else {
ui.add_llm_message(format!("✗ Model '{model}' not found"));
ui.add_llm_message("".to_string());
if models.is_empty() {
ui.add_llm_message("No models available. Pull a model first:".to_string());
ui.add_llm_message(" ollama pull llama3.2".to_string());
} else {
ui.add_llm_message("Available models:".to_string());
for available_model in &models {
ui.add_llm_message(format!(" {available_model}"));
}
ui.add_llm_message("".to_string());
ui.add_llm_message("Or pull the model:".to_string());
ui.add_llm_message(format!(" ollama pull {model}"));
}
}
}
Err(e) => {
ui.add_llm_message(format!("Failed to validate model: {e}"));
ui.add_llm_message("Make sure Ollama is running.".to_string());
}
}
Ok(())
}
async fn handle_show_backend(&mut self, ui: &mut App) -> Result<()> {
let backend_type = self.llm.backend_type();
let backend_url = self.llm.backend_url();
let current_model = self
.state
.get_ollama_model()
.await
.unwrap_or_else(|| "None".to_string());
ui.add_llm_message(format!("LLM Backend: {}", backend_type));
ui.add_llm_message(format!(" URL: {}", backend_url));
ui.add_llm_message(format!(" Model: {}", current_model));
ui.add_llm_message("".to_string());
ui.add_llm_message("To switch backend:".to_string());
ui.add_llm_message(" /backend ollama [url] - Switch to Ollama".to_string());
ui.add_llm_message(
" /backend openai <url> [api-key] - Switch to OpenAI-compatible".to_string(),
);
Ok(())
}
#[cfg(not(target_arch = "wasm32"))]
async fn handle_set_backend(&mut self, args: String, ui: &mut App) -> Result<()> {
let parts: Vec<&str> = args.splitn(3, ' ').collect();
match parts.first().map(|s| s.to_lowercase()).as_deref() {
Some("ollama") => {
let url = parts.get(1).unwrap_or(&"http://localhost:11434");
let new_client = crate::llm::OllamaClient::new(url.to_string())
.with_app_state(self.state.clone());
self.llm = new_client.clone();
self.state.set_llm_client(new_client).await;
ui.add_llm_message(format!("✓ Switched to Ollama backend: {}", url));
ui.add_llm_message(" Use /model to list and select a model.".to_string());
}
Some("openai") => {
if parts.len() < 2 {
ui.add_llm_message("✗ Usage: /backend openai <url> [api-key]".to_string());
ui.add_llm_message(" The API key can also be set via NETGET_API_KEY or OPENAI_API_KEY env vars.".to_string());
return Ok(());
}
let url = parts[1];
let api_key = if parts.len() >= 3 {
parts[2].to_string()
} else {
std::env::var("NETGET_API_KEY")
.or_else(|_| std::env::var("OPENAI_API_KEY"))
.unwrap_or_default()
};
if api_key.is_empty() {
ui.add_llm_message("✗ API key required. Provide as third argument or set NETGET_API_KEY/OPENAI_API_KEY env var.".to_string());
return Ok(());
}
let new_client = crate::llm::OllamaClient::new_openai(url, &api_key)
.with_app_state(self.state.clone());
self.llm = new_client.clone();
self.state.set_llm_client(new_client).await;
ui.add_llm_message(format!("✓ Switched to OpenAI-compatible backend: {}", url));
ui.add_llm_message(" Use /model <name> to set the model.".to_string());
}
_ => {
ui.add_llm_message("✗ Unknown backend. Use: /backend ollama [url] or /backend openai <url> [api-key]".to_string());
}
}
Ok(())
}
#[cfg(target_arch = "wasm32")]
async fn handle_set_backend(&mut self, _args: String, ui: &mut App) -> Result<()> {
ui.add_llm_message(
"✗ The browser build talks to the page's model bridge; pick the model on the page."
.to_string(),
);
Ok(())
}
async fn handle_show_docs(&mut self, protocol: Option<String>, ui: &mut App) -> Result<()> {
use crate::docs;
if let Some(protocol_name) = protocol {
match docs::show_protocol_docs(&protocol_name) {
Ok(docs_text) => {
for line in docs_text.lines() {
ui.add_llm_message(line.to_string());
}
}
Err(err_msg) => {
ui.add_llm_message(err_msg);
}
}
} else {
let docs_text = docs::list_all_protocols();
for line in docs_text.lines() {
ui.add_llm_message(line.to_string());
}
}
Ok(())
}
async fn handle_show_environment(&mut self, ui: &mut App) -> Result<()> {
use crate::protocol::{registry, CLIENT_REGISTRY};
ui.add_llm_message("".to_string());
ui.add_llm_message("=== NetGet Environment ===".to_string());
ui.add_llm_message("".to_string());
ui.add_llm_message("System Information:".to_string());
ui.add_llm_message(format!(" OS: {}", std::env::consts::OS));
ui.add_llm_message(format!(" Architecture: {}", std::env::consts::ARCH));
ui.add_llm_message(format!(
" Rust Version: {}",
env!("CARGO_PKG_RUST_VERSION", "unknown")
));
ui.add_llm_message(format!(" NetGet Version: {}", env!("CARGO_PKG_VERSION")));
ui.add_llm_message("".to_string());
let model = self
.state
.get_ollama_model()
.await
.unwrap_or_else(|| "None".to_string());
let web_search_mode = self.state.get_web_search_mode().await;
let scripting_mode = self.state.get_selected_scripting_mode().await;
ui.add_llm_message("LLM Configuration:".to_string());
ui.add_llm_message(format!(" Ollama Model: {}", model));
ui.add_llm_message(format!(" Web Search: {}", web_search_mode.as_str()));
ui.add_llm_message(format!(" Scripting: {}", scripting_mode.as_str()));
ui.add_llm_message("".to_string());
let caps = self.state.get_system_capabilities().await;
let server_excluded = registry().get_excluded_protocols(&caps);
let server_available = registry().get_available_protocols(&caps);
let client_excluded = CLIENT_REGISTRY.get_excluded_protocols(&caps);
let client_available = CLIENT_REGISTRY.get_available_protocols(&caps);
ui.add_llm_message("System Capabilities:".to_string());
ui.add_llm_message(format!(
" Root Access: {}",
if caps.is_root { "Yes" } else { "No" }
));
ui.add_llm_message(format!(
" Privileged Ports (<1024): {}",
if caps.can_bind_privileged_ports {
"Yes"
} else {
"No"
}
));
ui.add_llm_message(format!(
" Raw Socket Access (pcap): {}",
if caps.has_raw_socket_access {
"Yes"
} else {
"No"
}
));
ui.add_llm_message("".to_string());
let total_server_protocols = server_available.len() + server_excluded.len();
let total_client_protocols = client_available.len() + client_excluded.len();
ui.add_llm_message(format!(
"Server Protocols: {} available, {} excluded (total {})",
server_available.len(),
server_excluded.len(),
total_server_protocols
));
ui.add_llm_message(format!(
"Client Protocols: {} available, {} excluded (total {})",
client_available.len(),
client_excluded.len(),
total_client_protocols
));
ui.add_llm_message("".to_string());
if !server_excluded.is_empty() {
ui.add_llm_message("Excluded Server Protocols:".to_string());
let mut excluded_names: Vec<_> = server_excluded.keys().cloned().collect();
excluded_names.sort();
for protocol_name in excluded_names {
if let Some(missing_deps) = server_excluded.get(&protocol_name) {
ui.add_llm_message(format!(" {}", protocol_name));
for dep in missing_deps {
ui.add_llm_message(format!(" ✗ {}: {}", dep.name(), dep.description()));
ui.add_llm_message(format!(" → {}", dep.installation_hint()));
}
}
}
ui.add_llm_message("".to_string());
}
if !client_excluded.is_empty() {
ui.add_llm_message("Excluded Client Protocols:".to_string());
let mut excluded_names: Vec<_> = client_excluded.keys().cloned().collect();
excluded_names.sort();
for protocol_name in excluded_names {
if let Some(missing_deps) = client_excluded.get(&protocol_name) {
ui.add_llm_message(format!(" {}", protocol_name));
for dep in missing_deps {
ui.add_llm_message(format!(" ✗ {}: {}", dep.name(), dep.description()));
ui.add_llm_message(format!(" → {}", dep.installation_hint()));
}
}
}
ui.add_llm_message("".to_string());
}
if server_excluded.is_empty() && client_excluded.is_empty() {
ui.add_llm_message("✓ All protocols are available!".to_string());
ui.add_llm_message("".to_string());
}
Ok(())
}
async fn handle_save(&mut self, name: String, id: Option<u32>, ui: &mut App) -> Result<()> {
use crate::state::client::ClientId;
use crate::state::server::ServerId;
use crate::utils::save_load;
let path = if let Some(id_val) = id {
let server_id = ServerId::new(id_val);
if self.state.get_server(server_id).await.is_some() {
match save_load::save_server(&self.state, server_id, &name).await {
Ok(path) => {
ui.add_llm_message(format!(
"[SAVE] Saved server #{} to: {}",
id_val,
path.display()
));
path
}
Err(e) => {
ui.add_llm_message(format!(
"[ERROR] Failed to save server #{}: {}",
id_val, e
));
return Ok(());
}
}
} else {
let client_id = ClientId::new(id_val);
if self.state.get_client(client_id).await.is_some() {
match save_load::save_client(&self.state, client_id, &name).await {
Ok(path) => {
ui.add_llm_message(format!(
"[SAVE] Saved client #{} to: {}",
id_val,
path.display()
));
path
}
Err(e) => {
ui.add_llm_message(format!(
"[ERROR] Failed to save client #{}: {}",
id_val, e
));
return Ok(());
}
}
} else {
ui.add_llm_message(format!(
"[ERROR] No server or client found with ID #{}",
id_val
));
return Ok(());
}
}
} else {
match save_load::save_all(&self.state, &name).await {
Ok(path) => {
let servers = self.state.get_all_servers().await;
let clients = self.state.get_all_clients().await;
ui.add_llm_message(format!(
"[SAVE] Saved {} server(s) and {} client(s) to: {}",
servers.len(),
clients.len(),
path.display()
));
path
}
Err(e) => {
ui.add_llm_message(format!("[ERROR] Failed to save configuration: {}", e));
return Ok(());
}
}
};
ui.add_llm_message(format!(
"[INFO] Use '/load {}' to restore this configuration",
path.display()
));
Ok(())
}
async fn handle_load(&mut self, name: String, ui: &mut App) -> Result<()> {
use crate::utils::save_load;
let actions = match save_load::load_actions(&name).await {
Ok(actions) => actions,
Err(e) => {
ui.add_llm_message(format!("[ERROR] Failed to load file '{}': {}", name, e));
return Ok(());
}
};
if actions.is_empty() {
ui.add_llm_message(format!("[WARN] File '{}' contains no actions", name));
return Ok(());
}
ui.add_llm_message(format!(
"[LOAD] Loading {} action(s) from: {}",
actions.len(),
save_load::normalize_filename(&name)
));
for (i, action) in actions.iter().enumerate() {
if action.get("type").and_then(serde_json::Value::as_str) == Some("restore_session") {
match save_load::restore_session(&self.state, &self.llm, &action["session"]).await {
Ok(()) => ui.add_llm_message("[LOAD] Restored session relationships".into()),
Err(error) => {
ui.add_llm_message(format!("[ERROR] Session restoration failed: {error:#}"))
}
}
continue;
}
if let Ok(common_action) = crate::llm::actions::common::CommonAction::from_json(action)
{
use crate::llm::actions::common::CommonAction;
match common_action {
CommonAction::OpenServer {
mac_address,
interface,
host,
port,
protocol,
send_first,
initial_memory,
instruction,
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
let (status_tx, _status_rx) = mpsc::unbounded_channel();
match server_startup::start_server_from_action(
&self.state,
mac_address,
interface.clone(),
host,
port,
&protocol,
send_first,
initial_memory,
instruction.clone(),
startup_params,
event_handlers,
scheduled_tasks,
feedback_instructions,
status_tx,
)
.await
{
Ok(server_id) => {
let binding_desc = if let Some(iface) = &interface {
format!("interface {} ({})", iface, protocol)
} else if let Some(p) = port {
format!("port {} ({})", p, protocol)
} else {
format!("({})", protocol)
};
ui.add_llm_message(format!(
"[LOAD] Opened server #{} on {}",
server_id.as_u32(),
binding_desc
));
}
Err(e) => {
ui.add_llm_message(format!(
"[ERROR] Failed to open server (action {}): {}",
i + 1,
e
));
}
}
}
CommonAction::OpenClient {
protocol,
remote_addr,
instruction,
startup_params,
initial_memory,
event_handlers,
scheduled_tasks,
feedback_instructions,
} => {
use crate::cli::client_startup;
match client_startup::start_client_from_action(
&self.state,
&protocol,
&remote_addr,
instruction.clone(),
startup_params,
initial_memory,
event_handlers,
scheduled_tasks,
feedback_instructions,
self.llm.clone(),
None, )
.await
{
Ok(client_id) => {
ui.add_llm_message(format!(
"[LOAD] Opened client #{} to {} ({})",
client_id.as_u32(),
remote_addr,
protocol
));
}
Err(e) => {
ui.add_llm_message(format!(
"[ERROR] Failed to open client (action {}): {}",
i + 1,
e
));
}
}
}
_ => {
ui.add_llm_message(format!(
"[WARN] Skipping unsupported action type (action {})",
i + 1
));
}
}
} else {
ui.add_llm_message(format!("[WARN] Skipping invalid action (action {})", i + 1));
}
}
ui.add_llm_message("[LOAD] Configuration loaded successfully".to_string());
Ok(())
}
#[cfg(feature = "sqlite")]
async fn handle_sqlite(
&mut self,
db_id: Option<u32>,
query: Option<String>,
ui: &mut App,
) -> Result<()> {
match (db_id, query) {
(None, None) => {
let databases = self.state.get_all_databases().await;
if databases.is_empty() {
ui.add_llm_message("[DB] No databases".to_string());
} else {
ui.add_llm_message(format!("[DB] {} database(s):", databases.len()));
for db in databases {
ui.add_llm_message(format!(" {}", db.schema_summary()));
}
}
}
(Some(id), None) => {
let db_id_obj = crate::state::DatabaseId::new(id);
if let Some(db) = self.state.get_database(db_id_obj).await {
ui.add_llm_message(format!("[DB] Database {}:", db_id_obj));
ui.add_llm_message(db.schema_summary());
} else {
ui.add_llm_message(format!("[ERROR] Database {} not found", db_id_obj));
}
}
(Some(id), Some(sql)) => {
let db_id_obj = crate::state::DatabaseId::new(id);
match self.state.execute_sql(db_id_obj, &sql).await {
Ok(result) => {
let formatted = result.format();
ui.add_llm_message(format!("[DB] Query result:\n{}", formatted));
}
Err(e) => {
ui.add_llm_message(format!("[ERROR] SQL error: {}", e));
}
}
}
(None, Some(sql)) => {
let databases = self.state.get_all_databases().await;
if databases.is_empty() {
ui.add_llm_message("[ERROR] No databases available".to_string());
} else {
let db_id = databases[0].id;
match self.state.execute_sql(db_id, &sql).await {
Ok(result) => {
let formatted = result.format();
ui.add_llm_message(format!("[DB] Query result:\n{}", formatted));
}
Err(e) => {
ui.add_llm_message(format!("[ERROR] SQL error: {}", e));
}
}
}
}
}
Ok(())
}
}
const COMMON_ACTION_NAMES: &[&str] = &[
"show_message",
"open_server",
"close_server",
"close_all_servers",
"open_client",
"close_client",
"close_all_clients",
"close_connection_by_id",
"reconnect_client",
"update_client_instruction",
"update_instruction",
"change_model",
"set_memory",
"append_memory",
"append_to_log",
"schedule_task",
"cancel_task",
"provide_feedback",
#[cfg(feature = "sqlite")]
"create_database",
#[cfg(feature = "sqlite")]
"delete_database",
];
pub fn common_action_names() -> &'static [&'static str] {
COMMON_ACTION_NAMES
}
const MAX_LISTED_ACTIONS: usize = 40;
struct ActionCatalog {
names: std::collections::BTreeSet<String>,
scope: String,
}
fn describe_pattern(pattern: &crate::scripting::EventPattern) -> String {
use crate::scripting::EventPattern;
match pattern {
EventPattern::Specific(id) => format!("event '{}'", id),
EventPattern::Wildcard => "the wildcard event pattern".to_string(),
}
}
fn server_actions_for_pattern(
protocol: &dyn crate::llm::actions::protocol_trait::Protocol,
pattern: &crate::scripting::EventPattern,
) -> Option<Vec<String>> {
let event_types = protocol.get_event_types();
let matching: Vec<_> = event_types
.iter()
.filter(|et| pattern.matches(&et.id))
.collect();
if matching.is_empty() {
return None;
}
let mut names: Vec<String> = protocol
.get_sync_actions()
.into_iter()
.map(|a| a.name)
.collect();
for event_type in matching {
names.extend(event_type.actions.iter().map(|a| a.name.clone()));
}
Some(names)
}
fn event_handler_parameter_name(error_msg: &str) -> &'static str {
if error_msg.contains("instruction") {
"event_handlers[].handler.instruction"
} else if error_msg.contains("event_handlers") {
"event_handlers"
} else {
"unknown"
}
}
fn catalog_state() -> &'static AppState {
static CATALOG_STATE: std::sync::OnceLock<AppState> = std::sync::OnceLock::new();
CATALOG_STATE.get_or_init(AppState::new)
}
fn action_catalog_for_pattern(pattern: &crate::scripting::EventPattern) -> ActionCatalog {
let mut names: std::collections::BTreeSet<String> = COMMON_ACTION_NAMES
.iter()
.map(|s| (*s).to_string())
.collect();
let mut scopes: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
for (_, protocol) in crate::protocol::server_registry::registry().all_protocols() {
if let Some(actions) = server_actions_for_pattern(protocol.as_ref(), pattern) {
scopes.insert(protocol.protocol_name().to_string());
names.extend(actions);
}
}
for protocol in crate::protocol::client_registry::CLIENT_REGISTRY.get_all() {
if let Some(actions) = crate::llm::actions::client_trait::client_action_names_for_pattern(
protocol.as_ref(),
catalog_state(),
pattern,
) {
scopes.insert(protocol.protocol_name().to_string());
names.extend(actions);
}
}
if scopes.is_empty() {
return ActionCatalog {
names: std::collections::BTreeSet::new(),
scope: String::new(),
};
}
let scope = scopes.into_iter().collect::<Vec<_>>().join(", ");
ActionCatalog { names, scope }
}
fn validate_static_action_names(
pattern: &crate::scripting::EventPattern,
actions: &[serde_json::Value],
) -> Result<()> {
let catalog = action_catalog_for_pattern(pattern);
if catalog.names.is_empty() {
return Ok(());
}
for action in actions {
let Some(name) = action.get("type").and_then(|v| v.as_str()) else {
anyhow::bail!(
"Static handler action for {} has no \"type\" field: {}",
describe_pattern(pattern),
action
);
};
if catalog.names.contains(name) {
continue;
}
let listed: Vec<&str> = catalog
.names
.iter()
.take(MAX_LISTED_ACTIONS)
.map(|s| s.as_str())
.collect();
let overflow = catalog.names.len().saturating_sub(listed.len());
let mut valid = listed.join(", ");
if overflow > 0 {
valid.push_str(&format!(", … ({} more)", overflow));
}
anyhow::bail!(
"Unknown action \"{}\" in the static handler for {}. \
Valid actions for {}: {}",
name,
describe_pattern(pattern),
catalog.scope,
valid
);
}
Ok(())
}