crank/crates/crank-runtime/src/executor.rs

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use crank_core::{
    AdapterRegistry, CachedHeader, CachedResponse, ExecutionMode, HttpMethod, InvocationStatus,
    MeteringEvent, ResponseCacheStore, SharedMeteringSink, SharedProtocolAdapter, Target,
    TransportBehavior,
};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tracing::debug;

use crate::{
    AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation,
    RuntimeRequestContext, WindowExecutionResult,
};

#[derive(Clone)]
pub struct RuntimeExecutor {
    adapters: AdapterRegistry,
    limits: RuntimeLimits,
    unary_limiter: Arc<Semaphore>,
    window_limiter: Arc<Semaphore>,
    session_limiter: Arc<Semaphore>,
    job_limiter: Arc<Semaphore>,
    response_cache: Option<Arc<dyn ResponseCacheStore>>,
    metering_sink: SharedMeteringSink,
}

impl Default for RuntimeExecutor {
    fn default() -> Self {
        Self::new()
    }
}

impl RuntimeExecutor {
    pub fn new() -> Self {
        crate::executor_builder::community_default().build()
    }

    pub fn with_limits(limits: RuntimeLimits) -> Self {
        crate::executor_builder::community_default()
            .with_limits(limits)
            .build()
    }

    pub(crate) fn from_builder_parts(
        limits: RuntimeLimits,
        adapters: AdapterRegistry,
        response_cache: Option<Arc<dyn ResponseCacheStore>>,
        metering_sink: SharedMeteringSink,
    ) -> Self {
        Self {
            adapters,
            unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)),
            window_limiter: Arc::new(Semaphore::new(limits.max_concurrent_window)),
            session_limiter: Arc::new(Semaphore::new(limits.max_concurrent_sessions)),
            job_limiter: Arc::new(Semaphore::new(limits.max_concurrent_jobs)),
            limits,
            response_cache,
            metering_sink,
        }
    }

    pub fn with_response_cache_store(
        mut self,
        response_cache: Arc<dyn ResponseCacheStore>,
    ) -> Self {
        self.response_cache = Some(response_cache);
        self
    }

    pub async fn execute(
        &self,
        operation: &RuntimeOperation,
        input: &Value,
    ) -> Result<Value, RuntimeError> {
        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<Value, RuntimeError> {
        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<Value, RuntimeError> {
        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<Value, RuntimeError> {
        log_runtime_event("unary.execute", operation, request_context);
        let _permit = self.acquire_unary_permit(operation)?;
        let started_at = Instant::now();
        let prepared_request = self.prepare_request(operation, input)?;
        let prepared_request = apply_resolved_auth(prepared_request, resolved_auth);
        let result = self
            .execute_prepared(operation, prepared_request, request_context)
            .await;
        self.record_metering(operation, request_context, &result, started_at)
            .await;
        result
    }

    pub async fn execute_window(
        &self,
        operation: &RuntimeOperation,
        input: &Value,
    ) -> Result<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        log_runtime_event("window.execute", operation, request_context);
        let _permit = self.acquire_window_permit()?;
        let started_at = Instant::now();
        let Some(streaming) = operation.execution_config.streaming.as_ref() else {
            let result = Err(RuntimeError::MissingStreamingConfig {
                operation_id: operation.operation_id.as_str().to_owned(),
            });
            self.record_metering(operation, request_context, &result, started_at)
                .await;
            return result;
        };

        if streaming.mode != ExecutionMode::Window {
            let result = Err(RuntimeError::UnsupportedExecutionMode {
                operation_id: operation.operation_id.as_str().to_owned(),
                mode: streaming.mode,
            });
            self.record_metering(operation, request_context, &result, started_at)
                .await;
            return result;
        }

        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
        ) {
            self.execute_window_adapter(operation, prepared_request, request_context)
                .await?
        } else {
            self.execute_adapter(operation, prepared_request, request_context)
                .await?
        };

        let result = crate::aggregation::collect_window_result(&adapter_response.body, streaming);
        self.record_metering(operation, request_context, &result, started_at)
            .await;
        result
    }

    pub async fn execute_session_seed(
        &self,
        operation: &RuntimeOperation,
        input: &Value,
    ) -> Result<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        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<WindowExecutionResult, RuntimeError> {
        log_runtime_event("session.seed", operation, request_context);
        let _permit = self.acquire_session_permit()?;
        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<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,
        request_context: Option<&RuntimeRequestContext>,
    ) -> Result<Value, RuntimeError> {
        let adapter_response = match self
            .load_cached_adapter_response(operation, &prepared_request, request_context)
            .await
        {
            Some(response) => response,
            None => {
                let adapter_response = self
                    .execute_adapter(operation, prepared_request.clone(), request_context)
                    .await?;
                self.store_cached_adapter_response(
                    operation,
                    &prepared_request,
                    request_context,
                    &adapter_response,
                )
                .await;
                adapter_response
            }
        };
        let finalized_output = finalize_output(operation, &adapter_response)?;

        operation.output_schema.validate_shape(&finalized_output)?;

        Ok(finalized_output)
    }

    async fn record_metering<T>(
        &self,
        operation: &RuntimeOperation,
        request_context: Option<&RuntimeRequestContext>,
        result: &Result<T, RuntimeError>,
        started_at: Instant,
    ) {
        let Some(context) = request_context.and_then(RuntimeRequestContext::metering_context)
        else {
            return;
        };

        self.metering_sink
            .record(MeteringEvent {
                workspace_id: context.workspace_id.clone(),
                agent_id: context.agent_id.clone(),
                operation_id: operation.operation_id.clone(),
                source: context.source,
                status: if result.is_ok() {
                    InvocationStatus::Ok
                } else {
                    InvocationStatus::Error
                },
                duration_ms: started_at.elapsed().as_millis() as u64,
                created_at: OffsetDateTime::now_utc(),
            })
            .await;
    }

    async fn execute_adapter(
        &self,
        operation: &RuntimeOperation,
        prepared_request: PreparedRequest,
        request_context: Option<&RuntimeRequestContext>,
    ) -> Result<AdapterResponse, RuntimeError> {
        log_runtime_event("adapter.dispatch", operation, request_context);
        let adapter = self.adapter_for(operation)?;
        if !adapter.supports_mode(ExecutionMode::Unary) {
            return Err(RuntimeError::UnsupportedExecutionMode {
                operation_id: operation.operation_id.as_str().to_owned(),
                mode: ExecutionMode::Unary,
            });
        }

        let prepared_request = adapter_prepared_request(
            operation,
            &prepared_request,
            request_context,
            operation.execution_config.timeout_ms,
        );
        let adapter_context = adapter_request_context(request_context);

        adapter
            .invoke_unary(
                &operation.target,
                &prepared_request.into(),
                &adapter_context,
            )
            .await
            .map(Into::into)
            .map_err(|error| map_protocol_adapter_error(operation, error))
    }

    async fn execute_window_adapter(
        &self,
        operation: &RuntimeOperation,
        prepared_request: PreparedRequest,
        request_context: Option<&RuntimeRequestContext>,
    ) -> Result<AdapterResponse, RuntimeError> {
        log_runtime_event("window.adapter.dispatch", 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(),
            });
        };
        let adapter = self.adapter_for(operation)?;
        if !adapter.supports_mode(ExecutionMode::Window) {
            return Err(RuntimeError::UnsupportedExecutionMode {
                operation_id: operation.operation_id.as_str().to_owned(),
                mode: ExecutionMode::Window,
            });
        }

        let prepared_request = adapter_prepared_request(
            operation,
            &prepared_request,
            request_context,
            streaming
                .upstream_timeout_ms
                .unwrap_or(operation.execution_config.timeout_ms),
        );
        let adapter_context = adapter_request_context(request_context);
        let result = adapter
            .invoke_window(
                &operation.target,
                &prepared_request.into(),
                streaming.window_duration_ms.unwrap_or_default(),
                streaming.max_items,
                &adapter_context,
            )
            .await
            .map_err(|error| map_protocol_adapter_error(operation, error))?;

        Ok(AdapterResponse {
            status_code: 200,
            headers: BTreeMap::new(),
            body: json!({
                "summary": result.summary,
                "items": result.items,
                "cursor": result.cursor,
                "done": result.window_complete,
            }),
            data: Value::Null,
        })
    }

    fn acquire_unary_permit(
        &self,
        operation: &RuntimeOperation,
    ) -> Result<OwnedSemaphorePermit, RuntimeError> {
        let (kind, limit, limiter) = if operation
            .execution_config
            .streaming
            .as_ref()
            .is_some_and(|streaming| streaming.mode == ExecutionMode::AsyncJob)
        {
            (
                "async_job",
                self.limits.max_concurrent_jobs,
                Arc::clone(&self.job_limiter),
            )
        } else {
            (
                "unary",
                self.limits.max_concurrent_unary,
                Arc::clone(&self.unary_limiter),
            )
        };

        try_acquire_limit(limiter, kind, limit)
    }

    fn acquire_window_permit(&self) -> Result<OwnedSemaphorePermit, RuntimeError> {
        try_acquire_limit(
            Arc::clone(&self.window_limiter),
            "window",
            self.limits.max_concurrent_window,
        )
    }

    fn acquire_session_permit(&self) -> Result<OwnedSemaphorePermit, RuntimeError> {
        try_acquire_limit(
            Arc::clone(&self.session_limiter),
            "session",
            self.limits.max_concurrent_sessions,
        )
    }

    async fn load_cached_adapter_response(
        &self,
        operation: &RuntimeOperation,
        prepared_request: &PreparedRequest,
        request_context: Option<&RuntimeRequestContext>,
    ) -> Option<AdapterResponse> {
        let response_cache = self.response_cache.as_ref()?;
        let cache_key = response_cache_key(operation, prepared_request, request_context)?;
        let cached = match response_cache.get(&cache_key).await {
            Ok(cached) => cached?,
            Err(error) => {
                debug!(
                    operation_id = %operation.operation_id,
                    cache_key,
                    error = %error,
                    "response cache lookup skipped"
                );
                return None;
            }
        };

        match adapter_response_from_cached(cached) {
            Ok(response) => Some(response),
            Err(error) => {
                debug!(
                    operation_id = %operation.operation_id,
                    cache_key,
                    error,
                    "cached response payload was invalid"
                );
                let _ = response_cache.delete(&cache_key).await;
                None
            }
        }
    }

    async fn store_cached_adapter_response(
        &self,
        operation: &RuntimeOperation,
        prepared_request: &PreparedRequest,
        request_context: Option<&RuntimeRequestContext>,
        adapter_response: &AdapterResponse,
    ) {
        let Some(response_cache) = self.response_cache.as_ref() else {
            return;
        };

        if !(200..=299).contains(&adapter_response.status_code) {
            return;
        }

        let Some(cache_ttl) = response_cache_ttl(operation) else {
            return;
        };
        let Some(cache_key) = response_cache_key(operation, prepared_request, request_context)
        else {
            return;
        };
        let Some(cached_response) = cached_response_from_adapter(adapter_response) else {
            return;
        };

        if let Err(error) = response_cache
            .put(&cache_key, cached_response, cache_ttl)
            .await
        {
            debug!(
                operation_id = %operation.operation_id,
                cache_key,
                error = %error,
                "response cache write skipped"
            );
        }
    }
}

impl PreparedRequest {
    pub fn from_mapping_output(mapped: &Value) -> Result<Self, RuntimeError> {
        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(),
            timeout_ms: 0,
        })
    }
}

fn adapter_request_context(
    request_context: Option<&RuntimeRequestContext>,
) -> crank_core::RuntimeRequestContext {
    request_context
        .map(Into::into)
        .unwrap_or_else(|| crank_core::RuntimeRequestContext::new(String::new(), String::new()))
}

fn adapter_prepared_request(
    operation: &RuntimeOperation,
    prepared_request: &PreparedRequest,
    request_context: Option<&RuntimeRequestContext>,
    timeout_ms: u64,
) -> PreparedRequest {
    let empty_headers = BTreeMap::new();
    let static_headers = match &operation.target {
        Target::Rest(target) => &target.static_headers,
        _ => &empty_headers,
    };

    let mut prepared_request = prepared_request.clone();
    prepared_request.headers = merge_headers(
        static_headers,
        &operation.execution_config.headers,
        &prepared_request.headers,
        &runtime_context_headers(request_context),
    );
    prepared_request.timeout_ms = timeout_ms;
    prepared_request
}

fn map_protocol_adapter_error(
    operation: &RuntimeOperation,
    error: crank_core::ProtocolAdapterError,
) -> RuntimeError {
    match error {
        crank_core::ProtocolAdapterError::UnsupportedMode { mode, .. } => {
            RuntimeError::UnsupportedExecutionMode {
                operation_id: operation.operation_id.as_str().to_owned(),
                mode,
            }
        }
        crank_core::ProtocolAdapterError::Message(message) => RuntimeError::ProtocolAdapter(
            format!("operation {}: {message}", operation.operation_id),
        ),
    }
}

impl RuntimeExecutor {
    fn adapter_for(
        &self,
        operation: &RuntimeOperation,
    ) -> Result<SharedProtocolAdapter, RuntimeError> {
        self.adapters
            .get(operation.protocol)
            .ok_or(RuntimeError::UnsupportedProtocol {
                protocol: operation.protocol,
            })
    }
}

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 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<String, String>,
    execution_headers: &BTreeMap<String, String>,
    request_headers: &BTreeMap<String, String>,
    context_headers: &BTreeMap<String, String>,
) -> BTreeMap<String, String> {
    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<String, String> {
    request_context
        .map(RuntimeRequestContext::outbound_headers)
        .unwrap_or_default()
}

fn try_acquire_limit(
    limiter: Arc<Semaphore>,
    kind: &'static str,
    limit: usize,
) -> Result<OwnedSemaphorePermit, RuntimeError> {
    limiter
        .try_acquire_owned()
        .map_err(|_| RuntimeError::ConcurrencyLimitExceeded { kind, limit })
}

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 response_cache_ttl(operation: &RuntimeOperation) -> Option<Duration> {
    if operation.execution_config.streaming.is_some() {
        return None;
    }

    let is_cacheable_protocol =
        matches!(&operation.target, Target::Rest(target) if target.method == HttpMethod::Get);
    if !is_cacheable_protocol {
        return None;
    }

    let policy = operation.execution_config.response_cache.as_ref()?;
    if operation.execution_config.auth_profile_ref.is_some() || policy.ttl_ms == 0 {
        return None;
    }

    Some(Duration::from_millis(policy.ttl_ms))
}

fn response_cache_key(
    operation: &RuntimeOperation,
    prepared_request: &PreparedRequest,
    request_context: Option<&RuntimeRequestContext>,
) -> Option<String> {
    let _ = response_cache_ttl(operation)?;
    let scope = request_context?.response_cache_scope()?;
    let fingerprint = json!({
        "path_params": prepared_request.path_params,
        "query_params": prepared_request.query_params,
        "headers": prepared_request.headers,
        "body": prepared_request.body,
        "variables": prepared_request.variables,
        "grpc": prepared_request.grpc,
    });
    let fingerprint_bytes = serde_json::to_vec(&fingerprint).ok()?;
    let fingerprint_hash = URL_SAFE_NO_PAD.encode(Sha256::digest(fingerprint_bytes));

    Some(format!(
        "crank:response:workspace:{}:agent:{}:operation:{}:version:{}:request:{}",
        scope.workspace_key,
        scope.agent_key,
        operation.operation_id.as_str(),
        operation.operation_version,
        fingerprint_hash
    ))
}

fn cached_response_from_adapter(adapter_response: &AdapterResponse) -> Option<CachedResponse> {
    let body = serde_json::to_vec(&adapter_response.body).ok()?;
    Some(CachedResponse {
        status: adapter_response.status_code,
        headers: adapter_response
            .headers
            .iter()
            .map(|(name, value)| CachedHeader {
                name: name.clone(),
                value: value.clone(),
            })
            .collect(),
        body,
        data: serde_json::to_vec(&adapter_response.data).ok()?,
    })
}

fn adapter_response_from_cached(
    cached_response: CachedResponse,
) -> Result<AdapterResponse, String> {
    let body: Value =
        serde_json::from_slice(&cached_response.body).map_err(|error| error.to_string())?;
    let data: Value =
        serde_json::from_slice(&cached_response.data).map_err(|error| error.to_string())?;
    Ok(AdapterResponse {
        status_code: cached_response.status,
        headers: cached_response
            .headers
            .into_iter()
            .map(|header| (header.name, header.value))
            .collect(),
        body,
        data,
    })
}

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 {
            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::<Result<BTreeMap<_, _>, _>>()
        .map_err(|reason| RuntimeError::InvalidPreparedRequest {
            field: field_name.to_owned(),
            reason,
        })
}

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 std::io;
    use std::sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    };

    use async_trait::async_trait;
    use axum::{
        Json, Router,
        extract::Query,
        http::HeaderMap,
        routing::{get, post},
    };
    #[cfg(any())]
    use crank_adapter_grpc::test_support as grpc_test_support;
    use crank_core::{
        ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, HttpMethod, InvocationSource,
        InvocationStatus, MeteringEvent, MeteringSink, Operation, OperationId,
        OperationSecurityLevel, OperationStatus, Protocol, RestTarget, Samples, Target,
        ToolDescription, ToolExample, WorkspaceId,
    };
    use crank_mapping::{MappingRule, MappingSet};
    use crank_schema::{Schema, SchemaKind};
    use serde_json::{Value, json};
    use time::{OffsetDateTime, format_description::well_known::Rfc3339};
    use tokio::net::TcpListener;
    #[cfg(any())]
    use tokio_tungstenite::{
        accept_async, accept_hdr_async,
        tungstenite::handshake::server::{Request, Response},
    };
    use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*};

    use crate::{
        InMemoryResponseCacheStore, PreparedRequest, RuntimeError, RuntimeExecutor,
        RuntimeExecutorBuilder, RuntimeLimits, RuntimeOperation, RuntimeRequestContext,
    };

    fn timestamp(value: &str) -> OffsetDateTime {
        OffsetDateTime::parse(value, &Rfc3339).unwrap()
    }

    #[derive(Clone, Default)]
    struct SharedLogWriter {
        buffer: Arc<Mutex<Vec<u8>>>,
    }

    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<Mutex<Vec<u8>>>,
    }

    impl io::Write for SharedLogGuard {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            self.buffer.lock().unwrap().extend_from_slice(bytes);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    #[derive(Clone, Default)]
    struct RecordingMeteringSink {
        events: Arc<Mutex<Vec<MeteringEvent>>>,
    }

    impl RecordingMeteringSink {
        fn events(&self) -> Vec<MeteringEvent> {
            self.events.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl MeteringSink for RecordingMeteringSink {
        async fn record(&self, event: MeteringEvent) {
            self.events.lock().unwrap().push(event);
        }
    }

    #[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 records_metering_event_for_successful_invocation() {
        let base_url = spawn_runtime_server().await;
        let metering_sink = RecordingMeteringSink::default();
        let executor = RuntimeExecutorBuilder::new()
            .register_adapter(Arc::new(crank_adapter_rest::RestAdapter::new()))
            .with_metering_sink(Arc::new(metering_sink.clone()))
            .build();
        let operation = test_rest_operation(&base_url, false, false);
        let context = RuntimeRequestContext::from_request_id("req_meter_1").with_metering_context(
            WorkspaceId::new("ws_meter"),
            None,
            InvocationSource::AgentToolCall,
        );

        let output = executor
            .execute_with_context(
                &operation,
                &json!({ "email": "user@example.com" }),
                Some(&context),
            )
            .await
            .unwrap();

        assert_eq!(output, json!({ "id": "lead_123" }));

        let events = metering_sink.events();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].workspace_id, WorkspaceId::new("ws_meter"));
        assert_eq!(events[0].agent_id, None);
        assert_eq!(events[0].operation_id, operation.operation_id);
        assert_eq!(events[0].source, InvocationSource::AgentToolCall);
        assert_eq!(events[0].status, InvocationStatus::Ok);
        assert!(events[0].duration_ms <= 60_000);
    }

    #[tokio::test]
    async fn caches_rest_get_responses_within_agent_scope() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let base_url = spawn_cached_rest_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let operation = test_cached_rest_get_operation(&base_url);
        let first_context = RuntimeRequestContext::from_request_id("req_cache_1")
            .with_response_cache_scope("ws_cache", "agent_sales");
        let second_context = RuntimeRequestContext::from_request_id("req_cache_2")
            .with_response_cache_scope("ws_cache", "agent_sales");

        let first_output = executor
            .execute_with_context(&operation, &json!({ "city": "msk" }), Some(&first_context))
            .await
            .unwrap();
        let second_output = executor
            .execute_with_context(&operation, &json!({ "city": "msk" }), Some(&second_context))
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "city": "msk", "request_count": 1 }));
        assert_eq!(second_output, json!({ "city": "msk", "request_count": 1 }));
        assert_eq!(request_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn isolates_rest_get_cache_between_agents() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let base_url = spawn_cached_rest_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let operation = test_cached_rest_get_operation(&base_url);
        let agent_a_context = RuntimeRequestContext::from_request_id("req_cache_agent_a")
            .with_response_cache_scope("ws_cache", "agent_a");
        let agent_b_context = RuntimeRequestContext::from_request_id("req_cache_agent_b")
            .with_response_cache_scope("ws_cache", "agent_b");

        let first_output = executor
            .execute_with_context(
                &operation,
                &json!({ "city": "msk" }),
                Some(&agent_a_context),
            )
            .await
            .unwrap();
        let second_output = executor
            .execute_with_context(
                &operation,
                &json!({ "city": "msk" }),
                Some(&agent_b_context),
            )
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "city": "msk", "request_count": 1 }));
        assert_eq!(second_output, json!({ "city": "msk", "request_count": 2 }));
        assert_eq!(request_count.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn invalidates_rest_get_cache_on_operation_version_change() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let base_url = spawn_cached_rest_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let operation_v1 = test_cached_rest_get_operation(&base_url);
        let mut operation_v2 = operation_v1.clone();
        let context = RuntimeRequestContext::from_request_id("req_cache_version")
            .with_response_cache_scope("ws_cache", "agent_sales");

        let first_output = executor
            .execute_with_context(&operation_v1, &json!({ "city": "msk" }), Some(&context))
            .await
            .unwrap();
        operation_v2.operation_version = 2;
        let second_output = executor
            .execute_with_context(&operation_v2, &json!({ "city": "msk" }), Some(&context))
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "city": "msk", "request_count": 1 }));
        assert_eq!(second_output, json!({ "city": "msk", "request_count": 2 }));
        assert_eq!(request_count.load(Ordering::SeqCst), 2);
    }

    #[cfg(any())]
    #[tokio::test]
    async fn caches_graphql_query_responses_within_agent_scope() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let endpoint = spawn_cached_graphql_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let operation = test_cached_graphql_query_operation(&endpoint);
        let first_context = RuntimeRequestContext::from_request_id("req_cache_graphql_1")
            .with_response_cache_scope("ws_cache", "agent_sales");
        let second_context = RuntimeRequestContext::from_request_id("req_cache_graphql_2")
            .with_response_cache_scope("ws_cache", "agent_sales");

        let first_output = executor
            .execute_with_context(
                &operation,
                &json!({ "email": "user@example.com" }),
                Some(&first_context),
            )
            .await
            .unwrap();
        let second_output = executor
            .execute_with_context(
                &operation,
                &json!({ "email": "user@example.com" }),
                Some(&second_context),
            )
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "id": "lead_123" }));
        assert_eq!(second_output, json!({ "id": "lead_123" }));
        assert_eq!(request_count.load(Ordering::SeqCst), 1);
    }

    #[cfg(any())]
    #[tokio::test]
    async fn caches_read_only_grpc_unary_responses_within_agent_scope() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let server_addr =
            grpc_test_support::spawn_counting_unary_echo_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let operation = test_cached_grpc_operation(&server_addr);
        let first_context = RuntimeRequestContext::from_request_id("req_cache_grpc_1")
            .with_response_cache_scope("ws_cache", "agent_sales");
        let second_context = RuntimeRequestContext::from_request_id("req_cache_grpc_2")
            .with_response_cache_scope("ws_cache", "agent_sales");

        let first_output = executor
            .execute_with_context(
                &operation,
                &json!({ "message": "hello" }),
                Some(&first_context),
            )
            .await
            .unwrap();
        let second_output = executor
            .execute_with_context(
                &operation,
                &json!({ "message": "hello" }),
                Some(&second_context),
            )
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "message": "hello-1" }));
        assert_eq!(second_output, json!({ "message": "hello-1" }));
        assert_eq!(request_count.load(Ordering::SeqCst), 1);
    }

    #[cfg(any())]
    #[tokio::test]
    async fn skips_grpc_response_cache_without_read_only_flag() {
        let request_count = Arc::new(AtomicUsize::new(0));
        let server_addr =
            grpc_test_support::spawn_counting_unary_echo_server(Arc::clone(&request_count)).await;
        let executor = RuntimeExecutor::new()
            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
        let mut operation = test_cached_grpc_operation(&server_addr);
        if let Target::Grpc(target) = &mut operation.target {
            target.read_only = false;
        }
        let first_context = RuntimeRequestContext::from_request_id("req_cache_grpc_skip_1")
            .with_response_cache_scope("ws_cache", "agent_sales");
        let second_context = RuntimeRequestContext::from_request_id("req_cache_grpc_skip_2")
            .with_response_cache_scope("ws_cache", "agent_sales");

        let first_output = executor
            .execute_with_context(
                &operation,
                &json!({ "message": "hello" }),
                Some(&first_context),
            )
            .await
            .unwrap();
        let second_output = executor
            .execute_with_context(
                &operation,
                &json!({ "message": "hello" }),
                Some(&second_context),
            )
            .await
            .unwrap();

        assert_eq!(first_output, json!({ "message": "hello-1" }));
        assert_eq!(second_output, json!({ "message": "hello-2" }));
        assert_eq!(request_count.load(Ordering::SeqCst), 2);
    }

    #[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 rejects_unary_execution_when_unary_limit_is_exhausted() {
        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
            max_concurrent_unary: 0,
            ..RuntimeLimits::default()
        });
        let operation = test_rest_operation("http://example.invalid", false, false);

        let error = executor
            .execute(&operation, &json!({ "email": "user@example.com" }))
            .await
            .unwrap_err();

        assert!(matches!(
            error,
            RuntimeError::ConcurrencyLimitExceeded {
                kind: "unary",
                limit: 0
            }
        ));
    }

    #[cfg(any())]
    #[tokio::test]
    async fn rejects_window_execution_when_window_limit_is_exhausted() {
        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
            max_concurrent_window: 0,
            ..RuntimeLimits::default()
        });
        let operation = test_window_snapshot_operation(
            "http://example.invalid",
            AggregationMode::RawItems,
            None,
            None,
        );

        let error = executor
            .execute_window(&operation, &json!({}))
            .await
            .unwrap_err();

        assert!(matches!(
            error,
            RuntimeError::ConcurrencyLimitExceeded {
                kind: "window",
                limit: 0
            }
        ));
    }

    #[cfg(any())]
    #[tokio::test]
    async fn rejects_session_seed_when_session_limit_is_exhausted() {
        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
            max_concurrent_sessions: 0,
            ..RuntimeLimits::default()
        });
        let operation = test_session_seed_operation("http://example.invalid");

        let error = executor
            .execute_session_seed(&operation, &json!({}))
            .await
            .unwrap_err();

        assert!(matches!(
            error,
            RuntimeError::ConcurrencyLimitExceeded {
                kind: "session",
                limit: 0
            }
        ));
    }

    #[cfg(any())]
    #[tokio::test]
    async fn rejects_async_job_execution_when_job_limit_is_exhausted() {
        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
            max_concurrent_jobs: 0,
            ..RuntimeLimits::default()
        });
        let operation = test_async_job_operation("http://example.invalid");

        let error = executor.execute(&operation, &json!({})).await.unwrap_err();

        assert!(matches!(
            error,
            RuntimeError::ConcurrencyLimitExceeded {
                kind: "async_job",
                limit: 0
            }
        ));
    }

    #[cfg(any())]
    #[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" }));
    }

    #[cfg(any())]
    #[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" }));
    }

    #[cfg(any())]
    #[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" }));
    }

    #[cfg(any())]
    #[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" })
        );
    }

    #[cfg(any())]
    #[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" }));
    }

    #[cfg(any())]
    #[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" }));
    }

    #[cfg(any())]
    #[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);
    }

    #[cfg(any())]
    #[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);
    }

    #[cfg(any())]
    #[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::ProtocolAdapter(_)));
    }

    #[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(_)));
    }

    #[cfg(any())]
    #[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(_)));
    }

    #[cfg(any())]
    #[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);
    }

    #[cfg(any())]
    #[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());
    }

    #[cfg(any())]
    #[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..."));
    }

    #[cfg(any())]
    #[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(_)));
    }

    #[cfg(any())]
    #[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);
    }

    #[cfg(any())]
    #[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));
        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_cached_rest_server(request_count: Arc<AtomicUsize>) -> String {
        let app = Router::new().route(
            "/catalog",
            get(move |Query(params): Query<BTreeMap<String, String>>| {
                let request_count = Arc::clone(&request_count);
                async move {
                    let count = request_count.fetch_add(1, Ordering::SeqCst) + 1;
                    let city = params.get("city").cloned().unwrap_or_default();
                    (
                        axum::http::StatusCode::OK,
                        Json(json!({
                            "city": city,
                            "request_count": count,
                        })),
                    )
                }
            }),
        );
        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)
    }

    #[cfg(any())]
    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)
    }

    #[cfg(any())]
    async fn spawn_cached_graphql_server(request_count: Arc<AtomicUsize>) -> String {
        let app = Router::new().route(
            "/",
            post(move |Json(payload): Json<Value>| {
                let request_count = Arc::clone(&request_count);
                async move {
                    let count = request_count.fetch_add(1, Ordering::SeqCst) + 1;
                    let email = payload
                        .get("variables")
                        .and_then(|variables| variables.get("email"))
                        .and_then(Value::as_str)
                        .unwrap_or_default();

                    Json(json!({
                        "data": {
                            "lookupLead": {
                                "id": "lead_123",
                                "email": email,
                                "request_count": count,
                            }
                        }
                    }))
                }
            }),
        );
        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)
    }

    #[cfg(any())]
    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)
    }

    #[cfg(any())]
    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)
    }

    #[cfg(any())]
    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)
    }

    #[cfg(any())]
    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)
    }

    #[cfg(any())]
    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<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 create_lead_with_context(
        headers: HeaderMap,
        Json(_payload): Json<Value>,
    ) -> (axum::http::StatusCode, Json<Value>) {
        (
            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()
                ),
            })),
        )
    }

    #[cfg(any())]
    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
            })),
        )
    }

    #[cfg(any())]
    async fn slow_events_window() -> (axum::http::StatusCode, Json<Value>) {
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        events_window().await
    }

    #[cfg(any())]
    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())
    }

    #[cfg(any())]
    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())
    }

    #[cfg(any())]
    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"
                }
            }
        }))
    }

    #[cfg(any())]
    async fn graphql_context_handler(
        headers: HeaderMap,
        Json(_payload): Json<Value>,
    ) -> Json<Value> {
        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"
                }
            }
        }))
    }

    #[cfg(any())]
    async fn soap_handler(body: String) -> (axum::http::StatusCode, String) {
        assert!(body.contains("<email>user@example.com</email>"));

        (
            axum::http::StatusCode::OK,
            r#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
                <soap:Body>
                    <CreateLeadResponse>
                        <id>lead_123</id>
                        <status>created</status>
                    </CreateLeadResponse>
                </soap:Body>
            </soap:Envelope>"#
                .to_owned(),
        )
    }

    #[cfg(any())]
    async fn soap_context_handler(
        headers: HeaderMap,
        body: String,
    ) -> (axum::http::StatusCode, String) {
        assert!(body.contains("<email>user@example.com</email>"));
        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#"<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
                <soap:Body>
                    <CreateLeadResponse>
                        <id>{request_id}|{correlation_id}</id>
                        <status>created</status>
                    </CreateLeadResponse>
                </soap:Body>
            </soap:Envelope>"#
            ),
        )
    }

    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,
            security_level: OperationSecurityLevel::Standard,
            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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    fn test_cached_graphql_query_operation(endpoint: &str) -> RuntimeOperation {
        RuntimeOperation::from(Operation {
            id: OperationId::new("op_graphql_cached_runtime"),
            name: "crm_lookup_lead_graphql".to_owned(),
            display_name: "Lookup Lead GraphQL".to_owned(),
            category: "sales".to_owned(),
            protocol: Protocol::Graphql,
            security_level: OperationSecurityLevel::Standard,
            status: OperationStatus::Published,
            version: 1,
            target: Target::Graphql(GraphqlTarget {
                endpoint: endpoint.to_owned(),
                operation_type: GraphqlOperationType::Query,
                operation_name: "LookupLead".to_owned(),
                query_template:
                    "query LookupLead($email: String!) { lookupLead(email: $email) { id email request_count } }"
                        .to_owned(),
                response_path: "$.response.body.data.lookupLead".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,
                response_cache: Some(crank_core::ResponseCachePolicy { ttl_ms: 5_000 }),
                auth_profile_ref: None,
                headers: BTreeMap::new(),
                protocol_options: None,
                streaming: None,
            },
            tool_description: ToolDescription {
                title: "Lookup Lead GraphQL".to_owned(),
                description: "Reads 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()],
                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")),
        })
    }

    #[cfg(any())]
    fn test_cached_grpc_operation(server_addr: &str) -> RuntimeOperation {
        let mut operation = test_grpc_operation(server_addr);
        operation.operation_id = OperationId::new("op_grpc_cached_runtime");
        operation.tool_name = "echo_unary_grpc_cached".to_owned();
        if let Target::Grpc(target) = &mut operation.target {
            target.read_only = true;
        }
        operation.execution_config.response_cache =
            Some(crank_core::ResponseCachePolicy { ttl_ms: 5_000 });
        operation.tool_description = ToolDescription {
            title: "Unary Echo gRPC Cached".to_owned(),
            description: "Reads a cacheable unary gRPC payload".to_owned(),
            tags: vec!["grpc".to_owned(), "cache".to_owned()],
            examples: vec![ToolExample {
                input: json!({ "message": "hello" }),
            }],
        };
        operation
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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(),
                read_only: false,
                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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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(),
                read_only: false,
                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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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
    }

    #[cfg(any())]
    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,
            security_level: OperationSecurityLevel::Standard,
            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,
                response_cache: 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")),
        })
    }

    #[cfg(any())]
    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
    }

    #[cfg(any())]
    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 test_cached_rest_get_operation(base_url: &str) -> RuntimeOperation {
        RuntimeOperation::from(Operation {
            id: OperationId::new("op_rest_cache_runtime"),
            name: "catalog_lookup".to_owned(),
            display_name: "Catalog Lookup".to_owned(),
            category: "lookup".to_owned(),
            protocol: Protocol::Rest,
            security_level: OperationSecurityLevel::Standard,
            status: OperationStatus::Published,
            version: 1,
            target: Target::Rest(RestTarget {
                base_url: base_url.to_owned(),
                method: HttpMethod::Get,
                path_template: "/catalog".to_owned(),
                static_headers: BTreeMap::new(),
            }),
            input_schema: object_schema("city", SchemaKind::String),
            output_schema: Schema {
                kind: SchemaKind::Object,
                description: None,
                required: true,
                nullable: false,
                default_value: None,
                fields: BTreeMap::from([
                    (
                        "city".to_owned(),
                        Schema {
                            kind: SchemaKind::String,
                            description: None,
                            required: true,
                            nullable: false,
                            default_value: None,
                            fields: BTreeMap::new(),
                            items: None,
                            enum_values: Vec::new(),
                            variants: Vec::new(),
                        },
                    ),
                    (
                        "request_count".to_owned(),
                        Schema {
                            kind: SchemaKind::Number,
                            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(),
            },
            input_mapping: MappingSet {
                rules: vec![MappingRule {
                    source: "$.mcp.city".to_owned(),
                    target: "$.request.query.city".to_owned(),
                    required: true,
                    default_value: None,
                    transform: None,
                    condition: None,
                    notes: None,
                }],
            },
            output_mapping: MappingSet {
                rules: vec![
                    MappingRule {
                        source: "$.response.body.city".to_owned(),
                        target: "$.output.city".to_owned(),
                        required: true,
                        default_value: None,
                        transform: None,
                        condition: None,
                        notes: None,
                    },
                    MappingRule {
                        source: "$.response.body.request_count".to_owned(),
                        target: "$.output.request_count".to_owned(),
                        required: true,
                        default_value: None,
                        transform: None,
                        condition: None,
                        notes: None,
                    },
                ],
            },
            execution_config: ExecutionConfig {
                timeout_ms: 1_000,
                retry_policy: None,
                response_cache: Some(crank_core::ResponseCachePolicy { ttl_ms: 60_000 }),
                auth_profile_ref: None,
                headers: BTreeMap::new(),
                protocol_options: None,
                streaming: None,
            },
            tool_description: ToolDescription {
                title: "Catalog Lookup".to_owned(),
                description: "Loads a read-only catalog item".to_owned(),
                tags: vec!["lookup".to_owned()],
                examples: vec![ToolExample {
                    input: json!({ "city": "msk" }),
                }],
            },
            samples: Some(Samples::default()),
            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")),
        })
    }

    #[cfg(any())]
    fn test_session_seed_operation(base_url: &str) -> RuntimeOperation {
        let mut operation =
            test_window_snapshot_operation(base_url, AggregationMode::RawItems, Some(10), None);

        operation
            .execution_config
            .streaming
            .as_mut()
            .expect("streaming config")
            .mode = ExecutionMode::Session;

        operation
    }

    #[cfg(any())]
    fn test_async_job_operation(base_url: &str) -> RuntimeOperation {
        let mut operation = test_rest_operation(base_url, false, false);
        operation.execution_config.streaming = Some(StreamingConfig {
            mode: ExecutionMode::AsyncJob,
            transport_behavior: TransportBehavior::RequestResponse,
            window_duration_ms: None,
            poll_interval_ms: Some(5_000),
            upstream_timeout_ms: Some(operation.execution_config.timeout_ms),
            idle_timeout_ms: None,
            max_session_lifetime_ms: Some(300_000),
            max_items: None,
            max_bytes: None,
            aggregation_mode: AggregationMode::RawItems,
            summary_path: None,
            items_path: None,
            cursor_path: None,
            status_path: None,
            done_path: None,
            redacted_paths: Vec::new(),
            truncate_item_fields: false,
            max_field_length: None,
            drop_duplicates: false,
            sampling_rate: None,
            tool_family: ToolFamilyConfig::default(),
        });

        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(),
        }
    }
}