feat: add streaming end-to-end coverage

apps/mcp-server/src/main.rs

@@ -48,7 +48,12 @@ mod tests {
         time::{Duration, SystemTime, UNIX_EPOCH},
     };
 
-    use axum::{Json, Router, http::header, routing::post};
+    use axum::{
+        Json, Router,
+        http::header,
+        response::sse::{Event, KeepAlive, Sse},
+        routing::{get, post},
+    };
     use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
     use crank_adapter_grpc::test_support as grpc_test_support;
     use crank_core::{
@@ -64,6 +69,7 @@ mod tests {
         PublishAgentRequest, PublishRequest,
     };
     use crank_schema::{Schema, SchemaKind};
+    use futures_util::stream;
     use serde_json::{Value, json};
     use sha2::{Digest, Sha256};
     use sqlx::{Executor, postgres::PgPoolOptions};
@@ -1114,6 +1120,72 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn executes_window_operation_via_mcp() {
+        let registry = test_registry().await;
+        let upstream_base_url = spawn_upstream_server().await;
+        let operation = test_rest_window_operation(&upstream_base_url, "crm_window_logs");
+
+        registry
+            .create_operation(&test_workspace_id(), &operation, Some("alice"))
+            .await
+            .unwrap();
+        registry
+            .publish_operation(PublishRequest {
+                workspace_id: &test_workspace_id(),
+                operation_id: &operation.id,
+                version: 1,
+                published_at: "2026-03-26T10:00:00Z",
+                published_by: Some("alice"),
+            })
+            .await
+            .unwrap();
+        publish_agent_for_operation(&registry, &operation, "sales-window").await;
+        let api_key = create_platform_api_key(
+            &registry,
+            "mcp-window",
+            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
+        )
+        .await;
+
+        let base_url = spawn_mcp_server(build_app(
+            registry,
+            Duration::from_millis(0),
+            Some("https://crank.example.com".to_owned()),
+        ))
+        .await;
+        let client = reqwest::Client::new();
+        let mcp_url = agent_mcp_url(&base_url, "sales-window");
+        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+        let call_result = post_jsonrpc(
+            &client,
+            &mcp_url,
+            &api_key,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 3,
+                "method": "tools/call",
+                "params": {
+                    "name": "crm_window_logs",
+                    "arguments": {}
+                }
+            }),
+        )
+        .await;
+
+        assert_eq!(call_result["result"]["isError"], false);
+        assert_eq!(
+            call_result["result"]["structuredContent"]["items"][0]["message"],
+            json!("billing started")
+        );
+        assert_eq!(
+            call_result["result"]["structuredContent"]["window_complete"],
+            json!(true)
+        );
+    }
+
     async fn initialize_session(client: &reqwest::Client, mcp_url: &str, api_key: &str) -> String {
         let initialize_response = client
             .post(mcp_url)
@@ -1220,6 +1292,7 @@ mod tests {
 
     async fn spawn_upstream_server() -> String {
         let app = Router::new()
+            .route("/sse/logs", get(stream_logs))
             .route("/crm/leads", post(create_lead))
             .route("/crm/slow-leads", post(create_slow_lead));
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -1343,6 +1416,19 @@ mod tests {
         }))
     }
 
+    async fn stream_logs()
+    -> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+        let events = vec![
+            json!({ "level": "info", "message": "billing started" }),
+            json!({ "level": "warn", "message": "cache warmup slow" }),
+            json!({ "level": "error", "message": "invoice timeout" }),
+        ];
+        let stream = stream::iter(events.into_iter().map(|payload| {
+            Ok::<_, std::convert::Infallible>(Event::default().data(payload.to_string()))
+        }));
+        Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
+    }
+
     async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
         let email = payload
             .get("variables")
@@ -1670,6 +1756,65 @@ mod tests {
         operation
     }
 
+    fn test_rest_window_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
+        let mut operation = test_operation(base_url, name);
+        operation.display_name = "Window Logs".to_owned();
+        operation.target = Target::Rest(RestTarget {
+            base_url: base_url.to_owned(),
+            method: HttpMethod::Get,
+            path_template: "/sse/logs".to_owned(),
+            static_headers: BTreeMap::new(),
+        });
+        operation.input_schema = optional_object_schema("window");
+        operation.output_schema = empty_object_schema();
+        operation.input_mapping = MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.window".to_owned(),
+                target: "$.request.query.window".to_owned(),
+                required: false,
+                default_value: Some(json!("recent")),
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        };
+        operation.output_mapping = MappingSet { rules: Vec::new() };
+        operation.tool_description = ToolDescription {
+            title: "Window Logs".to_owned(),
+            description: "Collects a bounded SSE log window".to_owned(),
+            tags: vec![
+                "rest".to_owned(),
+                "streaming".to_owned(),
+                "window".to_owned(),
+            ],
+            examples: Vec::new(),
+        };
+        operation.execution_config.streaming = Some(StreamingConfig {
+            mode: ExecutionMode::Window,
+            transport_behavior: TransportBehavior::ServerStream,
+            window_duration_ms: Some(1_000),
+            poll_interval_ms: None,
+            upstream_timeout_ms: Some(1_000),
+            idle_timeout_ms: None,
+            max_session_lifetime_ms: None,
+            max_items: Some(3),
+            max_bytes: Some(16 * 1024),
+            aggregation_mode: AggregationMode::SummaryPlusSamples,
+            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(),
+        });
+        operation
+    }
+
     fn object_schema(field_name: &str) -> Schema {
         Schema {
             kind: SchemaKind::Object,
@@ -1696,4 +1841,45 @@ mod tests {
             variants: Vec::new(),
         }
     }
+
+    fn empty_object_schema() -> 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(),
+        }
+    }
+
+    fn optional_object_schema(field_name: &str) -> Schema {
+        Schema {
+            kind: SchemaKind::Object,
+            description: None,
+            required: true,
+            nullable: false,
+            default_value: None,
+            fields: BTreeMap::from([(
+                field_name.to_owned(),
+                Schema {
+                    kind: SchemaKind::String,
+                    description: None,
+                    required: false,
+                    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(),
+        }
+    }
 }