theway_daemon/trigger_engine/execution/
utils.rs

1//! Small shared helpers for the execution pipeline: audit-label / reason caps,
2//! prompt-request construction with payload validation, listener fan-out and
3//! banner preview truncation.
4
5use std::sync::Arc;
6
7use parking_lot::Mutex;
8
9use crate::trigger_engine::event::{TriggerEvent, TriggerListener};
10use crate::trigger_engine::types::Trigger;
11
12use super::promotion::{PROMOTION_BODY_CAP_BYTES, sha256_hex, truncate_on_char_boundary};
13use super::types::TriggerPromptRequest;
14
15const CONTROL_PLANE_PROMPT_LABEL_CAP_CHARS: usize = 200;
16
17pub(super) fn cap_control_plane_audit_label(label: &str) -> String {
18    if label.chars().count() <= CONTROL_PLANE_PROMPT_LABEL_CAP_CHARS {
19        return label.to_string();
20    }
21    let mut out: String = label
22        .chars()
23        .take(CONTROL_PLANE_PROMPT_LABEL_CAP_CHARS.saturating_sub(1))
24        .collect();
25    out.push('…');
26    out
27}
28
29pub(super) fn build_trigger_prompt_request(
30    trigger: &Trigger,
31    reason: String,
32) -> TriggerPromptRequest {
33    let receiver_agent_id = validated_payload_agent_id(trigger, &["_meta", "receiver_agent_id"])
34        .or_else(|| validated_payload_agent_id(trigger, &["receiver_agent_id"]));
35    let sender_agent_id = validated_payload_agent_id(trigger, &["_meta", "sender_agent_id"])
36        .or_else(|| validated_payload_agent_id(trigger, &["sender_agent_id"]))
37        .or_else(|| validated_payload_agent_id(trigger, &["agent_id"]))
38        .unwrap_or_else(|| cap_control_plane_audit_label(&trigger.authority.principal_id));
39    let action_class = validated_payload_action_class(trigger, &["_meta", "action_class"])
40        .or_else(|| validated_payload_action_class(trigger, &["action_class"]))
41        .unwrap_or_else(|| cap_control_plane_audit_label(&trigger.event_label));
42    let trigger_summary = trigger
43        .payload_summary
44        .clone()
45        .map(|summary| truncate_on_char_boundary(summary, PROMOTION_BODY_CAP_BYTES).0);
46    let payload = serde_json::json!({
47        "source_kind": trigger.source_kind,
48        "source_label": cap_control_plane_audit_label(&trigger.source_label),
49        "event_label": cap_control_plane_audit_label(&trigger.event_label),
50        "payload_visibility": trigger.payload_visibility,
51        "payload_summary": trigger_summary,
52        "authority": {
53            "principal_id": trigger.authority.principal_id.clone(),
54            "principal_label": cap_control_plane_audit_label(&trigger.authority.principal_label),
55            "credential_scope": trigger.authority.credential_scope,
56            "allowed_source_actions": trigger.authority.allowed_source_actions.clone(),
57        }
58    });
59    let binding = serde_json::json!([
60        "trigger_prompt:v1",
61        trigger.idempotency_key.clone(),
62        trigger.trace_id.clone(),
63        trigger.source_kind,
64        trigger.source_label.clone(),
65        trigger.event_label.clone(),
66        receiver_agent_id.clone(),
67        sender_agent_id.clone(),
68        action_class.clone(),
69    ]);
70    let trigger_prompt_id = sha256_hex(&binding.to_string());
71    TriggerPromptRequest {
72        trigger_prompt_id,
73        trace_id: trigger.trace_id.clone(),
74        source_label: cap_control_plane_audit_label(&trigger.source_label),
75        receiver_agent_id,
76        sender_agent_id,
77        action_class,
78        trigger_summary,
79        payload,
80        reason: cap_trigger_prompt_reason(&reason),
81    }
82}
83
84fn validated_payload_agent_id(trigger: &Trigger, path: &[&str]) -> Option<String> {
85    let value = trigger_json_string(trigger, path)?;
86    uuid::Uuid::parse_str(&value).ok()?;
87    Some(value)
88}
89
90fn validated_payload_action_class(trigger: &Trigger, path: &[&str]) -> Option<String> {
91    let value = trigger_json_string(trigger, path)?;
92    is_valid_action_class(&value).then_some(value)
93}
94
95fn trigger_json_string(trigger: &Trigger, path: &[&str]) -> Option<String> {
96    let mut value = trigger.payload.as_ref()?;
97    for key in path {
98        value = value.get(*key)?;
99    }
100    value.as_str().map(str::to_string)
101}
102
103fn is_valid_action_class(value: &str) -> bool {
104    let mut chars = value.chars();
105    let Some(first) = chars.next() else {
106        return false;
107    };
108    let lower = value.to_ascii_lowercase();
109    if lower.starts_with("sk-") || lower.contains("bearer") || lower.contains("token") {
110        return false;
111    }
112    value.len() <= 64
113        && first.is_ascii_lowercase()
114        && chars.all(|ch| {
115            ch.is_ascii_lowercase() || ch.is_ascii_digit() || matches!(ch, '_' | '-' | '.' | ':')
116        })
117}
118
119const TRIGGER_PROMPT_REASON_CAP_CHARS: usize = 512;
120
121pub(super) fn cap_trigger_prompt_reason(reason: &str) -> String {
122    if reason.chars().count() <= TRIGGER_PROMPT_REASON_CAP_CHARS {
123        return reason.to_string();
124    }
125    let mut out: String = reason
126        .chars()
127        .take(TRIGGER_PROMPT_REASON_CAP_CHARS.saturating_sub(1))
128        .collect();
129    out.push('…');
130    out
131}
132
133/// Emit a [`TriggerEvent`] to a snapshot of the listener registry, isolating each listener
134/// with `catch_unwind` so a single panicking listener cannot poison the others. Mirrors
135/// the contract of `TriggerExecutor::emit` but operates on a cloned `Arc` of listeners (so
136/// the spawned sub-agent task does not need a `TriggerExecutor` reference).
137pub(super) fn emit_from_listeners(
138    listeners: &Arc<Mutex<Vec<TriggerListener>>>,
139    event: TriggerEvent,
140) {
141    let snapshot: Vec<TriggerListener> = listeners.lock().clone();
142    for listener in snapshot {
143        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| listener(event.clone())));
144    }
145}
146
147/// Bounded preview text for status banners. Avoids panicking on multi-byte char boundaries
148/// by walking char count, not byte count.
149pub(super) fn preview_for_banner(text: &str, max_chars: usize) -> String {
150    if text.chars().count() <= max_chars {
151        return text.to_string();
152    }
153    let mut out: String = text.chars().take(max_chars).collect();
154    out.push('…');
155    out
156}
157
158#[cfg(test)]
159mod coverage_gap {
160    use super::*;
161    use crate::trigger_engine::types::{
162        CredentialScope, PayloadVisibility, ReplacementPolicy, SourceKind, TriggerAuthority,
163        TriggerSource,
164    };
165
166    fn sample_trigger(payload: Option<serde_json::Value>) -> Trigger {
167        Trigger {
168            source: TriggerSource::Mcp {
169                server_name: "github".into(),
170                method: "notifications/pr.merged".into(),
171            },
172            source_kind: SourceKind::Mcp,
173            source_label: "mcp:github".into(),
174            event_label: "pr_merged".into(),
175            payload_visibility: PayloadVisibility::Local,
176            payload_summary: None,
177            payload,
178            idempotency_key: "k".into(),
179            replacement_policy: ReplacementPolicy::Drop,
180            trace_id: "trace-utils".into(),
181            authority: TriggerAuthority {
182                principal_id: "mcp:github".into(),
183                principal_label: "github".into(),
184                credential_scope: CredentialScope::User,
185                allowed_source_actions: vec![],
186                expires_at: None,
187            },
188            received_at: chrono::Utc::now(),
189        }
190    }
191
192    #[test]
193    fn validated_payload_agent_id_rejects_non_string_and_bad_paths() {
194        let trigger = sample_trigger(Some(serde_json::json!({
195            "_meta": {"receiver_agent_id": 42},
196            "receiver_agent_id": "not-a-uuid",
197        })));
198        assert_eq!(
199            validated_payload_agent_id(&trigger, &["_meta", "receiver_agent_id"]),
200            None
201        );
202        assert_eq!(validated_payload_agent_id(&trigger, &["missing"]), None);
203    }
204
205    #[test]
206    fn validated_payload_action_class_rejects_invalid_shapes() {
207        let trigger = sample_trigger(Some(serde_json::json!({
208            "_meta": {"action_class": "Sk-secret"},
209            "action_class": "valid.class",
210        })));
211        assert_eq!(
212            validated_payload_action_class(&trigger, &["_meta", "action_class"]),
213            None
214        );
215        assert_eq!(
216            validated_payload_action_class(&trigger, &["action_class"]).as_deref(),
217            Some("valid.class")
218        );
219
220        let empty = sample_trigger(Some(serde_json::json!({"action_class": ""})));
221        assert_eq!(
222            validated_payload_action_class(&empty, &["action_class"]),
223            None
224        );
225    }
226
227    #[test]
228    fn is_valid_action_class_rejects_uppercase_length_and_bad_chars() {
229        assert!(!is_valid_action_class("Uppercase"));
230        assert!(!is_valid_action_class("a".repeat(65).as_str()));
231        assert!(!is_valid_action_class("bad class"));
232        assert!(!is_valid_action_class("contains-token-secret"));
233        assert!(!is_valid_action_class("sk-prefixed"));
234        assert!(is_valid_action_class("valid_action.class:with-all"));
235    }
236
237    #[test]
238    fn trigger_json_string_handles_missing_paths_and_scalars() {
239        let trigger = sample_trigger(Some(serde_json::json!({
240            "nested": {"value": 1}
241        })));
242        assert_eq!(trigger_json_string(&trigger, &["nested"]), None);
243        assert_eq!(trigger_json_string(&trigger, &["nested", "value"]), None);
244        assert_eq!(trigger_json_string(&trigger, &["nested", "missing"]), None);
245    }
246
247    #[test]
248    fn cap_control_plane_audit_label_truncates_long_labels() {
249        let label = "x".repeat(201);
250        let capped = cap_control_plane_audit_label(&label);
251        assert_eq!(capped.chars().count(), 200);
252        assert!(capped.ends_with('…'));
253        assert_eq!(cap_control_plane_audit_label("short"), "short");
254    }
255
256    #[test]
257    fn cap_trigger_prompt_reason_preserves_short_and_truncates_long() {
258        let short = "a short reason";
259        assert_eq!(cap_trigger_prompt_reason(short), short);
260        let long = "r".repeat(600);
261        let capped = cap_trigger_prompt_reason(&long);
262        assert_eq!(capped.chars().count(), 512);
263        assert!(capped.ends_with('…'));
264    }
265
266    #[test]
267    fn preview_for_banner_preserves_short_and_truncates_long() {
268        assert_eq!(preview_for_banner("hello", 10), "hello");
269        let capped = preview_for_banner(&"x".repeat(200), 80);
270        assert_eq!(capped.chars().count(), 81);
271        assert!(capped.ends_with('…'));
272    }
273
274    #[test]
275    fn is_valid_action_class_rejects_bearer_and_control_chars() {
276        assert!(!is_valid_action_class("Bearer invalid"));
277        assert!(!is_valid_action_class("bad!char"));
278        assert!(is_valid_action_class("ok.action_class:with-dashes"));
279    }
280
281    #[test]
282    fn build_trigger_prompt_request_resolves_nested_and_top_level_fields() {
283        let trigger = sample_trigger(Some(serde_json::json!({
284            "_meta": {
285                "receiver_agent_id": "11111111-1111-4111-8111-111111111111",
286                "sender_agent_id": "22222222-2222-4222-8222-222222222222",
287                "action_class": "nested.class",
288            },
289            "receiver_agent_id": "33333333-3333-4333-8333-333333333333",
290            "sender_agent_id": "44444444-4444-4444-8444-444444444444",
291            "agent_id": "55555555-5555-4555-8555-555555555555",
292            "action_class": "top.class",
293            "payload_summary": "summary",
294        })));
295        let request = build_trigger_prompt_request(&trigger, "why".into());
296        assert_eq!(
297            request.receiver_agent_id.as_deref(),
298            Some("11111111-1111-4111-8111-111111111111")
299        );
300        assert_eq!(
301            request.sender_agent_id.as_str(),
302            "22222222-2222-4222-8222-222222222222"
303        );
304        assert_eq!(request.action_class, "nested.class");
305    }
306
307    #[test]
308    fn emit_from_listeners_isolates_panicking_listener() {
309        use crate::trigger_engine::event::TriggerEvent;
310        let listeners: Arc<Mutex<Vec<TriggerListener>>> = Arc::new(Mutex::new(Vec::new()));
311        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
312        let calls_sink = calls.clone();
313        listeners.lock().push(Arc::new(move |_event: TriggerEvent| {
314            panic!("listener panic");
315        }));
316        listeners.lock().push(Arc::new(move |event: TriggerEvent| {
317            let _ = event;
318            calls_sink.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
319        }));
320        emit_from_listeners(
321            &listeners,
322            TriggerEvent::TriggerHandlingStart {
323                idempotency_key: "k".into(),
324                source_kind: SourceKind::Mcp,
325                source_label: "src".into(),
326                event_label: "evt".into(),
327                trace_id: "trace".into(),
328            },
329        );
330        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
331    }
332}
theway_daemon/trigger_engine/execution/utils.rs

theway_daemon/trigger_engine/execution/
utils.rs
