feat: add window execution to runtime

2026-04-06 10:38:23 +03:00
parent 7e4f3d142e
commit 8204a59dac
10 changed files with 708 additions and 27 deletions
@@ -3,10 +3,12 @@ use std::collections::BTreeMap;
 use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest};
 use crank_adapter_grpc::{GrpcAdapter, GrpcRequest};
 use crank_adapter_rest::{RestAdapter, RestRequest};
-use crank_core::Target;
+use crank_core::{ExecutionMode, Target};
 use serde_json::{Map, Value, json};

-use crate::{AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation};
+use crate::{
+    AdapterResponse, PreparedRequest, RuntimeError, RuntimeOperation, WindowExecutionResult,
+};

 #[derive(Clone, Debug)]
 pub struct RuntimeExecutor {
@@ -39,6 +41,30 @@ impl RuntimeExecutor {
        self.execute_prepared(operation, prepared_request).await
    }

+    pub async fn execute_window(
+        &self,
+        operation: &RuntimeOperation,
+        input: &Value,
+    ) -> Result<WindowExecutionResult, RuntimeError> {
+        let Some(streaming) = operation.execution_config.streaming.as_ref() else {
+            return Err(RuntimeError::MissingStreamingConfig {
+                operation_id: operation.operation_id.as_str().to_owned(),
+            });
+        };
+
+        if streaming.mode != ExecutionMode::Window {
+            return Err(RuntimeError::UnsupportedExecutionMode {
+                operation_id: operation.operation_id.as_str().to_owned(),
+                mode: streaming.mode,
+            });
+        }
+
+        let prepared_request = self.prepare_request(operation, input)?;
+        let adapter_response = self.execute_adapter(operation, prepared_request).await?;
+
+        crate::aggregation::collect_window_result(&adapter_response.body, streaming)
+    }
+
    pub fn prepare_request(
        &self,
        operation: &RuntimeOperation,
@@ -55,7 +81,20 @@ impl RuntimeExecutor {
        operation: &RuntimeOperation,
        prepared_request: PreparedRequest,
    ) -> Result<Value, RuntimeError> {
-        let adapter_response = match &operation.target {
+        let adapter_response = self.execute_adapter(operation, prepared_request).await?;
+        let finalized_output = finalize_output(operation, &adapter_response)?;
+
+        operation.output_schema.validate_shape(&finalized_output)?;
+
+        Ok(finalized_output)
+    }
+
+    async fn execute_adapter(
+        &self,
+        operation: &RuntimeOperation,
+        prepared_request: PreparedRequest,
+    ) -> Result<AdapterResponse, RuntimeError> {
+        match &operation.target {
            Target::Grpc(target) => {
                let request = GrpcRequest {
                    headers: merge_headers(
@@ -71,12 +110,12 @@ impl RuntimeExecutor {
                };
                let response = self.grpc_adapter.execute(target, &request).await?;

-                AdapterResponse {
+                Ok(AdapterResponse {
                    status_code: response.status_code,
                    headers: response.headers,
                    body: response.body.clone(),
                    data: response.body,
-                }
+                })
            }
            Target::Graphql(target) => {
                let request = GraphqlRequest {
@@ -90,12 +129,12 @@ impl RuntimeExecutor {
                };
                let response = self.graphql_adapter.execute(target, &request).await?;

-                AdapterResponse {
+                Ok(AdapterResponse {
                    status_code: response.status_code,
                    headers: response.headers,
                    body: response.body,
                    data: response.data,
-                }
+                })
            }
            Target::Rest(target) => {
                let request = RestRequest {
@@ -111,20 +150,14 @@ impl RuntimeExecutor {
                };
                let response = self.rest_adapter.execute(target, &request).await?;

-                AdapterResponse {
+                Ok(AdapterResponse {
                    status_code: response.status_code,
                    headers: response.headers,
                    body: response.body,
                    data: Value::Null,
-                }
+                })
            }
-        };
-
-        let finalized_output = finalize_output(operation, &adapter_response)?;
-
-        operation.output_schema.validate_shape(&finalized_output)?;
-
-        Ok(finalized_output)
+        }
    }
 }

@@ -234,9 +267,10 @@ mod tests {
    use axum::{Json, Router, routing::post};
    use crank_adapter_grpc::test_support as grpc_test_support;
    use crank_core::{
-        DescriptorId, ExecutionConfig, GeneratedDraft, GeneratedDraftStatus, GraphqlOperationType,
-        GraphqlTarget, GrpcTarget, HttpMethod, Operation, OperationId, OperationStatus, Protocol,
-        RestTarget, Samples, Target, ToolDescription, ToolExample,
+        AggregationMode, DescriptorId, ExecutionConfig, ExecutionMode, GeneratedDraft,
+        GeneratedDraftStatus, GraphqlOperationType, GraphqlTarget, GrpcTarget, HttpMethod,
+        Operation, OperationId, OperationStatus, Protocol, RestTarget, Samples, StreamingConfig,
+        Target, ToolDescription, ToolExample, ToolFamilyConfig, TransportBehavior,
    };
    use crank_mapping::{MappingRule, MappingSet};
    use crank_schema::{Schema, SchemaKind};
@@ -340,8 +374,81 @@ mod tests {
        assert!(matches!(error, RuntimeError::GraphqlAdapter(_)));
    }

+    #[tokio::test]
+    async fn executes_window_mode_with_raw_items() {
+        let base_url = spawn_runtime_server().await;
+        let executor = RuntimeExecutor::new();
+        let operation = test_window_operation(&base_url, AggregationMode::RawItems, None, None);
+
+        let result = executor
+            .execute_window(&operation, &json!({}))
+            .await
+            .unwrap();
+
+        assert_eq!(result.items.len(), 3);
+        assert_eq!(result.summary, Value::Null);
+        assert!(result.window_complete);
+        assert!(!result.truncated);
+    }
+
+    #[tokio::test]
+    async fn executes_window_mode_with_summary_only() {
+        let base_url = spawn_runtime_server().await;
+        let executor = RuntimeExecutor::new();
+        let operation =
+            test_window_operation(&base_url, AggregationMode::SummaryOnly, Some(10), None);
+
+        let result = executor
+            .execute_window(&operation, &json!({}))
+            .await
+            .unwrap();
+
+        assert_eq!(result.summary, json!({ "service": "billing", "errors": 1 }));
+        assert!(result.items.is_empty());
+    }
+
+    #[tokio::test]
+    async fn executes_window_mode_with_truncation_and_redaction() {
+        let base_url = spawn_runtime_server().await;
+        let executor = RuntimeExecutor::new();
+        let operation = test_window_operation(
+            &base_url,
+            AggregationMode::SummaryPlusSamples,
+            Some(2),
+            Some(120),
+        );
+
+        let result = executor
+            .execute_window(&operation, &json!({}))
+            .await
+            .unwrap();
+
+        assert!(result.truncated);
+        assert!(result.has_more);
+        assert_eq!(result.items.len(), 1);
+        assert_eq!(result.items[0]["secret"], json!("[REDACTED]"));
+        assert_eq!(result.items[0]["message"], json!("disk..."));
+    }
+
+    #[tokio::test]
+    async fn propagates_timeout_in_window_mode() {
+        let base_url = spawn_runtime_server().await;
+        let executor = RuntimeExecutor::new();
+        let operation = test_slow_window_operation(&base_url);
+
+        let error = executor
+            .execute_window(&operation, &json!({}))
+            .await
+            .unwrap_err();
+
+        assert!(matches!(error, RuntimeError::RestAdapter(_)));
+    }
+
    async fn spawn_runtime_server() -> String {
-        let app = Router::new().route("/leads", post(create_lead));
+        let app = Router::new()
+            .route("/leads", post(create_lead))
+            .route("/events", post(events_window))
+            .route("/slow-events", post(slow_events_window));
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();

@@ -383,6 +490,27 @@ mod tests {
        )
    }

+    async fn events_window() -> (axum::http::StatusCode, Json<Value>) {
+        (
+            axum::http::StatusCode::OK,
+            Json(json!({
+                "summary": { "service": "billing", "errors": 1 },
+                "items": [
+                    { "message": "disk pressure detected", "secret": "token-a" },
+                    { "message": "error budget exhausted", "secret": "token-b" },
+                    { "message": "node restarted", "secret": "token-c" }
+                ],
+                "cursor": "cursor_02",
+                "done": true
+            })),
+        )
+    }
+
+    async fn slow_events_window() -> (axum::http::StatusCode, Json<Value>) {
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        events_window().await
+    }
+
    async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
        let email = payload
            .get("variables")
@@ -653,6 +781,107 @@ mod tests {
        })
    }

+    fn test_window_operation(
+        base_url: &str,
+        aggregation_mode: AggregationMode,
+        max_items: Option<u32>,
+        max_bytes: Option<u32>,
+    ) -> RuntimeOperation {
+        RuntimeOperation::from(Operation {
+            id: OperationId::new("op_window_runtime"),
+            name: "billing_log_window".to_owned(),
+            display_name: "Billing Log Window".to_owned(),
+            category: "ops".to_owned(),
+            protocol: Protocol::Rest,
+            status: OperationStatus::Published,
+            version: 1,
+            target: Target::Rest(RestTarget {
+                base_url: base_url.to_owned(),
+                method: HttpMethod::Post,
+                path_template: "/events".to_owned(),
+                static_headers: BTreeMap::new(),
+            }),
+            input_schema: Schema {
+                kind: SchemaKind::Object,
+                description: None,
+                required: true,
+                nullable: false,
+                default_value: None,
+                fields: BTreeMap::new(),
+                items: None,
+                enum_values: Vec::new(),
+                variants: Vec::new(),
+            },
+            output_schema: object_schema("ignored", SchemaKind::String),
+            input_mapping: MappingSet {
+                rules: vec![MappingRule {
+                    source: "$.mcp".to_owned(),
+                    target: "$.request.body".to_owned(),
+                    required: false,
+                    default_value: None,
+                    transform: None,
+                    condition: None,
+                    notes: None,
+                }],
+            },
+            output_mapping: MappingSet { rules: Vec::new() },
+            execution_config: ExecutionConfig {
+                timeout_ms: 1_000,
+                retry_policy: None,
+                auth_profile_ref: None,
+                headers: BTreeMap::new(),
+                protocol_options: None,
+                streaming: Some(StreamingConfig {
+                    mode: ExecutionMode::Window,
+                    transport_behavior: TransportBehavior::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,
+                    aggregation_mode,
+                    summary_path: Some("$.summary".to_owned()),
+                    items_path: Some("$.items".to_owned()),
+                    cursor_path: Some("$.cursor".to_owned()),
+                    status_path: None,
+                    done_path: Some("$.done".to_owned()),
+                    redacted_paths: vec!["$.secret".to_owned()],
+                    truncate_item_fields: true,
+                    max_field_length: Some(4),
+                    drop_duplicates: false,
+                    sampling_rate: None,
+                    tool_family: ToolFamilyConfig::default(),
+                }),
+            },
+            tool_description: ToolDescription {
+                title: "Billing Log Window".to_owned(),
+                description: "Collects bounded billing events.".to_owned(),
+                tags: vec!["logs".to_owned()],
+                examples: Vec::new(),
+            },
+            samples: None,
+            generated_draft: None,
+            config_export: None,
+            created_at: "2026-04-06T12:00:00Z".to_owned(),
+            updated_at: "2026-04-06T12:00:00Z".to_owned(),
+            published_at: Some("2026-04-06T12:00:00Z".to_owned()),
+        })
+    }
+
+    fn test_slow_window_operation(base_url: &str) -> RuntimeOperation {
+        let mut operation =
+            test_window_operation(base_url, AggregationMode::RawItems, Some(10), None);
+
+        if let Target::Rest(target) = &mut operation.target {
+            target.path_template = "/slow-events".to_owned();
+        }
+
+        operation.execution_config.timeout_ms = 10;
+        operation
+    }
+
    fn object_schema(field_name: &str, kind: SchemaKind) -> Schema {
        Schema {
            kind: SchemaKind::Object,