a.tolmachev
1451 lines
54 KiB
Rust
1451 lines
54 KiB
Rust
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_websocket::{WebsocketAdapter, WebsocketWindowRequest};
|
|
use crank_core::{ExecutionMode, Target, TransportBehavior};
|
|
use serde_json::{Map, Value, json};
|
|
|
|
use crate::{
|
|
AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation, WindowExecutionResult,
|
|
};
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct RuntimeExecutor {
|
|
graphql_adapter: GraphqlAdapter,
|
|
grpc_adapter: GrpcAdapter,
|
|
rest_adapter: RestAdapter,
|
|
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(),
|
|
websocket_adapter: WebsocketAdapter::new(),
|
|
}
|
|
}
|
|
|
|
pub async fn execute(
|
|
&self,
|
|
operation: &RuntimeOperation,
|
|
input: &Value,
|
|
) -> Result<Value, RuntimeError> {
|
|
let prepared_request = self.prepare_request(operation, input)?;
|
|
self.execute_prepared(operation, prepared_request).await
|
|
}
|
|
|
|
pub async fn execute_window(
|
|
&self,
|
|
operation: &RuntimeOperation,
|
|
input: &Value,
|
|
) -> Result<WindowExecutionResult, RuntimeError> {
|
|
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 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)
|
|
.await?
|
|
} else {
|
|
self.execute_adapter(operation, prepared_request).await?
|
|
};
|
|
|
|
crate::aggregation::collect_window_result(&adapter_response.body, streaming)
|
|
}
|
|
|
|
pub async fn execute_session_seed(
|
|
&self,
|
|
operation: &RuntimeOperation,
|
|
input: &Value,
|
|
) -> Result<WindowExecutionResult, RuntimeError> {
|
|
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(&seeded_operation, input).await
|
|
}
|
|
|
|
pub fn prepare_request(
|
|
&self,
|
|
operation: &RuntimeOperation,
|
|
input: &Value,
|
|
) -> Result<PreparedRequest, RuntimeError> {
|
|
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,
|
|
) -> Result<Value, RuntimeError> {
|
|
let adapter_response = self.execute_adapter(operation, prepared_request).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,
|
|
) -> Result<AdapterResponse, RuntimeError> {
|
|
match &operation.target {
|
|
Target::Grpc(target) => {
|
|
let request = GrpcRequest {
|
|
headers: merge_headers(
|
|
&BTreeMap::new(),
|
|
&operation.execution_config.headers,
|
|
&prepared_request.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,
|
|
),
|
|
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,
|
|
),
|
|
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(_) => Err(RuntimeError::UnsupportedProtocol {
|
|
protocol: crank_core::Protocol::Soap,
|
|
}),
|
|
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,
|
|
) -> Result<AdapterResponse, RuntimeError> {
|
|
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,
|
|
),
|
|
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,
|
|
),
|
|
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,
|
|
),
|
|
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::UnsupportedProtocol {
|
|
protocol: crank_core::Protocol::Soap,
|
|
}),
|
|
_ => self.execute_adapter(operation, prepared_request).await,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl PreparedRequest {
|
|
pub fn from_mapping_output(mapped: &Value) -> Result<Self, RuntimeError> {
|
|
let request =
|
|
mapped
|
|
.get("request")
|
|
.ok_or_else(|| RuntimeError::InvalidPreparedRequest {
|
|
details: "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<Value, RuntimeError> {
|
|
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 merge_headers(
|
|
static_headers: &BTreeMap<String, String>,
|
|
execution_headers: &BTreeMap<String, String>,
|
|
request_headers: &BTreeMap<String, String>,
|
|
) -> BTreeMap<String, String> {
|
|
let mut headers = static_headers.clone();
|
|
headers.extend(execution_headers.clone());
|
|
headers.extend(request_headers.clone());
|
|
headers
|
|
}
|
|
|
|
fn read_string_map(
|
|
value: Option<&Value>,
|
|
field_name: &str,
|
|
) -> Result<BTreeMap<String, String>, RuntimeError> {
|
|
let Some(value) = value else {
|
|
return Ok(BTreeMap::new());
|
|
};
|
|
|
|
let Some(object) = value.as_object() else {
|
|
return Err(RuntimeError::InvalidPreparedRequest {
|
|
details: format!("{field_name} must be an object"),
|
|
});
|
|
};
|
|
|
|
object
|
|
.iter()
|
|
.map(|(key, value)| stringify_value(value).map(|value| (key.clone(), value)))
|
|
.collect::<Result<BTreeMap<_, _>, _>>()
|
|
.map_err(|details| RuntimeError::InvalidPreparedRequest { details })
|
|
}
|
|
|
|
fn stringify_value(value: &Value) -> Result<String, String> {
|
|
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<Value>) -> Option<Value> {
|
|
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 axum::{
|
|
Json, Router,
|
|
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,
|
|
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 tokio::net::TcpListener;
|
|
use tokio_tungstenite::accept_async;
|
|
|
|
use crate::{RuntimeError, RuntimeExecutor, RuntimeOperation};
|
|
|
|
#[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 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 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 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 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(_)));
|
|
}
|
|
|
|
#[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("/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_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 }),
|
|
] {
|
|
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<Value>) -> (axum::http::StatusCode, Json<Value>) {
|
|
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 events_window() -> (axum::http::StatusCode, Json<Value>) {
|
|
(
|
|
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<Value>) {
|
|
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
|
events_window().await
|
|
}
|
|
|
|
async fn events_stream()
|
|
-> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
|
|
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<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
|
|
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<Value>) -> Json<Value> {
|
|
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"
|
|
}
|
|
}
|
|
}))
|
|
}
|
|
|
|
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: "2026-03-25T20:00:00Z".to_owned(),
|
|
updated_at: "2026-03-25T20:00:00Z".to_owned(),
|
|
published_at: Some("2026-03-25T20:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
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: "2026-03-25T20:00:00Z".to_owned(),
|
|
updated_at: "2026-03-25T20:00:00Z".to_owned(),
|
|
published_at: Some("2026-03-25T20:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
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: "2026-03-25T20:00:00Z".to_owned(),
|
|
updated_at: "2026-03-25T20:00:00Z".to_owned(),
|
|
published_at: Some("2026-03-25T20:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
fn test_grpc_window_operation(
|
|
server_addr: &str,
|
|
aggregation_mode: AggregationMode,
|
|
max_items: Option<u32>,
|
|
) -> 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: "2026-04-06T12:00:00Z".to_owned(),
|
|
updated_at: "2026-04-06T12:00:00Z".to_owned(),
|
|
published_at: Some("2026-04-06T12:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
fn test_window_snapshot_operation(
|
|
base_url: &str,
|
|
aggregation_mode: AggregationMode,
|
|
max_items: Option<u32>,
|
|
max_bytes: Option<u32>,
|
|
) -> 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: "2026-04-06T12:00:00Z".to_owned(),
|
|
updated_at: "2026-04-06T12:00:00Z".to_owned(),
|
|
published_at: Some("2026-04-06T12:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
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<u32>,
|
|
) -> 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),
|
|
}),
|
|
}),
|
|
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: "2026-04-06T12:00:00Z".to_owned(),
|
|
updated_at: "2026-04-06T12:00:00Z".to_owned(),
|
|
published_at: Some("2026-04-06T12:00:00Z".to_owned()),
|
|
})
|
|
}
|
|
|
|
fn test_window_sse_operation(
|
|
base_url: &str,
|
|
aggregation_mode: AggregationMode,
|
|
max_items: Option<u32>,
|
|
) -> 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(),
|
|
}
|
|
}
|
|
}
|