nexo_core/agent/
remote_trigger_tool.rs

1//! `RemoteTrigger` tool: webhook + NATS publisher gated by a
2//! per-session allowlist (agent-level or per-binding override).
3//!
4//! A generic outbound publisher — webhook with HMAC sign + NATS
5//! publish, both gated by a YAML allowlist so URLs never travel
6//! through the model.
7//!
8//! Security model:
9//!   * The model never sees the URL or NATS subject — it refers to
10//!     a destination by `name`.
11//!   * Webhook bodies are HMAC-SHA256 signed when `secret_env` is
12//!     set; the runtime resolves the env var at call time.
13//!   * Per-trigger token-bucket rate limit (default 10 calls/min,
14//!     `0` = unlimited).
15//!   * Hard cap 256 KiB per payload.
16//!   * Plan-mode classified as `Outbound` (mutating) — refuses
17//!     while plan mode is on.
18
19use super::context::AgentContext;
20use super::tool_registry::ToolHandler;
21use async_trait::async_trait;
22use dashmap::DashMap;
23use hmac::{Hmac, Mac};
24use nexo_config::types::remote_triggers::{RemoteTriggerEntry, REMOTE_TRIGGER_MAX_BODY_BYTES};
25use nexo_llm::ToolDef;
26use serde_json::{json, Value};
27use sha2::Sha256;
28use std::sync::{Arc, Mutex};
29use std::time::{Duration, Instant};
30
31type HmacSha256 = Hmac<Sha256>;
32
33/// Process-shared per-trigger rate limiter. Sliding window by
34/// design: a `VecDeque` of timestamps trimmed to the last minute,
35/// then `len() < limit` to admit. Memory bounded by the configured
36/// limit; never grows unboundedly.
37#[derive(Default)]
38pub struct RemoteTriggerRateLimiter {
39    buckets: DashMap<String, Mutex<std::collections::VecDeque<Instant>>>,
40}
41
42impl RemoteTriggerRateLimiter {
43    /// `true` when the call is admitted; `false` when the trigger's
44    /// per-minute budget is exhausted. `limit_per_minute == 0`
45    /// always admits (no limit).
46    pub fn try_acquire(&self, trigger_name: &str, limit_per_minute: u32) -> bool {
47        if limit_per_minute == 0 {
48            return true;
49        }
50        let entry = self.buckets.entry(trigger_name.to_string()).or_default();
51        let mut q = entry.lock().unwrap();
52        let now = Instant::now();
53        let cutoff = now - Duration::from_secs(60);
54        while let Some(front) = q.front() {
55            if *front < cutoff {
56                q.pop_front();
57            } else {
58                break;
59            }
60        }
61        if q.len() < limit_per_minute as usize {
62            q.push_back(now);
63            true
64        } else {
65            false
66        }
67    }
68}
69
70/// Trait abstraction over the actual outbound webhook / NATS
71/// publish so tests can substitute a fake without standing up a
72/// real HTTP server / broker. Production wiring uses
73/// [`ReqwestSink`] for webhook + `AnyBroker` for NATS.
74#[async_trait]
75pub trait RemoteTriggerSink: Send + Sync {
76    /// Issue a POST with the supplied body + headers. Returns the
77    /// response body + status. Errors when the network fails or the
78    /// response status is ≥ 400.
79    async fn post_webhook(
80        &self,
81        url: &str,
82        body: &str,
83        headers: Vec<(String, String)>,
84        timeout: Duration,
85    ) -> anyhow::Result<u16>;
86
87    /// Publish a payload to a NATS subject. Errors propagate from
88    /// the broker.
89    async fn publish_nats(&self, subject: &str, payload: &[u8]) -> anyhow::Result<()>;
90}
91
92/// Production sink: reqwest + AnyBroker. Not used in tests.
93pub struct ReqwestSink {
94    pub broker: nexo_broker::AnyBroker,
95    pub client: reqwest::Client,
96}
97
98impl ReqwestSink {
99    pub fn new(broker: nexo_broker::AnyBroker) -> Self {
100        Self {
101            broker,
102            client: reqwest::Client::new(),
103        }
104    }
105}
106
107#[async_trait]
108impl RemoteTriggerSink for ReqwestSink {
109    async fn post_webhook(
110        &self,
111        url: &str,
112        body: &str,
113        headers: Vec<(String, String)>,
114        timeout: Duration,
115    ) -> anyhow::Result<u16> {
116        let mut req = self
117            .client
118            .post(url)
119            .timeout(timeout)
120            .header("Content-Type", "application/json")
121            .body(body.to_string());
122        for (k, v) in headers {
123            req = req.header(k, v);
124        }
125        let res = req
126            .send()
127            .await
128            .map_err(|e| anyhow::anyhow!("webhook POST failed: {e}"))?;
129        let status = res.status().as_u16();
130        if status >= 400 {
131            let text = res.text().await.unwrap_or_default();
132            anyhow::bail!(
133                "webhook returned HTTP {status}: {body}",
134                body = text.chars().take(200).collect::<String>()
135            );
136        }
137        Ok(status)
138    }
139
140    async fn publish_nats(&self, subject: &str, payload: &[u8]) -> anyhow::Result<()> {
141        use nexo_broker::BrokerHandle;
142        let json: serde_json::Value = serde_json::from_slice(payload)
143            .map_err(|e| anyhow::anyhow!("NATS payload not JSON: {e}"))?;
144        let event = nexo_broker::Event::new(subject, "remote_trigger", json);
145        self.broker
146            .publish(subject, event)
147            .await
148            .map_err(|e| anyhow::anyhow!("NATS publish failed: {e}"))
149    }
150}
151
152pub struct RemoteTriggerTool {
153    sink: Arc<dyn RemoteTriggerSink>,
154    rate_limiter: Arc<RemoteTriggerRateLimiter>,
155}
156
157impl RemoteTriggerTool {
158    pub fn new(sink: Arc<dyn RemoteTriggerSink>) -> Self {
159        Self {
160            sink,
161            rate_limiter: Arc::new(RemoteTriggerRateLimiter::default()),
162        }
163    }
164
165    pub fn tool_def() -> ToolDef {
166        ToolDef {
167            name: "RemoteTrigger".to_string(),
168            description: "Publish a JSON payload to a pre-configured outbound destination — webhook (HTTP POST, optionally HMAC-SHA256 signed) or NATS subject. Destinations are named in YAML (`agents[].remote_triggers` or binding override); the model passes only the name and payload, never URLs or subjects. Per-destination rate limit + 256 KiB body cap apply.".to_string(),
169            parameters: json!({
170                "type": "object",
171                "properties": {
172                    "name": {
173                        "type": "string",
174                        "description": "Name of the destination as configured in `remote_triggers[].name` for this session's effective policy."
175                    },
176                    "payload": {
177                        "description": "JSON payload to send. Object / array / scalar — any JSON. Capped at 256 KiB serialised."
178                    }
179                },
180                "required": ["name", "payload"]
181            }),
182        }
183    }
184}
185
186fn sign_body(secret: &[u8], body: &str) -> String {
187    let mut mac = HmacSha256::new_from_slice(secret).expect("HMAC accepts any key length");
188    mac.update(body.as_bytes());
189    let bytes = mac.finalize().into_bytes();
190    let mut hex = String::with_capacity(bytes.len() * 2);
191    for b in bytes {
192        use std::fmt::Write as _;
193        let _ = write!(hex, "{b:02x}");
194    }
195    format!("sha256={hex}")
196}
197
198#[async_trait]
199impl ToolHandler for RemoteTriggerTool {
200    async fn call(&self, ctx: &AgentContext, args: Value) -> anyhow::Result<Value> {
201        let name = args
202            .get("name")
203            .and_then(|v| v.as_str())
204            .ok_or_else(|| anyhow::anyhow!("RemoteTrigger requires `name` (string)"))?
205            .to_string();
206        let payload = args
207            .get("payload")
208            .ok_or_else(|| anyhow::anyhow!("RemoteTrigger requires `payload`"))?;
209
210        // Resolve via the effective per-session allowlist:
211        // `InboundBinding::remote_triggers` override when present,
212        // otherwise `AgentConfig::remote_triggers`.
213        let effective = ctx.effective_policy();
214        let entry = effective
215            .remote_triggers
216            .iter()
217            .find(|e| e.name() == name)
218            .cloned()
219            .ok_or_else(|| {
220                anyhow::anyhow!(
221                    "RemoteTrigger: no destination named `{name}` in this session allowlist. Operator must add it under `agents[].remote_triggers[]` or the matched `inbound_bindings[].remote_triggers[]` override."
222                )
223            })?;
224
225        // Serialise + cap before doing anything else so we never
226        // even hit the rate-limiter on a bad payload.
227        let body = serde_json::to_string(payload)
228            .map_err(|e| anyhow::anyhow!("RemoteTrigger: payload not serialisable: {e}"))?;
229        if body.len() > REMOTE_TRIGGER_MAX_BODY_BYTES {
230            return Err(anyhow::anyhow!(
231                "RemoteTrigger: payload too large ({actual} bytes > max {max})",
232                actual = body.len(),
233                max = REMOTE_TRIGGER_MAX_BODY_BYTES
234            ));
235        }
236
237        let rate_limit_key = match effective.binding_index {
238            Some(idx) => format!("{idx}:{}", entry.name()),
239            None => format!("agent:{}", entry.name()),
240        };
241        if !self
242            .rate_limiter
243            .try_acquire(&rate_limit_key, entry.rate_limit_per_minute())
244        {
245            return Err(anyhow::anyhow!(
246                "RemoteTrigger: rate limit exceeded for `{name}` ({} calls/min). Wait or raise `rate_limit_per_minute` in YAML.",
247                entry.rate_limit_per_minute()
248            ));
249        }
250
251        match &entry {
252            RemoteTriggerEntry::Webhook {
253                url,
254                secret_env,
255                timeout_ms,
256                ..
257            } => {
258                let now_unix = chrono::Utc::now().timestamp();
259                let mut headers: Vec<(String, String)> = Vec::with_capacity(3);
260                headers.push(("X-Nexo-Trigger-Name".to_string(), entry.name().to_string()));
261                headers.push(("X-Nexo-Timestamp".to_string(), now_unix.to_string()));
262                if let Some(env) = secret_env {
263                    let secret = std::env::var(env).map_err(|_| {
264                        anyhow::anyhow!(
265                            "RemoteTrigger: secret_env `{env}` is not set; refusing to send unsigned"
266                        )
267                    })?;
268                    headers.push((
269                        "X-Nexo-Signature".to_string(),
270                        sign_body(secret.as_bytes(), &body),
271                    ));
272                }
273                let timeout = Duration::from_millis(*timeout_ms);
274                let started = Instant::now();
275                let status = self.sink.post_webhook(url, &body, headers, timeout).await?;
276                Ok(json!({
277                    "ok": true,
278                    "kind": "webhook",
279                    "name": name,
280                    "status": status,
281                    "signed": secret_env.is_some(),
282                    "duration_ms": started.elapsed().as_millis() as u64,
283                    "bytes_sent": body.len(),
284                }))
285            }
286            RemoteTriggerEntry::Nats { subject, .. } => {
287                let started = Instant::now();
288                self.sink.publish_nats(subject, body.as_bytes()).await?;
289                Ok(json!({
290                    "ok": true,
291                    "kind": "nats",
292                    "name": name,
293                    "duration_ms": started.elapsed().as_millis() as u64,
294                    "bytes_sent": body.len(),
295                }))
296            }
297        }
298    }
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304    use crate::session::SessionManager;
305    use nexo_broker::AnyBroker;
306    use nexo_config::types::agents::{
307        AgentConfig, AgentRuntimeConfig, DreamingYamlConfig, HeartbeatConfig, ModelConfig,
308        OutboundAllowlistConfig, WorkspaceGitConfig,
309    };
310    use std::sync::Arc;
311    use std::sync::Mutex;
312
313    #[derive(Default)]
314    #[allow(dead_code)]
315    struct CapturedCall {
316        url: String,
317        body: String,
318        headers: Vec<(String, String)>,
319    }
320
321    #[derive(Default)]
322    struct FakeSink {
323        webhook_calls: Mutex<Vec<CapturedCall>>,
324        nats_calls: Mutex<Vec<(String, Vec<u8>)>>,
325        force_status: Mutex<u16>,
326    }
327
328    impl FakeSink {
329        fn new() -> Arc<Self> {
330            Arc::new(Self {
331                force_status: Mutex::new(200),
332                ..Default::default()
333            })
334        }
335    }
336
337    #[async_trait]
338    impl RemoteTriggerSink for FakeSink {
339        async fn post_webhook(
340            &self,
341            url: &str,
342            body: &str,
343            headers: Vec<(String, String)>,
344            _timeout: Duration,
345        ) -> anyhow::Result<u16> {
346            self.webhook_calls.lock().unwrap().push(CapturedCall {
347                url: url.to_string(),
348                body: body.to_string(),
349                headers,
350            });
351            let s = *self.force_status.lock().unwrap();
352            if s >= 400 {
353                anyhow::bail!("simulated HTTP {s}");
354            }
355            Ok(s)
356        }
357        async fn publish_nats(&self, subject: &str, payload: &[u8]) -> anyhow::Result<()> {
358            self.nats_calls
359                .lock()
360                .unwrap()
361                .push((subject.to_string(), payload.to_vec()));
362            Ok(())
363        }
364    }
365
366    fn agent_config_with_triggers(triggers: Vec<RemoteTriggerEntry>) -> AgentConfig {
367        AgentConfig {
368            id: "a".into(),
369            model: ModelConfig {
370                provider: "x".into(),
371                model: "y".into(),
372            },
373            plugins: Vec::new(),
374            heartbeat: HeartbeatConfig::default(),
375            config: AgentRuntimeConfig::default(),
376            system_prompt: String::new(),
377            workspace: String::new(),
378            skills: Vec::new(),
379            skills_dir: "./skills".into(),
380            skill_overrides: Default::default(),
381            transcripts_dir: String::new(),
382            dreaming: DreamingYamlConfig::default(),
383            workspace_git: WorkspaceGitConfig::default(),
384            tool_rate_limits: None,
385            tool_args_validation: None,
386            extra_docs: Vec::new(),
387            inbound_bindings: Vec::new(),
388            allowed_tools: Vec::new(),
389            sender_rate_limit: None,
390            allowed_delegates: Vec::new(),
391            accept_delegates_from: Vec::new(),
392            description: String::new(),
393            google_auth: None,
394            credentials: Default::default(),
395            link_understanding: serde_json::Value::Null,
396            web_search: serde_json::Value::Null,
397            pairing_policy: serde_json::Value::Null,
398            language: None,
399            locale_prompts: Default::default(),
400            outbound_allowlist: OutboundAllowlistConfig::default(),
401            context_optimization: None,
402            dispatch_policy: Default::default(),
403            plan_mode: Default::default(),
404            remote_triggers: triggers,
405            lsp: nexo_config::types::lsp::LspPolicy::default(),
406            config_tool: nexo_config::types::config_tool::ConfigToolPolicy::default(),
407            team: nexo_config::types::team::TeamPolicy::default(),
408            proactive: Default::default(),
409            repl: Default::default(),
410            auto_dream: None,
411            assistant_mode: None,
412            away_summary: None,
413            brief: None,
414            channels: None,
415            auto_approve: false,
416            extract_memories: None,
417            event_subscribers: Vec::new(),
418            tenant_id: None,
419            extensions_config: std::collections::BTreeMap::new(),
420            active: true,
421        }
422    }
423
424    fn ctx_with_triggers(triggers: Vec<RemoteTriggerEntry>) -> AgentContext {
425        let cfg = agent_config_with_triggers(triggers);
426        AgentContext::new(
427            "a",
428            Arc::new(cfg),
429            AnyBroker::local(),
430            Arc::new(SessionManager::new(std::time::Duration::from_secs(60), 8)),
431        )
432    }
433
434    fn ctx_with_binding_override(
435        agent_level: Vec<RemoteTriggerEntry>,
436        binding_level: Vec<RemoteTriggerEntry>,
437        binding_index: usize,
438    ) -> AgentContext {
439        let cfg = Arc::new(agent_config_with_triggers(agent_level));
440        let mut eff = crate::agent::EffectiveBindingPolicy::from_agent_defaults(&cfg);
441        eff.binding_index = Some(binding_index);
442        eff.remote_triggers = binding_level;
443        AgentContext::new(
444            "a",
445            Arc::clone(&cfg),
446            AnyBroker::local(),
447            Arc::new(SessionManager::new(std::time::Duration::from_secs(60), 8)),
448        )
449        .with_effective(Arc::new(eff))
450    }
451
452    fn webhook(name: &str, secret: Option<&str>, rate: u32) -> RemoteTriggerEntry {
453        RemoteTriggerEntry::Webhook {
454            name: name.into(),
455            url: format!("https://example.test/{name}"),
456            secret_env: secret.map(str::to_string),
457            timeout_ms: 5000,
458            rate_limit_per_minute: rate,
459        }
460    }
461
462    fn nats(name: &str, subject: &str, rate: u32) -> RemoteTriggerEntry {
463        RemoteTriggerEntry::Nats {
464            name: name.into(),
465            subject: subject.into(),
466            rate_limit_per_minute: rate,
467        }
468    }
469
470    fn tool(sink: Arc<dyn RemoteTriggerSink>) -> RemoteTriggerTool {
471        RemoteTriggerTool::new(sink)
472    }
473
474    #[tokio::test]
475    async fn refuses_name_not_in_allowlist() {
476        let ctx = ctx_with_triggers(vec![webhook("ops", None, 10)]);
477        let sink = FakeSink::new();
478        let err = tool(sink.clone())
479            .call(&ctx, json!({"name": "imaginary", "payload": {"a": 1}}))
480            .await
481            .unwrap_err()
482            .to_string();
483        assert!(err.contains("not in allowlist") || err.contains("no destination"));
484        assert!(sink.webhook_calls.lock().unwrap().is_empty());
485        assert!(sink.nats_calls.lock().unwrap().is_empty());
486    }
487
488    #[tokio::test]
489    async fn binding_override_allowlist_wins_over_agent_level() {
490        let ctx = ctx_with_binding_override(
491            vec![webhook("agent_only", None, 10)],
492            vec![webhook("binding_only", None, 10)],
493            4,
494        );
495        let sink = FakeSink::new();
496        let t = tool(sink.clone());
497
498        t.call(&ctx, json!({"name": "binding_only", "payload": {"a": 1}}))
499            .await
500            .unwrap();
501        let err = t
502            .call(&ctx, json!({"name": "agent_only", "payload": {"a": 1}}))
503            .await
504            .unwrap_err()
505            .to_string();
506
507        assert!(
508            err.contains("no destination"),
509            "agent-level destination must be hidden by binding override, got: {err}"
510        );
511        let calls = sink.webhook_calls.lock().unwrap();
512        assert_eq!(calls.len(), 1);
513        assert_eq!(calls[0].url, "https://example.test/binding_only");
514    }
515
516    #[tokio::test]
517    async fn rate_limit_isolated_per_binding_for_same_trigger_name() {
518        let ctx_a = ctx_with_binding_override(vec![], vec![webhook("ops", None, 1)], 0);
519        let ctx_b = ctx_with_binding_override(vec![], vec![webhook("ops", None, 1)], 1);
520        let sink = FakeSink::new();
521        let t = tool(sink.clone());
522
523        // Same trigger name, different binding index: each gets its own bucket.
524        t.call(&ctx_a, json!({"name": "ops", "payload": {}}))
525            .await
526            .unwrap();
527        t.call(&ctx_b, json!({"name": "ops", "payload": {}}))
528            .await
529            .unwrap();
530
531        let err_a = t
532            .call(&ctx_a, json!({"name": "ops", "payload": {}}))
533            .await
534            .unwrap_err()
535            .to_string();
536        let err_b = t
537            .call(&ctx_b, json!({"name": "ops", "payload": {}}))
538            .await
539            .unwrap_err()
540            .to_string();
541        assert!(err_a.contains("rate limit"), "got: {err_a}");
542        assert!(err_b.contains("rate limit"), "got: {err_b}");
543        assert_eq!(sink.webhook_calls.lock().unwrap().len(), 2);
544    }
545
546    #[tokio::test]
547    async fn webhook_unsigned_emits_no_signature_header() {
548        let ctx = ctx_with_triggers(vec![webhook("ops", None, 10)]);
549        let sink = FakeSink::new();
550        let res = tool(sink.clone())
551            .call(&ctx, json!({"name": "ops", "payload": {"a": 1}}))
552            .await
553            .unwrap();
554        assert_eq!(res["ok"], true);
555        assert_eq!(res["kind"], "webhook");
556        assert_eq!(res["signed"], false);
557        let calls = sink.webhook_calls.lock().unwrap();
558        assert_eq!(calls.len(), 1);
559        assert!(!calls[0]
560            .headers
561            .iter()
562            .any(|(k, _)| k == "X-Nexo-Signature"));
563        assert!(calls[0]
564            .headers
565            .iter()
566            .any(|(k, v)| k == "X-Nexo-Trigger-Name" && v == "ops"));
567        assert!(calls[0]
568            .headers
569            .iter()
570            .any(|(k, _)| k == "X-Nexo-Timestamp"));
571    }
572
573    #[tokio::test]
574    async fn webhook_signed_emits_valid_hmac() {
575        let env_var = "NEXO_TEST_RT_SECRET_VALID";
576        std::env::set_var(env_var, "topsecret");
577        let ctx = ctx_with_triggers(vec![webhook("ops", Some(env_var), 10)]);
578        let sink = FakeSink::new();
579        let _res = tool(sink.clone())
580            .call(&ctx, json!({"name": "ops", "payload": {"a": 1}}))
581            .await
582            .unwrap();
583        let calls = sink.webhook_calls.lock().unwrap();
584        let sig = calls[0]
585            .headers
586            .iter()
587            .find(|(k, _)| k == "X-Nexo-Signature")
588            .map(|(_, v)| v.clone())
589            .expect("signature header missing");
590        assert!(sig.starts_with("sha256="));
591        // Verify by recomputing.
592        let expected = sign_body(b"topsecret", &calls[0].body);
593        assert_eq!(sig, expected);
594        std::env::remove_var(env_var);
595    }
596
597    #[tokio::test]
598    async fn webhook_missing_secret_env_refuses() {
599        let env_var = "NEXO_TEST_RT_SECRET_MISSING_PLEASE_DO_NOT_SET";
600        std::env::remove_var(env_var);
601        let ctx = ctx_with_triggers(vec![webhook("ops", Some(env_var), 10)]);
602        let sink = FakeSink::new();
603        let err = tool(sink.clone())
604            .call(&ctx, json!({"name": "ops", "payload": {"a": 1}}))
605            .await
606            .unwrap_err()
607            .to_string();
608        assert!(err.contains("not set"), "got: {err}");
609        assert!(
610            sink.webhook_calls.lock().unwrap().is_empty(),
611            "must not send unsigned"
612        );
613    }
614
615    #[tokio::test]
616    async fn nats_publishes_to_subject() {
617        let ctx = ctx_with_triggers(vec![nats("ops", "agent.outbound.ops", 10)]);
618        let sink = FakeSink::new();
619        let res = tool(sink.clone())
620            .call(&ctx, json!({"name": "ops", "payload": {"x": 42}}))
621            .await
622            .unwrap();
623        assert_eq!(res["kind"], "nats");
624        let calls = sink.nats_calls.lock().unwrap();
625        assert_eq!(calls.len(), 1);
626        assert_eq!(calls[0].0, "agent.outbound.ops");
627        let body: Value = serde_json::from_slice(&calls[0].1).unwrap();
628        assert_eq!(body["x"], 42);
629    }
630
631    #[tokio::test]
632    async fn rate_limit_blocks_after_budget() {
633        let ctx = ctx_with_triggers(vec![webhook("ops", None, 2)]);
634        let sink = FakeSink::new();
635        let t = tool(sink.clone());
636        // 2 admitted, 3rd refused.
637        t.call(&ctx, json!({"name": "ops", "payload": {}}))
638            .await
639            .unwrap();
640        t.call(&ctx, json!({"name": "ops", "payload": {}}))
641            .await
642            .unwrap();
643        let err = t
644            .call(&ctx, json!({"name": "ops", "payload": {}}))
645            .await
646            .unwrap_err()
647            .to_string();
648        assert!(err.contains("rate limit"), "got: {err}");
649        assert_eq!(sink.webhook_calls.lock().unwrap().len(), 2);
650    }
651
652    #[tokio::test]
653    async fn rate_limit_zero_means_unlimited() {
654        let ctx = ctx_with_triggers(vec![webhook("ops", None, 0)]);
655        let sink = FakeSink::new();
656        let t = tool(sink.clone());
657        for _ in 0..50 {
658            t.call(&ctx, json!({"name": "ops", "payload": {}}))
659                .await
660                .unwrap();
661        }
662        assert_eq!(sink.webhook_calls.lock().unwrap().len(), 50);
663    }
664
665    #[tokio::test]
666    async fn payload_too_large_is_rejected_before_send() {
667        let ctx = ctx_with_triggers(vec![webhook("ops", None, 10)]);
668        let sink = FakeSink::new();
669        // 257 KiB string — exceeds 256 KiB cap.
670        let big = "x".repeat(REMOTE_TRIGGER_MAX_BODY_BYTES + 1);
671        let err = tool(sink.clone())
672            .call(&ctx, json!({"name": "ops", "payload": big}))
673            .await
674            .unwrap_err()
675            .to_string();
676        assert!(err.contains("too large"), "got: {err}");
677        assert!(sink.webhook_calls.lock().unwrap().is_empty());
678    }
679
680    #[tokio::test]
681    async fn webhook_4xx_propagates_as_error() {
682        let ctx = ctx_with_triggers(vec![webhook("ops", None, 10)]);
683        let sink = FakeSink::new();
684        *sink.force_status.lock().unwrap() = 503;
685        let err = tool(sink.clone())
686            .call(&ctx, json!({"name": "ops", "payload": {}}))
687            .await
688            .unwrap_err()
689            .to_string();
690        assert!(err.contains("503"), "got: {err}");
691    }
692
693    #[tokio::test]
694    async fn missing_name_arg_errors() {
695        let ctx = ctx_with_triggers(vec![]);
696        let sink = FakeSink::new();
697        let err = tool(sink)
698            .call(&ctx, json!({"payload": {}}))
699            .await
700            .unwrap_err()
701            .to_string();
702        assert!(err.contains("requires `name`"), "got: {err}");
703    }
704
705    #[tokio::test]
706    async fn missing_payload_arg_errors() {
707        let ctx = ctx_with_triggers(vec![]);
708        let sink = FakeSink::new();
709        let err = tool(sink)
710            .call(&ctx, json!({"name": "ops"}))
711            .await
712            .unwrap_err()
713            .to_string();
714        assert!(err.contains("requires `payload`"), "got: {err}");
715    }
716
717    #[tokio::test]
718    async fn sign_body_is_deterministic_and_hex() {
719        let s1 = sign_body(b"k", "{}");
720        let s2 = sign_body(b"k", "{}");
721        assert_eq!(s1, s2);
722        assert!(s1.starts_with("sha256="));
723        let hex = s1.trim_start_matches("sha256=");
724        assert_eq!(hex.len(), 64);
725        assert!(hex.chars().all(|c| c.is_ascii_hexdigit()));
726    }
727}
nexo_core/agent/remote_trigger_tool.rs

nexo_core/agent/
remote_trigger_tool.rs
