Skip to main content

axon/
http_tool.rs

1//! HTTP tool provider — executes tool calls as REST requests via reqwest.
2//!
3//! Tools declared with `provider: http` in .axon files dispatch their
4//! argument as the request body to the URL specified in `runtime`.
5//!
6//! Request format:
7//!   POST {runtime_url}
8//!   Content-Type: application/json
9//!   X-Axon-Tool: {tool_name}
10//!
11//!   Body: the tool argument (string, sent as JSON-wrapped if not already JSON)
12//!
13//! Response handling:
14//!   - 2xx: response body becomes tool output (success)
15//!   - 4xx/5xx: error message with status code (failure)
16//!   - Connection error: descriptive error (failure)
17//!
18//! Timeout: parsed from ToolEntry.timeout field (e.g., "10s", "500ms").
19//! Default timeout: 30 seconds.
20//!
21//! §Fase 34.e (v1.29.0) — Streaming surface via [`HttpStreamingTool`].
22//! The async-trait Tool impl drives the upstream HTTP request via
23//! `reqwest::Client` (async) + drains the response body chunk-by-chunk.
24//! Content-Type drives framing:
25//!   - `text/event-stream` → per-W3C-SSE-event ToolChunks
26//!   - `application/x-ndjson` / `application/jsonl` → per-line ToolChunks
27//!   - Other (raw bytes, JSON, etc.) → single-chunk wrap (D9 backwards-
28//!     compat for non-streaming HTTP endpoints)
29//! Per-chunk cancel poll honors the D5 ≤100ms budget.
30
31use std::time::Duration;
32
33use crate::tool_executor::ToolResult;
34use crate::tool_registry::ToolEntry;
35
36// ── Timeout parsing ───────────────────────────────────────────────────────
37
38/// Parse a timeout string like "10s", "500ms", "2m" into Duration.
39/// Returns None for empty or unparseable values.
40fn parse_timeout(s: &str) -> Option<Duration> {
41    let s = s.trim();
42    if s.is_empty() {
43        return None;
44    }
45
46    if let Some(secs) = s.strip_suffix("ms") {
47        secs.trim().parse::<u64>().ok().map(Duration::from_millis)
48    } else if let Some(secs) = s.strip_suffix('s') {
49        secs.trim().parse::<u64>().ok().map(Duration::from_secs)
50    } else if let Some(mins) = s.strip_suffix('m') {
51        mins.trim()
52            .parse::<u64>()
53            .ok()
54            .map(|m| Duration::from_secs(m * 60))
55    } else {
56        // Try as raw seconds
57        s.parse::<u64>().ok().map(Duration::from_secs)
58    }
59}
60
61/// Public accessor for timeout parsing (used by emcp module).
62pub fn parse_timeout_pub(s: &str) -> Option<Duration> {
63    parse_timeout(s)
64}
65
66const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
67
68// ── HTTP dispatch ─────────────────────────────────────────────────────────
69
70/// Execute an HTTP tool call.
71///
72/// - `entry`: the tool's registry entry (must have provider == "http")
73/// - `argument`: the argument string from the use_tool step
74///
75/// Returns a ToolResult with the HTTP response body on success,
76/// or an error description on failure.
77pub fn dispatch_http(entry: &ToolEntry, argument: &str) -> ToolResult {
78    let url = entry.runtime.trim();
79
80    if url.is_empty() {
81        return ToolResult {
82            success: false,
83            output: format!(
84                "HTTP tool '{}': no endpoint URL. Set runtime: \"https://...\" in tool definition.",
85                entry.name
86            ),
87            tool_name: entry.name.clone(),
88        };
89    }
90
91    // Validate URL scheme
92    if !url.starts_with("http://") && !url.starts_with("https://") {
93        return ToolResult {
94            success: false,
95            output: format!(
96                "HTTP tool '{}': invalid URL '{}'. Must start with http:// or https://.",
97                entry.name, url
98            ),
99            tool_name: entry.name.clone(),
100        };
101    }
102
103    let timeout = parse_timeout(&entry.timeout).unwrap_or(DEFAULT_TIMEOUT);
104
105    // Build the request body — wrap as JSON string if not already JSON
106    let body = if argument.trim_start().starts_with('{') || argument.trim_start().starts_with('[') {
107        argument.to_string()
108    } else {
109        serde_json::json!({ "input": argument }).to_string()
110    };
111
112    // Execute the HTTP request
113    match execute_request(url, &entry.name, &body, timeout) {
114        Ok(response) => response,
115        Err(e) => ToolResult {
116            success: false,
117            output: format!("HTTP tool '{}': {}", entry.name, e),
118            tool_name: entry.name.clone(),
119        },
120    }
121}
122
123/// Perform the actual HTTP POST request.
124fn execute_request(
125    url: &str,
126    tool_name: &str,
127    body: &str,
128    timeout: Duration,
129) -> Result<ToolResult, String> {
130    let client = reqwest::blocking::Client::builder()
131        .timeout(timeout)
132        .build()
133        .map_err(|e| format!("failed to create HTTP client: {e}"))?;
134
135    let response = client
136        .post(url)
137        .header("Content-Type", "application/json")
138        .header("X-Axon-Tool", tool_name)
139        .body(body.to_string())
140        .send()
141        .map_err(|e| {
142            if e.is_timeout() {
143                format!("request timed out after {}s", timeout.as_secs())
144            } else if e.is_connect() {
145                format!("connection failed to {url}")
146            } else {
147                format!("request failed: {e}")
148            }
149        })?;
150
151    let status = response.status();
152    let response_body = response
153        .text()
154        .map_err(|e| format!("failed to read response body: {e}"))?;
155
156    if status.is_success() {
157        Ok(ToolResult {
158            success: true,
159            output: response_body,
160            tool_name: tool_name.to_string(),
161        })
162    } else {
163        Ok(ToolResult {
164            success: false,
165            output: format!(
166                "HTTP {}: {}",
167                status.as_u16(),
168                if response_body.len() > 200 {
169                    format!("{}...", &response_body[..200])
170                } else {
171                    response_body
172                }
173            ),
174            tool_name: tool_name.to_string(),
175        })
176    }
177}
178
179// ════════════════════════════════════════════════════════════════════════════
180//  §Fase 34.e — HttpStreamingTool: async-trait Tool impl with per-chunk
181//  streaming wire emission via reqwest::Response::bytes_stream().
182// ════════════════════════════════════════════════════════════════════════════
183
184use async_trait::async_trait;
185use bytes::Bytes;
186use futures::StreamExt;
187
188use crate::backends::sse_streaming::{LineBuffer, SseEventParser};
189use crate::tool_trait::{Tool, ToolChunk, ToolContext, ToolFinishReason, ToolStream};
190
191/// HTTP tool with first-class streaming surface (Fase 34.e).
192///
193/// `stream()` drives a `reqwest::Client::post(url)` async request +
194/// drains `response.bytes_stream()` chunk-by-chunk. Content-Type
195/// header decides framing:
196///
197/// - **`text/event-stream`** → SSE per W3C spec. Each `data:` field
198///   from a complete SSE event emits as a `ToolChunk::intermediate`.
199///   `event:` / `id:` / `retry:` fields are dropped (the adopter sees
200///   only the data payload — the framing was for HTTP transport).
201/// - **`application/x-ndjson`** / **`application/jsonl`** → one
202///   `ToolChunk::intermediate` per LF-delimited line. Empty lines
203///   are skipped.
204/// - **Other** (`application/json`, `text/plain`, raw bytes) →
205///   single `ToolChunk::intermediate` with the full body accumulated,
206///   then a terminator. This is the D9 backwards-compat path for
207///   non-streaming HTTP endpoints — the same response shape
208///   [`dispatch_http`] returns synchronously, projected onto the
209///   single-chunk streaming surface.
210///
211/// # Cancel discipline (D5)
212///
213/// `ctx.cancel` is polled between every `bytes_stream().next().await`
214/// boundary. When fired, the stream drops the `reqwest::Response`
215/// (closing the connection) + emits a single
216/// `ToolFinishReason::Cancelled` terminator chunk. Wall-clock budget
217/// is bounded by the network roundtrip latency to the next chunk
218/// arrival (typically ≤100ms for SSE streams with regular keepalive).
219///
220/// # Error discipline
221///
222/// Every failure surface (URL invalid / client build / connect /
223/// timeout / non-2xx status / mid-stream byte error / I/O error)
224/// is captured as a `ToolFinishReason::Error { message }` terminator
225/// chunk — the consumer never sees a panic or a silently-truncated
226/// stream.
227pub struct HttpStreamingTool {
228    name: String,
229    url: String,
230    timeout: Duration,
231}
232
233impl HttpStreamingTool {
234    /// Construct from a registry [`ToolEntry`]. Validates the URL +
235    /// extracts the timeout. Returns `Err` with adopter-facing
236    /// diagnostic when the URL is missing or has an invalid scheme.
237    pub fn from_entry(entry: &ToolEntry) -> Result<Self, String> {
238        let url = entry.runtime.trim();
239        if url.is_empty() {
240            return Err(format!(
241                "HTTP tool '{}': no endpoint URL. Set runtime: \"https://...\" in tool definition.",
242                entry.name
243            ));
244        }
245        if !url.starts_with("http://") && !url.starts_with("https://") {
246            return Err(format!(
247                "HTTP tool '{}': invalid URL '{}'. Must start with http:// or https://.",
248                entry.name, url
249            ));
250        }
251        let timeout = parse_timeout(&entry.timeout).unwrap_or(DEFAULT_TIMEOUT);
252        Ok(Self {
253            name: entry.name.clone(),
254            url: url.to_string(),
255            timeout,
256        })
257    }
258
259    /// Public new() ctor for tests + adopters who construct directly
260    /// without a registry entry.
261    pub fn new(name: String, url: String, timeout: Duration) -> Self {
262        Self { name, url, timeout }
263    }
264}
265
266/// Build the request body — wrap as JSON `{ "input": args }` if not
267/// already JSON. Same logic as [`dispatch_http`].
268fn build_request_body(args: &str) -> String {
269    let trimmed = args.trim_start();
270    if trimmed.starts_with('{') || trimmed.starts_with('[') {
271        args.to_string()
272    } else {
273        serde_json::json!({ "input": args }).to_string()
274    }
275}
276
277/// Classify an HTTP Content-Type header into the framing mode the
278/// streaming tool will use.
279#[derive(Debug, Clone, Copy, PartialEq, Eq)]
280enum FramingMode {
281    /// W3C Server-Sent Events. Drain via [`LineBuffer`] +
282    /// [`SseEventParser`]; emit each event's `data:` field as a
283    /// `ToolChunk`.
284    Sse,
285    /// Newline-delimited JSON. Drain via [`LineBuffer`]; emit each
286    /// non-empty line as a `ToolChunk`.
287    Ndjson,
288    /// Anything else. Accumulate full body + emit as 1 chunk +
289    /// terminator. D9-style backwards-compat for non-streaming
290    /// HTTP endpoints.
291    Single,
292}
293
294fn classify_framing(content_type: &str) -> FramingMode {
295    let lc = content_type.to_ascii_lowercase();
296    if lc.contains("text/event-stream") {
297        FramingMode::Sse
298    } else if lc.contains("application/x-ndjson") || lc.contains("application/jsonl") {
299        FramingMode::Ndjson
300    } else {
301        FramingMode::Single
302    }
303}
304
305#[async_trait]
306impl Tool for HttpStreamingTool {
307    async fn execute(&self, args: String, _ctx: ToolContext) -> ToolResult {
308        // Synchronous path — adopters calling execute() directly get
309        // the legacy [`dispatch_http`] behavior verbatim. The
310        // streaming path drives `stream()` exclusively.
311        //
312        // [`dispatch_http`] uses `reqwest::blocking::Client` (it
313        // existed pre-async-trait). Calling blocking-reqwest from
314        // inside a tokio runtime panics; we wrap the call with
315        // `spawn_blocking` so the synchronous client runs on tokio's
316        // blocking pool. Output is byte-equal to dispatch_http (D9).
317        let entry = ToolEntry {
318            name: self.name.clone(),
319            provider: "http".to_string(),
320            timeout: format!("{}s", self.timeout.as_secs()),
321            runtime: self.url.clone(),
322            sandbox: None,
323            max_results: None,
324            output_schema: String::new(),
325            effect_row: Vec::new(),
326            // §Fase 58.f.2 — reconstructed entry for the legacy sync
327            // delegate; no typed input schema needed on this path.
328            parameters: Vec::new(),
329            secret: String::new(),
330            secret_partition: String::new(),
331            source: crate::tool_registry::ToolSource::Program,
332            is_streaming: false,
333            scrape: None,
334        };
335        match tokio::task::spawn_blocking(move || dispatch_http(&entry, &args)).await {
336            Ok(result) => result,
337            Err(e) => ToolResult {
338                success: false,
339                output: format!("HTTP tool '{}': blocking task join failed: {e}", self.name),
340                tool_name: self.name.clone(),
341            },
342        }
343    }
344
345    async fn stream(&self, args: String, ctx: ToolContext) -> ToolStream {
346        let url = self.url.clone();
347        let name = self.name.clone();
348        let timeout = self.timeout;
349        let cancel = ctx.cancel.clone();
350        let body = build_request_body(&args);
351
352        // mpsc + spawn pattern: the background task drives the HTTP
353        // request + drains chunks into the channel; the returned
354        // stream wraps the receiver. This gives us real per-chunk
355        // streaming (chunks reach the dispatcher AS they arrive from
356        // upstream) without requiring async-stream macro.
357        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<ToolChunk>();
358
359        tokio::spawn(async move {
360            // Helper: send the terminator + drop tx so the consumer's
361            // stream ends cleanly. Returning `Err` from a sub-step
362            // sends an Error-terminator; reaching the natural end of
363            // the body sends a Stop-terminator.
364            let send_terminator = |reason: ToolFinishReason| {
365                let _ = tx.send(ToolChunk::terminator("", reason));
366            };
367
368            // Pre-flight cancel check.
369            if cancel.is_cancelled() {
370                send_terminator(ToolFinishReason::Cancelled);
371                return;
372            }
373
374            // 1. Build async client.
375            let client = match reqwest::Client::builder().timeout(timeout).build() {
376                Ok(c) => c,
377                Err(e) => {
378                    send_terminator(ToolFinishReason::Error {
379                        message: format!(
380                            "HTTP tool '{name}': failed to build async client: {e}"
381                        ),
382                    });
383                    return;
384                }
385            };
386
387            // 2. Issue request.
388            let response = match client
389                .post(&url)
390                .header("Content-Type", "application/json")
391                .header("X-Axon-Tool", &name)
392                .body(body)
393                .send()
394                .await
395            {
396                Ok(r) => r,
397                Err(e) => {
398                    let message = if e.is_timeout() {
399                        format!(
400                            "HTTP tool '{name}': request timed out after {}s",
401                            timeout.as_secs()
402                        )
403                    } else if e.is_connect() {
404                        format!("HTTP tool '{name}': connection failed to {url}")
405                    } else {
406                        format!("HTTP tool '{name}': request failed: {e}")
407                    };
408                    send_terminator(ToolFinishReason::Error { message });
409                    return;
410                }
411            };
412
413            // 3. Non-2xx → error terminator with status code +
414            //    truncated body. Mirrors dispatch_http's diagnostic
415            //    shape.
416            let status = response.status();
417            if !status.is_success() {
418                let body_text = response.text().await.unwrap_or_default();
419                let truncated = if body_text.len() > 200 {
420                    format!("{}...", &body_text[..200])
421                } else {
422                    body_text
423                };
424                send_terminator(ToolFinishReason::Error {
425                    message: format!("HTTP {}: {}", status.as_u16(), truncated),
426                });
427                return;
428            }
429
430            // 4. Read Content-Type header → classify framing.
431            let content_type = response
432                .headers()
433                .get(reqwest::header::CONTENT_TYPE)
434                .and_then(|v| v.to_str().ok())
435                .unwrap_or("")
436                .to_string();
437            let framing = classify_framing(&content_type);
438
439            // 5. Drain the body byte-stream per framing mode.
440            let mut byte_stream = response.bytes_stream();
441            let drain_result = match framing {
442                FramingMode::Sse => {
443                    drain_sse(&mut byte_stream, &cancel, &tx).await
444                }
445                FramingMode::Ndjson => {
446                    drain_ndjson(&mut byte_stream, &cancel, &tx).await
447                }
448                FramingMode::Single => {
449                    drain_single(&mut byte_stream, &cancel, &tx).await
450                }
451            };
452
453            match drain_result {
454                DrainOutcome::Completed => send_terminator(ToolFinishReason::Stop),
455                DrainOutcome::Cancelled => send_terminator(ToolFinishReason::Cancelled),
456                DrainOutcome::Error(message) => {
457                    send_terminator(ToolFinishReason::Error { message })
458                }
459            }
460        });
461
462        // Wrap the receiver as a Stream. Each `recv().await` yields
463        // a ToolChunk + holds the channel open until the producer
464        // task drops `tx`.
465        Box::pin(futures::stream::unfold(rx, |mut rx| async move {
466            rx.recv().await.map(|chunk| (chunk, rx))
467        }))
468    }
469
470    fn is_streaming(&self) -> bool {
471        true
472    }
473}
474
475/// Per-framing-mode drain outcome. Drives the terminator decision
476/// in the spawned task without leaking implementation details.
477enum DrainOutcome {
478    Completed,
479    Cancelled,
480    Error(String),
481}
482
483/// Drain SSE framing. Reuses the battle-tested
484/// [`crate::backends::sse_streaming::LineBuffer`] +
485/// [`crate::backends::sse_streaming::SseEventParser`] from Fase 33.d
486/// so every adopter-emitted SSE shape (CRLF normalization, CR strip,
487/// multi-line data field, comment lines) is honored verbatim.
488async fn drain_sse<S>(
489    byte_stream: &mut S,
490    cancel: &crate::cancel_token::CancellationFlag,
491    tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
492) -> DrainOutcome
493where
494    S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
495{
496    let mut line_buf = LineBuffer::new();
497    let mut sse_parser = SseEventParser::new();
498    loop {
499        if cancel.is_cancelled() {
500            return DrainOutcome::Cancelled;
501        }
502        match byte_stream.next().await {
503            None => break,
504            Some(Err(e)) => {
505                return DrainOutcome::Error(format!("SSE stream chunk error: {e}"))
506            }
507            Some(Ok(bytes)) => {
508                let lines = line_buf.push(&bytes);
509                for line in lines {
510                    if let Some(event) = sse_parser.push_line(&line) {
511                        if let Some(data) = event.data {
512                            if tx
513                                .send(ToolChunk::intermediate(data))
514                                .is_err()
515                            {
516                                return DrainOutcome::Cancelled;
517                            }
518                        }
519                    }
520                }
521            }
522        }
523    }
524    // Flush trailing line (events without a final blank-line
525    // terminator) — push it into the parser; if it completes an
526    // event, emit it.
527    if let Some(line) = line_buf.flush() {
528        if let Some(event) = sse_parser.push_line(&line) {
529            if let Some(data) = event.data {
530                let _ = tx.send(ToolChunk::intermediate(data));
531            }
532        }
533    }
534    DrainOutcome::Completed
535}
536
537/// Drain NDJSON framing. Each LF-delimited line emits as a
538/// `ToolChunk::intermediate`. Empty lines are skipped per
539/// `application/x-ndjson` spec.
540async fn drain_ndjson<S>(
541    byte_stream: &mut S,
542    cancel: &crate::cancel_token::CancellationFlag,
543    tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
544) -> DrainOutcome
545where
546    S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
547{
548    let mut line_buf = LineBuffer::new();
549    loop {
550        if cancel.is_cancelled() {
551            return DrainOutcome::Cancelled;
552        }
553        match byte_stream.next().await {
554            None => break,
555            Some(Err(e)) => {
556                return DrainOutcome::Error(format!("NDJSON stream chunk error: {e}"))
557            }
558            Some(Ok(bytes)) => {
559                let lines = line_buf.push(&bytes);
560                for line in lines {
561                    if !line.is_empty()
562                        && tx.send(ToolChunk::intermediate(line)).is_err()
563                    {
564                        return DrainOutcome::Cancelled;
565                    }
566                }
567            }
568        }
569    }
570    if let Some(line) = line_buf.flush() {
571        if !line.is_empty() {
572            let _ = tx.send(ToolChunk::intermediate(line));
573        }
574    }
575    DrainOutcome::Completed
576}
577
578/// Drain single-chunk framing. Accumulate the full body + emit as
579/// 1 `ToolChunk::intermediate` (terminator follows from the caller).
580/// D9 backwards-compat for non-streaming HTTP endpoints.
581async fn drain_single<S>(
582    byte_stream: &mut S,
583    cancel: &crate::cancel_token::CancellationFlag,
584    tx: &tokio::sync::mpsc::UnboundedSender<ToolChunk>,
585) -> DrainOutcome
586where
587    S: futures::Stream<Item = reqwest::Result<Bytes>> + Unpin + Send,
588{
589    let mut acc: Vec<u8> = Vec::new();
590    loop {
591        if cancel.is_cancelled() {
592            return DrainOutcome::Cancelled;
593        }
594        match byte_stream.next().await {
595            None => break,
596            Some(Err(e)) => {
597                return DrainOutcome::Error(format!("HTTP body chunk error: {e}"))
598            }
599            Some(Ok(bytes)) => acc.extend_from_slice(&bytes),
600        }
601    }
602    let body_text = String::from_utf8_lossy(&acc).into_owned();
603    if !body_text.is_empty()
604        && tx
605            .send(ToolChunk::intermediate(body_text))
606            .is_err()
607    {
608        return DrainOutcome::Cancelled;
609    }
610    DrainOutcome::Completed
611}
612
613// ── Tests ─────────────────────────────────────────────────────────────────
614
615#[cfg(test)]
616mod tests {
617    use super::*;
618    use crate::tool_registry::{ToolEntry, ToolSource};
619
620    fn make_http_entry(name: &str, url: &str, timeout: &str) -> ToolEntry {
621        ToolEntry {
622            name: name.to_string(),
623            provider: "http".to_string(),
624            timeout: timeout.to_string(),
625            runtime: url.to_string(),
626            sandbox: None,
627            max_results: None,
628            output_schema: "JSON".to_string(),
629            effect_row: vec!["network".to_string()],
630            parameters: Vec::new(),
631            secret: String::new(),
632            secret_partition: String::new(),
633            source: ToolSource::Program,
634            // §Fase 34.c — HTTP tools default to non-streaming; effect_row
635            // carries `network` but no `stream:` prefix. HTTP streaming
636            // (SSE-aware adapter consuming upstream SSE) lands in Fase 34.e.
637            is_streaming: false,
638            scrape: None,
639        }
640    }
641
642    // ── Timeout parsing ───────────────────────────────────────────
643
644    #[test]
645    fn parse_timeout_seconds() {
646        assert_eq!(parse_timeout("10s"), Some(Duration::from_secs(10)));
647        assert_eq!(parse_timeout("30s"), Some(Duration::from_secs(30)));
648    }
649
650    #[test]
651    fn parse_timeout_milliseconds() {
652        assert_eq!(parse_timeout("500ms"), Some(Duration::from_millis(500)));
653        assert_eq!(parse_timeout("100ms"), Some(Duration::from_millis(100)));
654    }
655
656    #[test]
657    fn parse_timeout_minutes() {
658        assert_eq!(parse_timeout("2m"), Some(Duration::from_secs(120)));
659    }
660
661    #[test]
662    fn parse_timeout_raw_number() {
663        assert_eq!(parse_timeout("15"), Some(Duration::from_secs(15)));
664    }
665
666    #[test]
667    fn parse_timeout_empty() {
668        assert_eq!(parse_timeout(""), None);
669        assert_eq!(parse_timeout("  "), None);
670    }
671
672    #[test]
673    fn parse_timeout_invalid() {
674        assert_eq!(parse_timeout("abc"), None);
675        assert_eq!(parse_timeout("10x"), None);
676    }
677
678    // ── URL validation ────────────────────────────────────────────
679
680    #[test]
681    fn dispatch_empty_url_fails() {
682        let entry = make_http_entry("DataAPI", "", "10s");
683        let result = dispatch_http(&entry, "test query");
684        assert!(!result.success);
685        assert!(result.output.contains("no endpoint URL"));
686    }
687
688    #[test]
689    fn dispatch_invalid_url_scheme_fails() {
690        let entry = make_http_entry("DataAPI", "ftp://example.com", "10s");
691        let result = dispatch_http(&entry, "test query");
692        assert!(!result.success);
693        assert!(result.output.contains("invalid URL"));
694        assert!(result.output.contains("http://"));
695    }
696
697    // ── Connection errors (no server) ─────────────────────────────
698
699    #[test]
700    fn dispatch_connection_refused() {
701        // Port 1 is almost certainly not listening
702        let entry = make_http_entry("TestTool", "http://127.0.0.1:1/api", "2s");
703        let result = dispatch_http(&entry, "test");
704        assert!(!result.success);
705        assert!(
706            result.output.contains("connection failed")
707                || result.output.contains("request failed")
708                || result.output.contains("timed out"),
709            "unexpected error: {}",
710            result.output
711        );
712    }
713
714    // ── Body wrapping ─────────────────────────────────────────────
715
716    #[test]
717    fn json_body_passthrough() {
718        // If argument is already JSON, it should be sent as-is
719        let arg = r#"{"query": "test"}"#;
720        let body = if arg.trim_start().starts_with('{') {
721            arg.to_string()
722        } else {
723            serde_json::json!({ "input": arg }).to_string()
724        };
725        assert_eq!(body, r#"{"query": "test"}"#);
726    }
727
728    #[test]
729    fn plain_text_wrapped() {
730        // If argument is plain text, it should be wrapped
731        let arg = "search for cats";
732        let body = if arg.trim_start().starts_with('{') || arg.trim_start().starts_with('[') {
733            arg.to_string()
734        } else {
735            serde_json::json!({ "input": arg }).to_string()
736        };
737        let parsed: serde_json::Value = serde_json::from_str(&body).unwrap();
738        assert_eq!(parsed["input"], "search for cats");
739    }
740
741    #[test]
742    fn array_body_passthrough() {
743        let arg = r#"[1, 2, 3]"#;
744        let body = if arg.trim_start().starts_with('{') || arg.trim_start().starts_with('[') {
745            arg.to_string()
746        } else {
747            serde_json::json!({ "input": arg }).to_string()
748        };
749        assert_eq!(body, "[1, 2, 3]");
750    }
751}