use crate::llm::actions::{
executor::{execute_actions, ExecutionResult},
get_network_event_common_actions,
protocol_trait::Server,
ActionDefinition,
};
use crate::llm::ollama_client::OllamaClient;
use crate::llm::prompt::PromptBuilder;
use crate::protocol::{Event, EventLogContext};
use crate::state::app_state::AppState;
use crate::state::ServerId;
use anyhow::{Context as AnyhowContext, Result};
use std::sync::Arc;
use tracing::{debug, error, warn};
#[allow(clippy::too_many_arguments)]
pub async fn call_llm_with_actions(
llm_client: &OllamaClient,
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
event_description: &str,
context_json: serde_json::Value,
protocol: Option<&dyn Server>,
custom_actions: Vec<ActionDefinition>,
event_data: Option<serde_json::Value>,
) -> Result<ExecutionResult> {
let event_type_id = crate::scripting::ScriptManager::extract_context_type(event_description);
let handler_instruction: Option<String>;
match crate::llm::event_handler_executor::try_execute_event_handler(
state,
server_id,
connection_id,
&event_type_id,
event_description,
event_data.clone(),
protocol,
)
.await?
{
crate::llm::event_handler_executor::EventHandlerResult::Handled(result) => {
return Ok(result);
}
crate::llm::event_handler_executor::EventHandlerResult::FallbackToLlm { instruction } => {
handler_instruction = instruction;
}
}
let model = crate::llm::ensure_model_selected(
state.get_ollama_model().await,
&state.get_ollama_url().await,
)
.await
.context("Failed to ensure model is selected")?;
let mut all_actions = get_network_event_common_actions();
let has_feedback_instructions = state
.with_server_mut(server_id, |server| server.feedback_instructions.is_some())
.await
.unwrap_or(false);
if has_feedback_instructions {
all_actions.push(crate::llm::actions::common::provide_feedback_action());
}
if let Some(proto) = protocol {
all_actions.extend(proto.get_sync_actions());
}
all_actions.extend(custom_actions);
debug!(
"LLM call for event: {} (server #{}, {} actions available)",
event_description,
server_id.as_u32(),
all_actions.len()
);
let (system_prompt, advertised_actions) =
PromptBuilder::build_network_event_action_prompt_for_server_with_actions(
state,
server_id,
all_actions,
)
.await;
let truncated_desc = format!(
"LLM \"{}\"",
crate::utils::truncate_for_log(event_description, 27)
);
let rate_limiter = state.get_rate_limiter().await;
let mut conversation = crate::llm::ConversationHandler::new(
system_prompt,
std::sync::Arc::new(llm_client.clone()),
model,
rate_limiter,
crate::llm::RequestSource::Network, )
.with_tracking(
state.clone(),
crate::state::app_state::ConversationSource::Network {
server_id,
connection_id,
},
truncated_desc,
);
let event_trigger =
PromptBuilder::build_event_trigger_message(event_description, context_json.clone());
conversation.add_user_message(apply_handler_instruction(
event_trigger,
handler_instruction.as_deref(),
));
let web_search_mode = state.get_web_search_mode().await;
let approval_tx = state.get_web_approval_channel().await;
let action_values = conversation
.generate_with_tools_and_retry(approval_tx, web_search_mode, advertised_actions)
.await
.context("LLM generate with tools failed")?;
if action_values.is_empty() {
warn!(
"LLM returned empty actions array for event: {}",
event_description
);
}
let result = execute_actions(action_values, state, protocol, Some(server_id), None)
.await
.context("Failed to execute actions")?;
debug!(
"LLM call completed: {} messages, {} protocol results",
result.messages.len(),
result.protocol_results.len()
);
Ok(result)
}
pub async fn call_llm_with_protocol(
llm_client: &OllamaClient,
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
event_description: &str,
protocol: &dyn Server,
) -> Result<ExecutionResult> {
call_llm_with_actions(
llm_client,
state,
server_id,
connection_id,
event_description,
serde_json::json!({}), Some(protocol),
Vec::new(), None, )
.await
}
pub async fn call_llm_with_custom_actions(
llm_client: &OllamaClient,
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
event_description: &str,
custom_actions: Vec<ActionDefinition>,
) -> Result<ExecutionResult> {
call_llm_with_actions(
llm_client,
state,
server_id,
connection_id,
event_description,
serde_json::json!({}), None,
custom_actions,
None, )
.await
}
async fn record_event_access_log(
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
protocol: &dyn Server,
event: &Event,
result: &ExecutionResult,
) {
if let Some(summary) = result.failure_summary() {
error!(
"{} action(s) failed while handling '{}' on server #{} ({}): {}",
result.failures.len(),
event.id(),
server_id.as_u32(),
protocol.protocol_name(),
summary
);
}
state
.record_access_log(
crate::state::AccessLogOwner::Server(server_id.as_u32()),
protocol.protocol_name(),
connection_id.map(|c| c.as_u32()),
event.id(),
event.data.clone(),
if crate::utils::redact::contains_credentials(&event.data) {
vec![serde_json::json!({"type":"private_handler_result", "action_count":result.raw_actions.len(), "failed_actions":result.failures.len()})]
} else { result.access_log_actions() },
)
.await;
}
async fn record_failed_event_access_log(
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
protocol: &dyn Server,
event: &Event,
error: &anyhow::Error,
) {
state
.record_access_log(
crate::state::AccessLogOwner::Server(server_id.as_u32()),
protocol.protocol_name(),
connection_id.map(|c| c.as_u32()),
event.id(),
event.data.clone(),
vec![serde_json::json!({
"type": format!("FAILED: {}", event.id()),
"error": if crate::utils::redact::contains_credentials(&event.data) { "credential-bearing event failed; diagnostics hidden".to_string() } else { error.to_string() },
})],
)
.await;
}
pub async fn call_llm(
llm_client: &OllamaClient,
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
event: &Event,
protocol: &dyn Server,
) -> Result<ExecutionResult> {
anyhow::ensure!(
crate::utils::json_budget::within_budget(
&event.data,
crate::scripting::event_handler::MAX_INTERPOLATION_BYTES,
crate::scripting::event_handler::MAX_INTERPOLATION_NODES,
crate::scripting::event_handler::MAX_INTERPOLATION_DEPTH
),
"event exceeds the shared JSON budget"
);
let private = crate::utils::redact::contains_credentials(&event.data);
let run = call_llm_inner(
llm_client,
state,
server_id,
connection_id,
event,
protocol,
private,
);
let outcome = if private {
use tracing::instrument::WithSubscriber;
run.with_subscriber(tracing::subscriber::NoSubscriber::default())
.await
.map_err(crate::utils::redact::hide_error_details)
} else {
run.await
};
if let Err(e) = &outcome {
record_failed_event_access_log(state, server_id, connection_id, protocol, event, e).await;
}
outcome
}
#[allow(clippy::too_many_arguments)]
async fn call_llm_inner(
llm_client: &OllamaClient,
state: &AppState,
server_id: ServerId,
connection_id: Option<crate::server::connection::ConnectionId>,
event: &Event,
protocol: &dyn Server,
private_payloads: bool,
) -> Result<ExecutionResult> {
let client_addr = if let Some(conn_id) = connection_id {
state
.with_server_mut(server_id, |server| {
server.connections.get(&conn_id).map(|c| c.remote_addr)
})
.await
.flatten()
} else {
None
};
let log_ctx = EventLogContext::new(
event,
server_id,
connection_id,
client_addr,
protocol.protocol_name(),
);
let event_status_tx = llm_client.status_tx();
if !private_payloads {
log_ctx.log_start(event_status_tx);
}
crate::pipe::dispatch_pipes(state, server_id, event).await;
if let Some(easy_id) = state.get_easy_for_server(server_id).await {
use crate::protocol::EASY_REGISTRY;
if let Some(easy_instance) = state.get_easy_instance(easy_id).await {
if let Some(easy_protocol) = EASY_REGISTRY.get_by_name(&easy_instance.protocol_name) {
let actions = easy_protocol
.handle_event(
event.clone(),
easy_instance.user_instruction.clone(),
Arc::new(llm_client.clone()),
Arc::new(state.clone()),
)
.await
.context("Easy protocol handler failed")?;
let result = crate::llm::execute_actions(
actions,
state,
Some(protocol),
Some(server_id),
None, )
.await?;
if !private_payloads {
log_ctx.log_complete(event_status_tx, &result.protocol_results);
}
record_event_access_log(state, server_id, connection_id, protocol, event, &result)
.await;
return Ok(result);
}
}
}
let handler_instruction: Option<String>;
match crate::llm::event_handler_executor::try_execute_event_handler(
state,
server_id,
connection_id,
&event.event_type.id,
&event.event_type.description,
Some(event.data.clone()),
Some(protocol),
)
.await?
{
crate::llm::event_handler_executor::EventHandlerResult::Handled(result) => {
if !private_payloads {
log_ctx.log_complete(event_status_tx, &result.protocol_results);
}
record_event_access_log(state, server_id, connection_id, protocol, event, &result)
.await;
return Ok(result);
}
crate::llm::event_handler_executor::EventHandlerResult::FallbackToLlm { instruction } => {
handler_instruction = instruction;
}
}
let model = crate::llm::ensure_model_selected(
state.get_ollama_model().await,
&state.get_ollama_url().await,
)
.await
.context("Failed to ensure model is selected")?;
let mut all_actions = get_network_event_common_actions();
let has_feedback_instructions = state
.with_server_mut(server_id, |server| server.feedback_instructions.is_some())
.await
.unwrap_or(false);
if has_feedback_instructions {
all_actions.push(crate::llm::actions::common::provide_feedback_action());
}
if event.event_type.has_no_usable_actions() {
let fallback = protocol.get_sync_actions();
if !fallback.is_empty() {
error!(
"BUG: event '{}' of protocol '{}' declares no actions of its own, so the model \
would be offered none of the protocol's {} sync action(s) and anything \
protocol-specific it returned would be rejected as an unknown action. Falling \
back to the full sync action set. Fix by adding .with_actions(...) to the event \
type, or .with_no_actions() if it genuinely needs none.",
event.id(),
protocol.protocol_name(),
fallback.len()
);
all_actions.extend(fallback);
}
} else {
all_actions.extend(event.event_type.actions.clone());
}
debug!(
"LLM call for event '{}' (server #{}, {} actions available)",
event.id(),
server_id.as_u32(),
all_actions.len()
);
let event_description = event.to_prompt_description();
let (system_prompt, advertised_actions) =
PromptBuilder::build_network_event_action_prompt_for_server_with_actions(
state,
server_id,
all_actions,
)
.await;
let truncated_desc = if private_payloads {
"LLM credential-bearing event".to_string()
} else {
format!(
"LLM \"{}\"",
crate::utils::truncate_for_log(&event_description, 27)
)
};
let rate_limiter = state.get_rate_limiter().await;
let mut conversation = crate::llm::ConversationHandler::new(
system_prompt,
std::sync::Arc::new(llm_client.clone()),
model,
rate_limiter,
crate::llm::RequestSource::Network, )
.with_bridge_event(server_id, connection_id, protocol.protocol_name(), event)
.with_private_payloads(private_payloads)
.with_tracking(
state.clone(),
crate::state::app_state::ConversationSource::Network {
server_id,
connection_id,
},
truncated_desc,
);
let event_trigger = PromptBuilder::build_event_trigger_message_with_id(
event.id(),
&event_description,
event.data.clone(),
);
conversation.add_user_message(apply_handler_instruction(
event_trigger,
handler_instruction.as_deref(),
));
let actions = conversation
.generate_with_tools_and_retry(
None, crate::state::app_state::WebSearchMode::Off, advertised_actions,
)
.await
.context("✗ LLM failed to generate valid response after retries.")?;
if actions.is_empty() {
warn!("LLM returned empty actions array for event: {}", event.id());
}
let result = execute_actions(actions, state, Some(protocol), Some(server_id), None)
.await
.context("Failed to execute actions")?;
debug!(
"LLM call completed: {} messages, {} protocol results",
result.messages.len(),
result.protocol_results.len()
);
if !private_payloads {
log_ctx.log_complete(event_status_tx, &result.protocol_results);
}
record_event_access_log(state, server_id, connection_id, protocol, event, &result).await;
Ok(result)
}
#[derive(Debug, Clone)]
pub struct ClientLlmResult {
pub actions: Vec<serde_json::Value>,
pub memory_updates: Option<String>,
}
#[allow(clippy::too_many_arguments)]
pub async fn call_llm_for_client(
llm_client: &OllamaClient,
state: &AppState,
client_id: String,
instruction: &str,
memory: &str,
event: Option<&Event>,
protocol: &dyn crate::llm::actions::client_trait::Client,
status_tx: &tokio::sync::mpsc::UnboundedSender<String>,
) -> Result<ClientLlmResult> {
if let Some(event) = event {
anyhow::ensure!(
crate::utils::json_budget::within_budget(
&event.data,
crate::scripting::event_handler::MAX_INTERPOLATION_BYTES,
crate::scripting::event_handler::MAX_INTERPOLATION_NODES,
crate::scripting::event_handler::MAX_INTERPOLATION_DEPTH,
),
"event exceeds the shared JSON budget"
);
}
let private_payloads = event
.is_some_and(|event| crate::utils::redact::contains_credentials(&event.data))
|| crate::utils::redact::actions_have_credentials(
&crate::llm::actions::client_trait::client_llm_action_set(protocol, state, event),
);
let run = call_llm_for_client_inner(
llm_client,
state,
client_id,
instruction,
memory,
event,
protocol,
status_tx,
private_payloads,
);
if private_payloads {
use tracing::instrument::WithSubscriber;
run.with_subscriber(tracing::subscriber::NoSubscriber::default())
.await
.map_err(crate::utils::redact::hide_error_details)
} else {
run.await
}
}
#[allow(clippy::too_many_arguments)]
async fn call_llm_for_client_inner(
llm_client: &OllamaClient,
state: &AppState,
client_id: String,
instruction: &str,
memory: &str,
event: Option<&Event>,
protocol: &dyn crate::llm::actions::client_trait::Client,
status_tx: &tokio::sync::mpsc::UnboundedSender<String>,
private_payloads: bool,
) -> Result<ClientLlmResult> {
let mut all_actions =
crate::llm::actions::client_trait::client_llm_action_set(protocol, state, event);
all_actions.push(crate::llm::actions::common::set_memory_action());
all_actions.push(crate::llm::actions::common::append_memory_action());
all_actions.push(crate::llm::actions::common::show_message_action());
if let Some(cid) = crate::state::ClientId::from_string(&client_id) {
let has_feedback_instructions = state
.with_client_mut(cid, |client| client.feedback_instructions.is_some())
.await
.unwrap_or(false);
if has_feedback_instructions {
all_actions.push(crate::llm::actions::common::provide_feedback_action());
}
}
let system_prompt =
format!(
"You are controlling a network client ({}). Your instruction: {}\n\nAvailable actions:\n{}",
protocol.protocol_name(),
instruction,
all_actions.iter().map(|a| a.to_prompt_text()).collect::<Vec<_>>().join("\n\n")
);
let user_message = if let Some(ev) = event {
format!(
"Event: {}\nData: {}",
ev.id(),
serde_json::to_string_pretty(&ev.data).unwrap_or_default()
)
} else {
"Waiting for instructions".to_string()
};
let full_message = if !memory.is_empty() {
format!("Memory: {}\n\n{}", memory, user_message)
} else {
user_message
};
let current_model = state.get_ollama_model().await;
let model =
crate::llm::ensure_model_selected(current_model.clone(), &state.get_ollama_url().await)
.await
.context("Failed to ensure model is selected")?;
if current_model.is_none() {
let _ = status_tx.send(format!(
"⚠ Auto-selected model: {} (no model was configured)",
model
));
}
let rate_limiter = state.get_rate_limiter().await;
let mut conversation = crate::llm::ConversationHandler::new(
system_prompt,
std::sync::Arc::new(llm_client.clone()),
model,
rate_limiter,
crate::llm::RequestSource::Network, )
.with_status_tx(status_tx.clone())
.with_private_payloads(private_payloads);
if let Some(client_id) = crate::state::ClientId::from_string(&client_id) {
conversation = conversation.with_tracking(
state.clone(),
crate::state::app_state::ConversationSource::Client { client_id },
event
.map(|ev| ev.id().to_string())
.unwrap_or_else(|| "Client instructions".into()),
);
}
conversation.add_user_message(full_message);
let actions = conversation
.generate_with_tools_and_retry(
None,
crate::state::app_state::WebSearchMode::Off,
all_actions,
)
.await?;
let (common_actions, protocol_actions) = split_client_common_actions(actions);
for action in &common_actions {
if action.get("type").and_then(|t| t.as_str()) == Some("show_message") {
if let Some(message) = action.get("message").and_then(|m| m.as_str()) {
let _ = status_tx.send(format!("[CLIENT] {}", message));
}
}
}
if !common_actions.is_empty() {
match crate::state::ClientId::from_string(&client_id) {
Some(cid) => {
if let Err(e) = crate::llm::actions::executor::execute_actions(
common_actions,
state,
None, None, Some(cid),
)
.await
{
let error = if private_payloads {
crate::utils::redact::hide_error_details(e)
} else {
e
};
tracing::warn!("Client common action execution failed: {}", error);
let _ = status_tx.send(format!("[CLIENT] feedback action failed: {error}"));
}
}
None => {
tracing::warn!(
"Client returned a common action but client_id {:?} could not be parsed; \
the action was discarded",
client_id
);
}
}
}
let memory_updates = None;
Ok(ClientLlmResult {
actions: protocol_actions,
memory_updates,
})
}
pub const CLIENT_COMMON_ACTION_NAMES: &[&str] = &[
"provide_feedback",
"set_memory",
"append_memory",
"show_message",
];
pub fn split_client_common_actions(
actions: Vec<serde_json::Value>,
) -> (Vec<serde_json::Value>, Vec<serde_json::Value>) {
actions.into_iter().partition(|action| {
action
.get("type")
.and_then(|t| t.as_str())
.map(|name| CLIENT_COMMON_ACTION_NAMES.contains(&name))
.unwrap_or(false)
})
}
#[allow(clippy::too_many_arguments)]
pub async fn call_llm_for_feedback(
llm_client: &OllamaClient,
state: &AppState,
server_id: Option<crate::state::ServerId>,
client_id: Option<crate::state::ClientId>,
feedback_instructions: &str,
current_instruction: &str,
memory: &str,
feedback_entries: &[serde_json::Value],
status_tx: &tokio::sync::mpsc::UnboundedSender<String>,
) -> Result<Vec<serde_json::Value>> {
use crate::llm::actions::get_user_input_common_actions;
use crate::llm::prompt::PromptBuilder;
let selected_mode = state.get_selected_scripting_mode().await;
let scripting_env = state.get_scripting_env().await;
let is_open_server_enabled = true;
let is_open_client_enabled = true;
let available_actions = get_user_input_common_actions(
selected_mode,
&scripting_env,
is_open_server_enabled,
is_open_client_enabled,
);
let has_scripting = selected_mode != crate::state::app_state::ScriptingMode::Off;
let feedback_actions =
PromptBuilder::filter_actions_by_scripting_mode(available_actions, has_scripting);
let system_prompt = PromptBuilder::build_feedback_system_prompt(
state,
server_id,
client_id,
feedback_instructions,
current_instruction,
memory,
feedback_entries,
feedback_actions.clone(),
)
.await;
let current_model = state.get_ollama_model().await;
let model =
crate::llm::ensure_model_selected(current_model.clone(), &state.get_ollama_url().await)
.await
.context("Failed to ensure model is selected")?;
if current_model.is_none() {
let _ = status_tx.send(format!(
"⚠ Auto-selected model: {} (no model was configured)",
model
));
}
let instance_type = if server_id.is_some() {
"server"
} else {
"client"
};
let instance_id = server_id
.map(|id| id.as_u32())
.or_else(|| client_id.map(|id| id.as_u32()))
.unwrap_or(0);
debug!(
"LLM feedback processing for {} #{} ({} feedback entries)",
instance_type,
instance_id,
feedback_entries.len()
);
let rate_limiter = state.get_rate_limiter().await;
let conversation_source = if let Some(sid) = server_id {
crate::state::app_state::ConversationSource::Network {
server_id: sid,
connection_id: None,
}
} else if let Some(client_id) = client_id {
crate::state::app_state::ConversationSource::Client { client_id }
} else {
crate::state::app_state::ConversationSource::Task {
task_name: format!("feedback-client-{}", instance_id),
}
};
let mut conversation = crate::llm::ConversationHandler::new(
system_prompt,
std::sync::Arc::new(llm_client.clone()),
model,
rate_limiter,
crate::llm::RequestSource::User, )
.with_native_tools(&feedback_actions)
.with_status_tx(status_tx.clone())
.with_tracking(
state.clone(),
conversation_source,
format!("Feedback processing ({} entries)", feedback_entries.len()),
);
conversation
.add_user_message("Analyze the accumulated feedback and suggest adjustments.".to_string());
let web_search_mode = state.get_web_search_mode().await;
let actions = conversation
.generate_with_tools_and_retry(
state.get_web_approval_channel().await,
web_search_mode,
feedback_actions,
)
.await
.context("✗ LLM failed to generate feedback processing response after retries")?;
if actions.is_empty() {
warn!(
"LLM returned empty actions for {} #{} feedback processing (no adjustments needed)",
instance_type, instance_id
);
} else {
debug!(
"LLM feedback processing completed for {} #{}: {} adjustment actions",
instance_type,
instance_id,
actions.len()
);
}
Ok(actions)
}
fn apply_handler_instruction(event_trigger: String, instruction: Option<&str>) -> String {
match instruction {
Some(instruction) if !instruction.trim().is_empty() => {
debug!(
"Applying per-event handler instruction ({} bytes) to the event prompt",
instruction.len()
);
format!(
"{}\n\n{}:\n{}",
event_trigger,
crate::llm::event_handler_executor::HANDLER_INSTRUCTION_HEADER,
instruction
)
}
_ => event_trigger,
}
}