use std::collections::BTreeMap; use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest}; use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest}; use crank_adapter_rest::{RestAdapter, RestRequest, RestWindowRequest}; use crank_adapter_soap::{SoapAdapter, SoapRequest}; use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest}; use crank_core::{ExecutionMode, Target, TransportBehavior}; use serde_json::{Map, Value, json}; use tracing::debug; use crate::{ AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeOperation, RuntimeRequestContext, WindowExecutionResult, }; #[derive(Clone, Debug)] pub struct RuntimeExecutor { graphql_adapter: GraphqlAdapter, grpc_adapter: GrpcAdapter, rest_adapter: RestAdapter, soap_adapter: SoapAdapter, websocket_adapter: WebsocketAdapter, } impl Default for RuntimeExecutor { fn default() -> Self { Self::new() } } impl RuntimeExecutor { pub fn new() -> Self { Self { graphql_adapter: GraphqlAdapter::new(), grpc_adapter: GrpcAdapter::new(), rest_adapter: RestAdapter::new(), soap_adapter: SoapAdapter::new(), websocket_adapter: WebsocketAdapter::new(), } } pub async fn execute( &self, operation: &RuntimeOperation, input: &Value, ) -> Result { self.execute_with_auth_and_context(operation, input, None, None) .await } pub async fn execute_with_auth( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, ) -> Result { self.execute_with_auth_and_context(operation, input, resolved_auth, None) .await } pub async fn execute_with_context( &self, operation: &RuntimeOperation, input: &Value, request_context: Option<&RuntimeRequestContext>, ) -> Result { self.execute_with_auth_and_context(operation, input, None, request_context) .await } pub async fn execute_with_auth_and_context( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("unary.execute", operation, request_context); let prepared_request = self.prepare_request(operation, input)?; let prepared_request = apply_resolved_auth(prepared_request, resolved_auth); self.execute_prepared(operation, prepared_request, request_context) .await } pub async fn execute_window( &self, operation: &RuntimeOperation, input: &Value, ) -> Result { self.execute_window_with_auth_and_context(operation, input, None, None) .await } pub async fn execute_window_with_auth( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, ) -> Result { self.execute_window_with_auth_and_context(operation, input, resolved_auth, None) .await } pub async fn execute_window_with_context( &self, operation: &RuntimeOperation, input: &Value, request_context: Option<&RuntimeRequestContext>, ) -> Result { self.execute_window_with_auth_and_context(operation, input, None, request_context) .await } pub async fn execute_window_with_auth_and_context( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("window.execute", operation, request_context); let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), }); }; if streaming.mode != ExecutionMode::Window { return Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: streaming.mode, }); } let prepared_request = self.prepare_request(operation, input)?; let prepared_request = apply_resolved_auth(prepared_request, resolved_auth); let adapter_response = if matches!( streaming.transport_behavior, TransportBehavior::ServerStream ) && matches!( operation.target, Target::Rest(_) | Target::Grpc(_) | Target::Websocket(_) ) { self.execute_window_adapter(operation, prepared_request, request_context) .await? } else { self.execute_adapter(operation, prepared_request, request_context) .await? }; crate::aggregation::collect_window_result(&adapter_response.body, streaming) } pub async fn execute_session_seed( &self, operation: &RuntimeOperation, input: &Value, ) -> Result { self.execute_session_seed_with_auth_and_context(operation, input, None, None) .await } pub async fn execute_session_seed_with_auth( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, ) -> Result { self.execute_session_seed_with_auth_and_context(operation, input, resolved_auth, None) .await } pub async fn execute_session_seed_with_context( &self, operation: &RuntimeOperation, input: &Value, request_context: Option<&RuntimeRequestContext>, ) -> Result { self.execute_session_seed_with_auth_and_context(operation, input, None, request_context) .await } pub async fn execute_session_seed_with_auth_and_context( &self, operation: &RuntimeOperation, input: &Value, resolved_auth: Option<&ResolvedAuth>, request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("session.seed", operation, request_context); let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), }); }; if streaming.mode != ExecutionMode::Session { return Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: streaming.mode, }); } let batch_size = streaming.max_items.unwrap_or(10).max(1); let seed_limit = batch_size.saturating_mul(4); let mut seeded_operation = operation.clone(); if let Some(config) = seeded_operation.execution_config.streaming.as_mut() { config.mode = ExecutionMode::Window; config.window_duration_ms = config .window_duration_ms .or(config.poll_interval_ms) .or(config.upstream_timeout_ms) .or(Some(operation.execution_config.timeout_ms)); config.max_items = Some(seed_limit); } self.execute_window_with_auth_and_context( &seeded_operation, input, resolved_auth, request_context, ) .await } pub fn prepare_request( &self, operation: &RuntimeOperation, input: &Value, ) -> Result { operation.input_schema.validate_shape(input)?; let mapped_input = operation.input_mapping.apply(&json!({ "mcp": input }))?; PreparedRequest::from_mapping_output(&mapped_input) } async fn execute_prepared( &self, operation: &RuntimeOperation, prepared_request: PreparedRequest, request_context: Option<&RuntimeRequestContext>, ) -> Result { let adapter_response = self .execute_adapter(operation, prepared_request, request_context) .await?; let finalized_output = finalize_output(operation, &adapter_response)?; operation.output_schema.validate_shape(&finalized_output)?; Ok(finalized_output) } async fn execute_adapter( &self, operation: &RuntimeOperation, prepared_request: PreparedRequest, request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("adapter.dispatch", operation, request_context); let context_headers = runtime_context_headers(request_context); match &operation.target { Target::Grpc(target) => { let request = GrpcRequest { headers: merge_headers( &BTreeMap::new(), &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), body: prepared_request .grpc .clone() .unwrap_or(Value::Object(Map::new())), timeout_ms: operation.execution_config.timeout_ms, }; let response = self.grpc_adapter.execute(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body.clone(), data: response.body, }) } Target::Graphql(target) => { let request = GraphqlRequest { headers: merge_headers( &BTreeMap::new(), &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), variables: prepared_request.variables.clone(), timeout_ms: operation.execution_config.timeout_ms, }; let response = self.graphql_adapter.execute(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: response.data, }) } Target::Rest(target) => { let request = RestRequest { path_params: prepared_request.path_params.clone(), query_params: prepared_request.query_params.clone(), headers: merge_headers( &target.static_headers, &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), body: prepared_request.body.clone(), timeout_ms: operation.execution_config.timeout_ms, }; let response = self.rest_adapter.execute(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, }) } Target::Soap(target) => { let request = SoapRequest { headers: merge_headers( &BTreeMap::new(), &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), body: prepared_request .body .clone() .unwrap_or(Value::Object(Map::new())), timeout_ms: operation.execution_config.timeout_ms, }; let response = self.soap_adapter.execute(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, }) } Target::Websocket(_) => Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: ExecutionMode::Unary, }), } } async fn execute_window_adapter( &self, operation: &RuntimeOperation, prepared_request: PreparedRequest, request_context: Option<&RuntimeRequestContext>, ) -> Result { log_runtime_event("window.adapter.dispatch", operation, request_context); let context_headers = runtime_context_headers(request_context); match &operation.target { Target::Rest(target) => { let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), }); }; let request = RestWindowRequest { request: RestRequest { path_params: prepared_request.path_params.clone(), query_params: prepared_request.query_params.clone(), headers: merge_headers( &target.static_headers, &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), body: prepared_request.body.clone(), timeout_ms: streaming .upstream_timeout_ms .unwrap_or(operation.execution_config.timeout_ms), }, window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), max_items: streaming.max_items.map(|value| value.saturating_add(1)), }; let response = self.rest_adapter.execute_window(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, }) } Target::Grpc(target) => { let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), }); }; let request = GrpcWindowRequest { request: GrpcRequest { headers: merge_headers( &BTreeMap::new(), &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), body: prepared_request .grpc .clone() .unwrap_or(Value::Object(Map::new())), timeout_ms: streaming .upstream_timeout_ms .unwrap_or(operation.execution_config.timeout_ms), }, window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), max_items: streaming.max_items.map(|value| value.saturating_add(1)), }; let response = self.grpc_adapter.execute_window(target, &request).await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, }) } Target::Websocket(target) => { let Some(streaming) = operation.execution_config.streaming.as_ref() else { return Err(RuntimeError::MissingStreamingConfig { operation_id: operation.operation_id.as_str().to_owned(), }); }; let websocket_options = operation .execution_config .protocol_options .as_ref() .and_then(|options| options.websocket.as_ref()); let request = WebsocketWindowRequest { headers: merge_headers( &BTreeMap::new(), &operation.execution_config.headers, &prepared_request.headers, &context_headers, ), window_duration_ms: streaming.window_duration_ms.unwrap_or_default(), max_items: streaming.max_items.map(|value| value.saturating_add(1)), heartbeat_interval_ms: websocket_options .and_then(|options| options.heartbeat_interval_ms), reconnect_max_attempts: websocket_options .and_then(|options| options.reconnect_max_attempts) .unwrap_or_default(), reconnect_backoff_ms: websocket_options .and_then(|options| options.reconnect_backoff_ms) .unwrap_or_default(), }; let response = self .websocket_adapter .execute_window(target, &request) .await?; Ok(AdapterResponse { status_code: response.status_code, headers: response.headers, body: response.body, data: Value::Null, }) } Target::Soap(_) => Err(RuntimeError::UnsupportedExecutionMode { operation_id: operation.operation_id.as_str().to_owned(), mode: ExecutionMode::Window, }), _ => { self.execute_adapter(operation, prepared_request, request_context) .await } } } } impl PreparedRequest { pub fn from_mapping_output(mapped: &Value) -> Result { let request = mapped .get("request") .ok_or_else(|| RuntimeError::InvalidPreparedRequest { field: "request".to_owned(), reason: "mapped input must contain request root".to_owned(), })?; Ok(Self { path_params: read_string_map(request.get("path"), "request.path")?, query_params: read_string_map(request.get("query"), "request.query")?, headers: read_string_map(request.get("headers"), "request.headers")?, grpc: non_empty_payload(request.get("grpc").cloned()), variables: non_empty_payload(request.get("variables").cloned()), body: request.get("body").and_then(non_empty_body).cloned(), }) } } fn finalize_output( operation: &RuntimeOperation, response: &AdapterResponse, ) -> Result { let mapped = operation.output_mapping.apply(&json!({ "response": { "body": response.body, "data": response.data, "headers": response.headers, "status": response.status_code } }))?; Ok(mapped .get("output") .cloned() .unwrap_or_else(|| Value::Object(Map::new()))) } fn apply_resolved_auth( mut prepared_request: PreparedRequest, resolved_auth: Option<&ResolvedAuth>, ) -> PreparedRequest { if let Some(resolved_auth) = resolved_auth { resolved_auth.apply(&mut prepared_request); } prepared_request } fn merge_headers( static_headers: &BTreeMap, execution_headers: &BTreeMap, request_headers: &BTreeMap, context_headers: &BTreeMap, ) -> BTreeMap { let mut headers = static_headers.clone(); headers.extend(execution_headers.clone()); headers.extend(request_headers.clone()); headers.extend(context_headers.clone()); headers } fn runtime_context_headers( request_context: Option<&RuntimeRequestContext>, ) -> BTreeMap { request_context .map(RuntimeRequestContext::outbound_headers) .unwrap_or_default() } fn log_runtime_event( stage: &'static str, operation: &RuntimeOperation, request_context: Option<&RuntimeRequestContext>, ) { let request_id = request_context .map(|context| context.request_id.as_str()) .unwrap_or_default(); let correlation_id = request_context .map(|context| context.correlation_id.as_str()) .unwrap_or_default(); debug!( stage, operation_id = %operation.operation_id, protocol = ?operation.protocol, request_id, correlation_id, "runtime execution" ); } fn read_string_map( value: Option<&Value>, field_name: &str, ) -> Result, RuntimeError> { let Some(value) = value else { return Ok(BTreeMap::new()); }; let Some(object) = value.as_object() else { return Err(RuntimeError::InvalidPreparedRequest { field: field_name.to_owned(), reason: "must be an object".to_owned(), }); }; object .iter() .map(|(key, value)| stringify_value(value).map(|value| (key.clone(), value))) .collect::, _>>() .map_err(|reason| RuntimeError::InvalidPreparedRequest { field: field_name.to_owned(), reason, }) } fn stringify_value(value: &Value) -> Result { match value { Value::Null => Ok("null".to_owned()), Value::Bool(value) => Ok(value.to_string()), Value::Number(value) => Ok(value.to_string()), Value::String(value) => Ok(value.clone()), Value::Array(_) | Value::Object(_) => { Err("request path/query/headers accept only scalar values".to_owned()) } } } fn non_empty_body(value: &Value) -> Option<&Value> { match value { Value::Null => None, Value::Object(object) if object.is_empty() => None, _ => Some(value), } } fn non_empty_payload(value: Option) -> Option { match value { None | Some(Value::Null) => None, Some(Value::Object(object)) if object.is_empty() => None, other => other, } } #[cfg(test)] mod tests { use std::collections::BTreeMap; use std::io; use std::sync::{Arc, Mutex}; use axum::{ Json, Router, http::HeaderMap, response::sse::{Event, KeepAlive, Sse}, routing::post, }; use crank_adapter_grpc::test_support as grpc_test_support; use crank_core::{ AggregationMode, DescriptorId, ExecutionConfig, ExecutionMode, GeneratedDraft, GeneratedDraftStatus, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod, Operation, OperationId, OperationStatus, Protocol, ProtocolOptions, RestTarget, Samples, SoapBindingStyle, SoapOperationMetadata, SoapTarget, SoapVersion, StreamingConfig, Target, ToolDescription, ToolExample, ToolFamilyConfig, TransportBehavior, WebsocketProtocolOptions, WebsocketTarget, }; use crank_mapping::{MappingRule, MappingSet}; use crank_schema::{Schema, SchemaKind}; use futures_util::{SinkExt, StreamExt, stream}; use serde_json::{Value, json}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::net::TcpListener; use tokio_tungstenite::{ accept_async, accept_hdr_async, tungstenite::handshake::server::{Request, Response}, }; use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*}; use crate::{ PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation, RuntimeRequestContext, }; fn timestamp(value: &str) -> OffsetDateTime { OffsetDateTime::parse(value, &Rfc3339).unwrap() } #[derive(Clone, Default)] struct SharedLogWriter { buffer: Arc>>, } impl SharedLogWriter { fn output(&self) -> String { String::from_utf8(self.buffer.lock().unwrap().clone()).unwrap() } } impl<'a> MakeWriter<'a> for SharedLogWriter { type Writer = SharedLogGuard; fn make_writer(&'a self) -> Self::Writer { SharedLogGuard { buffer: Arc::clone(&self.buffer), } } } struct SharedLogGuard { buffer: Arc>>, } impl io::Write for SharedLogGuard { fn write(&mut self, bytes: &[u8]) -> io::Result { self.buffer.lock().unwrap().extend_from_slice(bytes); Ok(bytes.len()) } fn flush(&mut self) -> io::Result<()> { Ok(()) } } #[tokio::test] async fn executes_rest_operation_end_to_end() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_rest_operation(&base_url, false, false); let output = executor .execute(&operation, &json!({ "email": "user@example.com" })) .await .unwrap(); assert_eq!(output, json!({ "id": "lead_123" })); } #[tokio::test] async fn propagates_request_context_to_rest_headers() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let mut operation = test_rest_operation(&base_url, false, false); let context = RuntimeRequestContext::new("req_rest_123", "corr_rest_123"); if let Target::Rest(target) = &mut operation.target { target.path_template = "/leads-context".to_owned(); } let output = executor .execute_with_context( &operation, &json!({ "email": "user@example.com" }), Some(&context), ) .await .unwrap(); assert_eq!(output, json!({ "id": "req_rest_123|corr_rest_123" })); } #[tokio::test] async fn emits_runtime_tracing_with_request_context() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let mut operation = test_rest_operation(&base_url, false, false); let context = RuntimeRequestContext::new("req_log_123", "corr_log_123"); let writer = SharedLogWriter::default(); let subscriber = tracing_subscriber::registry().with( tracing_subscriber::fmt::layer() .with_writer(writer.clone()) .without_time() .with_ansi(false) .with_target(false) .compact() .with_filter(LevelFilter::DEBUG), ); let dispatch = tracing::Dispatch::new(subscriber); if let Target::Rest(target) = &mut operation.target { target.path_template = "/leads-context".to_owned(); } let _guard = tracing::dispatcher::set_default(&dispatch); let output = executor .execute_with_context( &operation, &json!({ "email": "user@example.com" }), Some(&context), ) .await .unwrap(); assert_eq!(output, json!({ "id": "req_log_123|corr_log_123" })); let logs = writer.output(); assert!(logs.contains("runtime execution")); assert!(logs.contains("unary.execute")); assert!(logs.contains("adapter.dispatch")); assert!(logs.contains("req_log_123")); assert!(logs.contains("corr_log_123")); assert!(logs.contains("op_rest_runtime")); } #[tokio::test] async fn executes_graphql_operation_end_to_end() { let endpoint = spawn_graphql_server().await; let executor = RuntimeExecutor::new(); let operation = test_graphql_operation(&endpoint, false); let output = executor .execute(&operation, &json!({ "email": "user@example.com" })) .await .unwrap(); assert_eq!(output, json!({ "id": "lead_123" })); } #[tokio::test] async fn propagates_request_context_to_graphql_headers() { let endpoint = spawn_graphql_context_server().await; let executor = RuntimeExecutor::new(); let operation = test_graphql_operation(&endpoint, false); let context = RuntimeRequestContext::new("req_graphql_123", "corr_graphql_123"); let output = executor .execute_with_context( &operation, &json!({ "email": "user@example.com" }), Some(&context), ) .await .unwrap(); assert_eq!(output, json!({ "id": "req_graphql_123|corr_graphql_123" })); } #[tokio::test] async fn executes_grpc_operation_end_to_end() { let server_addr = grpc_test_support::spawn_unary_echo_server().await; let executor = RuntimeExecutor::new(); let operation = test_grpc_operation(&server_addr); let output = executor .execute(&operation, &json!({ "message": "hello" })) .await .unwrap(); assert_eq!(output, json!({ "message": "hello" })); } #[tokio::test] async fn propagates_request_context_to_grpc_metadata() { let server_addr = grpc_test_support::spawn_metadata_echo_server().await; let executor = RuntimeExecutor::new(); let operation = test_grpc_operation(&server_addr); let context = RuntimeRequestContext::new("req_grpc_123", "corr_grpc_123"); let output = executor .execute_with_context(&operation, &json!({ "message": "hello" }), Some(&context)) .await .unwrap(); assert_eq!( output, json!({ "message": "hello|req_grpc_123|corr_grpc_123" }) ); } #[tokio::test] async fn executes_soap_operation_end_to_end() { let endpoint = spawn_soap_server().await; let executor = RuntimeExecutor::new(); let operation = test_soap_operation(&endpoint); let output = executor .execute(&operation, &json!({ "email": "user@example.com" })) .await .unwrap(); assert_eq!(output, json!({ "id": "lead_123" })); } #[tokio::test] async fn propagates_request_context_to_soap_headers() { let endpoint = spawn_soap_context_server().await; let executor = RuntimeExecutor::new(); let operation = test_soap_operation(&endpoint); let context = RuntimeRequestContext::new("req_soap_123", "corr_soap_123"); let output = executor .execute_with_context( &operation, &json!({ "email": "user@example.com" }), Some(&context), ) .await .unwrap(); assert_eq!(output, json!({ "id": "req_soap_123|corr_soap_123" })); } #[tokio::test] async fn executes_grpc_window_mode_with_server_stream() { let server_addr = grpc_test_support::spawn_unary_echo_server().await; let executor = RuntimeExecutor::new(); let operation = test_grpc_window_operation(&server_addr, AggregationMode::RawItems, Some(2)); let result = executor .execute_window(&operation, &json!({ "message": "hello" })) .await .unwrap(); assert_eq!(result.summary, Value::Null); assert_eq!(result.items.len(), 2); assert_eq!(result.items[0], json!({ "message": "hello-one" })); assert!(result.truncated); assert!(result.has_more); assert!(!result.window_complete); } #[tokio::test] async fn executes_websocket_window_mode_end_to_end() { let target_url = spawn_websocket_server().await; let executor = RuntimeExecutor::new(); let operation = test_websocket_window_operation(&target_url, AggregationMode::RawItems, Some(3)); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert_eq!(result.items.len(), 3); assert_eq!(result.items[0], json!({ "seq": 1, "value": 101 })); assert_eq!(result.items[2], json!({ "seq": 3, "value": 103 })); assert!(!result.window_complete); assert!(result.truncated); assert!(result.has_more); } #[tokio::test] async fn propagates_request_context_to_websocket_headers() { let target_url = spawn_request_context_websocket_server().await; let executor = RuntimeExecutor::new(); let operation = test_websocket_window_operation(&target_url, AggregationMode::RawItems, Some(2)); let context = RuntimeRequestContext::new("req_ws_123", "corr_ws_123"); let result = executor .execute_window_with_context(&operation, &json!({}), Some(&context)) .await .unwrap(); assert_eq!(result.items.len(), 2); assert_eq!(result.items[0], json!({ "seq": 1, "value": 101 })); } #[tokio::test] async fn rejects_invalid_input_shape() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_rest_operation(&base_url, false, false); let error = executor.execute(&operation, &json!({})).await.unwrap_err(); assert!(matches!(error, RuntimeError::Schema(_))); } #[test] fn rejects_missing_request_root_with_structured_error() { let error = PreparedRequest::from_mapping_output(&json!({})).unwrap_err(); match error { RuntimeError::InvalidPreparedRequest { field, reason } => { assert_eq!(field, "request"); assert_eq!(reason, "mapped input must contain request root"); } other => panic!("unexpected error: {other}"), } } #[tokio::test] async fn propagates_external_rest_errors() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_rest_operation(&base_url, true, false); let error = executor .execute(&operation, &json!({ "email": "user@example.com" })) .await .unwrap_err(); assert!(matches!(error, RuntimeError::RestAdapter(_))); } #[tokio::test] async fn fails_when_output_mapping_requires_missing_field() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_rest_operation(&base_url, false, true); let error = executor .execute(&operation, &json!({ "email": "user@example.com" })) .await .unwrap_err(); assert!(matches!(error, RuntimeError::Mapping(_))); } #[tokio::test] async fn propagates_graphql_operation_errors() { let endpoint = spawn_graphql_server().await; let executor = RuntimeExecutor::new(); let operation = test_graphql_operation(&endpoint, true); let error = executor .execute(&operation, &json!({ "email": "fail@example.com" })) .await .unwrap_err(); assert!(matches!(error, RuntimeError::GraphqlAdapter(_))); } #[tokio::test] async fn executes_window_mode_with_raw_items() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_window_snapshot_operation(&base_url, AggregationMode::RawItems, None, None); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert_eq!(result.items.len(), 3); assert_eq!(result.summary, Value::Null); assert!(result.window_complete); assert!(!result.truncated); } #[tokio::test] async fn executes_window_mode_with_summary_only() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_window_snapshot_operation(&base_url, AggregationMode::SummaryOnly, Some(10), None); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert_eq!(result.summary, json!({ "service": "billing", "errors": 1 })); assert!(result.items.is_empty()); } #[tokio::test] async fn executes_window_mode_with_truncation_and_redaction() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_window_snapshot_operation( &base_url, AggregationMode::SummaryPlusSamples, Some(2), Some(120), ); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert!(result.truncated); assert!(result.has_more); assert_eq!(result.items.len(), 1); assert_eq!(result.items[0]["secret"], json!("[REDACTED]")); assert_eq!(result.items[0]["message"], json!("disk...")); } #[tokio::test] async fn propagates_timeout_in_window_mode() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_slow_window_snapshot_operation(&base_url); let error = executor .execute_window(&operation, &json!({})) .await .unwrap_err(); assert!(matches!(error, RuntimeError::RestAdapter(_))); } #[tokio::test] async fn executes_window_mode_with_rest_sse_stream() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_window_sse_operation(&base_url, AggregationMode::RawItems, Some(2)); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert_eq!(result.items.len(), 2); assert_eq!(result.summary, Value::Null); assert!(result.truncated); assert!(result.has_more); assert!(!result.window_complete); } #[tokio::test] async fn completes_empty_window_for_idle_rest_sse_stream() { let base_url = spawn_runtime_server().await; let executor = RuntimeExecutor::new(); let operation = test_slow_window_sse_operation(&base_url); let result = executor .execute_window(&operation, &json!({})) .await .unwrap(); assert!(result.items.is_empty()); assert!(result.window_complete); assert!(!result.truncated); assert!(!result.has_more); } async fn spawn_runtime_server() -> String { let app = Router::new() .route("/leads", post(create_lead)) .route("/leads-context", post(create_lead_with_context)) .route("/events", post(events_window)) .route("/slow-events", post(slow_events_window)) .route("/events-sse", post(events_stream)) .route("/slow-events-sse", post(slow_events_stream)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); format!("http://{}", address) } async fn spawn_graphql_server() -> String { let app = Router::new().route("/", post(graphql_handler)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); format!("http://{}", address) } async fn spawn_graphql_context_server() -> String { let app = Router::new().route("/", post(graphql_context_handler)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); format!("http://{}", address) } async fn spawn_soap_server() -> String { let app = Router::new().route("/", post(soap_handler)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); format!("http://{}", address) } async fn spawn_soap_context_server() -> String { let app = Router::new().route("/", post(soap_context_handler)); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); format!("http://{}", address) } async fn spawn_websocket_server() -> String { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { loop { let (stream, _) = listener.accept().await.unwrap(); tokio::spawn(async move { let websocket = accept_async(stream).await.unwrap(); let (mut sink, mut source) = websocket.split(); if let Some(message) = source.next().await { let message = message.unwrap(); if !matches!(message, tokio_tungstenite::tungstenite::Message::Text(_)) { return; } } for payload in [ json!({ "seq": 1, "value": 101 }), json!({ "seq": 2, "value": 102 }), json!({ "seq": 3, "value": 103 }), json!({ "seq": 4, "value": 104 }), ] { sink.send(tokio_tungstenite::tungstenite::Message::Text( payload.to_string().into(), )) .await .unwrap(); } let _ = sink.close().await; }); } }); format!("ws://{}", address) } async fn spawn_request_context_websocket_server() -> String { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); tokio::spawn(async move { loop { let (stream, _) = listener.accept().await.unwrap(); tokio::spawn(async move { let websocket = accept_hdr_async(stream, |request: &Request, response: Response| { assert_eq!( request .headers() .get("x-request-id") .and_then(|value| value.to_str().ok()), Some("req_ws_123") ); assert_eq!( request .headers() .get("x-correlation-id") .and_then(|value| value.to_str().ok()), Some("corr_ws_123") ); Ok(response) }) .await .unwrap(); let (mut sink, mut source) = websocket.split(); if let Some(message) = source.next().await { let message = message.unwrap(); if !matches!(message, tokio_tungstenite::tungstenite::Message::Text(_)) { return; } } for payload in [ json!({ "seq": 1, "value": 101 }), json!({ "seq": 2, "value": 102 }), ] { sink.send(tokio_tungstenite::tungstenite::Message::Text( payload.to_string().into(), )) .await .unwrap(); } let _ = sink.close().await; }); } }); format!("ws://{}", address) } async fn create_lead(Json(payload): Json) -> (axum::http::StatusCode, Json) { let should_fail = payload .get("fail") .and_then(Value::as_bool) .unwrap_or(false); if should_fail { return ( axum::http::StatusCode::BAD_GATEWAY, Json(json!({ "error": "upstream failed" })), ); } ( axum::http::StatusCode::OK, Json(json!({ "id": "lead_123", "email": payload["email"] })), ) } async fn create_lead_with_context( headers: HeaderMap, Json(_payload): Json, ) -> (axum::http::StatusCode, Json) { ( axum::http::StatusCode::OK, Json(json!({ "id": format!( "{}|{}", headers .get("x-request-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default(), headers .get("x-correlation-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default() ), })), ) } async fn events_window() -> (axum::http::StatusCode, Json) { ( axum::http::StatusCode::OK, Json(json!({ "summary": { "service": "billing", "errors": 1 }, "items": [ { "message": "disk pressure detected", "secret": "token-a" }, { "message": "error budget exhausted", "secret": "token-b" }, { "message": "node restarted", "secret": "token-c" } ], "cursor": "cursor_02", "done": true })), ) } async fn slow_events_window() -> (axum::http::StatusCode, Json) { tokio::time::sleep(std::time::Duration::from_millis(100)).await; events_window().await } async fn events_stream() -> Sse>> { let events = vec![ Ok(Event::default() .data("{\"message\":\"disk pressure detected\",\"secret\":\"token-a\"}")), Ok(Event::default() .data("{\"message\":\"error budget exhausted\",\"secret\":\"token-b\"}")), Ok(Event::default().data("{\"message\":\"node restarted\",\"secret\":\"token-c\"}")), ]; Sse::new(stream::iter(events)).keep_alive(KeepAlive::default()) } async fn slow_events_stream() -> Sse>> { let delayed = stream::once(async { tokio::time::sleep(std::time::Duration::from_millis(100)).await; Ok(Event::default().data("{\"message\":\"late event\"}")) }); Sse::new(delayed).keep_alive(KeepAlive::default()) } async fn graphql_handler(Json(payload): Json) -> Json { let email = payload .get("variables") .and_then(|variables| variables.get("email")) .and_then(Value::as_str) .unwrap_or_default(); if email == "fail@example.com" { return Json(json!({ "data": { "createLead": null }, "errors": [{ "message": "lead creation failed" }] })); } Json(json!({ "data": { "createLead": { "id": "lead_123", "status": "created" } } })) } async fn graphql_context_handler( headers: HeaderMap, Json(_payload): Json, ) -> Json { Json(json!({ "data": { "createLead": { "id": format!( "{}|{}", headers .get("x-request-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default(), headers .get("x-correlation-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default() ), "status": "created" } } })) } async fn soap_handler(body: String) -> (axum::http::StatusCode, String) { assert!(body.contains("user@example.com")); ( axum::http::StatusCode::OK, r#" lead_123 created "# .to_owned(), ) } async fn soap_context_handler( headers: HeaderMap, body: String, ) -> (axum::http::StatusCode, String) { assert!(body.contains("user@example.com")); let request_id = headers .get("x-request-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default(); let correlation_id = headers .get("x-correlation-id") .and_then(|value| value.to_str().ok()) .unwrap_or_default(); ( axum::http::StatusCode::OK, format!( r#" {request_id}|{correlation_id} created "# ), ) } fn test_rest_operation( base_url: &str, should_fail: bool, invalid_output: bool, ) -> RuntimeOperation { let mut input_rules = vec![MappingRule { source: "$.mcp.email".to_owned(), target: "$.request.body.email".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }]; if should_fail { input_rules.push(MappingRule { source: "$.mcp.fail".to_owned(), target: "$.request.body.fail".to_owned(), required: false, default_value: Some(Value::Bool(true)), transform: None, condition: None, notes: None, }); } let output_source = if invalid_output { "$.response.body.missing" } else { "$.response.body.id" }; RuntimeOperation::from(Operation { id: OperationId::new("op_rest_runtime"), name: "crm_create_lead".to_owned(), display_name: "Create Lead".to_owned(), category: "sales".to_owned(), protocol: Protocol::Rest, status: OperationStatus::Published, version: 1, target: Target::Rest(RestTarget { base_url: base_url.to_owned(), method: HttpMethod::Post, path_template: "/leads".to_owned(), static_headers: BTreeMap::new(), }), input_schema: object_schema("email", SchemaKind::String), output_schema: object_schema("id", SchemaKind::String), input_mapping: MappingSet { rules: input_rules }, output_mapping: MappingSet { rules: vec![MappingRule { source: output_source.to_owned(), target: "$.output.id".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: None, }, tool_description: ToolDescription { title: "Create Lead".to_owned(), description: "Creates a CRM lead".to_owned(), tags: vec!["crm".to_owned()], examples: vec![ToolExample { input: json!({ "email": "user@example.com" }), }], }, samples: Some(Samples::default()), generated_draft: Some(GeneratedDraft { status: GeneratedDraftStatus::Available, source_types: vec!["input_json".to_owned()], generated_at: Some("2026-03-25T20:00:00Z".to_owned()), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }), config_export: None, created_at: timestamp("2026-03-25T20:00:00Z"), updated_at: timestamp("2026-03-25T20:00:00Z"), published_at: Some(timestamp("2026-03-25T20:00:00Z")), }) } fn test_graphql_operation(endpoint: &str, _should_fail: bool) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_graphql_runtime"), name: "crm_create_lead_graphql".to_owned(), display_name: "Create Lead GraphQL".to_owned(), category: "sales".to_owned(), protocol: Protocol::Graphql, status: OperationStatus::Published, version: 1, target: Target::Graphql(GraphqlTarget { endpoint: endpoint.to_owned(), operation_type: GraphqlOperationType::Mutation, operation_name: "CreateLead".to_owned(), query_template: "mutation CreateLead($email: String!) { createLead(email: $email) { id status } }" .to_owned(), response_path: "$.response.body.data.createLead".to_owned(), }), input_schema: object_schema("email", SchemaKind::String), output_schema: object_schema("id", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp.email".to_owned(), target: "$.request.variables.email".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: vec![MappingRule { source: "$.response.data.id".to_owned(), target: "$.output.id".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: None, }, tool_description: ToolDescription { title: "Create Lead GraphQL".to_owned(), description: "Creates a CRM lead through GraphQL".to_owned(), tags: vec!["crm".to_owned(), "graphql".to_owned()], examples: vec![ToolExample { input: json!({ "email": "user@example.com" }), }], }, samples: Some(Samples::default()), generated_draft: Some(GeneratedDraft { status: GeneratedDraftStatus::Available, source_types: vec!["input_json".to_owned(), "output_json".to_owned()], generated_at: Some("2026-03-25T20:00:00Z".to_owned()), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }), config_export: None, created_at: timestamp("2026-03-25T20:00:00Z"), updated_at: timestamp("2026-03-25T20:00:00Z"), published_at: Some(timestamp("2026-03-25T20:00:00Z")), }) } fn test_grpc_operation(server_addr: &str) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_grpc_runtime"), name: "echo_unary_grpc".to_owned(), display_name: "Unary Echo gRPC".to_owned(), category: "support".to_owned(), protocol: Protocol::Grpc, status: OperationStatus::Published, version: 1, target: Target::Grpc(GrpcTarget { server_addr: server_addr.to_owned(), package: "echo".to_owned(), service: "EchoService".to_owned(), method: "UnaryEcho".to_owned(), descriptor_ref: DescriptorId::new("desc_echo"), descriptor_set_b64: grpc_test_support::descriptor_set_b64(), }), input_schema: object_schema("message", SchemaKind::String), output_schema: object_schema("message", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp.message".to_owned(), target: "$.request.grpc.message".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: vec![MappingRule { source: "$.response.data.message".to_owned(), target: "$.output.message".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: None, }, tool_description: ToolDescription { title: "Unary Echo gRPC".to_owned(), description: "Echoes a unary gRPC payload".to_owned(), tags: vec!["grpc".to_owned()], examples: vec![ToolExample { input: json!({ "message": "hello" }), }], }, samples: Some(Samples::default()), generated_draft: Some(GeneratedDraft { status: GeneratedDraftStatus::Available, source_types: vec!["descriptor_set".to_owned()], generated_at: Some("2026-03-25T20:00:00Z".to_owned()), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }), config_export: None, created_at: timestamp("2026-03-25T20:00:00Z"), updated_at: timestamp("2026-03-25T20:00:00Z"), published_at: Some(timestamp("2026-03-25T20:00:00Z")), }) } fn test_grpc_window_operation( server_addr: &str, aggregation_mode: AggregationMode, max_items: Option, ) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_grpc_window_runtime"), name: "echo_stream_grpc".to_owned(), display_name: "Server Stream Echo gRPC".to_owned(), category: "support".to_owned(), protocol: Protocol::Grpc, status: OperationStatus::Published, version: 1, target: Target::Grpc(GrpcTarget { server_addr: server_addr.to_owned(), package: "echo".to_owned(), service: "EchoService".to_owned(), method: "ServerEcho".to_owned(), descriptor_ref: DescriptorId::new("desc_echo"), descriptor_set_b64: grpc_test_support::descriptor_set_b64(), }), input_schema: object_schema("message", SchemaKind::String), output_schema: object_schema("ignored", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp.message".to_owned(), target: "$.request.grpc.message".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: Vec::new() }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: Some(StreamingConfig { mode: ExecutionMode::Window, transport_behavior: TransportBehavior::ServerStream, window_duration_ms: Some(3_000), poll_interval_ms: None, upstream_timeout_ms: Some(1_000), idle_timeout_ms: None, max_session_lifetime_ms: None, max_items, max_bytes: None, aggregation_mode, summary_path: None, items_path: Some("$.items".to_owned()), cursor_path: None, status_path: None, done_path: Some("$.done".to_owned()), redacted_paths: Vec::new(), truncate_item_fields: false, max_field_length: None, drop_duplicates: false, sampling_rate: None, tool_family: ToolFamilyConfig::default(), }), }, tool_description: ToolDescription { title: "Server Stream Echo gRPC".to_owned(), description: "Collects bounded gRPC stream messages.".to_owned(), tags: vec!["grpc".to_owned(), "stream".to_owned()], examples: vec![ToolExample { input: json!({ "message": "hello" }), }], }, samples: None, generated_draft: None, config_export: None, created_at: timestamp("2026-04-06T12:00:00Z"), updated_at: timestamp("2026-04-06T12:00:00Z"), published_at: Some(timestamp("2026-04-06T12:00:00Z")), }) } fn test_soap_operation(endpoint: &str) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_soap_runtime"), name: "crm_create_lead_soap".to_owned(), display_name: "Create Lead SOAP".to_owned(), category: "sales".to_owned(), protocol: Protocol::Soap, status: OperationStatus::Published, version: 1, target: Target::Soap(SoapTarget { wsdl_ref: "sample_wsdl".into(), service_name: "LeadService".to_owned(), port_name: "LeadPort".to_owned(), operation_name: "CreateLead".to_owned(), endpoint_override: Some(endpoint.to_owned()), soap_version: SoapVersion::Soap11, soap_action: Some("urn:createLead".to_owned()), binding_style: SoapBindingStyle::DocumentLiteral, headers: Vec::new(), fault_contract: None, metadata: SoapOperationMetadata { input_part_names: vec!["CreateLeadRequest".to_owned()], output_part_names: vec!["CreateLeadResponse".to_owned()], namespaces: vec!["urn:crm".to_owned()], }, }), input_schema: object_schema("email", SchemaKind::String), output_schema: object_schema("id", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp.email".to_owned(), target: "$.request.body.email".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: vec![MappingRule { source: "$.response.body.id".to_owned(), target: "$.output.id".to_owned(), required: true, default_value: None, transform: None, condition: None, notes: None, }], }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: None, }, tool_description: ToolDescription { title: "Create Lead SOAP".to_owned(), description: "Creates a CRM lead through SOAP".to_owned(), tags: vec!["crm".to_owned(), "soap".to_owned()], examples: vec![ToolExample { input: json!({ "email": "user@example.com" }), }], }, samples: Some(Samples::default()), generated_draft: Some(GeneratedDraft { status: GeneratedDraftStatus::Available, source_types: vec!["wsdl".to_owned()], generated_at: Some("2026-04-06T12:00:00Z".to_owned()), input_schema_generated: true, output_schema_generated: true, input_mapping_generated: true, output_mapping_generated: true, warnings: Vec::new(), }), config_export: None, created_at: timestamp("2026-04-06T12:00:00Z"), updated_at: timestamp("2026-04-06T12:00:00Z"), published_at: Some(timestamp("2026-04-06T12:00:00Z")), }) } fn test_window_snapshot_operation( base_url: &str, aggregation_mode: AggregationMode, max_items: Option, max_bytes: Option, ) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_window_runtime"), name: "billing_log_window".to_owned(), display_name: "Billing Log Window".to_owned(), category: "ops".to_owned(), protocol: Protocol::Rest, status: OperationStatus::Published, version: 1, target: Target::Rest(RestTarget { base_url: base_url.to_owned(), method: HttpMethod::Post, path_template: "/events".to_owned(), static_headers: BTreeMap::new(), }), input_schema: Schema { kind: SchemaKind::Object, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::new(), items: None, enum_values: Vec::new(), variants: Vec::new(), }, output_schema: object_schema("ignored", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp".to_owned(), target: "$.request.body".to_owned(), required: false, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: Vec::new() }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: None, streaming: Some(StreamingConfig { mode: ExecutionMode::Window, transport_behavior: TransportBehavior::RequestResponse, window_duration_ms: Some(3_000), poll_interval_ms: None, upstream_timeout_ms: Some(1_000), idle_timeout_ms: None, max_session_lifetime_ms: None, max_items, max_bytes, aggregation_mode, summary_path: Some("$.summary".to_owned()), items_path: Some("$.items".to_owned()), cursor_path: Some("$.cursor".to_owned()), status_path: None, done_path: Some("$.done".to_owned()), redacted_paths: vec!["$.secret".to_owned()], truncate_item_fields: true, max_field_length: Some(4), drop_duplicates: false, sampling_rate: None, tool_family: ToolFamilyConfig::default(), }), }, tool_description: ToolDescription { title: "Billing Log Window".to_owned(), description: "Collects bounded billing events.".to_owned(), tags: vec!["logs".to_owned()], examples: Vec::new(), }, samples: None, generated_draft: None, config_export: None, created_at: timestamp("2026-04-06T12:00:00Z"), updated_at: timestamp("2026-04-06T12:00:00Z"), published_at: Some(timestamp("2026-04-06T12:00:00Z")), }) } fn test_slow_window_snapshot_operation(base_url: &str) -> RuntimeOperation { let mut operation = test_window_snapshot_operation(base_url, AggregationMode::RawItems, Some(10), None); if let Target::Rest(target) = &mut operation.target { target.path_template = "/slow-events".to_owned(); } operation.execution_config.timeout_ms = 10; operation } fn test_websocket_window_operation( target_url: &str, aggregation_mode: AggregationMode, max_items: Option, ) -> RuntimeOperation { RuntimeOperation::from(Operation { id: OperationId::new("op_websocket_window_runtime"), name: "telemetry_window_ws".to_owned(), display_name: "Telemetry Window WebSocket".to_owned(), category: "ops".to_owned(), protocol: Protocol::Websocket, status: OperationStatus::Published, version: 1, target: Target::Websocket(WebsocketTarget { url: target_url.to_owned(), subprotocols: Vec::new(), subscribe_message_template: Some(json!({"type":"subscribe","topic":"telemetry"})), unsubscribe_message_template: Some( json!({"type":"unsubscribe","topic":"telemetry"}), ), static_headers: BTreeMap::new(), }), input_schema: Schema { kind: SchemaKind::Object, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::new(), items: None, enum_values: Vec::new(), variants: Vec::new(), }, output_schema: object_schema("ignored", SchemaKind::String), input_mapping: MappingSet { rules: vec![MappingRule { source: "$.mcp".to_owned(), target: "$.request.body".to_owned(), required: false, default_value: None, transform: None, condition: None, notes: None, }], }, output_mapping: MappingSet { rules: Vec::new() }, execution_config: ExecutionConfig { timeout_ms: 1_000, retry_policy: None, auth_profile_ref: None, headers: BTreeMap::new(), protocol_options: Some(ProtocolOptions { grpc: None, websocket: Some(WebsocketProtocolOptions { heartbeat_interval_ms: Some(100), reconnect_max_attempts: Some(1), reconnect_backoff_ms: Some(10), }), soap: None, }), streaming: Some(StreamingConfig { mode: ExecutionMode::Window, transport_behavior: TransportBehavior::ServerStream, window_duration_ms: Some(2_000), poll_interval_ms: None, upstream_timeout_ms: Some(1_000), idle_timeout_ms: None, max_session_lifetime_ms: None, max_items, max_bytes: None, aggregation_mode, summary_path: None, items_path: Some("$.items".to_owned()), cursor_path: None, status_path: None, done_path: Some("$.done".to_owned()), redacted_paths: Vec::new(), truncate_item_fields: false, max_field_length: None, drop_duplicates: false, sampling_rate: None, tool_family: ToolFamilyConfig::default(), }), }, tool_description: ToolDescription { title: "Telemetry Window WebSocket".to_owned(), description: "Collects bounded WebSocket telemetry.".to_owned(), tags: vec!["websocket".to_owned(), "stream".to_owned()], examples: Vec::new(), }, samples: None, generated_draft: None, config_export: None, created_at: timestamp("2026-04-06T12:00:00Z"), updated_at: timestamp("2026-04-06T12:00:00Z"), published_at: Some(timestamp("2026-04-06T12:00:00Z")), }) } fn test_window_sse_operation( base_url: &str, aggregation_mode: AggregationMode, max_items: Option, ) -> RuntimeOperation { let mut operation = test_window_snapshot_operation(base_url, aggregation_mode, max_items, None); if let Target::Rest(target) = &mut operation.target { target.path_template = "/events-sse".to_owned(); } operation.execution_config.streaming = Some(StreamingConfig { transport_behavior: TransportBehavior::ServerStream, summary_path: None, cursor_path: None, done_path: Some("$.done".to_owned()), ..operation.execution_config.streaming.clone().unwrap() }); operation } fn test_slow_window_sse_operation(base_url: &str) -> RuntimeOperation { let mut operation = test_window_sse_operation(base_url, AggregationMode::RawItems, Some(10)); if let Target::Rest(target) = &mut operation.target { target.path_template = "/slow-events-sse".to_owned(); } operation .execution_config .streaming .as_mut() .expect("streaming config") .window_duration_ms = Some(20); operation.execution_config.timeout_ms = 1_000; operation } fn object_schema(field_name: &str, kind: SchemaKind) -> Schema { Schema { kind: SchemaKind::Object, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::from([( field_name.to_owned(), Schema { kind, description: None, required: true, nullable: false, default_value: None, fields: BTreeMap::new(), items: None, enum_values: Vec::new(), variants: Vec::new(), }, )]), items: None, enum_values: Vec::new(), variants: Vec::new(), } } }