use super::contract::metadata_key as meta;
use crate::{agent::cancellation::AgentCancellation, output::redact_sensitive_text};
use dom_smoothie::{Config, Readability, TextMode};
use reqwest::{StatusCode, Url, blocking::Client, redirect::Policy};
use serde_json::json;
use std::{
io::Read,
net::{IpAddr, SocketAddr, ToSocketAddrs},
sync::mpsc,
thread::{self, JoinHandle},
time::{Duration, Instant},
};
const URL_FETCH_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
const URL_FETCH_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
const URL_FETCH_MAX_BYTES: usize = 512 * 1024;
const URL_FETCH_OUTPUT_MAX_BYTES: usize = 48 * 1024;
const URL_FETCH_DEFAULT_MAX_TOKENS: u64 = 5_000;
const URL_FETCH_MIN_MAX_TOKENS: u64 = 1_000;
const URL_FETCH_MAX_MAX_TOKENS: u64 = 10_000;
const URL_FETCH_READABILITY_MIN_LENGTH: usize = 200;
const TRUNCATION_MARKER: &str = "\n[truncated]";
const URL_FETCH_CANCEL_POLL_INTERVAL: Duration = Duration::from_millis(25);
const URL_FETCH_WORKER_JOIN_TIMEOUT: Duration = Duration::from_millis(500);
#[derive(Debug, Clone)]
struct PinnedTarget {
host: String,
addrs: Vec<SocketAddr>,
}
#[derive(Debug, Clone)]
struct ExtractedContent {
text: String,
title: Option<String>,
}
#[derive(Debug, Clone)]
pub(crate) struct UrlFetchInput {
pub(crate) url: String,
pub(crate) max_tokens: Option<u64>,
}
impl UrlFetchInput {
pub(crate) fn validate(mut self) -> anyhow::Result<Self> {
self.url = self.url.trim().to_string();
if self.url.is_empty() {
anyhow::bail!("url must not be empty");
}
let parsed = Url::parse(&self.url)
.map_err(|_| anyhow::anyhow!("url must be an http(s) URL with a host"))?;
if !matches!(parsed.scheme(), "http" | "https") || parsed.host_str().is_none() {
anyhow::bail!("url must be an http(s) URL with a host");
}
if let Some(max_tokens) = self.max_tokens
&& !(URL_FETCH_MIN_MAX_TOKENS..=URL_FETCH_MAX_MAX_TOKENS).contains(&max_tokens)
{
anyhow::bail!(
"maxTokens must be between {URL_FETCH_MIN_MAX_TOKENS} and {URL_FETCH_MAX_MAX_TOKENS}"
);
}
Ok(self)
}
fn max_tokens(&self) -> u64 {
self.max_tokens.unwrap_or(URL_FETCH_DEFAULT_MAX_TOKENS)
}
}
#[derive(Debug, Clone)]
pub(crate) struct UrlFetchOutput {
pub(crate) success: bool,
pub(crate) content: String,
pub(crate) metadata: serde_json::Value,
}
pub(crate) fn fetch_url(
input: UrlFetchInput,
cancellation: &AgentCancellation,
) -> anyhow::Result<UrlFetchOutput> {
cancellation.check()?;
let url = Url::parse(&input.url)?;
let target = resolve_public_target(&url)?;
cancellation.check()?;
let client = build_pinned_client(&target)?;
cancellation.check()?;
fetch_with_client(input, url, &client, cancellation)
}
fn build_pinned_client(target: &PinnedTarget) -> anyhow::Result<Client> {
Client::builder()
.connect_timeout(URL_FETCH_CONNECT_TIMEOUT)
.timeout(URL_FETCH_REQUEST_TIMEOUT)
.user_agent(format!("magi-code/{}", env!("CARGO_PKG_VERSION")))
.redirect(Policy::none())
.resolve_to_addrs(&target.host, &target.addrs)
.build()
.map_err(|error| anyhow::anyhow!("url_fetch HTTP client setup failed: {error}"))
}
struct FetchHttpResponse {
status: StatusCode,
content_type: String,
bytes: Vec<u8>,
}
struct UrlFetchWorkerHandle {
done_receiver: mpsc::Receiver<()>,
join_handle: JoinHandle<()>,
}
impl UrlFetchWorkerHandle {
fn join_or_warn(self) {
match self
.done_receiver
.recv_timeout(URL_FETCH_WORKER_JOIN_TIMEOUT)
{
Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => {
if self.join_handle.join().is_err() {
eprintln!("magi-code warning: url_fetch HTTP worker panicked");
}
}
Err(mpsc::RecvTimeoutError::Timeout) => {
eprintln!(
"magi-code warning: url_fetch HTTP worker did not exit within cleanup grace period; detaching worker"
);
}
}
}
}
fn spawn_url_fetch_worker(
client: Client,
url: Url,
) -> anyhow::Result<(
mpsc::Receiver<anyhow::Result<FetchHttpResponse>>,
UrlFetchWorkerHandle,
)> {
let (sender, receiver) = mpsc::sync_channel(1);
let (done_sender, done_receiver) = mpsc::sync_channel(1);
let join_handle = thread::Builder::new()
.name("url-fetch-http".to_string())
.spawn(move || {
let _ = sender.send(blocking_fetch_response(client, url));
let _ = done_sender.send(());
})?;
Ok((
receiver,
UrlFetchWorkerHandle {
done_receiver,
join_handle,
},
))
}
fn recv_cancellable<T>(
receiver: &mpsc::Receiver<T>,
timeout_label: &str,
idle_timeout: Duration,
cancellation: &AgentCancellation,
) -> anyhow::Result<T> {
let start = Instant::now();
loop {
cancellation.check()?;
if start.elapsed() >= idle_timeout {
anyhow::bail!("{timeout_label} after {}s", idle_timeout.as_secs());
}
let remaining = idle_timeout.saturating_sub(start.elapsed());
match receiver.recv_timeout(remaining.min(URL_FETCH_CANCEL_POLL_INTERVAL)) {
Ok(value) => return Ok(value),
Err(mpsc::RecvTimeoutError::Timeout) => continue,
Err(mpsc::RecvTimeoutError::Disconnected) => {
cancellation.check()?;
anyhow::bail!("url_fetch HTTP worker disconnected");
}
}
}
}
fn blocking_fetch_response(client: Client, url: Url) -> anyhow::Result<FetchHttpResponse> {
let response = client.get(url).send().map_err(|error| {
if error.is_timeout() {
anyhow::anyhow!("url_fetch request timed out")
} else {
anyhow::anyhow!("url_fetch network error: {error}")
}
})?;
let status = response.status();
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string();
if response
.content_length()
.is_some_and(|length| length > URL_FETCH_MAX_BYTES as u64)
{
anyhow::bail!("url_fetch response exceeded {URL_FETCH_MAX_BYTES} byte limit");
}
let bytes = read_bounded_body(response)?;
Ok(FetchHttpResponse {
status,
content_type,
bytes,
})
}
fn fetch_with_client(
args: UrlFetchInput,
url: Url,
client: &Client,
cancellation: &AgentCancellation,
) -> anyhow::Result<UrlFetchOutput> {
let (receiver, worker) = spawn_url_fetch_worker(client.clone(), url.clone())?;
let response = match recv_cancellable(
&receiver,
"url_fetch response timeout",
URL_FETCH_REQUEST_TIMEOUT,
cancellation,
) {
Ok(response) => response,
Err(error) => {
drop(receiver);
worker.join_or_warn();
return Err(error);
}
}?;
worker.join_or_warn();
cancellation.check()?;
let status = response.status;
let content_type = response.content_type;
let byte_count = response.bytes.len();
let raw = String::from_utf8_lossy(&response.bytes).into_owned();
if !status.is_success() {
let (content, truncated) = sanitize_and_bound(&raw, args.max_tokens());
return Ok(result(FetchResultParts {
success: false,
content,
url: &url,
status,
content_type: &content_type,
title: None,
bytes: byte_count,
truncated,
}));
}
let extracted = extract_content(&raw, &content_type, url.as_str());
cancellation.check()?;
let (content, truncated) = sanitize_and_bound(&extracted.text, args.max_tokens());
cancellation.check()?;
Ok(result(FetchResultParts {
success: true,
content,
url: &url,
status,
content_type: &content_type,
title: extracted.title.as_deref(),
bytes: byte_count,
truncated,
}))
}
struct FetchResultParts<'a> {
success: bool,
content: String,
url: &'a Url,
status: StatusCode,
content_type: &'a str,
title: Option<&'a str>,
bytes: usize,
truncated: bool,
}
fn result(parts: FetchResultParts<'_>) -> UrlFetchOutput {
UrlFetchOutput {
success: parts.success,
content: parts.content,
metadata: json!({
(meta::URL): parts.url.as_str(),
(meta::STATUS): parts.status.as_u16(),
(meta::CONTENT_TYPE): parts.content_type,
(meta::TITLE): parts.title,
(meta::BYTES): parts.bytes,
(meta::TRUNCATED): parts.truncated,
}),
}
}
fn read_bounded_body(mut response: reqwest::blocking::Response) -> anyhow::Result<Vec<u8>> {
let mut bytes = Vec::new();
response
.by_ref()
.take((URL_FETCH_MAX_BYTES + 1) as u64)
.read_to_end(&mut bytes)
.map_err(|error| anyhow::anyhow!("url_fetch response read failed: {error}"))?;
if bytes.len() > URL_FETCH_MAX_BYTES {
anyhow::bail!("url_fetch response exceeded {URL_FETCH_MAX_BYTES} byte limit");
}
Ok(bytes)
}
fn extract_content(body: &str, content_type: &str, url: &str) -> ExtractedContent {
if !content_type.to_ascii_lowercase().contains("html") {
return ExtractedContent {
text: body.to_string(),
title: None,
};
}
if let Some(extracted) = readability_extract(body, url) {
return extracted;
}
ExtractedContent {
text: html2md::rewrite_html(body, false),
title: None,
}
}
fn readability_extract(html: &str, url: &str) -> Option<ExtractedContent> {
let cfg = Config {
text_mode: TextMode::Markdown,
..Default::default()
};
let mut readability = Readability::new(html, Some(url), Some(cfg)).ok()?;
let article = readability.parse().ok()?;
let text = article.text_content.to_string();
let text = text.trim().to_string();
if article.length <= URL_FETCH_READABILITY_MIN_LENGTH || text.is_empty() {
return None;
}
let title = (!article.title.trim().is_empty()).then(|| article.title.trim().to_string());
Some(ExtractedContent { text, title })
}
fn sanitize_and_bound(text: &str, max_tokens: u64) -> (String, bool) {
let cap = URL_FETCH_OUTPUT_MAX_BYTES.min(max_tokens.saturating_mul(4) as usize);
let mut truncated = false;
let mut content = truncate_with_marker(text.trim(), cap, &mut truncated);
content = redact_sensitive_text(&content);
if content.len() > cap {
truncated = true;
content = truncate_with_marker(&content, cap, &mut truncated);
}
(content, truncated)
}
fn truncate_with_marker(text: &str, max_bytes: usize, truncated: &mut bool) -> String {
if text.len() <= max_bytes {
return text.to_string();
}
*truncated = true;
if max_bytes <= TRUNCATION_MARKER.len() {
return truncate_to_char_boundary(TRUNCATION_MARKER, max_bytes).to_string();
}
let text_budget = max_bytes - TRUNCATION_MARKER.len();
format!(
"{}{}",
truncate_to_char_boundary(text, text_budget),
TRUNCATION_MARKER
)
}
fn truncate_to_char_boundary(text: &str, max_bytes: usize) -> &str {
if text.len() <= max_bytes {
return text;
}
let mut end = max_bytes;
while end > 0 && !text.is_char_boundary(end) {
end -= 1;
}
&text[..end]
}
fn resolve_public_target(url: &Url) -> anyhow::Result<PinnedTarget> {
if !matches!(url.scheme(), "http" | "https") {
anyhow::bail!("url_fetch only supports http and https URLs");
}
let host = url
.host_str()
.ok_or_else(|| anyhow::anyhow!("url_fetch URL must include a host"))?
.to_string();
let port = url.port_or_known_default().unwrap_or(80);
let addrs = resolve_host_addrs(&host, port)?;
validate_public_addrs(&host, addrs)
}
fn resolve_host_addrs(host: &str, port: u16) -> anyhow::Result<Vec<SocketAddr>> {
if let Ok(ip) = host.parse::<IpAddr>() {
return Ok(vec![SocketAddr::new(ip, port)]);
}
(host, port)
.to_socket_addrs()
.map(|iter| iter.collect::<Vec<_>>())
.map_err(|error| anyhow::anyhow!("url_fetch DNS resolution failed for {host}: {error}"))
}
fn validate_public_addrs(host: &str, addrs: Vec<SocketAddr>) -> anyhow::Result<PinnedTarget> {
if addrs.is_empty() {
anyhow::bail!("url_fetch DNS resolution returned no addresses for {host}");
}
for addr in &addrs {
if !is_public_ip(addr.ip()) {
anyhow::bail!(
"url_fetch target resolves to unsafe IP address: {}",
addr.ip()
);
}
}
Ok(PinnedTarget {
host: host.to_string(),
addrs,
})
}
fn is_shared_ipv4(ip: std::net::Ipv4Addr) -> bool {
let octets = ip.octets();
octets[0] == 100 && (octets[1] & 0b1100_0000) == 64
}
fn is_benchmarking_ipv4(ip: std::net::Ipv4Addr) -> bool {
let octets = ip.octets();
octets[0] == 198 && matches!(octets[1], 18 | 19)
}
fn is_documentation_ipv4(ip: std::net::Ipv4Addr) -> bool {
let octets = ip.octets();
matches!(
octets,
[192, 0, 2, _] | [198, 51, 100, _] | [203, 0, 113, _]
)
}
fn is_public_ipv4(ip: std::net::Ipv4Addr) -> bool {
let octets = ip.octets();
!(ip.is_private()
|| ip.is_loopback()
|| ip.is_link_local()
|| ip.is_unspecified()
|| ip.is_multicast()
|| octets == [255, 255, 255, 255]
|| is_shared_ipv4(ip)
|| is_benchmarking_ipv4(ip)
|| is_documentation_ipv4(ip)
|| octets[0] == 127)
}
fn is_public_ip(ip: IpAddr) -> bool {
match ip {
IpAddr::V4(ip) => is_public_ipv4(ip),
IpAddr::V6(ip) => {
if let Some(mapped) = ip.to_ipv4_mapped() {
return is_public_ipv4(mapped);
}
!(ip.is_loopback()
|| ip.is_unspecified()
|| ip.is_multicast()
|| ((ip.segments()[0] & 0xfe00) == 0xfc00)
|| ((ip.segments()[0] & 0xffc0) == 0xfe80))
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::agent::cancellation::AgentCancellationHandle;
use std::{
io::{Read, Write},
net::{Ipv4Addr, Ipv6Addr, TcpListener},
sync::mpsc,
thread,
};
fn http_response(content_type: &str, body: &str) -> String {
format!(
"HTTP/1.1 200 OK\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\n\r\n{body}",
body.len()
)
}
fn fake_server(
response: String,
) -> (SocketAddr, mpsc::Receiver<String>, thread::JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let (tx, rx) = mpsc::channel();
let handle = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut buf = [0_u8; 8192];
let n = stream.read(&mut buf).unwrap();
let _ = tx.send(String::from_utf8_lossy(&buf[..n]).to_string());
let _ = stream.write_all(response.as_bytes());
});
(addr, rx, handle)
}
fn pinned_loopback_client(addr: SocketAddr) -> Client {
build_pinned_client(&PinnedTarget {
host: "example.test".to_string(),
addrs: vec![addr],
})
.unwrap()
}
#[test]
fn rejects_private_loopback_link_local_unspecified_and_multicast_ips() {
for ip in [
IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)),
IpAddr::V4(Ipv4Addr::new(172, 16, 0, 1)),
IpAddr::V4(Ipv4Addr::new(192, 168, 0, 1)),
IpAddr::V4(Ipv4Addr::new(169, 254, 169, 254)),
IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)),
IpAddr::V4(Ipv4Addr::new(224, 0, 0, 1)),
IpAddr::V4(Ipv4Addr::new(255, 255, 255, 255)),
IpAddr::V4(Ipv4Addr::new(100, 64, 0, 1)),
IpAddr::V4(Ipv4Addr::new(100, 127, 255, 254)),
IpAddr::V4(Ipv4Addr::new(198, 18, 0, 1)),
IpAddr::V4(Ipv4Addr::new(198, 19, 255, 254)),
IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)),
IpAddr::V4(Ipv4Addr::new(198, 51, 100, 1)),
IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)),
IpAddr::V6(Ipv6Addr::LOCALHOST),
IpAddr::V6("fc00::1".parse().unwrap()),
IpAddr::V6("fe80::1".parse().unwrap()),
] {
assert!(!is_public_ip(ip), "{ip}");
assert!(validate_public_addrs("bad", vec![SocketAddr::new(ip, 80)]).is_err());
}
assert!(is_public_ip(IpAddr::V4(Ipv4Addr::new(93, 184, 216, 34))));
assert!(is_public_ip(IpAddr::V4(Ipv4Addr::new(100, 128, 0, 1))));
assert!(is_public_ip(IpAddr::V4(Ipv4Addr::new(198, 20, 0, 1))));
assert!(resolve_public_target(&Url::parse("http://127.0.0.1/").unwrap()).is_err());
assert!(resolve_public_target(&Url::parse("http://localhost/").unwrap()).is_err());
}
#[test]
fn rejects_private_ipv4_mapped_ipv6_addresses() {
for ip in [
"::ffff:127.0.0.1",
"::ffff:10.0.0.1",
"::ffff:192.168.0.1",
"::ffff:172.16.0.1",
] {
let ip = IpAddr::V6(ip.parse().unwrap());
assert!(!is_public_ip(ip), "{ip}");
assert!(validate_public_addrs("bad", vec![SocketAddr::new(ip, 80)]).is_err());
}
assert!(is_public_ip(IpAddr::V6("::ffff:8.8.8.8".parse().unwrap())));
}
#[test]
fn pinned_resolution_is_used_for_request() {
let (addr, rx, handle) = fake_server(http_response("text/plain", "ok"));
let client = pinned_loopback_client(addr);
let args = UrlFetchInput {
url: "http://example.test/hello".to_string(),
max_tokens: None,
};
let result = fetch_with_client(
args,
Url::parse("http://example.test/hello").unwrap(),
&client,
&AgentCancellation::default(),
)
.unwrap();
handle.join().unwrap();
let request = rx.recv().unwrap();
assert!(request.starts_with("GET /hello HTTP/1.1"), "{request}");
assert!(
request.to_ascii_lowercase().contains("host: example.test"),
"{request}"
);
assert!(result.success);
assert_eq!(result.content, "ok");
}
#[test]
fn redirects_are_not_followed() {
let response = "HTTP/1.1 302 Found\r\nLocation: http://example.test/final\r\nContent-Length: 8\r\n\r\nredirect".to_string();
let (addr, rx, handle) = fake_server(response);
let client = pinned_loopback_client(addr);
let result = fetch_with_client(
UrlFetchInput {
url: "http://example.test/redirect".to_string(),
max_tokens: None,
},
Url::parse("http://example.test/redirect").unwrap(),
&client,
&AgentCancellation::default(),
)
.unwrap();
handle.join().unwrap();
let request = rx.recv().unwrap();
assert!(request.starts_with("GET /redirect HTTP/1.1"), "{request}");
assert!(!result.success);
assert_eq!(result.metadata[meta::STATUS], 302);
}
#[test]
fn content_length_over_max_bytes_fails_before_reading_body() {
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\nsmall",
URL_FETCH_MAX_BYTES + 1
);
let (addr, _, handle) = fake_server(response);
let client = pinned_loopback_client(addr);
let error = fetch_with_client(
UrlFetchInput {
url: "http://example.test/too-large".to_string(),
max_tokens: None,
},
Url::parse("http://example.test/too-large").unwrap(),
&client,
&AgentCancellation::default(),
)
.unwrap_err();
handle.join().unwrap();
assert!(error.to_string().contains("response exceeded"), "{error}");
}
#[test]
fn streamed_body_over_max_bytes_fails_without_content_length() {
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nConnection: close\r\n\r\n{}",
"x".repeat(URL_FETCH_MAX_BYTES + 1)
);
let (addr, _, handle) = fake_server(response);
let client = pinned_loopback_client(addr);
let error = fetch_with_client(
UrlFetchInput {
url: "http://example.test/stream-too-large".to_string(),
max_tokens: None,
},
Url::parse("http://example.test/stream-too-large").unwrap(),
&client,
&AgentCancellation::default(),
)
.unwrap_err();
handle.join().unwrap();
assert!(error.to_string().contains("response exceeded"), "{error}");
}
#[test]
fn output_truncation_preserves_utf8_char_boundaries() {
let (content, truncated) = sanitize_and_bound(&"界".repeat(2_000), 1_024);
assert!(truncated);
assert!(content.ends_with(TRUNCATION_MARKER));
assert!(content.len() <= 1_024 * 4);
assert!(content.is_char_boundary(content.len() - TRUNCATION_MARKER.len()));
}
#[test]
fn non_success_status_returns_bounded_redacted_body() {
let secret = "sk-webFetchSecret123456";
let body = format!(
"failure token={secret} {}",
"x".repeat(URL_FETCH_OUTPUT_MAX_BYTES)
);
let response = format!(
"HTTP/1.1 500 Internal Server Error\r\nContent-Type: text/plain\r\nContent-Length: {}\r\n\r\n{}",
body.len(),
body
);
let (addr, _, handle) = fake_server(response);
let client = pinned_loopback_client(addr);
let result = fetch_with_client(
UrlFetchInput {
url: "http://example.test/error".to_string(),
max_tokens: Some(10_000),
},
Url::parse("http://example.test/error").unwrap(),
&client,
&AgentCancellation::default(),
)
.unwrap();
handle.join().unwrap();
assert!(!result.success);
assert_eq!(result.metadata[meta::STATUS], 500);
assert_eq!(result.metadata[meta::TRUNCATED], true);
assert!(result.content.len() <= URL_FETCH_OUTPUT_MAX_BYTES);
assert!(!result.content.contains(secret));
assert!(result.content.contains("<redacted>"));
}
#[test]
fn html_uses_readability_when_article_is_long_enough() {
let html = format!(
"<html><head><title>Noise</title></head><body><article><h1>Readable Title</h1><p>{}</p></article></body></html>",
"Readable article sentence. ".repeat(20)
);
let extracted = extract_content(&html, "text/html", "https://example.com/page");
assert!(
extracted.text.contains("Readable article sentence"),
"{}",
extracted.text
);
assert!(extracted.title.is_some());
}
#[test]
fn short_html_falls_back_to_fast_html2md() {
let extracted = extract_content(
"<html><body><p>Short <strong>fallback</strong></p></body></html>",
"text/html; charset=utf-8",
"https://example.com",
);
assert!(extracted.text.contains("Short"));
assert!(extracted.text.contains("fallback"));
assert_eq!(extracted.title, None);
}
#[test]
fn non_html_returns_raw_redacted_text_and_truncates() {
let secret = "sk-webFetchSecret123456";
let (content, truncated) = sanitize_and_bound(
&format!(
"{{\"token\":\"{secret}\"}}{}",
"x".repeat(URL_FETCH_OUTPUT_MAX_BYTES)
),
10_000,
);
assert!(truncated);
assert!(content.ends_with(TRUNCATION_MARKER));
assert!(!content.contains(secret));
assert!(content.contains("<redacted>"));
assert!(content.len() <= URL_FETCH_OUTPUT_MAX_BYTES);
let extracted = extract_content("{\"ok\":true}", "application/json", "https://example.com");
assert_eq!(extracted.text, "{\"ok\":true}");
}
#[test]
fn canceled_during_blocking_body_read_returns_without_waiting_for_request_timeout() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
let (release_tx, release_rx) = mpsc::channel::<()>();
let server = thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut buf = [0_u8; 8192];
let _ = stream.read(&mut buf).unwrap();
let _ = stream.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 4\r\n\r\n",
);
let _ = release_rx.recv();
});
let client = pinned_loopback_client(addr);
let (cancellation, handle): (AgentCancellation, AgentCancellationHandle) =
AgentCancellation::default().child_token();
let canceler = thread::spawn(move || {
thread::sleep(Duration::from_millis(100));
handle.cancel();
});
let start = Instant::now();
let error = fetch_with_client(
UrlFetchInput {
url: "http://example.test/hang".to_string(),
max_tokens: None,
},
Url::parse("http://example.test/hang").unwrap(),
&client,
&cancellation,
)
.unwrap_err();
release_tx.send(()).unwrap();
canceler.join().unwrap();
server.join().unwrap();
assert_eq!(error.to_string(), "prompt canceled");
assert!(start.elapsed() < Duration::from_secs(2));
}
}