use crate::console_trace;
use crate::llm::ActionResult;
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::{Request, Response};
use std::collections::HashMap;
use std::convert::Infallible;
use tokio::sync::mpsc;
use tracing::{debug, error, trace, warn};
#[derive(Debug)]
pub struct RequestData {
pub method: String,
pub uri: String,
pub version: String,
pub headers: HashMap<String, String>,
pub body_bytes: Bytes,
}
pub const MAX_REQUEST_BODY_BYTES: usize = 8 * 1024 * 1024;
#[derive(Debug)]
pub struct BodyTooLarge {
pub method: String,
pub uri: String,
}
pub async fn extract_request_data(
req: Request<Incoming>,
protocol_label: &str,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<RequestData, BodyTooLarge> {
let method = req.method().to_string();
let uri = req
.uri()
.path_and_query()
.map(|pq| pq.as_str())
.unwrap_or(req.uri().path())
.to_string();
let version = format!("{:?}", req.version());
let mut headers = HashMap::new();
for (name, value) in req.headers() {
if let Ok(value_str) = value.to_str() {
headers.insert(name.to_string(), value_str.to_string());
}
}
let limited = http_body_util::Limited::new(req.into_body(), MAX_REQUEST_BODY_BYTES);
let body_bytes = match limited.collect().await {
Ok(collected) => collected.to_bytes(),
Err(e) => {
error!(
"{} {} {}: refusing request body ({}); limit is {} bytes",
protocol_label, method, uri, e, MAX_REQUEST_BODY_BYTES
);
let _ = status_tx.send(format!(
"✗ {} {} {} → 413 (request body over {} bytes)",
protocol_label, method, uri, MAX_REQUEST_BODY_BYTES
));
return Err(BodyTooLarge { method, uri });
}
};
debug!(
"{} request (action-based): {} {} {} ({} bytes)",
protocol_label,
method,
uri,
version,
body_bytes.len(),
);
let _ = status_tx.send(format!(
"[DEBUG] {} request: {} {} {} ({} bytes)",
protocol_label,
method,
uri,
version,
body_bytes.len()
));
trace!("{} request headers:", protocol_label);
for (name, value) in &headers {
trace!(" {}: {}", name, value);
let _ = status_tx.send(format!(
"[TRACE] {} header: {}: {}",
protocol_label, name, value
));
}
if !body_bytes.is_empty() {
if let Ok(body_str) = std::str::from_utf8(&body_bytes) {
if let Ok(json) = serde_json::from_str::<serde_json::Value>(body_str) {
let pretty = serde_json::to_string_pretty(&json).unwrap_or(body_str.to_string());
trace!("{} request body (JSON):\n{}", protocol_label, pretty);
let _ = status_tx.send(format!(
"[TRACE] {} request body (JSON):\r\n{}",
protocol_label,
pretty.replace('\n', "\r\n")
));
} else {
trace!("{} request body:\n{}", protocol_label, body_str);
let _ = status_tx.send(format!(
"[TRACE] {} request body:\r\n{}",
protocol_label,
body_str.replace('\n', "\r\n")
));
}
} else {
console_trace!(
status_tx,
"{} request body (binary): {} bytes",
protocol_label,
body_bytes.len()
);
}
}
Ok(RequestData {
method,
uri,
version,
headers,
body_bytes,
})
}
pub fn build_payload_too_large_response(
too_large: &BodyTooLarge,
protocol_label: &str,
status_tx: &mpsc::UnboundedSender<String>,
) -> Response<Full<Bytes>> {
warn!(
"{} {} {} decision=refused_body_too_large: over {} bytes",
protocol_label, too_large.method, too_large.uri, MAX_REQUEST_BODY_BYTES
);
let _ = status_tx.send(format!(
"→ {} {} {} → 413",
protocol_label, too_large.method, too_large.uri
));
build_safe_response(
413,
[("Content-Type".to_string(), "text/plain".to_string())],
format!(
"Payload Too Large: request bodies are limited to {} bytes\n",
MAX_REQUEST_BODY_BYTES
),
protocol_label,
)
}
pub fn build_safe_response(
status: u16,
headers: impl IntoIterator<Item = (String, String)>,
body: String,
context: &str,
) -> Response<Full<Bytes>> {
let status_code = match hyper::StatusCode::from_u16(status) {
Ok(code) => code,
Err(_) => {
error!(
"{}: invalid HTTP status {} (must be 100-599), sending 500 instead",
context, status
);
hyper::StatusCode::INTERNAL_SERVER_ERROR
}
};
let mut builder = Response::builder().status(status_code);
for (name, value) in headers {
match (
hyper::header::HeaderName::from_bytes(name.as_bytes()),
hyper::header::HeaderValue::from_str(&value),
) {
(Ok(n), Ok(v)) => builder = builder.header(n, v),
_ => warn!(
"{}: dropping invalid response header {:?} (name or value is not a legal HTTP header)",
context, name
),
}
}
match builder.body(Full::new(Bytes::from(body))) {
Ok(response) => response,
Err(e) => {
error!(
"{}: failed to build response ({}), sending bare 500",
context, e
);
let mut fallback =
Response::new(Full::new(Bytes::from_static(b"Internal Server Error")));
*fallback.status_mut() = hyper::StatusCode::INTERNAL_SERVER_ERROR;
fallback
}
}
}
const MAX_RESULT_NESTING: usize = 8;
fn for_each_output(results: &[ActionResult], visit: &mut impl FnMut(&[u8])) {
for_each_output_at(results, 0, visit)
}
fn for_each_output_at(results: &[ActionResult], depth: usize, visit: &mut impl FnMut(&[u8])) {
if depth >= MAX_RESULT_NESTING {
tracing::warn!(
"http_common: result nesting deeper than {MAX_RESULT_NESTING}, not descending \
further. An executor is producing nested Multiple results; the outputs found so \
far are still used."
);
return;
}
for result in results {
match result {
ActionResult::Output(data) => visit(data),
ActionResult::Multiple(inner) => for_each_output_at(inner, depth + 1, visit),
_ => {}
}
}
}
pub fn produced_http_response(results: &[ActionResult]) -> bool {
let mut produced = false;
for_each_output(results, &mut |data| {
if serde_json::from_slice::<serde_json::Value>(data).is_ok() {
produced = true;
}
});
produced
}
pub fn build_response(
protocol_results: Vec<ActionResult>,
protocol_label: &str,
method: &str,
uri: &str,
status_tx: &mpsc::UnboundedSender<String>,
default_response: Option<(u16, Vec<(String, String)>, String)>,
) -> Result<Response<Full<Bytes>>, Infallible> {
let mut status_code = 200;
let mut response_headers = HashMap::new();
let mut response_body = String::new();
let mut produced_response = false;
for_each_output(&protocol_results, &mut |output_data| {
if let Ok(json_value) = serde_json::from_slice::<serde_json::Value>(output_data) {
produced_response = true;
if let Some(status) = json_value.get("status").and_then(|v| v.as_u64()) {
status_code = u16::try_from(status).unwrap_or(500);
}
if let Some(headers_obj) = json_value.get("headers").and_then(|v| v.as_object()) {
for (k, v) in headers_obj {
if let Some(v_str) = v.as_str() {
response_headers.insert(k.clone(), v_str.to_string());
}
}
}
if let Some(body) = json_value.get("body").and_then(|v| v.as_str()) {
response_body = body.to_string();
}
}
});
if !produced_response {
if let Some((status, headers, body)) = default_response {
status_code = status;
response_headers = headers.into_iter().collect();
response_body = body;
}
}
let _ = status_tx.send(format!(
"→ {} {} {} → {} ({} bytes)",
protocol_label,
method,
uri,
status_code,
response_body.len()
));
Ok(build_safe_response(
status_code,
response_headers,
response_body,
protocol_label,
))
}
pub fn build_error_response(
error: anyhow::Error,
protocol_label: &str,
method: &str,
uri: &str,
status_tx: &mpsc::UnboundedSender<String>,
) -> Result<Response<Full<Bytes>>, Infallible> {
let failure = crate::utils::WireFailure::classify(&error);
let (status, decision) = match failure {
crate::utils::WireFailure::Overloaded => (503, "fail_closed_llm_overloaded"),
crate::utils::WireFailure::Unavailable => (500, "fail_closed_llm_error"),
};
error!(
"{} {} {} decision={} status={}: {}",
protocol_label, method, uri, decision, status, error
);
let _ = status_tx.send(format!(
"[ERROR] ✗ LLM error for {} {} → {} decision={}: {}",
method, uri, status, decision, error
));
if status == 503 {
return Ok(Response::builder()
.status(503)
.header(hyper::header::RETRY_AFTER, "1")
.body(Full::new(Bytes::from(failure.prefixed_text())))
.unwrap());
}
Ok(Response::builder()
.status(500)
.body(Full::new(Bytes::from(failure.prefixed_text())))
.unwrap())
}
#[derive(Debug, Clone)]
enum HeaderExpect {
Present,
Contains(String),
}
#[derive(Debug, Clone)]
struct HeaderMatch {
name: String,
expect: HeaderExpect,
}
#[derive(Debug)]
struct FilterRule {
methods: Option<Vec<String>>,
path: Option<regex::Regex>,
headers: Vec<HeaderMatch>,
}
impl FilterRule {
fn matches(&self, req: &RequestData, path: &str) -> bool {
if let Some(methods) = &self.methods {
let m = req.method.to_ascii_uppercase();
if !methods.iter().any(|allowed| allowed == &m) {
return false;
}
}
if let Some(re) = &self.path {
if !re.is_match(path) {
return false;
}
}
for hm in &self.headers {
let value = req
.headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case(&hm.name))
.map(|(_, v)| v);
match (&hm.expect, value) {
(HeaderExpect::Present, Some(_)) => {}
(HeaderExpect::Contains(needle), Some(v))
if v.to_ascii_lowercase()
.contains(&needle.to_ascii_lowercase()) => {}
_ => return false,
}
}
true
}
}
#[derive(Debug, Clone)]
struct FilteredResponse {
status: u16,
body: String,
headers: Vec<(String, String)>,
}
impl Default for FilteredResponse {
fn default() -> Self {
Self {
status: 404,
body: String::new(),
headers: Vec::new(),
}
}
}
#[derive(Debug)]
pub struct RequestFilter {
rules: Vec<FilterRule>,
filtered_response: FilteredResponse,
default_response: Option<FilteredResponse>,
warnings: Vec<String>,
}
impl RequestFilter {
pub fn from_startup_params(params: Option<&serde_json::Value>) -> Self {
let mut rules = Vec::new();
let mut warnings = Vec::new();
match params.and_then(|p| p.get("request_filter")) {
None | Some(serde_json::Value::Null) => {}
Some(serde_json::Value::Array(arr)) => {
for (i, raw) in arr.iter().enumerate() {
match Self::parse_rule(raw) {
Ok(rule) => rules.push(rule),
Err(e) => {
let msg =
format!("invalid request_filter rule #{}: {} — rule ignored", i, e);
error!("{}", msg);
warnings.push(msg);
}
}
}
if !arr.is_empty() && rules.is_empty() {
let msg = "request_filter was configured but EVERY rule was invalid — \
the filter is disabled and every request will reach the LLM \
(fail-open)"
.to_string();
error!("{}", msg);
warnings.push(msg);
}
}
Some(other) => {
let msg = format!(
"request_filter must be an array of rule objects, got {} — \
filter ignored, every request will reach the LLM (fail-open)",
kind_of(other)
);
error!("{}", msg);
warnings.push(msg);
}
}
let filtered_response = params
.and_then(|p| p.get("filtered_response"))
.map(|raw| Self::parse_filtered_response(raw, &mut warnings))
.unwrap_or_default();
if rules.is_empty() && params.and_then(|p| p.get("filtered_response")).is_some() {
let msg = "filtered_response is set but no valid request_filter rules exist — \
it will never be used"
.to_string();
warn!("{}", msg);
warnings.push(msg);
}
let default_response = params
.and_then(|p| p.get("default_response"))
.map(|raw| Self::parse_filtered_response(raw, &mut warnings));
Self {
rules,
filtered_response,
default_response,
warnings,
}
}
pub fn default_response_parts(&self) -> Option<(u16, Vec<(String, String)>, String)> {
self.default_response
.as_ref()
.map(|r| (r.status, r.headers.clone(), r.body.clone()))
}
pub fn warnings(&self) -> &[String] {
&self.warnings
}
fn parse_rule(raw: &serde_json::Value) -> Result<FilterRule, String> {
let obj = raw.as_object().ok_or("rule must be an object")?;
let methods = match obj.get("methods") {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::Array(arr)) => Some(
arr.iter()
.filter_map(|m| m.as_str().map(|s| s.to_ascii_uppercase()))
.collect::<Vec<_>>(),
),
Some(serde_json::Value::String(s)) => Some(vec![s.to_ascii_uppercase()]),
Some(_) => return Err("`methods` must be a string or array of strings".into()),
};
let path = match obj.get("path") {
None | Some(serde_json::Value::Null) => None,
Some(serde_json::Value::String(pattern)) => Some(
regex::Regex::new(pattern)
.map_err(|e| format!("invalid `path` regex {:?}: {}", pattern, e))?,
),
Some(_) => return Err("`path` must be a regex string".into()),
};
let mut headers = Vec::new();
if let Some(hdrs) = obj.get("headers") {
let map = hdrs.as_object().ok_or("`headers` must be an object")?;
for (name, matcher) in map {
let expect = match matcher {
serde_json::Value::Bool(true) => HeaderExpect::Present,
serde_json::Value::Bool(false) => continue, serde_json::Value::String(s) => HeaderExpect::Contains(s.clone()),
_ => {
return Err(format!(
"header `{}` matcher must be true or a string",
name
))
}
};
headers.push(HeaderMatch {
name: name.to_ascii_lowercase(),
expect,
});
}
}
if methods.is_none() && path.is_none() && headers.is_empty() {
debug!("request_filter rule has no conditions — matches all requests");
}
Ok(FilterRule {
methods,
path,
headers,
})
}
fn parse_filtered_response(
raw: &serde_json::Value,
warnings: &mut Vec<String>,
) -> FilteredResponse {
let mut resp = FilteredResponse::default();
if let Some(status) = raw.get("status") {
match status.as_u64() {
Some(s) if (100..=599).contains(&s) => resp.status = s as u16,
_ => {
let msg = format!(
"filtered_response.status {} is not a valid HTTP status (100-599) — using {}",
status, resp.status
);
error!("{}", msg);
warnings.push(msg);
}
}
}
if let Some(body) = raw.get("body").and_then(|v| v.as_str()) {
resp.body = body.to_string();
}
if let Some(headers) = raw.get("headers").and_then(|v| v.as_object()) {
resp.headers = headers
.iter()
.filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
.collect();
}
resp
}
pub fn is_pass_through(&self) -> bool {
self.rules.is_empty()
}
pub fn allows(&self, req: &RequestData, path: &str) -> bool {
self.is_pass_through() || self.rules.iter().any(|r| r.matches(req, path))
}
pub fn rejection_parts(&self) -> (u16, Vec<(String, String)>, String) {
(
self.filtered_response.status,
self.filtered_response.headers.clone(),
self.filtered_response.body.clone(),
)
}
pub fn rejection(&self) -> Response<Full<Bytes>> {
build_safe_response(
self.filtered_response.status,
self.filtered_response.headers.iter().cloned(),
self.filtered_response.body.clone(),
"filtered_response",
)
}
}
fn kind_of(v: &serde_json::Value) -> &'static str {
match v {
serde_json::Value::Null => "null",
serde_json::Value::Bool(_) => "a boolean",
serde_json::Value::Number(_) => "a number",
serde_json::Value::String(_) => "a string",
serde_json::Value::Array(_) => "an array",
serde_json::Value::Object(_) => "an object",
}
}
pub fn request_handling_startup_parameters() -> Vec<crate::llm::actions::ParameterDefinition> {
use crate::llm::actions::ParameterDefinition;
vec![
ParameterDefinition {
name: "request_filter".to_string(),
type_hint: "array".to_string(),
description: "RECOMMENDED for any HTTP server a browser or the public internet will \
reach: set this so you do not spend a slow LLM call on favicon.ico, CORS \
preflights, and scanner noise. A good default that only reasons about real page \
loads is [{\"methods\":[\"GET\"],\"headers\":{\"accept\":\"text/html\"}}] — favicon \
and image probes carry `Accept: image/*`, and preflights are not GET, so they are \
auto-answered without a model call. \
Allowlist of request-match rules deciding which requests reach the LLM. \
A request is handled by the LLM only if it matches at least one rule; requests \
matching no rule get `filtered_response` (default 404) with NO LLM call. Omit this \
to send every request to the LLM. Each rule is an object; all present conditions \
must hold (AND), and rules are OR'd together. Conditions: `methods` (array of HTTP \
methods, case-insensitive; omit for any), `path` (a regular expression matched \
against the URL path, e.g. \"^/$\" or \"^/api/\"; omit for any), `headers` (object \
mapping header name to either true = must be present, or a string = value must \
contain that substring case-insensitively, e.g. {\"accept\": \"text/html\"}). \
Example filter that only sends real browser page loads to the LLM (favicon/OPTIONS/\
XHR are auto-404'd): [{\"methods\":[\"GET\"],\"headers\":{\"accept\":\"text/html\"}}]. \
Fail-open: a malformed rule (e.g. an invalid `path` regex) is dropped with a loud \
error instead of failing startup, so a typo means MORE requests reach the LLM, not \
fewer. Check the server status output after starting."
.to_string(),
required: false,
example: serde_json::json!([
{ "methods": ["GET"], "path": "^/$", "headers": { "accept": "text/html" } },
{ "methods": ["POST"], "path": "^/api/" }
]),
default: None,
},
ParameterDefinition {
name: "filtered_response".to_string(),
type_hint: "object".to_string(),
description: "Response returned (without any LLM call) for requests that match no \
`request_filter` rule. Fields: `status` (number, default 404), `body` (string, \
default empty), `headers` (object of name→value). Ignored when no request_filter \
is set."
.to_string(),
required: false,
example: serde_json::json!({ "status": 404, "body": "Not Found" }),
default: None,
},
ParameterDefinition {
name: "default_response".to_string(),
type_hint: "object".to_string(),
description: "Response used when the model IS consulted for a request but returns no \
send_http_response (an empty answer). Without this, such a request gets a blank \
200, which looks broken. Set a sensible fallback — commonly a 404 page — so a \
declined request looks intentional. Same shape as filtered_response: `status` \
(number), `body` (string), `headers` (object of name→value)."
.to_string(),
required: false,
example: serde_json::json!({
"status": 404,
"headers": { "Content-Type": "text/html" },
"body": "<!doctype html><title>Not Found</title><h1>404</h1>"
}),
default: None,
},
]
}