1use crate::llm::actions::executor::{execute_actions, ExecutionResult};
10use crate::llm::actions::protocol_trait::Server;
11use crate::scripting::EventHandlerType;
12use crate::state::app_state::AppState;
13use crate::state::ServerId;
14use anyhow::{Context as AnyhowContext, Result};
15use tracing::{debug, error, warn};
16
17pub const HANDLER_INSTRUCTION_HEADER: &str = "Handler instruction for this event";
23
24pub enum EventHandlerResult {
26 Handled(ExecutionResult),
28 FallbackToLlm { instruction: Option<String> },
36}
37
38pub async fn try_execute_event_handler(
54 state: &AppState,
55 server_id: ServerId,
56 connection_id: Option<crate::server::connection::ConnectionId>,
57 event_type_id: &str,
58 event_description: &str,
59 event_data: Option<serde_json::Value>,
60 protocol: Option<&dyn Server>,
61) -> Result<EventHandlerResult> {
62 if event_data.as_ref().is_some_and(|value| {
63 !crate::utils::json_budget::within_budget(
64 value,
65 crate::scripting::event_handler::MAX_INTERPOLATION_BYTES,
66 crate::scripting::event_handler::MAX_INTERPOLATION_NODES,
67 crate::scripting::event_handler::MAX_INTERPOLATION_DEPTH,
68 )
69 }) {
70 if let Some(value) = event_data {
71 crate::utils::json_budget::drop_iteratively(value);
72 }
73 anyhow::bail!("event exceeds the shared JSON budget");
74 }
75 let private = event_data
76 .as_ref()
77 .is_some_and(crate::utils::redact::contains_credentials);
78 let run = try_execute_event_handler_inner(
79 state,
80 server_id,
81 connection_id,
82 event_type_id,
83 event_description,
84 event_data,
85 protocol,
86 private,
87 );
88 if private {
89 use tracing::instrument::WithSubscriber;
90 run.with_subscriber(tracing::subscriber::NoSubscriber::default())
91 .await
92 .map_err(|_| anyhow::anyhow!("credential-bearing handler failed; diagnostics hidden"))
93 } else {
94 run.await
95 }
96}
97
98#[allow(clippy::too_many_arguments)]
99async fn try_execute_event_handler_inner(
100 state: &AppState,
101 server_id: ServerId,
102 connection_id: Option<crate::server::connection::ConnectionId>,
103 event_type_id: &str,
104 event_description: &str,
105 event_data: Option<serde_json::Value>,
106 protocol: Option<&dyn Server>,
107 private_payloads: bool,
108) -> Result<EventHandlerResult> {
109 let event_handler_config = state.get_event_handler_config(server_id).await;
111
112 let Some(config) = event_handler_config else {
113 return Ok(EventHandlerResult::FallbackToLlm { instruction: None });
115 };
116
117 let Some(handler_type) = config.find_handler(event_type_id) else {
119 debug!("No handler matches event '{}', using LLM", event_type_id);
121 return Ok(EventHandlerResult::FallbackToLlm { instruction: None });
122 };
123
124 match handler_type {
125 EventHandlerType::Llm { instruction } => {
126 debug!(
129 "LLM handler configured for event '{}' with instruction: {}",
130 event_type_id, instruction
131 );
132 Ok(EventHandlerResult::FallbackToLlm {
133 instruction: Some(instruction.to_string()),
134 })
135 }
136
137 EventHandlerType::Script {
138 language,
139 code,
140 resident,
141 scope,
142 } => {
143 execute_script_handler(
145 state,
146 server_id,
147 connection_id,
148 event_type_id,
149 event_description,
150 event_data,
151 language,
152 code,
153 *resident,
154 scope.as_deref(),
155 protocol,
156 private_payloads,
157 )
158 .await
159 }
160
161 EventHandlerType::Static { actions } => {
162 execute_static_handler(
164 state,
165 server_id,
166 event_type_id,
167 event_description,
168 actions,
169 event_data.as_ref(),
170 protocol,
171 )
172 .await
173 }
174
175 EventHandlerType::Manual { timeout_secs } => {
176 let actions = await_manual_answer(
177 state,
178 crate::state::intercepts::InterceptOwner::Server(server_id),
179 connection_id.map(|c| c.as_u32()),
180 event_type_id,
181 event_description,
182 event_data.clone(),
183 *timeout_secs,
184 )
185 .await?;
186 execute_static_handler(
190 state,
191 server_id,
192 event_type_id,
193 event_description,
194 &actions,
195 event_data.as_ref(),
196 protocol,
197 )
198 .await
199 }
200 }
201}
202
203async fn await_manual_answer(
211 state: &AppState,
212 owner: crate::state::intercepts::InterceptOwner,
213 connection_id: Option<u32>,
214 event_type_id: &str,
215 event_description: &str,
216 event_data: Option<serde_json::Value>,
217 timeout_secs: u64,
218) -> Result<Vec<serde_json::Value>> {
219 let (id, reply_rx) = state
220 .park_intercept(
221 owner,
222 connection_id,
223 event_type_id,
224 event_description,
225 event_data,
226 timeout_secs,
227 )
228 .await;
229 tracing::info!(
230 "manual handler: event '{}' parked as intercept #{} — waiting up to {}s for an \
231 operator answer at the dashboard",
232 event_type_id,
233 id,
234 timeout_secs
235 );
236
237 match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), reply_rx).await {
238 Ok(Ok(actions)) => {
239 debug!(
240 "manual handler: intercept #{} answered with {} action(s)",
241 id,
242 actions.len()
243 );
244 Ok(actions)
245 }
246 Ok(Err(_)) => {
247 anyhow::bail!(
249 "manual handler: the operator dismissed request #{id} for event '{event_type_id}'"
250 )
251 }
252 Err(_) => {
253 state.remove_intercept(id).await;
254 warn!(
255 "manual handler: intercept #{} for event '{}' got no operator answer within \
256 {}s — failing closed",
257 id, event_type_id, timeout_secs
258 );
259 anyhow::bail!(
260 "manual handler: no operator answer within {timeout_secs}s for event \
261 '{event_type_id}'"
262 )
263 }
264 }
265}
266
267pub enum ClientEventHandlerResult {
277 Handled { actions: Vec<serde_json::Value> },
279 FallbackToLlm { instruction: Option<String> },
283}
284
285pub async fn try_execute_client_event_handler(
298 state: &AppState,
299 client_id: crate::state::ClientId,
300 event_type_id: &str,
301 event_description: &str,
302 event_data: Option<serde_json::Value>,
303) -> Result<ClientEventHandlerResult> {
304 try_execute_client_event_handler_with_privacy(
305 state,
306 client_id,
307 event_type_id,
308 event_description,
309 event_data,
310 false,
311 )
312 .await
313}
314
315pub async fn try_execute_client_event_handler_with_privacy(
317 state: &AppState,
318 client_id: crate::state::ClientId,
319 event_type_id: &str,
320 event_description: &str,
321 event_data: Option<serde_json::Value>,
322 private_payloads: bool,
323) -> Result<ClientEventHandlerResult> {
324 if event_data.as_ref().is_some_and(|value| {
325 !crate::utils::json_budget::within_budget(
326 value,
327 crate::scripting::event_handler::MAX_INTERPOLATION_BYTES,
328 crate::scripting::event_handler::MAX_INTERPOLATION_NODES,
329 crate::scripting::event_handler::MAX_INTERPOLATION_DEPTH,
330 )
331 }) {
332 if let Some(value) = event_data {
333 crate::utils::json_budget::drop_iteratively(value);
334 }
335 anyhow::bail!("event exceeds the shared JSON budget");
336 }
337 let private_payloads = private_payloads
338 || event_data
339 .as_ref()
340 .is_some_and(crate::utils::redact::contains_credentials);
341 let run = try_execute_client_event_handler_inner(
342 state,
343 client_id,
344 event_type_id,
345 event_description,
346 event_data,
347 private_payloads,
348 );
349 if private_payloads {
350 use tracing::instrument::WithSubscriber;
351 run.with_subscriber(tracing::subscriber::NoSubscriber::default())
352 .await
353 .map_err(crate::utils::redact::hide_error_details)
354 } else {
355 run.await
356 }
357}
358
359async fn try_execute_client_event_handler_inner(
360 state: &AppState,
361 client_id: crate::state::ClientId,
362 event_type_id: &str,
363 event_description: &str,
364 event_data: Option<serde_json::Value>,
365 private_payloads: bool,
366) -> Result<ClientEventHandlerResult> {
367 let Some(config) = state.get_client_event_handler_config(client_id).await else {
368 return Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None });
369 };
370
371 let Some(handler_type) = config.find_handler(event_type_id) else {
372 debug!(
373 "No client handler matches event '{}', using LLM",
374 event_type_id
375 );
376 return Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None });
377 };
378
379 match handler_type {
380 EventHandlerType::Llm { instruction } => Ok(ClientEventHandlerResult::FallbackToLlm {
381 instruction: Some(instruction.to_string()),
382 }),
383
384 EventHandlerType::Static { actions } => {
385 debug!(
386 "Static client handler executing for event '{}' ({} actions)",
387 event_type_id,
388 actions.len()
389 );
390 let actions =
391 crate::scripting::event_handler::interpolate_actions(actions, event_data.as_ref())
392 .with_context(|| {
393 format!(
394 "Static handler for client event '{}' could not be rendered",
395 event_type_id
396 )
397 })?;
398 Ok(ClientEventHandlerResult::Handled { actions })
399 }
400
401 EventHandlerType::Script {
402 language,
403 code,
404 resident,
405 scope,
406 } => {
407 execute_client_script_handler(
408 state,
409 client_id,
410 event_type_id,
411 event_description,
412 event_data,
413 language,
414 code,
415 *resident,
416 scope.as_deref(),
417 private_payloads,
418 )
419 .await
420 }
421
422 EventHandlerType::Manual { timeout_secs } => {
423 let actions = await_manual_answer(
424 state,
425 crate::state::intercepts::InterceptOwner::Client(client_id),
426 None,
427 event_type_id,
428 event_description,
429 event_data.clone(),
430 *timeout_secs,
431 )
432 .await?;
433 let actions =
436 crate::scripting::event_handler::interpolate_actions(&actions, event_data.as_ref())
437 .with_context(|| {
438 format!(
439 "Manual answer for client event '{}' could not be rendered",
440 event_type_id
441 )
442 })?;
443 Ok(ClientEventHandlerResult::Handled { actions })
444 }
445 }
446}
447
448#[allow(clippy::too_many_arguments)]
451async fn execute_client_script_handler(
452 state: &AppState,
453 client_id: crate::state::ClientId,
454 event_type_id: &str,
455 event_description: &str,
456 event_data: Option<serde_json::Value>,
457 language: &str,
458 code: &str,
459 resident: bool,
460 scope: Option<&str>,
461 private_payloads: bool,
462) -> Result<ClientEventHandlerResult> {
463 let Some(client) = state.get_client(client_id).await else {
464 warn!(
465 "Client #{} not found for script execution",
466 client_id.as_u32()
467 );
468 return Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None });
469 };
470
471 let event_json =
472 event_data.unwrap_or_else(|| serde_json::json!({"description": event_description}));
473
474 let script_input = crate::scripting::types::ScriptInput {
475 event_type_id: event_type_id.to_string(),
476 server: None,
477 client: Some(crate::scripting::types::ClientContext {
478 id: client.id.as_u32(),
479 remote_addr: client.remote_addr.clone(),
480 protocol: client.protocol_name.clone(),
481 memory: client.memory.clone(),
482 instruction: client.instruction.clone(),
483 }),
484 connection: None,
485 event: event_json,
486 };
487
488 let script_language = match language.to_lowercase().as_str() {
489 "python" => crate::scripting::ScriptLanguage::Python,
490 "javascript" | "js" => crate::scripting::ScriptLanguage::JavaScript,
491 "go" => crate::scripting::ScriptLanguage::Go,
492 "perl" => crate::scripting::ScriptLanguage::Perl,
493 _ => {
494 warn!(
495 "Unknown script language '{}', falling back to LLM",
496 language
497 );
498 return Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None });
499 }
500 };
501
502 let scripting_env = state.get_scripting_env().await;
503 if !scripting_env.is_available(script_language) {
504 warn!(
505 "Script language {} not available, falling back to LLM",
506 script_language.as_str()
507 );
508 return Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None });
509 }
510
511 let script_config = crate::scripting::types::ScriptConfig {
512 language: script_language,
513 source: crate::scripting::types::ScriptSource::Inline(code.to_string()),
514 handles_contexts: vec![event_type_id.to_string()],
515 };
516
517 let use_resident =
518 resident && crate::scripting::resident::resident_language_supported(script_language);
519 if resident && !use_resident {
520 warn!(
521 "Resident mode requested for '{}' but that language runs per-event only; \
522 falling back to per-event execution",
523 script_language.as_str()
524 );
525 }
526
527 let script_result = if private_payloads {
528 use tracing::instrument::WithSubscriber;
529 let run = async {
530 if use_resident {
531 crate::scripting::ResidentScriptManager::dispatch_private(
532 &script_config,
533 &script_input,
534 crate::scripting::ResidentScope::parse(scope),
535 )
536 .await
537 } else {
538 crate::scripting::executor::execute_script_async(&script_config, &script_input)
539 .await
540 }
541 };
542 run.with_subscriber(tracing::subscriber::NoSubscriber::default())
543 .await
544 .map_err(|_| anyhow::anyhow!("credential-bearing script failed; diagnostics hidden"))
545 } else if use_resident {
546 crate::scripting::ResidentScriptManager::dispatch(
547 &script_config,
548 &script_input,
549 crate::scripting::ResidentScope::parse(scope),
550 )
551 .await
552 } else {
553 crate::scripting::executor::execute_script_async(&script_config, &script_input).await
554 };
555
556 match script_result {
557 Ok(response) => {
558 debug!(
559 "Script handled client event '{}' ({} actions)",
560 event_type_id,
561 response.actions.len()
562 );
563 Ok(ClientEventHandlerResult::Handled {
564 actions: response.actions,
565 })
566 }
567 Err(e) => {
568 warn!("Client script execution failed: {}, falling back to LLM", e);
569 Ok(ClientEventHandlerResult::FallbackToLlm { instruction: None })
570 }
571 }
572}
573
574#[allow(clippy::too_many_arguments)]
576async fn execute_script_handler(
577 state: &AppState,
578 server_id: ServerId,
579 connection_id: Option<crate::server::connection::ConnectionId>,
580 event_type_id: &str,
581 event_description: &str,
582 event_data: Option<serde_json::Value>,
583 language: &str,
584 code: &str,
585 resident: bool,
586 scope: Option<&str>,
587 protocol: Option<&dyn Server>,
588 private_payloads: bool,
589) -> Result<EventHandlerResult> {
590 let server_info = state.get_server(server_id).await;
592
593 let Some(server) = server_info else {
594 warn!(
595 "Server #{} not found for script execution",
596 server_id.as_u32()
597 );
598 return Ok(EventHandlerResult::FallbackToLlm { instruction: None });
599 };
600
601 let connection_context = if let Some(conn_id) = connection_id {
603 server.connections.get(&conn_id).map(|conn_state| {
604 crate::scripting::types::ConnectionContext {
605 id: conn_id.to_string(),
606 remote_addr: conn_state.remote_addr.to_string(),
607 bytes_received: conn_state.bytes_received,
608 bytes_sent: conn_state.bytes_sent,
609 }
610 })
611 } else {
612 None
613 };
614
615 let event_json =
617 event_data.unwrap_or_else(|| serde_json::json!({"description": event_description}));
618
619 let script_input = crate::scripting::types::ScriptInput {
620 event_type_id: event_type_id.to_string(),
621 client: None,
622 server: Some(crate::scripting::types::ServerContext {
623 id: server.id.as_u32(),
624 port: server.port,
625 stack: crate::protocol::server_registry::registry()
626 .stack_name_by_protocol(&server.protocol_name)
627 .unwrap_or("UNKNOWN")
628 .to_string(),
629 memory: server.memory.clone(),
630 instruction: server.instruction.clone(),
631 }),
632 connection: connection_context,
633 event: event_json,
634 };
635
636 let script_language = match language.to_lowercase().as_str() {
638 "python" => crate::scripting::ScriptLanguage::Python,
639 "javascript" | "js" => crate::scripting::ScriptLanguage::JavaScript,
640 "go" => crate::scripting::ScriptLanguage::Go,
641 "perl" => crate::scripting::ScriptLanguage::Perl,
642 _ => {
643 warn!(
644 "Unknown script language '{}', falling back to LLM",
645 language
646 );
647 return Ok(EventHandlerResult::FallbackToLlm { instruction: None });
648 }
649 };
650
651 let scripting_env = state.get_scripting_env().await;
653 if !scripting_env.is_available(script_language) {
654 warn!(
655 "Script language {} not available, falling back to LLM",
656 script_language.as_str()
657 );
658 return Ok(EventHandlerResult::FallbackToLlm { instruction: None });
659 }
660
661 let script_config = crate::scripting::types::ScriptConfig {
663 language: script_language,
664 source: crate::scripting::types::ScriptSource::Inline(code.to_string()),
665 handles_contexts: vec![event_type_id.to_string()],
666 };
667
668 let use_resident =
675 resident && crate::scripting::resident::resident_language_supported(script_language);
676 if resident && !use_resident {
677 warn!(
678 "Resident mode requested for '{}' but that language runs per-event only; \
679 falling back to per-event execution",
680 script_language.as_str()
681 );
682 }
683
684 let script_result = if private_payloads {
685 use tracing::instrument::WithSubscriber;
686 let run = async {
687 if use_resident {
688 crate::scripting::ResidentScriptManager::dispatch_private(
689 &script_config,
690 &script_input,
691 crate::scripting::ResidentScope::parse(scope),
692 )
693 .await
694 } else {
695 crate::scripting::executor::execute_script_async(&script_config, &script_input)
696 .await
697 }
698 };
699 run.with_subscriber(tracing::subscriber::NoSubscriber::default())
700 .await
701 .map_err(|_| anyhow::anyhow!("credential-bearing script failed; diagnostics hidden"))
702 } else if use_resident {
703 let resident_scope = crate::scripting::ResidentScope::parse(scope);
704 crate::scripting::ResidentScriptManager::dispatch(
705 &script_config,
706 &script_input,
707 resident_scope,
708 )
709 .await
710 } else {
711 crate::scripting::executor::execute_script_async(&script_config, &script_input).await
712 };
713
714 match script_result {
716 Ok(response) => {
717 debug!(
718 "Script handled event '{}' ({} actions)",
719 event_type_id,
720 response.actions.len()
721 );
722
723 let truncated_desc = if private_payloads {
725 "SCRIPT credential-bearing event".to_string()
726 } else {
727 format!(
728 "SCRIPT \"{}\"",
729 crate::utils::truncate_for_log(event_description, 27)
730 )
731 };
732 let conv_id = format!(
733 "script-{}-{:x}",
734 crate::utils::clock::SystemTime::now()
735 .duration_since(crate::utils::clock::UNIX_EPOCH)
736 .unwrap()
737 .as_millis(),
738 rand::random::<u32>()
739 );
740 state
741 .register_conversation(
742 conv_id.clone(),
743 crate::state::app_state::ConversationSource::Network {
744 server_id,
745 connection_id,
746 },
747 truncated_desc,
748 )
749 .await;
750
751 let result = execute_actions(response.actions, state, protocol, Some(server_id), None)
753 .await
754 .context("Failed to execute script actions")?;
755
756 state.end_conversation(&conv_id).await;
758
759 Ok(EventHandlerResult::Handled(result))
760 }
761 Err(e) => {
762 warn!("Script execution failed: {}, falling back to LLM", e);
763 Ok(EventHandlerResult::FallbackToLlm { instruction: None })
764 }
765 }
766}
767
768#[allow(clippy::too_many_arguments)]
779async fn execute_static_handler(
780 state: &AppState,
781 server_id: ServerId,
782 event_type_id: &str,
783 event_description: &str,
784 actions: &[serde_json::Value],
785 event_data: Option<&serde_json::Value>,
786 protocol: Option<&dyn Server>,
787) -> Result<EventHandlerResult> {
788 debug!(
789 "Static handler executing for event '{}' ({} actions)",
790 event_type_id,
791 actions.len()
792 );
793
794 let actions = match crate::scripting::event_handler::interpolate_actions(actions, event_data) {
796 Ok(actions) => actions,
797 Err(e) => {
798 error!(
799 "Static handler for event '{}' has an unresolvable reference: {}",
800 event_type_id, e
801 );
802 return Err(e).with_context(|| {
803 format!(
804 "Static handler for event '{}' could not be rendered",
805 event_type_id
806 )
807 });
808 }
809 };
810
811 let result = execute_actions(actions, state, protocol, Some(server_id), None)
813 .await
814 .context("Failed to execute static actions")?;
815
816 debug!("Static handler completed: {}", event_description);
818
819 Ok(EventHandlerResult::Handled(result))
820}