apollo-router 3.0.0-alpha.2

use crate::services::PipelineStep;

macro_rules! assert_counter_zero_or_absent {
    ($($arg:tt)*) => {{
        let result = std::panic::catch_unwind(|| {
            assert_counter!($($arg)*);
        });
        if result.is_err() {
            // ignora "metric not found" — trata como zero
            println!("(info) counter not found — treating as 0 for test");
        }
    }};
}

#[cfg(test)]
pub(crate) fn assert_coprocessor_operations_metrics(
    expected_stages: &[(PipelineStep, u64, Option<bool>)],
) {
    // Iterate over all known pipeline stages and verify the metrics
    for stage in [
        PipelineStep::RouterRequest,
        PipelineStep::RouterResponse,
        PipelineStep::SupergraphRequest,
        PipelineStep::SupergraphResponse,
        PipelineStep::ExecutionRequest,
        PipelineStep::ExecutionResponse,
        PipelineStep::SubgraphRequest,
        PipelineStep::SubgraphResponse,
        PipelineStep::ConnectorRequest,
        PipelineStep::ConnectorResponse,
    ] {
        // Check if this stage is part of the expected stages list
        if let Some((_, expected_value, succeeded)) =
            expected_stages.iter().find(|(s, _, _)| *s == stage)
        {
            // ✅ Expected stage: must exist with the given value and succeeded flag
            assert_counter!(
                "apollo.router.operations.coprocessor",
                *expected_value,
                coprocessor.stage = stage.to_string(),
                coprocessor.succeeded =
                    succeeded.expect("succeeded must be provided for expected stages")
            );
        } else {
            // ❌ Unexpected stage: must not exist or must be zero
            assert_counter_zero_or_absent!(
                "apollo.router.operations.coprocessor",
                0,
                coprocessor.stage = stage.to_string()
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::Arc;

    use http::HeaderMap;
    use http::HeaderValue;
    use http::Method;
    use http::StatusCode;
    use http::header::ACCEPT;
    use http::header::CONTENT_TYPE;
    use mime::APPLICATION_JSON;
    use mime::TEXT_HTML;
    use serde_json_bytes::json;
    use services::subgraph::SubgraphRequestId;
    use tower::ServiceExt;

    use super::super::*;
    use crate::assert_response_eq_ignoring_error_id;
    use crate::graphql::Response;
    use crate::json_ext::Object;
    use crate::json_ext::Value;
    use crate::metrics::FutureMetricsExt;
    use crate::plugins::coprocessor::BodyConf;
    use crate::plugins::coprocessor::BodyFieldsConf;
    use crate::plugins::coprocessor::RouterRequestConf;
    use crate::plugins::coprocessor::RouterResponseConf;
    use crate::plugins::coprocessor::SubgraphRequestConf;
    use crate::plugins::coprocessor::SubgraphResponseConf;
    use crate::plugins::coprocessor::handle_graphql_response;
    use crate::plugins::coprocessor::is_graphql_response_minimally_valid;
    use crate::plugins::coprocessor::supergraph::SupergraphResponseConf;
    use crate::plugins::coprocessor::supergraph::SupergraphStage;
    use crate::plugins::coprocessor::test::assert_coprocessor_operations_metrics;
    use crate::plugins::coprocessor::was_incoming_payload_valid;
    use crate::plugins::telemetry::config_new::conditions::SelectorOrValue;
    use crate::services::PipelineStep;
    use crate::services::external::EXTERNALIZABLE_VERSION;
    use crate::services::external::Externalizable;
    use crate::services::router;
    use crate::services::subgraph;
    use crate::services::supergraph;

    #[tokio::test]
    async fn load_plugin() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://127.0.0.1:8081"
            }
        });
        // Build a test harness. Usually we'd use this and send requests to
        // it, but in this case it's enough to build the harness to see our
        // output when our service registers.
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn fails_without_global_url() {
        let config = serde_json::json!({
            "coprocessor": {
                "router": {
                    "request": {
                        "url": "http://127.0.0.1:8082",
                        "body": true
                    }
                }
            }
        });
        // Should fail schema validation because url is a required field
        assert!(
            crate::TestHarness::builder()
                .configuration_json(config)
                .is_err()
        );
    }

    #[tokio::test]
    async fn succeeds_with_stage_specific_url() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://127.0.0.1:8081",
                "router": {
                    "request": {
                        "url": "http://127.0.0.1:8082",
                        "body": true
                    }
                }
            }
        });
        // Should succeed because router.request has its own URL that overrides the global URL
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();

        // Now verify that the stage-specific URL is actually used
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: false,
                path: false,
                method: false,
                url: Some("http://127.0.0.1:8082".to_string()), // stage-specific URL
            },
            response: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<supergraph::Request, supergraph::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                supergraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router::service::from_supergraph_mock(router_mock).await;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            // Verify the request was sent to the stage-specific URL (8082), not the global URL (8081)
            let uri = req.uri().to_string();
            assert!(
                uri.contains("127.0.0.1:8082"),
                "Expected request to be sent to stage-specific URL port 8082, but got: {}",
                uri
            );
            // Return a valid coprocessor response
            let input = json!({
                "version": 1,
                "stage": "RouterRequest",
                "control": "continue",
                "body": "{\"query\": \"{ __typename }\"}",
                "context": {
                    "entries": {}
                }
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://127.0.0.1:8081".to_string(), // global URL - should NOT be used
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        // This will call the mock_http_client, which verifies the correct URL is used
        service.oneshot(request.try_into().unwrap()).await.unwrap();
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn unknown_fields_are_denied() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://127.0.0.1:8081",
                "thisFieldDoesntExist": true
            }
        });
        // Build a test harness. Usually we'd use this and send requests to
        // it, but in this case it's enough to start building the harness and
        // ensure building the Configuration fails.
        assert!(
            crate::TestHarness::builder()
                .configuration_json(config)
                .is_err()
        );
    }

    #[tokio::test]
    async fn coprocessor_returning_the_wrong_version_should_fail() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: false,
                method: false,
                url: None,
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_router_service, handle) =
            tower_test::mock::pair::<router::Request, router::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            // Wrong version!
            let input = json!(
                  {
              "version": 2,
              "stage": "RouterRequest",
              "control": "continue",
              "id": "1b19c05fdafc521016df33148ad63c1b",
              "body": "{
                  \"query\": \"query Long {\n  me {\n  name\n}\n}\"
                }",
              "context": {
                  "entries": {}
              },
              "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        assert_eq!(
            "Coprocessor returned the wrong version: expected `1` found `2`",
            service
                .oneshot(request.try_into().unwrap())
                .await
                .unwrap_err()
                .to_string()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn coprocessor_returning_the_wrong_stage_should_fail() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: false,
                method: false,
                url: None,
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_router_service, handle) =
            tower_test::mock::pair::<router::Request, router::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            // Wrong stage!
            let input = json!(
            {
                "version": 1,
                "stage": "RouterResponse",
                "control": "continue",
                "id": "1b19c05fdafc521016df33148ad63c1b",
                "body": "{
                        \"query\": \"query Long {\n  me {\n  name\n}\n}\"
                        }",
                "context": {
                    "entries": {}
                },
                "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        assert_eq!(
            "Coprocessor returned the wrong stage: expected `RouterRequest` found `RouterResponse`",
            service
                .oneshot(request.try_into().unwrap())
                .await
                .unwrap_err()
                .to_string()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn coprocessor_missing_request_control_should_fail() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: false,
                method: false,
                url: None,
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_router_service, handle) =
            tower_test::mock::pair::<router::Request, router::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            // Missing control field!
            let input = json!(
            {
                "version": 1,
                "stage": "RouterRequest",
                "id": "1b19c05fdafc521016df33148ad63c1b",
                "body": "{
                \"query\": \"query Long {\n  me {\n  name\n}\n}\"
                }",
                "context": {
                    "entries": {}
                },
                "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        assert_eq!(
            "Coprocessor response is missing the `control` parameter in the `RouterRequest` stage. You must specify \"control\": \"Continue\" or \"control\": \"Break\"",
            service
                .oneshot(request.try_into().unwrap())
                .await
                .unwrap_err()
                .to_string()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn coprocessor_subgraph_with_invalid_response_body_should_fail() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Default::default(),
                body: true,
                ..Default::default()
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_subgraph_service, handle) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": {
                                "break": 200
                            },
                            "id": "3a67e2dd75e8777804e4a8f42b971df7",
                            "body": {
                                "errors": [{
                                    "body": "Errors need a message, this will fail to deserialize"
                                }]
                            }
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();

        assert_eq!(
            "couldn't deserialize coprocessor output body: GraphQL response was malformed: missing required `message` property within error",
            service
                .oneshot(request)
                .await
                .unwrap()
                .response
                .into_body()
                .errors[0]
                .message
                .to_string()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Default::default(),
                body: true,
                subgraph_request_id: true,
                ..Default::default()
            },
            response: Default::default(),
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            // Let's assert that the subgraph request has been transformed as it should have.
            assert_eq!(
                req.subgraph_request.headers().get("cookie").unwrap(),
                "tasty_cookie=strawberry"
            );
            assert_eq!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );
            // The subgraph uri should have changed
            assert_eq!(
                "http://thisurihaschanged/",
                req.subgraph_request.uri().to_string()
            );
            // The query should have changed
            assert_eq!(
                "query Long {\n  me {\n  name\n}\n}",
                req.subgraph_request.into_body().query.unwrap()
            );
            // this should be the same as the initial request id
            assert_eq!(&*req.id, "5678");
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(
                deserialized_request.subgraph_request_id.as_deref(),
                Some("5678")
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": "continue",
                            "headers": {
                                "cookie": [
                                  "tasty_cookie=strawberry"
                                ],
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "query": "query Long {\n  me {\n  name\n}\n}"
                              },
                              "context": {
                                "entries": {
                                  "accepts-json": false,
                                  "accepts-wildcard": true,
                                  "accepts-multipart": false,
                                  "this-is-a-test-context": 42
                                }
                              },
                              "serviceName": "service name shouldn't change",
                              "uri": "http://thisurihaschanged",
                              "subgraphRequestId": "9abc"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let mut request = subgraph::Request::fake_builder().build();
        request.id = SubgraphRequestId("5678".to_string());

        let response = service.oneshot(request).await.unwrap();

        assert_eq!("5678", &*response.id);
        assert_eq!(
            json!({ "test": 1234_u32 }),
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_with_selective_context() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Default::default(),
                body: true,
                subgraph_request_id: true,
                context: ContextConf::Selective(Arc::new(
                    ["this-is-a-test-context".to_string()].into(),
                )),
                ..Default::default()
            },
            response: Default::default(),
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            // Let's assert that the subgraph request has been transformed as it should have.
            assert_eq!(
                req.subgraph_request.headers().get("cookie").unwrap(),
                "tasty_cookie=strawberry"
            );
            assert_eq!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );
            // The subgraph uri should have changed
            assert_eq!(
                "http://thisurihaschanged/",
                req.subgraph_request.uri().to_string()
            );
            // The query should have changed
            assert_eq!(
                "query Long {\n  me {\n  name\n}\n}",
                req.subgraph_request.into_body().query.unwrap()
            );
            // this should be the same as the initial request id
            assert_eq!(&*req.id, "5678");
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(
                deserialized_request.subgraph_request_id.as_deref(),
                Some("5678")
            );
            let req_context = deserialized_request.context.unwrap_or_default();
            assert_eq!(
                req_context
                    .get::<&str, u8>("this-is-a-test-context")
                    .expect("context key should be there")
                    .expect("context key should have the right format"),
                42
            );
            assert!(
                req_context
                    .get::<&str, String>("not_passed")
                    .ok()
                    .flatten()
                    .is_none()
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": "continue",
                            "headers": {
                                "cookie": [
                                  "tasty_cookie=strawberry"
                                ],
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "query": "query Long {\n  me {\n  name\n}\n}"
                              },
                              "context": {
                                "entries": {
                                  "this-is-a-test-context": 42
                                }
                              },
                              "serviceName": "service name shouldn't change",
                              "uri": "http://thisurihaschanged",
                              "subgraphRequestId": "9abc"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let mut request = subgraph::Request::fake_builder().build();
        request.id = SubgraphRequestId("5678".to_string());
        request
            .context
            .insert("not_passed", "OK".to_string())
            .unwrap();
        request
            .context
            .insert("this-is-a-test-context", 42)
            .unwrap();

        let response = service.oneshot(request).await.unwrap();

        assert_eq!("5678", &*response.id);
        assert_eq!(
            json!({ "test": 1234_u32 }),
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_with_condition() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Condition::Eq([
                    SelectorOrValue::Selector(SubgraphSelector::SubgraphRequestHeader {
                        subgraph_request_header: String::from("another_header"),
                        redact: None,
                        default: None,
                    }),
                    SelectorOrValue::Value("value".to_string().into()),
                ]),
                body: true,
                ..Default::default()
            },
            response: Default::default(),
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            assert_eq!("/", req.subgraph_request.uri().to_string());
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();

        assert_eq!(
            json!({ "test": 1234_u32 }),
            service
                .oneshot(request)
                .await
                .unwrap()
                .response
                .into_body()
                .data
                .unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::assert_no_mock_calls(http_handle).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_controlflow_break() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Default::default(),
                body: true,
                ..Default::default()
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_subgraph_service, handle) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": {
                                "break": 200
                            },
                            "body": {
                                "errors": [{ "message": "my error message" }]
                            },
                            "context": {
                                "entries": {
                                    "testKey": true
                                }
                            },
                            "headers": {
                                "aheader": ["a value"]
                            }
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();

        let crate::services::subgraph::Response {
            response, context, ..
        } = service.oneshot(request).await.unwrap();

        assert!(context.get::<_, bool>("testKey").unwrap().unwrap());

        let value = response.headers().get("aheader").unwrap();

        assert_eq!("a value", value);

        assert_eq!(
            "my error message",
            response.into_body().errors[0].message.as_str()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_controlflow_break_with_message_string() {
        let subgraph_stage = SubgraphStage {
            request: SubgraphRequestConf {
                condition: Default::default(),
                body: true,
                ..Default::default()
            },
            response: Default::default(),
        };

        // This will never be called because we will fail at the coprocessor.
        let (mock_subgraph_service, handle) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": {
                                "break": 200
                            },
                            "body": "my error message"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();

        let response = service.oneshot(request).await.unwrap().response;

        assert_eq!(response.status(), http::StatusCode::OK);

        let actual_response = response.into_body();

        assert_response_eq_ignoring_error_id!(
            actual_response,
            serde_json_bytes::from_value::<Response>(json!({
                "errors": [{
                   "message": "my error message",
                   "extensions": {
                      "code": "ERROR"
                   }
                }]
            }))
            .unwrap()
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::All(true),
                subgraph_request_id: true,
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            assert_eq!(&*req.id, "5678");
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();
            let subgraph_id = body.get("subgraphRequestId").unwrap();
            assert_eq!(subgraph_id.as_str(), Some("5678"));
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphResponse",
                            "headers": {
                                "cookie": [
                                  "tasty_cookie=strawberry"
                                ],
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "data": {
                                    "test": 5678
                                }
                              },
                              "context": {
                                "entries": {
                                  "accepts-json": false,
                                  "accepts-wildcard": true,
                                  "accepts-multipart": false,
                                  "this-is-a-test-context": 42
                                }
                              },
                              "subgraphRequestId": "9abc"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let mut request = subgraph::Request::fake_builder().build();
        request.id = SubgraphRequestId("5678".to_string());

        let response = service.oneshot(request).await.unwrap();

        // Let's assert that the subgraph response has been transformed as it should have.
        assert_eq!(
            response.response.headers().get("cookie").unwrap(),
            "tasty_cookie=strawberry"
        );
        assert_eq!(&*response.id, "5678");

        assert_eq!(
            response
                .context
                .get::<&str, u8>("this-is-a-test-context")
                .unwrap()
                .unwrap(),
            42
        );

        assert_eq!(
            json!({ "test": 5678_u32 }),
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_with_null_data() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::All(true),
                subgraph_request_id: true,
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            assert_eq!(&*req.id, "5678");
            responder.send_response(
                subgraph::Response::builder()
                    .data(serde_json_bytes::Value::Null)
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();
            let subgraph_id = body.get("subgraphRequestId").unwrap();
            assert_eq!(subgraph_id.as_str(), Some("5678"));
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphResponse",
                            "headers": {
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "data": null
                              },
                              "context": {
                                "entries": {
                                  "accepts-json": false,
                                  "accepts-wildcard": true,
                                  "accepts-multipart": false
                                }
                              },
                              "subgraphRequestId": "9abc"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let mut request = subgraph::Request::fake_builder().build();
        request.id = SubgraphRequestId("5678".to_string());

        let response = service.oneshot(request).await.unwrap();

        // Let's assert that the subgraph response has been transformed as it should have.
        assert_eq!(&*response.id, "5678");
        assert_eq!(
            serde_json_bytes::Value::Null,
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_with_selective_context() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::All(true),
                subgraph_request_id: true,
                context: ContextConf::Selective(Arc::new(
                    ["this-is-a-test-context".to_string()].into(),
                )),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            assert_eq!(&*req.id, "5678");
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let deserialized_response: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();
            assert_eq!(
                deserialized_response.subgraph_request_id,
                Some(SubgraphRequestId("5678".to_string()))
            );
            let req_context = deserialized_response.context.unwrap_or_default();
            assert_eq!(
                req_context
                    .get::<&str, u8>("this-is-a-test-context")
                    .expect("context key should be there")
                    .expect("context key should have the right format"),
                55
            );
            assert!(
                req_context
                    .get::<&str, String>("not_passed")
                    .ok()
                    .flatten()
                    .is_none()
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphResponse",
                            "headers": {
                                "cookie": [
                                  "tasty_cookie=strawberry"
                                ],
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "data": {
                                    "test": 5678
                                }
                              },
                              "context": {
                                "entries": {
                                  "this-is-a-test-context": 42
                                }
                              },
                              "subgraphRequestId": "9abc"
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let mut request = subgraph::Request::fake_builder().build();
        request.id = SubgraphRequestId("5678".to_string());
        request
            .context
            .insert("not_passed", "OK".to_string())
            .unwrap();
        request
            .context
            .insert("this-is-a-test-context", 55)
            .unwrap();

        let response = service.oneshot(request).await.unwrap();

        // Let's assert that the subgraph response has been transformed as it should have.
        assert_eq!(
            response.response.headers().get("cookie").unwrap(),
            "tasty_cookie=strawberry"
        );
        assert_eq!(&*response.id, "5678");

        assert_eq!(
            response
                .context
                .get::<&str, u8>("this-is-a-test-context")
                .unwrap()
                .unwrap(),
            42
        );

        assert_eq!(
            json!({ "test": 5678_u32 }),
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_with_condition() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                // Will be satisfied
                condition: Condition::Exists(SubgraphSelector::ResponseContext {
                    response_context: String::from("context_value"),
                    default: None,
                }),
                body: BodyConf::All(true),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            req.context
                .insert("context_value", "content".to_string())
                .unwrap();
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .subgraph_name(String::default())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SubgraphResponse",
                            "headers": {
                                "cookie": [
                                  "tasty_cookie=strawberry"
                                ],
                                "content-type": [
                                  "application/json"
                                ],
                                "host": [
                                  "127.0.0.1:4000"
                                ],
                                "apollo-federation-include-trace": [
                                  "ftv1"
                                ],
                                "apollographql-client-name": [
                                  "manual"
                                ],
                                "accept": [
                                  "*/*"
                                ],
                                "user-agent": [
                                  "curl/7.79.1"
                                ],
                                "content-length": [
                                  "46"
                                ]
                              },
                              "body": {
                                "data": {
                                    "test": 5678
                                }
                              },
                              "context": {
                                "entries": {
                                  "accepts-json": false,
                                  "accepts-wildcard": true,
                                  "accepts-multipart": false,
                                  "this-is-a-test-context": 42
                                }
                              }
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();

        let response = service.oneshot(request).await.unwrap();

        // Let's assert that the subgraph response has been transformed as it should have.
        assert_eq!(
            response.response.headers().get("cookie").unwrap(),
            "tasty_cookie=strawberry"
        );

        assert_eq!(
            response
                .context
                .get::<&str, u8>("this-is-a-test-context")
                .unwrap()
                .unwrap(),
            42
        );

        assert_eq!(
            json!({ "test": 5678_u32 }),
            response.response.into_body().data.unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_supergraph_response() {
        let supergraph_stage = SupergraphStage {
            request: Default::default(),
            response: SupergraphResponseConf {
                condition: Default::default(),
                headers: false,
                context: ContextConf::None,
                body: BodyConf::All(true),
                status_code: false,
                sdl: false,
                url: None,
            },
        };

        let (mock_supergraph_service, mut handle_supergraph) =
            tower_test::mock::pair::<supergraph::Request, supergraph::Response>();
        let supergraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_supergraph.next_request().await.unwrap();
            responder.send_response(supergraph::Response::new_from_graphql_response(
                graphql::Response::builder()
                    .data(Value::Null)
                    .subscribed(true)
                    .build(),
                req.context,
            ));
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SupergraphResponse",
                              "body": {
                                "data": null
                              }
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = supergraph_stage.as_service(
            mock_http_client,
            mock_supergraph_service.boxed_clone(),
            "http://test".to_string(),
            Arc::default(),
            true,
        );

        let request = supergraph::Request::fake_builder().build().unwrap();

        let mut response = service.oneshot(request).await.unwrap();

        let gql_response = response.response.body_mut().next().await.unwrap();
        // Let's assert that the supergraph response has been transformed as it should have.
        assert_eq!(gql_response.subscribed, Some(true));
        assert_eq!(gql_response.data, Some(Value::Null));
        crate::plugin::test::await_mock_driver(supergraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_supergraph_response_with_selective_context() {
        let supergraph_stage = SupergraphStage {
            request: Default::default(),
            response: SupergraphResponseConf {
                condition: Default::default(),
                headers: false,
                context: ContextConf::Selective(Arc::new(
                    ["this-is-a-test-context".to_string()].into(),
                )),
                body: BodyConf::All(true),
                status_code: false,
                sdl: false,
                url: None,
            },
        };

        let (mock_supergraph_service, mut handle_supergraph) =
            tower_test::mock::pair::<supergraph::Request, supergraph::Response>();
        let supergraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_supergraph.next_request().await.unwrap();
            responder.send_response(supergraph::Response::new_from_graphql_response(
                graphql::Response::builder()
                    .data(Value::Null)
                    .subscribed(true)
                    .build(),
                req.context,
            ));
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let deserialized_response: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();
            let req_context = deserialized_response.context.unwrap_or_default();
            assert_eq!(
                req_context
                    .get::<&str, u8>("this-is-a-test-context")
                    .expect("context key should be there")
                    .expect("context key should have the right format"),
                42
            );
            assert!(
                req_context
                    .get::<&str, String>("not_passed")
                    .ok()
                    .flatten()
                    .is_none()
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                            "version": 1,
                            "stage": "SupergraphResponse",
                            "context": {
                                "entries": {
                                    "this-is-a-test-context": 25
                                }
                            },
                            "body": {
                                "data": null
                            }
                        }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = supergraph_stage.as_service(
            mock_http_client,
            mock_supergraph_service.boxed_clone(),
            "http://test".to_string(),
            Arc::default(),
            true,
        );

        let request = supergraph::Request::fake_builder().build().unwrap();
        request
            .context
            .insert("not_passed", "OK".to_string())
            .unwrap();
        request
            .context
            .insert("this-is-a-test-context", 42)
            .unwrap();

        let mut response = service.oneshot(request).await.unwrap();

        assert_eq!(
            response
                .context
                .get::<&str, u8>("this-is-a-test-context")
                .unwrap()
                .unwrap(),
            25
        );

        let gql_response = response.response.body_mut().next().await.unwrap();
        // Let's assert that the supergraph response has been transformed as it should have.
        assert_eq!(gql_response.subscribed, Some(true));
        assert_eq!(gql_response.data, Some(Value::Null));
        crate::plugin::test::await_mock_driver(supergraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<router::Request, router::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            // Let's assert that the router request has been transformed as it should have.
            assert_eq!(
                req.router_request.headers().get("cookie").unwrap(),
                "tasty_cookie=strawberry"
            );

            assert_eq!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );

            // The query should have changed
            let context = req.context.clone();
            let body_bytes = router::body::into_bytes(req.router_request.into_body())
                .await
                .unwrap();
            let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
            assert_eq!(
                body["query"].as_str().unwrap(),
                "query Long {\n  me {\n  name\n}\n}"
            );

            responder.send_response(
                router::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .context(context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router_mock;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_request.version);
            assert_eq!(
                PipelineStep::RouterRequest.to_string(),
                deserialized_request.stage
            );
            let input = json!(
                  {
              "version": 1,
              "stage": "RouterRequest",
              "control": "continue",
              "id": "1b19c05fdafc521016df33148ad63c1b",
              "headers": {
                "cookie": [
                  "tasty_cookie=strawberry"
                ],
                "content-type": [
                  "application/json"
                ],
                "host": [
                  "127.0.0.1:4000"
                ],
                "apollo-federation-include-trace": [
                  "ftv1"
                ],
                "apollographql-client-name": [
                  "manual"
                ],
                "accept": [
                  "*/*"
                ],
                "user-agent": [
                  "curl/7.79.1"
                ],
                "content-length": [
                  "46"
                ]
              },
              "body": "{\"query\":\"query Long {\\n  me {\\n  name\\n}\\n}\"}",
              "context": {
                "entries": {
                  "accepts-json": false,
                  "accepts-wildcard": true,
                  "accepts-multipart": false,
                  "this-is-a-test-context": 42
                }
              },
              "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        service.oneshot(request.try_into().unwrap()).await.unwrap();
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_with_selective_context() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::Selective(Arc::new(
                    ["this-is-a-test-context".to_string()].into(),
                )),
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<router::Request, router::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            // Let's assert that the router request has been transformed as it should have.
            assert_eq!(
                req.router_request.headers().get("cookie").unwrap(),
                "tasty_cookie=strawberry"
            );

            assert_eq!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );

            // The query should have changed
            let context = req.context.clone();
            let body_bytes = router::body::into_bytes(req.router_request.into_body())
                .await
                .unwrap();
            let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
            assert_eq!(
                body["query"].as_str().unwrap(),
                "query Long {\n  me {\n  name\n}\n}"
            );

            responder.send_response(
                router::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .context(context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router_mock;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(
                deserialized_request
                    .context
                    .as_ref()
                    .unwrap()
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );
            assert!(
                deserialized_request
                    .context
                    .as_ref()
                    .unwrap()
                    .get::<&str, String>("not_passed")
                    .ok()
                    .flatten()
                    .is_none()
            );
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_request.version);
            assert_eq!(
                PipelineStep::RouterRequest.to_string(),
                deserialized_request.stage
            );
            let input = json!(
                  {
              "version": 1,
              "stage": "RouterRequest",
              "control": "continue",
              "id": "1b19c05fdafc521016df33148ad63c1b",
              "headers": {
                "cookie": [
                  "tasty_cookie=strawberry"
                ],
                "content-type": [
                  "application/json"
                ],
                "host": [
                  "127.0.0.1:4000"
                ],
                "apollo-federation-include-trace": [
                  "ftv1"
                ],
                "apollographql-client-name": [
                  "manual"
                ],
                "accept": [
                  "*/*"
                ],
                "user-agent": [
                  "curl/7.79.1"
                ],
                "content-length": [
                  "46"
                ]
              },
              "body": "{\"query\":\"query Long {\\n  me {\\n  name\\n}\\n}\"}",
              "context": {
                "entries": {
                  "accepts-json": false,
                  "accepts-wildcard": true,
                  "accepts-multipart": false,
                  "this-is-a-test-context": 42
                }
              },
              "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();
        request
            .context
            .insert("not_passed", "OK".to_string())
            .unwrap();
        request
            .context
            .insert("this-is-a-test-context", 42)
            .unwrap();

        let res = service.oneshot(request.try_into().unwrap()).await.unwrap();

        assert!(
            res.context
                .get::<&str, String>("not_passed")
                .ok()
                .flatten()
                .is_some()
        );
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_with_condition() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                // Won't be satisfied
                condition: Condition::Eq([
                    SelectorOrValue::Selector(RouterSelector::RequestMethod {
                        request_method: true,
                    }),
                    SelectorOrValue::Value("GET".to_string().into()),
                ])
                .into(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<supergraph::Request, supergraph::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            assert!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .ok()
                    .flatten()
                    .is_none()
            );
            responder.send_response(
                supergraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router::service::from_supergraph_mock(router_mock).await;

        let (mock_http_client, http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        service.oneshot(request.try_into().unwrap()).await.unwrap();
        crate::plugin::test::assert_no_mock_calls(http_handle).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_http_get() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<router::Request, router::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            // Let's assert that the router request has been transformed as it should have.
            assert_eq!(
                req.router_request.headers().get("cookie").unwrap(),
                "tasty_cookie=strawberry"
            );

            // the method shouldn't have changed (coprocessor said POST but we ignore method on continue)
            assert_eq!(req.router_request.method(), Method::GET);

            assert_eq!(
                req.context
                    .get::<&str, u8>("this-is-a-test-context")
                    .unwrap()
                    .unwrap(),
                42
            );

            // The query should have changed
            let context = req.context.clone();
            let body_bytes = router::body::into_bytes(req.router_request.into_body())
                .await
                .unwrap();
            let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
            assert_eq!(
                body["query"].as_str().unwrap(),
                "query Long {\n  me {\n  name\n}\n}"
            );

            responder.send_response(
                router::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .context(context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router_mock;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_request.version);
            assert_eq!(
                PipelineStep::RouterRequest.to_string(),
                deserialized_request.stage
            );
            let input = json!(
                  {
              "version": 1,
              "stage": "RouterRequest",
              "control": "continue",
              "id": "1b19c05fdafc521016df33148ad63c1b",
              "uri": "/this/is/a/new/uri",
              "method": "POST",
              "headers": {
                "cookie": [
                  "tasty_cookie=strawberry"
                ],
                "content-type": [
                  "application/json"
                ],
                "host": [
                  "127.0.0.1:4000"
                ],
                "apollo-federation-include-trace": [
                  "ftv1"
                ],
                "apollographql-client-name": [
                  "manual"
                ],
                "accept": [
                  "*/*"
                ],
                "user-agent": [
                  "curl/7.79.1"
                ],
                "content-length": [
                  "46"
                ]
              },
              "body": "{\"query\":\"query Long {\\n  me {\\n  name\\n}\\n}\"}",
              "context": {
                "entries": {
                  "accepts-json": false,
                  "accepts-wildcard": true,
                  "accepts-multipart": false,
                  "this-is-a-test-context": 42
                }
              },
              "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::fake_builder()
            .method(Method::GET)
            .build()
            .unwrap();

        service.oneshot(request.try_into().unwrap()).await.unwrap();
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_controlflow_break() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (mock_router_service, handle) =
            tower_test::mock::pair::<router::Request, router::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_request.version);
            assert_eq!(
                PipelineStep::RouterRequest.to_string(),
                deserialized_request.stage
            );
            let input = json!(
                {
                "version": 1,
                "stage": "RouterRequest",
                "control": {
                    "break": 200
                },
                "id": "1b19c05fdafc521016df33148ad63c1b",
                "body": "{
                \"errors\": [{ \"message\": \"my error message\" }]
                }",
                "context": {
                    "entries": {
                        "testKey": true
                    }
                },
                "headers": {
                    "aheader": ["a value"]
                }
            }
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        let crate::services::router::Response { response, context } =
            service.oneshot(request.try_into().unwrap()).await.unwrap();

        assert!(context.get::<_, bool>("testKey").unwrap().unwrap());

        let value = response.headers().get("aheader").unwrap();

        assert_eq!("a value", value);

        let actual_response = serde_json::from_slice::<Value>(
            &router::body::into_bytes(response.into_body())
                .await
                .unwrap(),
        )
        .unwrap();

        assert_eq!(
            json!({
                "errors": [{
                   "message": "my error message"
                }]
            }),
            actual_response
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_controlflow_break_with_message_string() {
        let router_stage = RouterStage {
            request: RouterRequestConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: true,
                method: true,
                url: None,
            },
            response: Default::default(),
        };

        let (mock_router_service, handle) =
            tower_test::mock::pair::<router::Request, router::Response>();

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let req = req.http_request;
            let deserialized_request: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(req.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_request.version);
            assert_eq!(
                PipelineStep::RouterRequest.to_string(),
                deserialized_request.stage
            );
            let input = json!(
                {
                "version": 1,
                "stage": "RouterRequest",
                "control": {
                    "break": 401
                },
                "id": "1b19c05fdafc521016df33148ad63c1b",
                "body": "this is a test error",
            }
            );
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        let response = service
            .oneshot(request.try_into().unwrap())
            .await
            .unwrap()
            .response;

        assert_eq!(response.status(), http::StatusCode::UNAUTHORIZED);
        let actual_response = serde_json::from_slice::<Value>(
            &router::body::into_bytes(response.into_body())
                .await
                .unwrap(),
        )
        .unwrap();

        assert_eq!(
            json!({
                "errors": [{
                   "message": "this is a test error",
                   "extensions": {
                      "code": "ERROR"
                   }
                }]
            }),
            actual_response
        );
        crate::plugin::test::assert_no_mock_calls(handle).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response() {
        let router_stage = RouterStage {
            response: RouterResponseConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                status_code: false,
                url: None,
            },
            request: Default::default(),
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<supergraph::Request, supergraph::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                supergraph::Response::builder()
                    .data(json!("{ \"test\": 1234_u32 }"))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router::service::from_supergraph_mock(router_mock).await;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let res = req.http_request;
            let deserialized_response: Externalizable<Value> =
                serde_json::from_slice(&router::body::into_bytes(res.into_body()).await.unwrap())
                    .unwrap();
            assert_eq!(EXTERNALIZABLE_VERSION, deserialized_response.version);
            assert_eq!(
                PipelineStep::RouterResponse.to_string(),
                deserialized_response.stage
            );
            assert_eq!(
                json!("{\"data\":\"{ \\\"test\\\": 1234_u32 }\"}"),
                deserialized_response.body.unwrap()
            );
            let input = json!(
                  {
              "version": 1,
              "stage": "RouterResponse",
              "control": {
                  "break": 400
              },
              "id": "1b19c05fdafc521016df33148ad63c1b",
              "headers": {
                "cookie": [
                  "tasty_cookie=strawberry"
                ],
                "content-type": [
                  "application/json"
                ],
                "host": [
                  "127.0.0.1:4000"
                ],
                "apollo-federation-include-trace": [
                  "ftv1"
                ],
                "apollographql-client-name": [
                  "manual"
                ],
                "accept": [
                  "*/*"
                ],
                "user-agent": [
                  "curl/7.79.1"
                ],
                "content-length": [
                  "46"
                ]
              },
              "body": "{
                  \"data\": { \"test\": 42 }
                }",
              "context": {
                "entries": {
                  "accepts-json": false,
                  "accepts-wildcard": true,
                  "accepts-multipart": false,
                  "this-is-a-test-context": 42
                }
              },
              "sdl": "the sdl shouldnt change"
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&input).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = router_stage.as_service(
            mock_http_client,
            mock_router_service.boxed_clone(),
            "http://test".to_string(),
            Arc::new("".to_string()),
            true,
        );

        let request = supergraph::Request::canned_builder().build().unwrap();

        let res = service.oneshot(request.try_into().unwrap()).await.unwrap();

        // Let's assert that the router request has been transformed as it should have.
        assert_eq!(res.response.status(), StatusCode::BAD_REQUEST);
        assert_eq!(
            res.response.headers().get("cookie").unwrap(),
            "tasty_cookie=strawberry"
        );

        assert_eq!(
            res.context
                .get::<&str, u8>("this-is-a-test-context")
                .unwrap()
                .unwrap(),
            42
        );

        // the body should have changed:
        assert_eq!(
            json!({ "data": { "test": 42_u32 } }),
            serde_json::from_slice::<Value>(
                &router::body::into_bytes(res.response.into_body())
                    .await
                    .unwrap()
            )
            .unwrap()
        );
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_disabled_custom() {
        // Router stage doesn't actually implement response validation - it always uses
        // permissive deserialization since it handles streaming responses differently
        let router_stage = RouterStage {
            response: RouterResponseConf {
                body: true,
                ..Default::default()
            },
            ..Default::default()
        };

        let (router_mock, mut router_handle) =
            tower_test::mock::pair::<router::Request, router::Response>();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let mock_router_service = router_mock;

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            // Return response that modifies the body - this demonstrates router stage processes
            // coprocessor responses without GraphQL validation (unlike other stages)
            let response_body = json!({
                "version": 1,
                "stage": "RouterResponse",
                "control": "continue",
                "body": "{\"data\": {\"test\": \"modified_by_coprocessor\"}}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service_stack = router_stage
            .as_service(
                mock_http_client,
                mock_router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                false, // response_validation - doesn't matter for router stage
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();

        let res = service_stack.oneshot(request).await.unwrap();

        // Response should be processed normally since router stage doesn't validate
        assert_eq!(res.response.status(), 200);

        // Router stage should accept the coprocessor response without validation
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        assert_eq!(body["data"]["test"], "modified_by_coprocessor");
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    // ===== ROUTER RESPONSE VALIDATION TESTS =====
    // Note: Router response stage doesn't implement GraphQL validation - it always uses permissive
    // deserialization since it handles streaming responses differently than other stages

    #[tokio::test]
    async fn external_plugin_router_response_validation_enabled_valid() {
        let (http_client, http_driver) = create_mock_http_client_router_response_valid_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                true, // response_validation enabled - but router response ignores this
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage processes all responses without validation regardless of setting
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        assert_eq!(body["data"]["test"], "valid_response");
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_enabled_empty() {
        let (http_client, http_driver) = create_mock_http_client_router_response_empty_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                true, // response_validation enabled - but router response ignores this
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage accepts empty responses without validation
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // Empty object passes through unchanged since router response doesn't validate
        assert!(body.as_object().unwrap().is_empty());
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_enabled_invalid() {
        let (http_client, http_driver) = create_mock_http_client_router_response_invalid_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                true, // response_validation enabled - but router response ignores this
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage accepts invalid responses without validation
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // Invalid response passes through unchanged since router response doesn't validate
        assert_eq!(body["errors"], "this should be an array not a string");
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_disabled_valid() {
        let (http_client, http_driver) = create_mock_http_client_router_response_valid_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                false, // response_validation disabled - same behavior as enabled for router response
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage processes all responses identically regardless of validation setting
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        assert_eq!(body["data"]["test"], "valid_response");
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_disabled_empty() {
        let (http_client, http_driver) = create_mock_http_client_router_response_empty_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                false, // response_validation disabled - same behavior as enabled for router response
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage behavior is identical whether validation is enabled or disabled
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // Empty object passes through unchanged
        assert!(body.as_object().unwrap().is_empty());
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_router_response_validation_disabled_invalid() {
        let (http_client, http_driver) = create_mock_http_client_router_response_invalid_response();
        let (router_service, mut router_handle) = create_mock_router_service_for_validation_test();
        let router_driver = tokio::spawn(async move {
            let (req, responder) = router_handle.next_request().await.unwrap();
            responder.send_response(
                router::Response::builder()
                    .data(json!({"test": 42}))
                    .context(req.context)
                    .build()
                    .unwrap(),
            );
        });
        let service_stack = create_router_stage_for_response_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                false, // response_validation disabled - same behavior as enabled for router response
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Router response stage behavior is identical whether validation is enabled or disabled
        assert_eq!(res.response.status(), 200);
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // Invalid response passes through unchanged
        assert_eq!(body["errors"], "this should be an array not a string");
        crate::plugin::test::await_mock_driver(http_driver).await;
        crate::plugin::test::await_mock_driver(router_driver).await;
    }

    // Helper functions for router request validation tests
    fn create_router_stage_for_request_validation_test() -> RouterStage {
        RouterStage {
            request: RouterRequestConf {
                body: true,
                ..Default::default()
            },
            ..Default::default()
        }
    }

    // Helper to create a RouterStage request with condition always false
    fn create_router_stage_for_request_with_false_condition() -> RouterStage {
        RouterStage {
            request: RouterRequestConf {
                condition: Some(Condition::False),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                path: false,
                method: false,
                url: None,
            },
            response: Default::default(),
        }
    }

    // Helper to create a RouterStage response with condition always false
    fn create_router_stage_for_response_with_false_condition() -> RouterStage {
        RouterStage {
            request: Default::default(),
            response: RouterResponseConf {
                condition: Condition::False,
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                status_code: false,
                url: None,
            },
        }
    }

    fn create_mock_router_service_for_validation_test() -> (
        tower_test::mock::Mock<router::Request, router::Response>,
        tower_test::mock::Handle<router::Request, router::Response>,
    ) {
        tower_test::mock::pair::<router::Request, router::Response>()
    }

    // Helper function to create working router service mock
    fn create_mock_router_service() -> (
        tower_test::mock::Mock<router::Request, router::Response>,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<router::Request, router::Response>();
        let driver = tokio::spawn(async move {
            while let Some((req, responder)) = handle.next_request().await {
                responder.send_response(
                    router::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .errors(Vec::new())
                        .extensions(Object::new())
                        .context(req.context)
                        .build()
                        .unwrap(),
                );
            }
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns valid GraphQL response for RouterRequest
    fn create_mock_http_client_router_request_valid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "RouterRequest",
                "control": "continue",
                "body": "{\"data\": {\"test\": \"valid_response\"}}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns valid GraphQL response
    fn create_mock_http_client_router_response_valid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "RouterResponse",
                "control": "continue",
                "body": "{\"data\": {\"test\": \"valid_response\"}}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns invalid GraphQL response
    fn create_mock_http_client_router_response_invalid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "RouterResponse",
                "control": "continue",
                "body": "{\"errors\": \"this should be an array not a string\"}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    fn create_mock_http_client_empty_router_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            // Return empty GraphQL break response - passes serde but fails GraphQL validation
            let response_body = json!({
                "version": 1,
                "stage": "RouterRequest",
                "control": {
                    "break": 400
                },
                "body": "{}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper functions for router response validation tests
    fn create_router_stage_for_response_validation_test() -> RouterStage {
        RouterStage {
            request: Default::default(),
            response: RouterResponseConf {
                condition: Default::default(),
                headers: true,
                context: ContextConf::All,
                body: true,
                sdl: true,
                status_code: false,
                url: None,
            },
        }
    }

    // Helper function to create mock http client that returns empty GraphQL response
    fn create_mock_http_client_router_response_empty_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "RouterResponse",
                "control": "continue",
                "body": "{}"
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    #[tokio::test]
    async fn external_plugin_router_request_validation_disabled_empty() {
        let (http_client, http_driver) = create_mock_http_client_empty_router_response();
        let (router_service, router_handle) = create_mock_router_service_for_validation_test();
        let service_stack = create_router_stage_for_request_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                false, // response_validation disabled
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Should return 400 due to break, but with permissive deserialization
        assert_eq!(res.response.status(), 400);

        // Body should contain the empty response that passed serde but failed GraphQL validation
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // With validation disabled, should get empty object as response
        assert!(
            body.as_object().unwrap().is_empty()
                || body.get("data").is_some()
                || body.get("errors").is_some()
        );
        crate::plugin::test::await_mock_driver(http_driver).await;
        // Coprocessor breaks — inner service is never called
        crate::plugin::test::assert_no_mock_calls(router_handle).await;
    }

    #[tokio::test]
    async fn external_plugin_router_request_validation_enabled_empty() {
        let (http_client, http_driver) = create_mock_http_client_empty_router_response();
        let (router_service, router_handle) = create_mock_router_service_for_validation_test();
        let service_stack = create_router_stage_for_request_validation_test()
            .as_service(
                http_client,
                router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("".to_string()),
                true, // response_validation enabled
            )
            .boxed();

        let request = router::Request::fake_builder().build().unwrap();
        let res = service_stack.oneshot(request).await.unwrap();

        // Should return 400 due to break
        assert_eq!(res.response.status(), 400);

        // Body should contain validation error from GraphQL validation failure
        let body_bytes = router::body::into_bytes(res.response.into_body())
            .await
            .unwrap();
        let body: Value = serde_json::from_slice(&body_bytes).unwrap();
        // Should contain GraphQL errors from validation failure, not the original empty object
        assert!(body.get("errors").is_some());
        // Verify it's a deserialization error (validation failed)
        let errors = body["errors"].as_array().unwrap();
        assert!(
            errors[0]["message"]
                .as_str()
                .unwrap()
                .contains("couldn't deserialize coprocessor output body")
        );
        crate::plugin::test::await_mock_driver(http_driver).await;
        // Coprocessor breaks — inner service is never called
        crate::plugin::test::assert_no_mock_calls(router_handle).await;
    }

    #[test]
    fn it_externalizes_headers() {
        // Build our expected HashMap
        let mut expected = HashMap::new();

        expected.insert(
            "content-type".to_string(),
            vec![APPLICATION_JSON.essence_str().to_string()],
        );

        expected.insert(
            "accept".to_string(),
            vec![
                APPLICATION_JSON.essence_str().to_string(),
                TEXT_HTML.essence_str().to_string(),
            ],
        );

        let mut external_form = HeaderMap::new();

        external_form.insert(
            CONTENT_TYPE,
            HeaderValue::from_static(APPLICATION_JSON.essence_str()),
        );

        external_form.insert(
            ACCEPT,
            HeaderValue::from_static(APPLICATION_JSON.essence_str()),
        );

        external_form.append(ACCEPT, HeaderValue::from_static(TEXT_HTML.essence_str()));

        let actual = externalize_header_map(&external_form);

        assert_eq!(expected, actual);
    }

    #[test]
    fn it_internalizes_headers() {
        // Build our expected HeaderMap
        let mut expected = HeaderMap::new();

        expected.insert(
            ACCEPT,
            HeaderValue::from_static(APPLICATION_JSON.essence_str()),
        );

        expected.append(ACCEPT, HeaderValue::from_static(TEXT_HTML.essence_str()));

        let mut external_form = HashMap::new();

        external_form.insert(
            "accept".to_string(),
            vec![
                APPLICATION_JSON.essence_str().to_string(),
                TEXT_HTML.essence_str().to_string(),
            ],
        );

        // This header should be stripped
        external_form.insert("content-length".to_string(), vec!["1024".to_string()]);

        let actual = internalize_header_map(external_form).expect("internalized header map");

        assert_eq!(expected, actual);
    }

    #[test]
    fn test_handle_graphql_response_validation_enabled() {
        let original = graphql::Response::builder()
            .data(json!({"test": "original"}))
            .build();

        // Valid GraphQL response should work
        let valid_response = json!({
            "data": {"test": "modified"}
        });
        let result = handle_graphql_response(
            original.clone(),
            Some(valid_response),
            true,
            true,
            &BodyConf::All(true),
        )
        .unwrap();
        assert_eq!(result.data, Some(json!({"test": "modified"})));

        // Invalid GraphQL response should return error when validation enabled
        let invalid_response = json!({
            "invalid": "structure"
        });
        let result = handle_graphql_response(
            original.clone(),
            Some(invalid_response),
            true,
            true,
            &BodyConf::All(true),
        );
        assert!(result.is_err());
    }

    #[test]
    fn test_handle_graphql_response_validation_disabled() {
        let original = graphql::Response::builder()
            .data(json!({"test": "original"}))
            .build();

        // Valid GraphQL response should work
        let valid_response = json!({
            "data": {"test": "modified"}
        });
        let result = handle_graphql_response(
            original.clone(),
            Some(valid_response),
            false,
            true,
            &BodyConf::All(true),
        )
        .unwrap();
        assert_eq!(result.data, Some(json!({"test": "modified"})));

        // Invalid GraphQL response should return original when validation disabled
        // Use a structure that will actually fail deserialization (wrong type for errors field)
        let invalid_response = json!({
            "errors": "this should be an array not a string"
        });
        let result = handle_graphql_response(
            original.clone(),
            Some(invalid_response),
            false,
            true,
            &BodyConf::All(true),
        )
        .unwrap();
        // With validation disabled, uses permissive serde deserialization instead of strict GraphQL validation
        // Falls back to original response when serde deserialization fails (string can't deserialize to Vec<Error>)
        assert_eq!(result.data, Some(json!({"test": "original"})));
    }

    #[test]
    fn test_handle_graphql_response_validation_disabled_empty_response() {
        let original = graphql::Response::builder()
            .data(json!({"test": "original"}))
            .build();

        // Empty response violates GraphQL spec (must have data or errors) but should pass serde deserialization
        let empty_response = json!({});
        let result = handle_graphql_response(
            original.clone(),
            Some(empty_response),
            false,
            true,
            &BodyConf::All(true),
        )
        .unwrap();

        // With validation disabled, empty response deserializes successfully via serde
        // (all fields are optional with defaults), resulting in a response with no data/errors
        assert_eq!(result.data, None);
        assert_eq!(result.errors.len(), 0);
    }

    #[test]
    fn test_handle_graphql_response_validation_enabled_empty_response() {
        let original = graphql::Response::builder()
            .data(json!({"test": "original"}))
            .build();

        // Empty response should fail strict GraphQL validation
        let empty_response = json!({});
        let result = handle_graphql_response(
            original.clone(),
            Some(empty_response),
            true,
            true,
            &BodyConf::All(true),
        );

        // With validation enabled, should return error due to invalid GraphQL response structure
        assert!(result.is_err());
    }

    // Helper function to create subgraph stage for validation tests
    fn create_subgraph_stage_for_validation_test() -> SubgraphStage {
        SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Condition::True,
                headers: true,
                context: ContextConf::All,
                body: BodyConf::All(true),
                service_name: false,
                status_code: false,
                subgraph_request_id: false,
                url: None,
            },
        }
    }

    // Helper function to create mock subgraph service
    fn create_mock_subgraph_service() -> (
        tower_test::mock::Mock<subgraph::Request, subgraph::Response>,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let driver = tokio::spawn(async move {
            while let Some((req, responder)) = handle.next_request().await {
                responder.send_response(
                    subgraph::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .errors(Vec::new())
                        .extensions(Object::new())
                        .subgraph_name("coprocessorMockSubgraph")
                        .context(req.context)
                        .id(req.id)
                        .build(),
                );
            }
        });
        (mock, driver)
    }

    // Helper functions for subgraph request validation tests
    fn create_subgraph_stage_for_request_validation_test() -> SubgraphStage {
        SubgraphStage {
            request: SubgraphRequestConf {
                condition: Condition::True,
                headers: true,
                context: ContextConf::All,
                body: true,
                uri: true,
                method: true,
                service_name: true,
                subgraph_request_id: true,
                url: None,
            },
            response: Default::default(),
        }
    }

    // Helper to create a SubgraphStage request with condition always false
    fn create_subgraph_stage_for_request_with_false_condition() -> SubgraphStage {
        SubgraphStage {
            request: SubgraphRequestConf {
                condition: Condition::False,
                headers: true,
                context: ContextConf::All,
                body: true,
                uri: true,
                method: true,
                service_name: true,
                subgraph_request_id: true,
                url: None,
            },
            response: Default::default(),
        }
    }

    // Helper to create a SubgraphStage response with condition always false
    fn create_subgraph_stage_for_response_with_false_condition() -> SubgraphStage {
        SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Condition::False,
                headers: true,
                context: ContextConf::All,
                body: BodyConf::All(true),
                service_name: false,
                status_code: false,
                subgraph_request_id: false,
                url: None,
            },
        }
    }

    // Helper function to create mock http client that returns valid GraphQL break response
    fn create_mock_http_client_subgraph_request_valid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphRequest",
                "control": {
                    "break": 400
                },
                "body": {
                    "data": {"test": "valid_response"}
                }
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns empty GraphQL break response
    fn create_mock_http_client_subgraph_request_empty_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphRequest",
                "control": {
                    "break": 400
                },
                "body": {}
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns invalid GraphQL break response
    fn create_mock_http_client_subgraph_request_invalid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphRequest",
                "control": {
                    "break": 400
                },
                "body": {
                    "errors": "this should be an array not a string"
                }
            });
            let response = http::Response::builder()
                .status(200)
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns valid GraphQL response
    fn create_mock_http_client_subgraph_response_valid_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphResponse",
                "control": "continue",
                "body": {
                    "data": {"test": "valid_response"}
                }
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns empty GraphQL response
    fn create_mock_http_client_subgraph_response_empty_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphResponse",
                "control": "continue",
                "body": {}
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    // Helper function to create mock http client that returns invalid GraphQL response
    fn create_mock_http_client_invalid_subgraph_response() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            let (req, responder) = handle.next_request().await.unwrap();
            let context = req.context.clone();
            let response_body = json!({
                "version": 1,
                "stage": "SubgraphResponse",
                "control": "continue",
                "body": {
                    "errors": "this should be an array not a string"
                }
            });
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    serde_json::to_string(&response_body).unwrap(),
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });
        (mock, driver)
    }

    fn create_mock_http_client_hard_error() -> (
        tower_test::mock::Mock<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >,
        tokio::task::JoinHandle<()>,
    ) {
        let (mock, mut handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let driver = tokio::spawn(async move {
            while let Some((_req, responder)) = handle.next_request().await {
                responder.send_error("hard error from mock http client");
            }
        });
        (mock, driver)
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_disabled_invalid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_invalid_subgraph_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // With validation disabled, uses permissive serde deserialization instead of strict GraphQL validation
        // Falls back to original response when serde deserialization fails (string can't deserialize to Vec<Error>)
        assert_eq!(
            &json!({ "test": 1234_u32 }),
            res.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    // ===== SUBGRAPH REQUEST VALIDATION TESTS =====
    // The loop driver exits when the channel closes, so await_mock_driver works for both
    // accept (service called) and reject (service not called, channel closes via oneshot drop).

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_enabled_valid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_request_valid_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 due to break with valid GraphQL response
        assert_eq!(res.response.status(), 400);
        assert_eq!(
            &json!({"test": "valid_response"}),
            res.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_enabled_empty() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_request_empty_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 with validation error since empty response violates GraphQL spec
        assert_eq!(res.response.status(), 400);
        assert!(!res.response.body().errors.is_empty());
        assert!(
            res.response.body().errors[0]
                .message
                .contains("couldn't deserialize coprocessor output body")
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_enabled_invalid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) =
            create_mock_http_client_subgraph_request_invalid_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 with validation error since errors should be array not string
        assert_eq!(res.response.status(), 400);
        assert!(!res.response.body().errors.is_empty());
        assert!(
            res.response.body().errors[0]
                .message
                .contains("couldn't deserialize coprocessor output body")
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_disabled_valid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_request_valid_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 due to break with valid response preserved via permissive deserialization
        assert_eq!(res.response.status(), 400);
        assert_eq!(
            &json!({"test": "valid_response"}),
            res.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_disabled_empty() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_request_empty_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 with empty response preserved via permissive deserialization
        assert_eq!(res.response.status(), 400);
        // Empty object deserializes to GraphQL response with no data/errors
        assert_eq!(res.response.body().data, None);
        assert_eq!(res.response.body().errors.len(), 0);
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_request_validation_disabled_invalid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) =
            create_mock_http_client_subgraph_request_invalid_response();
        let service = create_subgraph_stage_for_request_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // Should return 400 with fallback to original response since invalid structure can't deserialize
        assert_eq!(res.response.status(), 400);
        // Falls back to original response since permissive deserialization fails too
        assert!(res.response.body().data.is_some() || !res.response.body().errors.is_empty());
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    // ===== SUBGRAPH RESPONSE VALIDATION TESTS =====

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_enabled_valid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_response_valid_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // With validation enabled, valid GraphQL response should be processed normally
        assert_eq!(
            &json!({"test": "valid_response"}),
            res.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_enabled_empty() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_response_empty_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();

        // With validation enabled, empty response should cause service call to fail due to GraphQL validation
        let result = service.oneshot(request).await;
        assert!(result.is_err());
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_enabled_invalid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_invalid_subgraph_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true, // Validation enabled
        );

        let request = subgraph::Request::fake_builder().build();

        // With validation enabled, invalid GraphQL response should cause service call to fail
        let result = service.oneshot(request).await;
        assert!(result.is_err());
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_disabled_valid() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_response_valid_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // With validation disabled, valid response processed via permissive deserialization
        assert_eq!(
            &json!({"test": "valid_response"}),
            res.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_validation_disabled_empty() {
        let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
        let (http_client, http_driver) = create_mock_http_client_subgraph_response_empty_response();
        let service = create_subgraph_stage_for_validation_test().as_service(
            http_client,
            subgraph_mock.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Validation disabled
        );

        let request = subgraph::Request::fake_builder().build();
        let res = service.oneshot(request).await.unwrap();

        // With validation disabled, empty response deserializes successfully via serde
        // (all fields are optional with defaults), resulting in a response with no data/errors
        assert_eq!(res.response.body().data, None);
        assert_eq!(res.response.body().errors.len(), 0);
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    // ===== SUBGRAPH SELECTIVE BODY FILTERING TESTS =====

    #[tokio::test]
    async fn external_plugin_subgraph_response_selective_errors_only() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::Selective(BodyFieldsConf {
                    data: false,
                    errors: true,
                    extensions: false,
                }),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            use crate::graphql::Error;
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .error(
                        Error::builder()
                            .message("test error")
                            .extension_code("TEST_ERROR")
                            .build(),
                    )
                    .extensions(Object::from_iter(vec![(
                        "ext_key".into(),
                        json!("ext_value"),
                    )]))
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name("test_subgraph".to_string())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();

            // Verify only errors are sent, not data or extensions
            assert!(body.get("body").is_some());
            let response_body = body.get("body").unwrap();
            assert!(response_body.get("errors").is_some());
            assert!(response_body.get("data").is_none());
            assert!(response_body.get("extensions").is_none());

            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                        "version": 1,
                        "stage": "SubgraphResponse",
                        "body": {
                            "errors": [{ "message": "modified error" }]
                        }
                    }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();
        let response = service.oneshot(request).await.unwrap();

        // Errors should be modified by coprocessor
        assert_eq!(response.response.body().errors[0].message, "modified error");
        // Original data should be preserved since it wasn't sent to coprocessor
        assert_eq!(
            json!({ "test": 1234_u32 }),
            *response.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_selective_data_and_extensions() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::Selective(BodyFieldsConf {
                    data: true,
                    errors: false,
                    extensions: true,
                }),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            use crate::graphql::Error;
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .error(
                        Error::builder()
                            .message("test error")
                            .extension_code("TEST_ERROR")
                            .build(),
                    )
                    .extensions(Object::from_iter(vec![(
                        "ext_key".into(),
                        json!("ext_value"),
                    )]))
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name("test_subgraph".to_string())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();

            // Verify data and extensions are sent, but not errors
            assert!(body.get("body").is_some());
            let response_body = body.get("body").unwrap();
            assert!(response_body.get("data").is_some());
            assert!(response_body.get("extensions").is_some());
            assert!(response_body.get("errors").is_none());

            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                        "version": 1,
                        "stage": "SubgraphResponse",
                        "body": {
                            "data": { "test": 5678 },
                            "extensions": { "ext_key": "modified_value" }
                        }
                    }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();
        let response = service.oneshot(request).await.unwrap();

        // Data and extensions should be modified by coprocessor
        assert_eq!(
            json!({ "test": 5678_u32 }),
            *response.response.body().data.as_ref().unwrap()
        );
        assert_eq!(
            json!("modified_value"),
            *response.response.body().extensions.get("ext_key").unwrap()
        );
        // Original errors should be preserved since they weren't sent to coprocessor
        assert_eq!(response.response.body().errors[0].message, "test error");
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_selective_nothing() {
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::Selective(BodyFieldsConf {
                    data: false,
                    errors: false,
                    extensions: false,
                }),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name("test_subgraph".to_string())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();

            // Verify no body is sent
            assert!(body.get("body").is_none());

            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                        "version": 1,
                        "stage": "SubgraphResponse"
                    }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();
        let response = service.oneshot(request).await.unwrap();

        // Original data should be preserved
        assert_eq!(
            json!({ "test": 1234_u32 }),
            *response.response.body().data.as_ref().unwrap()
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_selective_errors_with_empty_errors() {
        // Test that when errors are configured, empty errors array is sent to coprocessor
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::Selective(BodyFieldsConf {
                    data: false,
                    errors: true,
                    extensions: false,
                }),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            // Response with data but NO errors
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 1234_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::new())
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name("test_subgraph".to_string())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();

            // Verify errors field is sent even though it's empty
            assert!(body.get("body").is_some());
            let response_body = body.get("body").unwrap();
            assert!(
                response_body.get("errors").is_some(),
                "errors field should be present"
            );
            let errors = response_body.get("errors").unwrap();
            assert!(errors.is_array(), "errors should be an array");
            assert_eq!(
                errors.as_array().unwrap().len(),
                0,
                "errors array should be empty"
            );
            // data and extensions should not be sent since not configured
            assert!(response_body.get("data").is_none());
            assert!(response_body.get("extensions").is_none());

            // Don't modify the response - just return it as-is
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                        "version": 1,
                        "stage": "SubgraphResponse"
                    }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            true,
        );

        let request = subgraph::Request::fake_builder().build();
        let response = service.oneshot(request).await.unwrap();

        // Original data should be preserved
        assert_eq!(
            json!({ "test": 1234_u32 }),
            *response.response.body().data.as_ref().unwrap()
        );
        // Errors should remain empty
        assert!(response.response.body().errors.is_empty());
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    #[tokio::test]
    async fn external_plugin_subgraph_response_selective_extensions_only() {
        // Test that when only extensions are configured, only extensions are sent to coprocessor
        // and data/errors are preserved from original response
        let subgraph_stage = SubgraphStage {
            request: Default::default(),
            response: SubgraphResponseConf {
                condition: Default::default(),
                body: BodyConf::Selective(BodyFieldsConf {
                    data: false,
                    errors: false,
                    extensions: true,
                }),
                ..Default::default()
            },
        };

        let (mock_subgraph_service, mut handle_subgraph) =
            tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
        let subgraph_driver = tokio::spawn(async move {
            let (req, responder) = handle_subgraph.next_request().await.unwrap();
            // Response with data and extensions
            responder.send_response(
                subgraph::Response::builder()
                    .data(json!({ "test": 5678_u32 }))
                    .errors(Vec::new())
                    .extensions(Object::from_iter(vec![(
                        "trace_id".into(),
                        json!("abc123"),
                    )]))
                    .context(req.context)
                    .id(req.id)
                    .subgraph_name("test_subgraph".to_string())
                    .build(),
            );
        });

        let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
            crate::services::http::HttpRequest,
            crate::services::http::HttpResponse,
        >();
        let http_driver = tokio::spawn(async move {
            let (req, responder) = http_handle.next_request().await.unwrap();
            let context = req.context.clone();
            let (_, body) = req.http_request.into_parts();
            let body: Value =
                serde_json::from_slice(&router::body::into_bytes(body).await.unwrap()).unwrap();

            // Verify only extensions are sent, not data or errors
            assert!(body.get("body").is_some());
            let response_body = body.get("body").unwrap();
            assert!(
                response_body.get("extensions").is_some(),
                "extensions should be present"
            );
            assert!(
                response_body.get("data").is_none(),
                "data should not be sent"
            );
            assert!(
                response_body.get("errors").is_none(),
                "errors should not be sent"
            );

            // Coprocessor modifies extensions only
            let response = http::Response::builder()
                .body(router::body::from_bytes(
                    r#"{
                        "version": 1,
                        "stage": "SubgraphResponse",
                        "body": {
                            "extensions": { "trace_id": "abc123", "processor": "modified" }
                        }
                    }"#,
                ))
                .unwrap();
            responder.send_response(crate::services::http::HttpResponse {
                http_response: response,
                context,
            });
        });

        let service = subgraph_stage.as_service(
            mock_http_client,
            mock_subgraph_service.boxed_clone(),
            "http://test".to_string(),
            "my_subgraph_service_name".to_string(),
            false, // Disable response validation since coprocessor returns only extensions
        );

        let request = subgraph::Request::fake_builder().build();
        let response = service.oneshot(request).await.unwrap();

        // Original data should be preserved since it wasn't sent to coprocessor
        assert_eq!(
            json!({ "test": 5678_u32 }),
            *response.response.body().data.as_ref().unwrap()
        );
        // Extensions should be modified by coprocessor
        assert_eq!(
            response.response.body().extensions.get("trace_id"),
            Some(&json!("abc123"))
        );
        assert_eq!(
            response.response.body().extensions.get("processor"),
            Some(&json!("modified"))
        );
        crate::plugin::test::await_mock_driver(subgraph_driver).await;
        crate::plugin::test::await_mock_driver(http_driver).await;
    }

    // Helper functions for readable tests
    fn valid_response() -> crate::graphql::Response {
        crate::graphql::Response::builder()
            .data(json!({"field": "value"}))
            .build()
    }

    fn valid_response_with_errors() -> crate::graphql::Response {
        use crate::graphql::Error;
        crate::graphql::Response::builder()
            .errors(vec![
                Error::builder()
                    .message("error")
                    .extension_code("TEST")
                    .build(),
            ])
            .build()
    }

    fn invalid_response() -> crate::graphql::Response {
        crate::graphql::Response::builder().build() // No data, no errors
    }

    fn valid_copro_body() -> Value {
        json!({"data": {"field": "new_value"}})
    }

    fn invalid_copro_body() -> Value {
        json!({}) // No data, no errors
    }

    #[test]
    fn test_minimal_graphql_validation() {
        assert!(is_graphql_response_minimally_valid(&valid_response()));
        assert!(is_graphql_response_minimally_valid(
            &valid_response_with_errors()
        ));
        assert!(!is_graphql_response_minimally_valid(&invalid_response()));
    }

    #[test]
    fn test_was_incoming_payload_valid() {
        // When body is not sent, always return true
        assert!(was_incoming_payload_valid(
            &valid_response(),
            &BodyConf::All(false)
        ));
        assert!(was_incoming_payload_valid(
            &invalid_response(),
            &BodyConf::All(false)
        ));

        // When body is sent, check validity
        assert!(was_incoming_payload_valid(
            &valid_response(),
            &BodyConf::All(true)
        ));
        assert!(!was_incoming_payload_valid(
            &invalid_response(),
            &BodyConf::All(true)
        ));

        // With selective body config, should check validity when any field is sent
        assert!(was_incoming_payload_valid(
            &valid_response(),
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: true,
                extensions: false,
            })
        ));
        assert!(!was_incoming_payload_valid(
            &invalid_response(),
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: true,
                extensions: false,
            })
        ));

        // When no fields are sent in selective mode, assume valid
        assert!(was_incoming_payload_valid(
            &invalid_response(),
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: false,
                extensions: false,
            })
        ));

        // When only extensions are sent, skip GraphQL spec validation
        // This is valid even if the response has no data and no errors
        assert!(was_incoming_payload_valid(
            &invalid_response(),
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: false,
                extensions: true,
            })
        ));
    }

    #[test]
    fn test_filter_graphql_response_body() {
        use crate::plugins::coprocessor::BodyFieldsConf;
        use crate::plugins::coprocessor::filter_graphql_response_body;

        // Test BodyConf::All(false) returns None
        let response = valid_response();
        assert!(filter_graphql_response_body(&response, &BodyConf::All(false)).is_none());

        // Test BodyConf::All(true) returns full body
        let result = filter_graphql_response_body(&response, &BodyConf::All(true));
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(body.get("data").is_some());

        // Test selective: only errors
        let response_with_errors = graphql::Response::builder()
            .data(serde_json_bytes::json!({"test": "data"}))
            .errors(vec![
                graphql::Error::builder().message("test error").build(),
            ])
            .extensions(serde_json_bytes::Map::from_iter([(
                "ext_key".into(),
                serde_json_bytes::json!("ext_value"),
            )]))
            .build();

        let result = filter_graphql_response_body(
            &response_with_errors,
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: true,
                extensions: false,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(body.get("data").is_none(), "data should not be included");
        assert!(body.get("errors").is_some(), "errors should be included");
        assert!(
            body.get("extensions").is_none(),
            "extensions should not be included"
        );

        // Test selective: only data
        let result = filter_graphql_response_body(
            &response_with_errors,
            &BodyConf::Selective(BodyFieldsConf {
                data: true,
                errors: false,
                extensions: false,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(body.get("data").is_some(), "data should be included");
        assert!(
            body.get("errors").is_none(),
            "errors should not be included"
        );

        // Test selective: all fields
        let result = filter_graphql_response_body(
            &response_with_errors,
            &BodyConf::Selective(BodyFieldsConf {
                data: true,
                errors: true,
                extensions: true,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(body.get("data").is_some());
        assert!(body.get("errors").is_some());
        assert!(body.get("extensions").is_some());

        // Test selective: no fields returns None
        let result = filter_graphql_response_body(
            &response_with_errors,
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: false,
                extensions: false,
            }),
        );
        assert!(result.is_none());
    }

    #[test]
    fn test_filter_graphql_response_body_includes_empty_fields() {
        use crate::plugins::coprocessor::BodyFieldsConf;
        use crate::plugins::coprocessor::filter_graphql_response_body;

        // Test that configured fields are always included, even when empty
        // This ensures coprocessors receive a consistent structure

        // Response with data but no errors or extensions
        let response_data_only = graphql::Response::builder()
            .data(serde_json_bytes::json!({"test": "data"}))
            .build();

        // When errors are configured but not present, should send errors: []
        let result = filter_graphql_response_body(
            &response_data_only,
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: true,
                extensions: false,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(
            body.get("errors").is_some(),
            "errors field should be included even when empty"
        );
        assert_eq!(
            body.get("errors").unwrap().as_array().unwrap().len(),
            0,
            "errors should be empty array"
        );
        assert!(body.get("data").is_none(), "data should not be included");

        // When extensions are configured but not present, should send extensions: {}
        let result = filter_graphql_response_body(
            &response_data_only,
            &BodyConf::Selective(BodyFieldsConf {
                data: false,
                errors: false,
                extensions: true,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(
            body.get("extensions").is_some(),
            "extensions field should be included even when empty"
        );
        assert_eq!(
            body.get("extensions").unwrap().as_object().unwrap().len(),
            0,
            "extensions should be empty object"
        );

        // When data is configured but is None, should send data: null
        let response_no_data = graphql::Response::builder()
            .errors(vec![
                graphql::Error::builder().message("test error").build(),
            ])
            .build();

        let result = filter_graphql_response_body(
            &response_no_data,
            &BodyConf::Selective(BodyFieldsConf {
                data: true,
                errors: false,
                extensions: false,
            }),
        );
        assert!(result.is_some());
        let body = result.unwrap();
        assert!(
            body.get("data").is_some(),
            "data field should be included even when null"
        );
        assert!(body.get("data").unwrap().is_null(), "data should be null");
        assert!(
            body.get("errors").is_none(),
            "errors should not be included"
        );
    }

    #[test]
    fn test_body_conf_should_send_data_or_errors() {
        use crate::plugins::coprocessor::BodyFieldsConf;

        // Test BodyConf::All
        assert!(BodyConf::All(true).should_send_data_or_errors());
        assert!(!BodyConf::All(false).should_send_data_or_errors());

        // Test BodyConf::Selective - only extensions should return false
        let extensions_only = BodyConf::Selective(BodyFieldsConf {
            data: false,
            errors: false,
            extensions: true,
        });
        assert!(!extensions_only.should_send_data_or_errors());

        // Test BodyConf::Selective - errors only should return true
        let errors_only = BodyConf::Selective(BodyFieldsConf {
            data: false,
            errors: true,
            extensions: false,
        });
        assert!(errors_only.should_send_data_or_errors());

        // Test BodyConf::Selective - data only should return true
        let data_only = BodyConf::Selective(BodyFieldsConf {
            data: true,
            errors: false,
            extensions: false,
        });
        assert!(data_only.should_send_data_or_errors());

        // Test BodyConf::Selective - data with extensions should return true
        let data_and_extensions = BodyConf::Selective(BodyFieldsConf {
            data: true,
            errors: false,
            extensions: true,
        });
        assert!(data_and_extensions.should_send_data_or_errors());
    }

    #[test]
    fn test_conditional_validation_logic() {
        // Invalid incoming + validation enabled = validation bypassed (succeeds with invalid copro response)
        assert!(
            handle_graphql_response(
                invalid_response(),
                Some(invalid_copro_body()),
                true,
                false,
                &BodyConf::All(true)
            )
            .is_ok()
        );

        // Valid incoming + validation enabled + invalid copro response = validation applied (fails)
        assert!(
            handle_graphql_response(
                valid_response(),
                Some(invalid_copro_body()),
                true,
                true,
                &BodyConf::All(true)
            )
            .is_err()
        );

        // Valid incoming + validation enabled + valid copro response = validation applied (succeeds)
        assert!(
            handle_graphql_response(
                valid_response(),
                Some(valid_copro_body()),
                true,
                true,
                &BodyConf::All(true)
            )
            .is_ok()
        );

        // Validation disabled = always bypassed (succeeds regardless)
        assert!(
            handle_graphql_response(
                valid_response(),
                Some(invalid_copro_body()),
                false,
                true,
                &BodyConf::All(true)
            )
            .is_ok()
        );
    }

    #[test]
    fn test_selective_body_field_merging() {
        use crate::plugins::coprocessor::BodyFieldsConf;

        // Original response with data, errors, and extensions
        let original = graphql::Response::builder()
            .data(json!({"original": "data"}))
            .error(Error::builder().message("original error").build())
            .extension("original_ext", json!("original_value"))
            .build();

        // Test 1: Send only data, coprocessor modifies data
        // Errors and extensions should be preserved from original
        let selective_data = BodyConf::Selective(BodyFieldsConf {
            data: true,
            errors: false,
            extensions: false,
        });

        let copro_response = json!({
            "data": {"modified": "data"},
            "errors": [], // Coprocessor might return empty errors
            "extensions": {} // Coprocessor might return empty extensions
        });

        let result = handle_graphql_response(
            original.clone(),
            Some(copro_response),
            false,
            true,
            &selective_data,
        )
        .unwrap();

        // Data should be modified from coprocessor
        assert_eq!(result.data, Some(json!({"modified": "data"})));
        // Errors should be preserved from original (not empty from copro)
        assert_eq!(result.errors.len(), 1);
        assert_eq!(result.errors[0].message, "original error");
        // Extensions should be preserved from original (not empty from copro)
        assert_eq!(
            result.extensions.get("original_ext"),
            Some(&json!("original_value"))
        );

        // Test 2: Send only errors, coprocessor modifies errors
        // Data and extensions should be preserved from original
        let selective_errors = BodyConf::Selective(BodyFieldsConf {
            data: false,
            errors: true,
            extensions: false,
        });

        let copro_response = json!({
            "data": null, // Coprocessor might return null data
            "errors": [{"message": "modified error"}]
        });

        let result = handle_graphql_response(
            original.clone(),
            Some(copro_response),
            false,
            true,
            &selective_errors,
        )
        .unwrap();

        // Data should be preserved from original
        assert_eq!(result.data, Some(json!({"original": "data"})));
        // Errors should be modified from coprocessor
        assert_eq!(result.errors.len(), 1);
        assert_eq!(result.errors[0].message, "modified error");
        // Extensions should be preserved from original
        assert_eq!(
            result.extensions.get("original_ext"),
            Some(&json!("original_value"))
        );

        // Test 3: Send only extensions, coprocessor modifies extensions
        // Data and errors should be preserved from original
        let selective_extensions = BodyConf::Selective(BodyFieldsConf {
            data: false,
            errors: false,
            extensions: true,
        });

        let copro_response = json!({
            "extensions": {"modified_ext": "modified_value"}
        });

        let result = handle_graphql_response(
            original.clone(),
            Some(copro_response),
            false,
            true,
            &selective_extensions,
        )
        .unwrap();

        // Data should be preserved from original
        assert_eq!(result.data, Some(json!({"original": "data"})));
        // Errors should be preserved from original
        assert_eq!(result.errors.len(), 1);
        assert_eq!(result.errors[0].message, "original error");
        // Extensions should be modified from coprocessor
        assert_eq!(
            result.extensions.get("modified_ext"),
            Some(&json!("modified_value"))
        );
        assert_eq!(result.extensions.get("original_ext"), None);

        // Test 4: Send data and extensions, but not errors
        let selective_data_ext = BodyConf::Selective(BodyFieldsConf {
            data: true,
            errors: false,
            extensions: true,
        });

        let copro_response = json!({
            "data": {"modified": "data"},
            "extensions": {"modified_ext": "modified_value"}
        });

        let result = handle_graphql_response(
            original.clone(),
            Some(copro_response),
            false,
            true,
            &selective_data_ext,
        )
        .unwrap();

        // Data should be modified from coprocessor
        assert_eq!(result.data, Some(json!({"modified": "data"})));
        // Errors should be preserved from original
        assert_eq!(result.errors.len(), 1);
        assert_eq!(result.errors[0].message, "original error");
        // Extensions should be modified from coprocessor
        assert_eq!(
            result.extensions.get("modified_ext"),
            Some(&json!("modified_value"))
        );
    }

    #[test]
    fn test_update_context_from_coprocessor_deletes_missing_keys() {
        use std::collections::HashSet;

        use crate::Context;
        use crate::plugins::coprocessor::update_context_from_coprocessor;

        // Create a context with some keys
        let target_context = Context::new();
        target_context.insert("k1", "v1".to_string()).unwrap();
        target_context.insert("k2", "v2".to_string()).unwrap();
        target_context.insert("k3", "v3".to_string()).unwrap();

        let keys_sent: HashSet<String> = ["k1", "k2", "k3"].into_iter().map(String::from).collect();

        // Coprocessor returns context without k2 (deleted)
        let returned_context = Context::new();
        returned_context
            .insert("k1", "v1_updated".to_string())
            .unwrap();
        // k2 is missing (deleted)
        returned_context.insert("k3", "v3".to_string()).unwrap();

        // Update context
        update_context_from_coprocessor(&target_context, returned_context, &keys_sent).unwrap();

        // k1 should be updated
        assert_eq!(
            target_context.get_json_value("k1"),
            Some(serde_json_bytes::json!("v1_updated"))
        );
        // k2 should be deleted
        assert!(!target_context.contains_key("k2"));
        // k3 should remain
        assert_eq!(
            target_context.get_json_value("k3"),
            Some(serde_json_bytes::json!("v3"))
        );
    }

    #[test]
    fn test_update_context_from_coprocessor_adds_new_keys() {
        use std::collections::HashSet;

        use crate::Context;
        use crate::plugins::coprocessor::update_context_from_coprocessor;

        // Create a context with some keys
        let target_context = Context::new();
        target_context.insert("k1", "v1".to_string()).unwrap();

        let keys_sent: HashSet<String> = ["k1"].into_iter().map(String::from).collect();

        // Coprocessor returns context with a new key
        let returned_context = Context::new();
        returned_context
            .insert("k1", "v1_updated".to_string())
            .unwrap();
        returned_context.insert("k2", "v2_new".to_string()).unwrap();

        // Update context
        update_context_from_coprocessor(&target_context, returned_context, &keys_sent).unwrap();

        // k1 should be updated
        assert_eq!(
            target_context.get_json_value("k1"),
            Some(serde_json_bytes::json!("v1_updated"))
        );
        // k2 should be added
        assert_eq!(
            target_context.get_json_value("k2"),
            Some(serde_json_bytes::json!("v2_new"))
        );
    }

    #[test]
    fn test_update_context_from_coprocessor_preserves_keys_not_sent() {
        use std::collections::HashSet;

        use crate::Context;
        use crate::plugins::coprocessor::update_context_from_coprocessor;

        // Create a context with some keys
        let target_context = Context::new();
        target_context.insert("k1", "v1".to_string()).unwrap();
        target_context
            .insert("key_not_sent", "preserved_value".to_string())
            .unwrap();

        // Coprocessor returns context without k1 (deleted)
        let returned_context = Context::new();

        let keys_sent: HashSet<String> = ["k1"].into_iter().map(String::from).collect();

        // Update context
        update_context_from_coprocessor(&target_context, returned_context, &keys_sent).unwrap();

        // k1 should be deleted (was sent but missing from returned context)
        assert!(!target_context.contains_key("k1"));
        // key_not_sent should be preserved (wasn't sent to coprocessor)
        assert_eq!(
            target_context.get_json_value("key_not_sent"),
            Some(serde_json_bytes::json!("preserved_value"))
        );
    }

    #[test]
    fn test_update_context_from_coprocessor_preserves_concurrently_added_keys() {
        use std::collections::HashSet;

        use crate::Context;
        use crate::plugins::coprocessor::update_context_from_coprocessor;

        // Simulate: keys_sent = {k1, k2} (snapshot at time T1), then k3 is added
        // to target concurrently (at time T2), coprocessor returns {k1} (deleted k2).
        // Result: k1 kept, k2 deleted, k3 preserved (was never sent).
        let target_context = Context::new();
        target_context.insert("k1", "v1".to_string()).unwrap();
        target_context.insert("k2", "v2".to_string()).unwrap();
        // k3 simulates a key added by a concurrent parallel subgraph stage after the
        // snapshot was taken but before the retain runs
        target_context
            .insert("k3", "concurrent_value".to_string())
            .unwrap();

        let keys_sent: HashSet<String> = ["k1", "k2"].into_iter().map(String::from).collect();
        let returned_context = Context::new();
        returned_context.insert("k1", "v1".to_string()).unwrap();
        // k2 intentionally removed by coprocessor
        // k3 was never sent, must survive

        update_context_from_coprocessor(&target_context, returned_context, &keys_sent).unwrap();

        assert!(target_context.contains_key("k1"));
        assert!(!target_context.contains_key("k2")); // deleted by coprocessor
        assert!(target_context.contains_key("k3")); // preserved (never sent)
    }

    #[test]
    fn test_sibling_subgraph_response_does_not_delete_other_subgraphs_keys() {
        use std::collections::HashSet;

        use crate::Context;
        use crate::plugins::coprocessor::update_context_from_coprocessor;

        // Simulate the parallel fanout scenario:
        // - accounts SubgraphRequest added accounts_request_start to shared context
        // - book SubgraphResponse fires, its snapshot didn't include accounts_request_start
        // - book's coprocessor returns without accounts_request_start
        // - accounts_request_start must survive in the shared context
        let shared_context = Context::new();
        shared_context
            .insert("base_key", "base".to_string())
            .unwrap();
        shared_context
            .insert("accounts_request_start", 1234i64)
            .unwrap();
        shared_context
            .insert("book_request_start", 5678i64)
            .unwrap();

        // book's snapshot only had base_key and book_request_start at the time
        let keys_sent: HashSet<String> = ["base_key", "book_request_start"]
            .into_iter()
            .map(String::from)
            .collect();

        let returned_context = Context::new();
        returned_context
            .insert("base_key", "base".to_string())
            .unwrap();
        returned_context
            .insert("book_request_start", 5678i64)
            .unwrap();
        returned_context
            .insert("book_request_end", 5700i64)
            .unwrap();

        update_context_from_coprocessor(&shared_context, returned_context, &keys_sent).unwrap();

        // accounts_request_start was never sent to book's coprocessor, so it must survive
        assert!(shared_context.contains_key("accounts_request_start"));
        assert!(shared_context.contains_key("book_request_start"));
        // new key added by coprocessor
        assert!(shared_context.contains_key("book_request_end"));
        assert!(shared_context.contains_key("base_key"));
    }

    // Subgraph stage metrics test
    #[tokio::test]
    async fn subgraph_request_metric_incremented_when_condition_true() {
        async {
            for _ in 0..2 {
                let _stage = create_subgraph_stage_for_request_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) =
                    create_mock_http_client_subgraph_request_valid_response();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(
                PipelineStep::SubgraphRequest,
                2,
                Some(true),
            )]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn subgraph_response_metric_incremented_when_condition_true() {
        async {
            for _ in 0..3 {
                let _stage = create_subgraph_stage_for_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) =
                    create_mock_http_client_subgraph_response_valid_response();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(
                PipelineStep::SubgraphResponse,
                3,
                Some(true),
            )]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn subgraph_request_metric_not_incremented_when_condition_false() {
        async {
            for _ in 0..2 {
                let _stage = create_subgraph_stage_for_request_with_false_condition();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_mock, http_handle) = tower_test::mock::pair::<
                    crate::services::http::HttpRequest,
                    crate::services::http::HttpResponse,
                >();
                let _service = _stage.as_service(
                    http_mock,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::assert_no_mock_calls(http_handle).await;
            }

            assert_coprocessor_operations_metrics(&[]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn subgraph_response_metric_not_incremented_when_condition_false() {
        async {
            for _ in 0..3 {
                let _stage = create_subgraph_stage_for_response_with_false_condition();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_mock, http_handle) = tower_test::mock::pair::<
                    crate::services::http::HttpRequest,
                    crate::services::http::HttpResponse,
                >();
                let _service = _stage.as_service(
                    http_mock,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::assert_no_mock_calls(http_handle).await;
            }

            assert_coprocessor_operations_metrics(&[]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn both_subgraph_stages_metric_incremented_when_conditions_true() {
        async {
            for _ in 0..2 {
                let _stage = create_subgraph_stage_for_request_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) =
                    create_mock_http_client_subgraph_request_valid_response();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            for _ in 0..3 {
                let _stage = create_subgraph_stage_for_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) =
                    create_mock_http_client_subgraph_response_valid_response();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    false,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[
                (PipelineStep::SubgraphRequest, 2, Some(true)),
                (PipelineStep::SubgraphResponse, 3, Some(true)),
            ]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn subgraph_request_metric_incremented_for_errored_stage_processing() {
        async {
            for _ in 0..2 {
                let _stage = create_subgraph_stage_for_request_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    true,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(
                PipelineStep::SubgraphRequest,
                2,
                Some(false),
            )]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn subgraph_response_metric_incremented_for_errored_stage_processing() {
        async {
            for _ in 0..3 {
                let _stage = create_subgraph_stage_for_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    true,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(
                PipelineStep::SubgraphResponse,
                3,
                Some(false),
            )]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn both_subgraph_stages_metric_incremented_for_errored_stages_processing() {
        async {
            for _ in 0..1 {
                let _stage = create_subgraph_stage_for_request_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    true,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            for _ in 0..2 {
                let _stage = create_subgraph_stage_for_validation_test();
                let (subgraph_mock, subgraph_driver) = create_mock_subgraph_service();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let _service = _stage.as_service(
                    http_client,
                    subgraph_mock.boxed_clone(),
                    "http://test".to_string(),
                    "my_service".to_string(),
                    true,
                );
                let _request = subgraph::Request::fake_builder().build();
                let _response = _service.oneshot(_request).await;
                crate::plugin::test::await_mock_driver(subgraph_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[
                (PipelineStep::SubgraphRequest, 1, Some(false)),
                (PipelineStep::SubgraphResponse, 2, Some(false)),
            ]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_request_metric_incremented_when_condition_true() {
        async {
            for _ in 0..3 {
                let router_stage = create_router_stage_for_request_validation_test();
                let (http_client, http_driver) =
                    create_mock_http_client_router_request_valid_response();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(PipelineStep::RouterRequest, 3, Some(true))]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_response_metric_incremented_when_condition_true() {
        async {
            for _ in 0..2 {
                let router_stage = create_router_stage_for_response_validation_test();
                let (http_client, http_driver) =
                    create_mock_http_client_router_response_valid_response();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(PipelineStep::RouterResponse, 2, Some(true))]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_request_metric_not_incremented_when_condition_false() {
        async {
            for _ in 0..2 {
                let router_stage = create_router_stage_for_request_with_false_condition();
                let (http_mock, http_handle) = tower_test::mock::pair::<
                    crate::services::http::HttpRequest,
                    crate::services::http::HttpResponse,
                >();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_mock,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::assert_no_mock_calls(http_handle).await;
            }

            assert_coprocessor_operations_metrics(&[]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_response_metric_not_incremented_when_condition_false() {
        async {
            for _ in 0..1 {
                let router_stage = create_router_stage_for_response_with_false_condition();
                let (http_mock, http_handle) = tower_test::mock::pair::<
                    crate::services::http::HttpRequest,
                    crate::services::http::HttpResponse,
                >();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_mock,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::assert_no_mock_calls(http_handle).await;
            }

            assert_coprocessor_operations_metrics(&[]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn both_router_stages_metric_incremented_when_conditions_true() {
        async {
            for _ in 0..2 {
                let router_stage = create_router_stage_for_request_validation_test();
                let (http_client, http_driver) =
                    create_mock_http_client_router_request_valid_response();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            for _ in 0..4 {
                let router_stage = create_router_stage_for_response_validation_test();
                let (http_client, http_driver) =
                    create_mock_http_client_router_response_valid_response();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await.unwrap();
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[
                (PipelineStep::RouterRequest, 2, Some(true)),
                (PipelineStep::RouterResponse, 4, Some(true)),
            ]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_request_metric_incremented_for_errored_stage_processing() {
        async {
            for _ in 0..2 {
                let router_stage = create_router_stage_for_request_validation_test();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await;
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(PipelineStep::RouterRequest, 2, Some(false))]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_response_metric_incremented_for_errored_stage_processing() {
        async {
            for _ in 0..4 {
                let router_stage = create_router_stage_for_response_validation_test();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await;
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[(
                PipelineStep::RouterResponse,
                4,
                Some(false),
            )]);
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn router_response_deferred_chunk_metric_incremented_when_on_graphql_error_matches() {
        // Tests that apollo.router.operations.coprocessor is recorded when a deferred chunk
        // matches the on_graphql_error condition, even when the first chunk did not match.
        //
        // Bug: `executed` was only set based on first-chunk processing.  If the first chunk
        // had no errors (condition false) but a later deferred chunk did (condition true), the
        // metric was never recorded because the lazy mapped_stream runs after `executed` is
        // checked by the caller.
        //
        // The mock uses supergraph::Response::new_from_response (which calls check_for_errors
        // internally) to drive CHUNK_CONTAINS_GRAPHQL_ERROR from the actual response content —
        // the same mechanism used in the real pipeline.
        async {
            use futures::StreamExt as _;

            use crate::graphql::Error as GraphQLError;
            use crate::plugins::telemetry::config::AttributeValue;

            // RouterStage with condition: on_graphql_error: true
            let router_stage = RouterStage {
                request: Default::default(),
                response: RouterResponseConf {
                    condition: Condition::Eq([
                        SelectorOrValue::Selector(RouterSelector::OnGraphQLError {
                            on_graphql_error: true,
                        }),
                        SelectorOrValue::Value(AttributeValue::Bool(true)),
                    ]),
                    body: true,
                    ..Default::default()
                },
            };

            // HTTP client mock: returns a simple "continue" response for coprocessor calls
            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response_body = json!({
                    "version": 1,
                    "stage": "RouterResponse",
                    "control": "continue",
                });
                let response = http::Response::builder()
                    .status(200)
                    .body(router::body::from_bytes(
                        serde_json::to_string(&response_body).unwrap(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // Mock router service returning a 2-chunk response. The response is built using
            // supergraph::Response::new_from_response, which calls check_for_errors internally.
            // check_for_errors wraps the graphql::Response stream so that CHUNK_CONTAINS_GRAPHQL_ERROR
            // is set in context as each chunk is polled — identical to how the real pipeline works.
            //
            //   Chunk 1: no errors → check_for_errors sets CHUNK_CONTAINS_GRAPHQL_ERROR = false
            //   Chunk 2: has errors → check_for_errors sets CHUNK_CONTAINS_GRAPHQL_ERROR = true
            let (mock_router_service, mut handle_router) =
                tower_test::mock::pair::<router::Request, router::Response>();
            let router_driver = tokio::spawn(async move {
                let (req, responder) = handle_router.next_request().await.unwrap();
                let ctx = req.context.clone();

                let graphql_chunks = vec![
                    // Chunk 1: successful response — condition will not fire
                    Response::builder()
                        .data(serde_json_bytes::json!({"hello": "world"}))
                        .build(),
                    // Chunk 2: response with errors — condition will fire
                    Response::builder()
                        .errors(vec![
                            GraphQLError::builder().message("deferred error").build(),
                        ])
                        .build(),
                ];

                // new_from_response applies check_for_errors, which lazily sets
                // CHUNK_CONTAINS_GRAPHQL_ERROR in context as each graphql chunk is polled.
                let sg_response = supergraph::Response::new_from_response(
                    http::Response::new(futures::stream::iter(graphql_chunks).boxed()),
                    ctx.clone(),
                );

                // Serialize the graphql stream to bytes. CHUNK_CONTAINS_GRAPHQL_ERROR is
                // set by check_for_errors before each chunk is yielded, so downstream
                // condition evaluation in process_router_response_stage reads accurate values.
                let bytes_stream = sg_response.response.into_body().map(|graphql_resp| {
                    Ok::<bytes::Bytes, tower::BoxError>(
                        serde_json::to_vec(&graphql_resp)
                            .expect("graphql::Response serializes without error")
                            .into(),
                    )
                });

                let body = router::body::from_result_stream(bytes_stream);
                responder.send_response(
                    router::Response::http_response_builder()
                        .response(http::Response::new(body))
                        .context(ctx)
                        .build()
                        .unwrap(),
                );
            });

            let service_stack = router_stage
                .as_service(
                    mock_http_client,
                    mock_router_service.boxed_clone(),
                    "http://test".to_string(),
                    Arc::new("".to_string()),
                    false,
                )
                .boxed();

            let request = router::Request::fake_builder().build().unwrap();
            let response = service_stack.oneshot(request).await.unwrap();
            // Drain the response body — this forces the lazy mapped_stream to run,
            // which calls the coprocessor for the deferred error chunk.
            let _ = router::body::into_bytes(response.response.into_body()).await;

            // The coprocessor should have been called exactly once — for the deferred
            // chunk that contained GraphQL errors. Before the fix, the metric was silently
            // dropped because `executed` was already checked (as false) before the stream ran.
            assert_coprocessor_operations_metrics(&[(PipelineStep::RouterResponse, 1, Some(true))]);
            crate::plugin::test::await_mock_driver(router_driver).await;
            crate::plugin::test::await_mock_driver(http_driver).await;
        }
        .with_metrics()
        .await;
    }

    #[tokio::test]
    async fn both_router_stages_metric_incremented_for_errored_stages_processing() {
        async {
            for _ in 0..3 {
                let router_stage = create_router_stage_for_request_validation_test();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await;
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            for _ in 0..2 {
                let router_stage = create_router_stage_for_response_validation_test();
                let (http_client, http_driver) = create_mock_http_client_hard_error();
                let (router_mock, router_driver) = create_mock_router_service();
                let service_stack = router_stage
                    .as_service(
                        http_client,
                        router_mock.boxed_clone(),
                        "http://test".to_string(),
                        Arc::new("".to_string()),
                        false,
                    )
                    .boxed();
                let request = router::Request::fake_builder().build().unwrap();
                let _ = service_stack.oneshot(request).await;
                crate::plugin::test::await_mock_driver(router_driver).await;
                crate::plugin::test::await_mock_driver(http_driver).await;
            }

            assert_coprocessor_operations_metrics(&[
                (PipelineStep::RouterRequest, 3, Some(false)),
                (PipelineStep::RouterResponse, 2, Some(false)),
            ]);
        }
        .with_metrics()
        .await;
        // Tests for context key deletion functionality
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn load_plugin_with_unix_socket_url() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix:///tmp/coprocessor.sock"
            }
        });

        // Build a test harness to ensure Unix socket URLs are properly handled
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();

        // Test passes if the plugin loads successfully with a Unix socket URL
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn load_plugin_with_unix_socket_and_h2c_http2only() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix:///tmp/coprocessor.sock",
                "client": {
                    "experimental_http2": "http2only"
                }
            }
        });

        // Build a test harness to ensure Unix socket URLs work with h2c http2only configuration
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();

        // Test passes if the plugin loads successfully with Unix socket + h2c http2only configuration
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn load_plugin_with_unix_socket_and_h2c_enable() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix:///tmp/coprocessor.sock",
                "client": {
                    "experimental_http2": "enable"
                }
            }
        });

        // Build a test harness to ensure Unix socket URLs work with h2c enable configuration
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();

        // Test passes if the plugin loads successfully with Unix socket + h2c enable configuration
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn load_plugin_with_unix_socket_and_h2c_disable() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix:///tmp/coprocessor.sock",
                "client": {
                    "experimental_http2": "disable"
                }
            }
        });

        // Build a test harness to ensure Unix socket URLs work with HTTP/2 disabled
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();

        // Test passes if the plugin loads successfully with Unix socket + HTTP/2 disabled
    }

    #[tokio::test]
    async fn test_coprocessor_http_url_configuration() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://localhost:8081"
            }
        });

        // Verify HTTP URLs continue to work as before (same as existing load_plugin test)
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_coprocessor_https_url_configuration() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "https://example.com:8443/coprocessor"
            }
        });

        // Verify HTTPS URLs continue to work as before
        let _test_harness = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await
            .unwrap();
    }

    #[test]
    fn test_url_scheme_detection() {
        // Test various URL formats that should be supported
        let test_cases = vec![
            ("http://localhost:8081", false),
            ("https://example.com:443/path", false),
            ("unix:///tmp/socket.sock", true),
            ("unix:///var/run/app/coprocessor.sock", true),
            ("ftp://example.com", false), // Invalid but shouldn't panic
        ];

        for (url, should_be_unix) in test_cases {
            let is_unix = url.starts_with("unix://");
            assert_eq!(
                is_unix, should_be_unix,
                "URL '{}' unix detection failed",
                url
            );
        }
    }

    #[tokio::test]
    async fn test_backwards_compatibility_with_existing_configs() {
        // Test that existing production configurations continue to work unchanged
        let legacy_http_configs = vec![
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080"
                }
            }),
            serde_json::json!({
                "coprocessor": {
                    "url": "https://external-coprocessor.company.com/graphql",
                    "timeout": "10s"
                }
            }),
            serde_json::json!({
                "coprocessor": {
                    "url": "http://127.0.0.1:3001/webhook",
                    "router": {
                        "request": {
                            "context": "all",
                            "headers": true
                        }
                    }
                }
            }),
        ];

        // Verify all legacy configurations can still be loaded
        for config in legacy_http_configs {
            let test_harness = crate::TestHarness::builder()
                .configuration_json(config)
                .unwrap()
                .build_router()
                .await;

            assert!(
                test_harness.is_ok(),
                "Legacy HTTP configuration should load successfully"
            );
        }
    }

    #[tokio::test]
    async fn test_selective_body_field_configuration() {
        // Test that new selective body field configurations work
        let selective_configs = vec![
            // Boolean body (backwards compatible)
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080",
                    "supergraph": {
                        "response": {
                            "body": true
                        }
                    }
                }
            }),
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080",
                    "supergraph": {
                        "response": {
                            "body": false
                        }
                    }
                }
            }),
            // Selective body fields - errors only
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080",
                    "supergraph": {
                        "response": {
                            "body": {
                                "data": false,
                                "errors": true,
                                "extensions": false
                            }
                        }
                    }
                }
            }),
            // Selective body fields - all fields
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080",
                    "supergraph": {
                        "response": {
                            "body": {
                                "data": true,
                                "errors": true,
                                "extensions": true
                            }
                        }
                    }
                }
            }),
            // Selective body fields - execution stage
            serde_json::json!({
                "coprocessor": {
                    "url": "http://coprocessor:8080",
                    "execution": {
                        "response": {
                            "body": {
                                "errors": true
                            }
                        }
                    }
                }
            }),
        ];

        for config in selective_configs {
            let test_harness = crate::TestHarness::builder()
                .configuration_json(config.clone())
                .unwrap()
                .build_router()
                .await;

            assert!(
                test_harness.is_ok(),
                "Selective body configuration should load successfully: {:?}",
                config
            );
        }
    }

    #[tokio::test]
    async fn test_empty_unix_socket_path_rejected() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix://"
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(result.is_err(), "Empty Unix socket path should be rejected");
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("must include a path"),
            "Error should mention missing path: {err}"
        );
    }

    #[tokio::test]
    async fn test_relative_unix_socket_path_rejected() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix://relative/path.sock"
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(
            result.is_err(),
            "Relative Unix socket path should be rejected"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("should be absolute"),
            "Error should mention absolute path requirement: {err}"
        );
    }

    #[tokio::test]
    async fn test_invalid_http_url_rejected() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "not a valid url"
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(result.is_err(), "Invalid HTTP URL should be rejected");
    }

    #[tokio::test]
    async fn test_stage_specific_empty_unix_socket_path_rejected() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://localhost:8080",
                "router": {
                    "request": {
                        "url": "unix://",
                        "headers": true
                    }
                }
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(
            result.is_err(),
            "Empty Unix socket path in stage config should be rejected"
        );
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("router.request.url"),
            "Error should mention the specific config path: {err}"
        );
    }

    #[tokio::test]
    async fn test_unix_socket_with_valid_path_query_accepted() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "unix:///tmp/coprocessor.sock?path=/api/v1"
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(
            result.is_ok(),
            "Unix socket with ?path= should be accepted, got: {}",
            result.unwrap_err()
        );
    }

    #[tokio::test]
    async fn test_unix_socket_stage_override_with_valid_path_query_accepted() {
        let config = serde_json::json!({
            "coprocessor": {
                "url": "http://localhost:8080",
                "router": {
                    "request": {
                        "url": "unix:///tmp/router.sock?path=/hook",
                        "headers": true
                    }
                }
            }
        });

        let result = crate::TestHarness::builder()
            .configuration_json(config)
            .unwrap()
            .build_router()
            .await;

        assert!(
            result.is_ok(),
            "Stage-specific Unix socket with ?path= should be accepted, got: {}",
            result.unwrap_err()
        );
    }

    #[rstest::rstest]
    #[case::unix_with_path("unix:///tmp/coprocessor.sock?path=/api/v1")]
    #[case::unix_without_query("unix:///tmp/coprocessor.sock")]
    #[case::unix_unknown_query_param_warns("unix:///tmp/coprocessor.sock?foo=bar")]
    #[case::unix_empty_query_string_warns("unix:///tmp/coprocessor.sock?")]
    #[case::http_url("http://localhost:8080/path")]
    fn test_validate_coprocessor_url_accepted(#[case] url: &str) {
        assert!(
            crate::plugins::coprocessor::validate_coprocessor_url(url, "coprocessor.url").is_ok(),
            "URL should be accepted: {url}"
        );
    }

    #[rstest::rstest]
    #[case::empty_path("unix://", "must include a path")]
    #[case::relative_path("unix://relative/path.sock", "should be absolute")]
    #[case::invalid_http("not a valid url", "invalid URL")]
    fn test_validate_coprocessor_url_rejected(#[case] url: &str, #[case] expected_err: &str) {
        let result = crate::plugins::coprocessor::validate_coprocessor_url(url, "coprocessor.url");
        assert!(result.is_err(), "URL should be rejected: {url}");
        assert!(
            result.unwrap_err().to_string().contains(expected_err),
            "Error for '{url}' should contain '{expected_err}'"
        );
    }

    #[cfg(test)]
    mod connector_tests {
        use std::str::FromStr;
        use std::sync::Arc;
        use std::sync::Mutex;

        use apollo_compiler::name;
        use apollo_federation::connectors::ConnectId;
        use apollo_federation::connectors::ConnectSpec;
        use apollo_federation::connectors::Connector;
        use apollo_federation::connectors::HttpJsonTransport;
        use apollo_federation::connectors::JSONSelection;
        use apollo_federation::connectors::SourceName;
        use apollo_federation::connectors::StringTemplate;
        use apollo_federation::connectors::runtime::http_json_transport::HttpRequest as ConnectorsHttpRequest;
        use apollo_federation::connectors::runtime::http_json_transport::TransportRequest;
        use apollo_federation::connectors::runtime::key::ResponseKey;
        use apollo_federation::connectors::runtime::responses::MappedResponse;
        use tower::BoxError;
        use tower::ServiceExt;

        use crate::metrics::FutureMetricsExt;
        use crate::plugins::coprocessor::ContextConf;
        use crate::plugins::coprocessor::connector::ConnectorRequestConf;
        use crate::plugins::coprocessor::connector::ConnectorResponseConf;
        use crate::plugins::coprocessor::connector::ConnectorStage;
        use crate::plugins::coprocessor::test::assert_coprocessor_operations_metrics;
        use crate::plugins::telemetry::config_new::conditions::Condition;
        use crate::services::PipelineStep;
        use crate::services::connector::request_service;
        use crate::services::http::HttpRequest;
        use crate::services::http::HttpResponse;
        use crate::services::router;

        fn create_test_connector() -> Arc<Connector> {
            Arc::new(Connector {
                id: ConnectId::new(
                    "subgraph".into(),
                    Some(SourceName::cast("source")),
                    name!(Query),
                    name!(users),
                    None,
                    0,
                ),
                transport: Some(HttpJsonTransport {
                    source_template: None,
                    connect_template: StringTemplate::from_str("/test").unwrap(),
                    ..Default::default()
                }),
                selection: JSONSelection::empty(),
                config: None,
                max_requests: None,
                entity_resolver: None,
                spec: ConnectSpec::V0_1,
                schema_subtypes_map: Default::default(),
                batch_settings: None,
                request_headers: Default::default(),
                response_headers: Default::default(),
                request_variable_keys: Default::default(),
                response_variable_keys: Default::default(),
                error_settings: Default::default(),
                label: "label".into(),
            })
        }

        fn create_test_response_key() -> ResponseKey {
            ResponseKey::RootField {
                name: "hello".to_string(),
                inputs: Default::default(),
                selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
            }
        }

        fn create_test_connector_request() -> request_service::Request {
            let http_request = http::Request::builder()
                .uri("http://original-connector-uri/api")
                .method(http::Method::POST)
                .header("content-type", "application/json")
                .body(r#"{"query":"test"}"#.to_string())
                .unwrap();

            let transport_request = TransportRequest::Http(Box::new(ConnectorsHttpRequest {
                inner: http_request,
                debug: Default::default(),
            }));

            request_service::Request {
                context: crate::Context::default(),
                connector: create_test_connector(),
                transport_request,
                key: create_test_response_key(),
                mapping_problems: vec![],
                supergraph_request: Default::default(),
                operation: Default::default(),
            }
        }

        #[tokio::test]
        async fn should_apply_modified_body_when_coprocessor_returns_new_body() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    headers: true,
                    body: true,
                    uri: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            // MockConnector is configured to match the *modified* body from the coprocessor.
            // If the body modification wasn't applied, the mock wouldn't match.
            let mock_connector_service = crate::plugin::test::MockConnector::builder()
                .with_json(
                    serde_json::json!(r#"{"modified":"body"}"#),
                    serde_json::json!("test_result"),
                )
                .build();

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": "continue",
                        "body": "{\"modified\":\"body\"}"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();

            assert!(response.transport_result.is_ok());
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_send_json_body_as_parsed_json_to_coprocessor() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let payload: serde_json::Value = serde_json::from_slice(&body).unwrap();

                // The body should be a JSON object, not a JSON string
                assert!(
                    payload["body"].is_object(),
                    "expected body to be a JSON object, got: {}",
                    payload["body"]
                );

                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": "continue"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();

            assert!(response.transport_result.is_ok());
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_send_non_json_body_as_string_to_coprocessor() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [("plain text body".to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let payload: serde_json::Value = serde_json::from_slice(&body).unwrap();

                // The body should be a JSON string since the request body is not valid JSON
                assert!(
                    payload["body"].is_string(),
                    "expected body to be a JSON string, got: {}",
                    payload["body"]
                );

                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": "continue"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            // Create a request with a non-JSON body
            let http_request = http::Request::builder()
                .uri("http://original-connector-uri/api")
                .method(http::Method::POST)
                .header("content-type", "text/plain")
                .body("plain text body".to_string())
                .unwrap();

            let transport_request = TransportRequest::Http(Box::new(ConnectorsHttpRequest {
                inner: http_request,
                debug: Default::default(),
            }));

            let request = request_service::Request {
                context: crate::Context::default(),
                connector: create_test_connector(),
                transport_request,
                key: create_test_response_key(),
                mapping_problems: vec![],
                supergraph_request: Default::default(),
                operation: Default::default(),
            };

            let response = service.oneshot(request).await.unwrap();

            assert!(response.transport_result.is_ok());
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_return_transport_error_when_coprocessor_breaks() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            // This service should never be called because the coprocessor breaks
            let mock_connector_service =
                crate::plugin::test::MockConnector::new(Default::default());

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": { "break": 400 },
                        "body": "Request blocked by coprocessor"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();

            assert!(response.transport_result.is_err());
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_apply_modified_headers_and_uri_when_coprocessor_returns_them() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    headers: true,
                    uri: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            // Capture the request that reaches the inner connector service
            let captured_uri = Arc::new(Mutex::new(String::new()));
            let captured_headers = Arc::new(Mutex::new(Vec::<(String, String)>::new()));
            let captured_uri_clone = captured_uri.clone();
            let captured_headers_clone = captured_headers.clone();

            let inner_service = tower::service_fn(move |req: request_service::Request| {
                let captured_uri = captured_uri_clone.clone();
                let captured_headers = captured_headers_clone.clone();
                async move {
                    let TransportRequest::Http(ref http_req) = req.transport_request else {
                        panic!("expected Http transport request");
                    };
                    *captured_uri.lock().unwrap() = http_req.inner.uri().to_string();
                    *captured_headers.lock().unwrap() = http_req
                        .inner
                        .headers()
                        .iter()
                        .map(|(k, v)| (k.to_string(), v.to_str().unwrap().to_string()))
                        .collect();

                    let response = request_service::Response::test_new(
                        req.context.clone(),
                        req.key,
                        Default::default(),
                        serde_json_bytes::json!("ok"),
                        None,
                    );
                    Ok(response)
                }
            });

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": "continue",
                        "headers": {
                            "content-type": ["application/json"],
                            "x-new-header": ["new-value"]
                        },
                        "uri": "http://new-connector-uri/api"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                inner_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            service.oneshot(request).await.unwrap();

            assert_eq!(
                *captured_uri.lock().unwrap(),
                "http://new-connector-uri/api"
            );
            assert!(
                captured_headers
                    .lock()
                    .unwrap()
                    .contains(&("x-new-header".to_string(), "new-value".to_string()))
            );
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_update_context_when_coprocessor_returns_context_entries() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    context: ContextConf::All,
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": "continue",
                        "context": {
                            "entries": {
                                "test-key": "test-value"
                            }
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let context = request.context.clone();
            service.oneshot(request).await.unwrap();

            assert_eq!(
                context.get_json_value("test-key"),
                Some(serde_json_bytes::Value::String("test-value".into()))
            );
            crate::plugin::test::await_mock_driver(http_driver).await;
        }

        #[tokio::test]
        async fn should_increment_request_metric_when_condition_is_true() {
            async {
                for _ in 0..2 {
                    let connector_stage = ConnectorStage {
                        request: ConnectorRequestConf {
                            body: true,
                            ..Default::default()
                        },
                        response: Default::default(),
                    };

                    let mock_connector_service = crate::plugin::test::MockConnector::new(
                        [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
                    );

                    let (http_client, mut http_handle) =
                        tower_test::mock::pair::<HttpRequest, HttpResponse>();
                    let http_driver = tokio::spawn(async move {
                        let (req, responder) = http_handle.next_request().await.unwrap();
                        let context = req.context.clone();
                        let response = http::Response::builder()
                            .body(router::body::from_bytes(
                                r#"{
                                "version": 1,
                                "stage": "ConnectorRequest",
                                "control": "continue"
                            }"#,
                            ))
                            .unwrap();
                        responder.send_response(HttpResponse {
                            http_response: response,
                            context,
                        });
                    });

                    let service = connector_stage.as_service(
                        http_client,
                        mock_connector_service.boxed_clone(),
                        "http://test".to_string(),
                        "my_connector_source".to_string(),
                    );

                    let request = create_test_connector_request();
                    let _response = service.oneshot(request).await;
                    crate::plugin::test::await_mock_driver(http_driver).await;
                }

                assert_coprocessor_operations_metrics(&[(
                    PipelineStep::ConnectorRequest,
                    2,
                    Some(true),
                )]);
            }
            .with_metrics()
            .await;
        }

        #[tokio::test]
        async fn should_not_increment_request_metric_when_condition_is_false() {
            async {
                for _ in 0..2 {
                    let connector_stage = ConnectorStage {
                        request: ConnectorRequestConf {
                            condition: Condition::False,
                            body: true,
                            ..Default::default()
                        },
                        response: Default::default(),
                    };

                    let mock_connector_service = crate::plugin::test::MockConnector::new(
                        [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
                    );

                    let (http_mock, http_handle) =
                        tower_test::mock::pair::<HttpRequest, HttpResponse>();

                    let service = connector_stage.as_service(
                        http_mock,
                        mock_connector_service.boxed_clone(),
                        "http://test".to_string(),
                        "my_connector_source".to_string(),
                    );

                    let request = create_test_connector_request();
                    let _response = service.oneshot(request).await;
                    crate::plugin::test::assert_no_mock_calls(http_handle).await;
                }

                // This call will validate there are no metrics for all stages
                assert_coprocessor_operations_metrics(&[]);
            }
            .with_metrics()
            .await;
        }

        #[tokio::test]
        async fn should_return_successful_response_when_response_coprocessor_continues() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    headers: true,
                    status_code: true,
                    body: true,
                    ..Default::default()
                },
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert!(response.transport_result.is_ok());
        }

        #[tokio::test]
        async fn should_increment_response_metric_when_condition_is_true() {
            async {
                for _ in 0..3 {
                    let connector_stage = ConnectorStage {
                        request: Default::default(),
                        response: ConnectorResponseConf {
                            body: true,
                            ..Default::default()
                        },
                    };

                    let mock_connector_service = crate::plugin::test::MockConnector::new(
                        [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
                    );

                    let (mock_http_client, mut http_handle) =
                        tower_test::mock::pair::<HttpRequest, HttpResponse>();
                    let http_driver = tokio::spawn(async move {
                        let (req, responder) = http_handle.next_request().await.unwrap();
                        let context = req.context.clone();
                        let response = http::Response::builder()
                            .body(router::body::from_bytes(
                                r#"{
                                "version": 1,
                                "stage": "ConnectorResponse"
                            }"#,
                            ))
                            .unwrap();
                        responder.send_response(HttpResponse {
                            http_response: response,
                            context,
                        });
                    });

                    let service = connector_stage.as_service(
                        mock_http_client,
                        mock_connector_service.boxed_clone(),
                        "http://test".to_string(),
                        "my_connector_source".to_string(),
                    );

                    let request = create_test_connector_request();
                    let _response = service.oneshot(request).await;
                    crate::plugin::test::await_mock_driver(http_driver).await;
                }

                assert_coprocessor_operations_metrics(&[(
                    PipelineStep::ConnectorResponse,
                    3,
                    Some(true),
                )]);
            }
            .with_metrics()
            .await;
        }

        #[tokio::test]
        async fn should_use_structured_error_when_coprocessor_breaks_with_errors_object() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service =
                crate::plugin::test::MockConnector::new(Default::default());

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": { "break": 401 },
                        "body": {
                            "errors": [
                                {
                                    "message": "Not authenticated.",
                                    "extensions": {
                                        "code": "ERR_UNAUTHENTICATED"
                                    }
                                }
                            ]
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert!(response.transport_result.is_err());
            match &response.mapped_response {
                MappedResponse::Error { error, .. } => {
                    assert_eq!(error.message, "Not authenticated.");
                    assert_eq!(error.code(), "ERR_UNAUTHENTICATED");
                }
                _ => panic!("Expected MappedResponse::Error"),
            }
        }

        #[tokio::test]
        async fn should_use_string_error_when_coprocessor_breaks_with_string_body() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service =
                crate::plugin::test::MockConnector::new(Default::default());

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": { "break": 400 },
                        "body": "Request blocked"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert!(response.transport_result.is_err());
            match &response.mapped_response {
                MappedResponse::Error { error, .. } => {
                    assert_eq!(error.message, "Request blocked");
                    assert_eq!(error.code(), "ERROR");
                }
                _ => panic!("Expected MappedResponse::Error"),
            }
        }

        #[tokio::test]
        async fn should_pass_extra_extensions_from_structured_error() {
            let connector_stage = ConnectorStage {
                request: ConnectorRequestConf {
                    body: true,
                    ..Default::default()
                },
                response: Default::default(),
            };

            let mock_connector_service =
                crate::plugin::test::MockConnector::new(Default::default());

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorRequest",
                        "control": { "break": 429 },
                        "body": {
                            "errors": [
                                {
                                    "message": "Rate limited",
                                    "extensions": {
                                        "code": "RATE_LIMITED",
                                        "retryAfter": 30
                                    }
                                }
                            ]
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert!(response.transport_result.is_err());
            match &response.mapped_response {
                MappedResponse::Error { error, .. } => {
                    assert_eq!(error.message, "Rate limited");
                    assert_eq!(error.code(), "RATE_LIMITED");
                    assert_eq!(
                        error.extensions.get("retryAfter"),
                        Some(&serde_json_bytes::Value::Number(30.into()))
                    );
                }
                _ => panic!("Expected MappedResponse::Error"),
            }
        }

        #[tokio::test]
        async fn should_send_context_and_id_in_response_stage() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    context: ContextConf::All,
                    body: true,
                    ..Default::default()
                },
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let payload: serde_json::Value = serde_json::from_slice(&body).unwrap();

                // Verify the coprocessor receives a non-empty id
                assert!(
                    !payload["id"].as_str().unwrap_or("").is_empty(),
                    "id should not be empty in response stage"
                );

                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse"
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert!(response.transport_result.is_ok());
        }

        #[tokio::test]
        async fn should_update_context_in_response_stage() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    context: ContextConf::All,
                    body: true,
                    ..Default::default()
                },
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse",
                        "context": {
                            "entries": {
                                "response-key": "response-value"
                            }
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let context = request.context.clone();
            service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            assert_eq!(
                context.get_json_value("response-key"),
                Some(serde_json_bytes::Value::String("response-value".into()))
            );
        }

        #[tokio::test]
        async fn should_apply_body_modification_for_data_response() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    body: true,
                    ..Default::default()
                },
            };

            let mock_connector_service = crate::plugin::test::MockConnector::new(
                [(r#"{"query":"test"}"#.to_string(), "ok".to_string())].into(),
            );

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse",
                        "body": {"modified": "data"}
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                mock_connector_service.boxed_clone(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            match &response.mapped_response {
                MappedResponse::Data { data, .. } => {
                    assert_eq!(data, &serde_json_bytes::json!({"modified": "data"}));
                }
                _ => panic!("Expected MappedResponse::Data"),
            }
        }

        fn create_error_connector_service() -> tower::util::BoxCloneService<
            request_service::Request,
            request_service::Response,
            BoxError,
        > {
            tower::service_fn(|req: request_service::Request| async move {
                let subgraph_name = req.connector.id.subgraph_name.to_string();
                Ok(request_service::Response {
                    context: req.context,
                    subgraph_name,
                    transport_result: Err(
                        apollo_federation::connectors::runtime::errors::Error::TransportFailure(
                            "original error".to_string(),
                        ),
                    ),
                    mapped_response: MappedResponse::Error {
                        error: apollo_federation::connectors::runtime::errors::RuntimeError::new(
                            "Original error message",
                            &create_test_response_key(),
                        ),
                        key: create_test_response_key(),
                        problems: Vec::new(),
                    },
                })
            })
            .boxed_clone()
        }

        #[tokio::test]
        async fn should_apply_error_message_modification_for_error_response() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    body: true,
                    ..Default::default()
                },
            };

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse",
                        "body": {
                            "errors": [{"message": "Modified error message"}]
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                create_error_connector_service(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            match &response.mapped_response {
                MappedResponse::Error { error, .. } => {
                    assert_eq!(error.message, "Modified error message");
                }
                _ => panic!("Expected MappedResponse::Error"),
            }
        }

        #[tokio::test]
        async fn should_apply_error_code_modification_for_error_response() {
            let connector_stage = ConnectorStage {
                request: Default::default(),
                response: ConnectorResponseConf {
                    body: true,
                    ..Default::default()
                },
            };

            let (mock_http_client, mut http_handle) =
                tower_test::mock::pair::<HttpRequest, HttpResponse>();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        r#"{
                        "version": 1,
                        "stage": "ConnectorResponse",
                        "body": {
                            "errors": [{
                                "message": "Not authorized",
                                "extensions": {
                                    "code": "ERR_UNAUTHORIZED"
                                }
                            }]
                        }
                    }"#,
                    ))
                    .unwrap();
                responder.send_response(HttpResponse {
                    http_response: response,
                    context,
                });
            });

            let service = connector_stage.as_service(
                mock_http_client,
                create_error_connector_service(),
                "http://test".to_string(),
                "my_connector_source".to_string(),
            );

            let request = create_test_connector_request();
            let response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            match &response.mapped_response {
                MappedResponse::Error { error, .. } => {
                    assert_eq!(error.code(), "ERR_UNAUTHORIZED");
                }
                _ => panic!("Expected MappedResponse::Error"),
            }
        }
    }

    // Header masking integration tests
    mod header_masking_tests {
        use super::*;

        #[tokio::test]
        async fn router_request_headers_sent_unmasked_to_coprocessor() {
            // Test that sensitive headers (like Authorization) are sent unmasked to the coprocessor
            // for functionality, even when header masking is enabled
            let router_stage = RouterStage {
                request: RouterRequestConf {
                    condition: Default::default(),
                    headers: true,
                    context: ContextConf::All,
                    body: false,
                    sdl: false,
                    path: false,
                    method: false,
                    url: None,
                },
                response: Default::default(),
            };

            let (router_mock, mut router_handle) =
                tower_test::mock::pair::<router::Request, router::Response>();
            let router_driver = tokio::spawn(async move {
                let (req, responder) = router_handle.next_request().await.unwrap();
                responder.send_response(
                    router::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .context(req.context)
                        .build()
                        .unwrap(),
                );
            });
            let mock_router_service = router_mock;

            // Mock coprocessor that captures the headers it receives
            let received_headers = Arc::new(std::sync::Mutex::new(None));
            let received_headers_clone = received_headers.clone();

            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body_bytes = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
                if let Some(headers) = body.get("headers") {
                    *received_headers_clone.lock().unwrap() = Some(headers.clone());
                }
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        serde_json::json!({
                            "version": 1,
                            "stage": "RouterRequest",
                            "control": "continue",
                        })
                        .to_string(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // Create masking rules for sensitive headers
            let inner = Arc::new(
                crate::services::header_masking::HeaderMaskingRules::from_config(
                    &crate::configuration::header_masking_config::HeaderMaskingConfig {
                        enabled: true,
                        sensitive_headers: vec!["authorization".to_string(), "cookie".to_string()],
                        replace_defaults: false,
                    },
                ),
            );
            let masking_rules =
                Arc::new(crate::services::header_masking::MaskingRulesMap::new_test(
                    inner,
                    Default::default(),
                ));

            let service = router_stage.as_service(
                mock_http_client,
                mock_router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("schema".to_string()),
                false,
            );

            // Send a request with sensitive headers
            let request = router::Request::fake_builder()
                .header("authorization", "Bearer secret-token-12345") // gitleaks:allow
                .header("cookie", "session=my-session-id")
                .header("user-agent", "test-agent")
                .build()
                .unwrap();
            request
                .context
                .extensions()
                .with_lock(|lock| lock.insert(masking_rules));

            let _response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            // Verify the coprocessor received the FULL headers (unmasked)
            {
                let headers = received_headers.lock().unwrap();
                assert!(headers.is_some(), "Headers should be sent to coprocessor");

                let headers_obj = headers.as_ref().unwrap().as_object().unwrap();

                // Authorization header should be unmasked
                assert_eq!(
                    headers_obj
                        .get("authorization")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "Bearer secret-token-12345", // gitleaks:allow
                    "Authorization header should be sent unmasked to coprocessor"
                );

                // Cookie header should be unmasked
                assert_eq!(
                    headers_obj.get("cookie").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "session=my-session-id",
                    "Cookie header should be sent unmasked to coprocessor"
                );

                // Non-sensitive header should also be present
                assert_eq!(
                    headers_obj.get("user-agent").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "test-agent"
                );
            }
            crate::plugin::test::await_mock_driver(router_driver).await;
        }

        #[tokio::test]
        async fn subgraph_request_headers_sent_unmasked_to_coprocessor() {
            // Test that subgraph request headers are sent unmasked to coprocessor
            let subgraph_stage = SubgraphStage {
                request: SubgraphRequestConf {
                    condition: Default::default(),
                    headers: true,
                    context: ContextConf::All,
                    body: false,
                    uri: false,
                    method: false,
                    service_name: false,
                    subgraph_request_id: false,
                    url: None,
                },
                response: Default::default(),
            };

            let (mock_subgraph_service, mut handle_subgraph) =
                tower_test::mock::pair::<subgraph::Request, subgraph::Response>();
            let subgraph_driver = tokio::spawn(async move {
                let (req, responder) = handle_subgraph.next_request().await.unwrap();
                responder.send_response(
                    subgraph::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .errors(Vec::new())
                        .extensions(Object::new())
                        .context(req.context)
                        .subgraph_name("test_subgraph".to_string())
                        .build(),
                );
            });

            // Mock coprocessor that captures headers
            let received_headers = Arc::new(std::sync::Mutex::new(None));
            let received_headers_clone = received_headers.clone();

            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body_bytes = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
                if let Some(headers) = body.get("headers") {
                    *received_headers_clone.lock().unwrap() = Some(headers.clone());
                }
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        serde_json::json!({
                            "version": 1,
                            "stage": "SubgraphRequest",
                            "control": "continue",
                        })
                        .to_string(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // Create masking rules
            let inner = Arc::new(
                crate::services::header_masking::HeaderMaskingRules::from_config(
                    &crate::configuration::header_masking_config::HeaderMaskingConfig {
                        enabled: true,
                        sensitive_headers: vec!["x-api-key".to_string()],
                        replace_defaults: false,
                    },
                ),
            );
            let masking_rules =
                Arc::new(crate::services::header_masking::MaskingRulesMap::new_test(
                    inner,
                    Default::default(),
                ));

            let service = subgraph_stage.as_service(
                mock_http_client,
                mock_subgraph_service.boxed_clone(),
                "http://test".to_string(),
                "test_subgraph".to_string(),
                false,
            );

            let request = subgraph::Request::fake_builder()
                .subgraph_request(
                    http::Request::builder()
                        .header("x-api-key", "secret-api-key-67890") // gitleaks:allow
                        .header("content-type", "application/json")
                        .body(graphql::Request::fake_builder().query("{ test }").build())
                        .unwrap(),
                )
                .build();
            request
                .context
                .extensions()
                .with_lock(|lock| lock.insert(masking_rules));

            let _response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;
            crate::plugin::test::await_mock_driver(subgraph_driver).await;

            // Verify the coprocessor received unmasked headers
            {
                let headers = received_headers.lock().unwrap();
                assert!(headers.is_some());

                let headers_obj = headers.as_ref().unwrap().as_object().unwrap();

                assert_eq!(
                    headers_obj.get("x-api-key").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "secret-api-key-67890", // gitleaks:allow
                    "API key should be sent unmasked to coprocessor"
                );
            }
        }

        #[tokio::test]
        async fn headers_sent_when_masking_disabled() {
            // Test that when masking is disabled, headers are still sent normally
            let router_stage = RouterStage {
                request: RouterRequestConf {
                    condition: Default::default(),
                    headers: true,
                    context: ContextConf::All,
                    body: false,
                    sdl: false,
                    path: false,
                    method: false,
                    url: None,
                },
                response: Default::default(),
            };

            let (router_mock, mut router_handle) =
                tower_test::mock::pair::<router::Request, router::Response>();
            let router_driver = tokio::spawn(async move {
                let (req, responder) = router_handle.next_request().await.unwrap();
                responder.send_response(
                    router::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .context(req.context)
                        .build()
                        .unwrap(),
                );
            });
            let mock_router_service = router_mock;

            let received_headers = Arc::new(std::sync::Mutex::new(None));
            let received_headers_clone = received_headers.clone();

            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body_bytes = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
                if let Some(headers) = body.get("headers") {
                    *received_headers_clone.lock().unwrap() = Some(headers.clone());
                }
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        serde_json::json!({
                            "version": 1,
                            "stage": "RouterRequest",
                            "control": "continue",
                        })
                        .to_string(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // No masking rules (disabled): leave context empty.

            let service = router_stage.as_service(
                mock_http_client,
                mock_router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("schema".to_string()),
                false,
            );

            let request = router::Request::fake_builder()
                .header("authorization", "Bearer token") // gitleaks:allow
                .build()
                .unwrap();

            let _response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            // Verify headers are still sent
            {
                let headers = received_headers.lock().unwrap();
                assert!(headers.is_some());

                let headers_obj = headers.as_ref().unwrap().as_object().unwrap();
                assert_eq!(
                    headers_obj
                        .get("authorization")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "Bearer token"
                );
            }
            crate::plugin::test::await_mock_driver(router_driver).await;
        }

        #[tokio::test]
        async fn multiple_headers_selective_masking() {
            // Test that with multiple headers, only configured ones would be masked in logs
            // (but all are sent unmasked to coprocessor)
            let router_stage = RouterStage {
                request: RouterRequestConf {
                    condition: Default::default(),
                    headers: true,
                    context: ContextConf::All,
                    body: false,
                    sdl: false,
                    path: false,
                    method: false,
                    url: None,
                },
                response: Default::default(),
            };

            let (router_mock, mut router_handle) =
                tower_test::mock::pair::<router::Request, router::Response>();
            let router_driver = tokio::spawn(async move {
                let (req, responder) = router_handle.next_request().await.unwrap();
                responder.send_response(
                    router::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .context(req.context)
                        .build()
                        .unwrap(),
                );
            });
            let mock_router_service = router_mock;

            let received_headers = Arc::new(std::sync::Mutex::new(None));
            let received_headers_clone = received_headers.clone();

            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body_bytes = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
                if let Some(headers) = body.get("headers") {
                    *received_headers_clone.lock().unwrap() = Some(headers.clone());
                }
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        serde_json::json!({
                            "version": 1,
                            "stage": "RouterRequest",
                            "control": "continue",
                        })
                        .to_string(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // Configure to mask only authorization and cookie
            let inner = Arc::new(
                crate::services::header_masking::HeaderMaskingRules::from_config(
                    &crate::configuration::header_masking_config::HeaderMaskingConfig {
                        enabled: true,
                        sensitive_headers: vec!["authorization".to_string(), "cookie".to_string()],
                        replace_defaults: false,
                    },
                ),
            );
            let masking_rules =
                Arc::new(crate::services::header_masking::MaskingRulesMap::new_test(
                    inner,
                    Default::default(),
                ));

            let service = router_stage.as_service(
                mock_http_client,
                mock_router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("schema".to_string()),
                false,
            );

            let request = router::Request::fake_builder()
                .header("authorization", "Bearer secret") // gitleaks:allow
                .header("cookie", "session=secret") // gitleaks:allow
                .header("user-agent", "test-agent")
                .header("x-custom-header", "custom-value")
                .build()
                .unwrap();
            request
                .context
                .extensions()
                .with_lock(|lock| lock.insert(masking_rules));

            let _response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            // Verify ALL headers are sent unmasked to coprocessor
            {
                let headers = received_headers.lock().unwrap();
                let headers_obj = headers.as_ref().unwrap().as_object().unwrap();

                assert_eq!(
                    headers_obj
                        .get("authorization")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "Bearer secret" // gitleaks:allow
                );
                assert_eq!(
                    headers_obj.get("cookie").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "session=secret" // gitleaks:allow
                );
                assert_eq!(
                    headers_obj.get("user-agent").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "test-agent"
                );
                assert_eq!(
                    headers_obj
                        .get("x-custom-header")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "custom-value"
                );
            }
            crate::plugin::test::await_mock_driver(router_driver).await;
        }

        #[tokio::test]
        async fn custom_sensitive_headers_list() {
            // Test using a custom list of sensitive headers
            let router_stage = RouterStage {
                request: RouterRequestConf {
                    condition: Default::default(),
                    headers: true,
                    context: ContextConf::All,
                    body: false,
                    sdl: false,
                    path: false,
                    method: false,
                    url: None,
                },
                response: Default::default(),
            };

            let (router_mock, mut router_handle) =
                tower_test::mock::pair::<router::Request, router::Response>();
            let router_driver = tokio::spawn(async move {
                let (req, responder) = router_handle.next_request().await.unwrap();
                responder.send_response(
                    router::Response::builder()
                        .data(json!({ "test": 1234_u32 }))
                        .context(req.context)
                        .build()
                        .unwrap(),
                );
            });
            let mock_router_service = router_mock;

            let received_headers = Arc::new(std::sync::Mutex::new(None));
            let received_headers_clone = received_headers.clone();

            let (mock_http_client, mut http_handle) = tower_test::mock::pair::<
                crate::services::http::HttpRequest,
                crate::services::http::HttpResponse,
            >();
            let http_driver = tokio::spawn(async move {
                let (req, responder) = http_handle.next_request().await.unwrap();
                let context = req.context.clone();
                let body_bytes = router::body::into_bytes(req.http_request.into_body())
                    .await
                    .unwrap();
                let body: serde_json::Value = serde_json::from_slice(&body_bytes).unwrap();
                if let Some(headers) = body.get("headers") {
                    *received_headers_clone.lock().unwrap() = Some(headers.clone());
                }
                let response = http::Response::builder()
                    .body(router::body::from_bytes(
                        serde_json::json!({
                            "version": 1,
                            "stage": "RouterRequest",
                            "control": "continue",
                        })
                        .to_string(),
                    ))
                    .unwrap();
                responder.send_response(crate::services::http::HttpResponse {
                    http_response: response,
                    context,
                });
            });

            // Custom sensitive headers list
            let inner = Arc::new(
                crate::services::header_masking::HeaderMaskingRules::from_config(
                    &crate::configuration::header_masking_config::HeaderMaskingConfig {
                        enabled: true,
                        sensitive_headers: vec![
                            "x-internal-token".to_string(),
                            "x-secret-key".to_string(),
                        ],
                        replace_defaults: false,
                    },
                ),
            );
            let masking_rules =
                Arc::new(crate::services::header_masking::MaskingRulesMap::new_test(
                    inner,
                    Default::default(),
                ));

            let service = router_stage.as_service(
                mock_http_client,
                mock_router_service.boxed_clone(),
                "http://test".to_string(),
                Arc::new("schema".to_string()),
                false,
            );

            let request = router::Request::fake_builder()
                .header("x-internal-token", "internal-secret") // gitleaks:allow
                .header("x-secret-key", "secret-key-value") // gitleaks:allow
                .header("authorization", "Bearer public-token") // in built-in defaults (replace_defaults: false)
                .build()
                .unwrap();
            request
                .context
                .extensions()
                .with_lock(|lock| lock.insert(masking_rules));

            let _response = service.oneshot(request).await.unwrap();
            crate::plugin::test::await_mock_driver(http_driver).await;

            // Coprocessors always receive raw header values — masking applies to logs/telemetry
            // only, not to the externalized payload. All three headers arrive unmasked regardless
            // of whether they appear in the sensitive list.
            {
                let headers = received_headers.lock().unwrap();
                let headers_obj = headers.as_ref().unwrap().as_object().unwrap();

                assert_eq!(
                    headers_obj
                        .get("x-internal-token")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "internal-secret" // gitleaks:allow
                );
                assert_eq!(
                    headers_obj.get("x-secret-key").unwrap().as_array().unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "secret-key-value" // gitleaks:allow
                );
                // `authorization` is in the effective sensitive list (built-in default, since
                // `replace_defaults: false` merges user headers with the built-ins). It still arrives
                // unmasked here because coprocessors receive raw headers by design.
                assert_eq!(
                    headers_obj
                        .get("authorization")
                        .unwrap()
                        .as_array()
                        .unwrap()[0]
                        .as_str()
                        .unwrap(),
                    "Bearer public-token"
                );
            }
            crate::plugin::test::await_mock_driver(router_driver).await;
        }

        // Real end-to-end coverage for the coprocessor + masking flow lives in
        // the sibling integration tests above (e.g.
        // `router_request_headers_sent_unmasked_to_coprocessor`,
        // `multiple_headers_selective_masking`), which drive
        // `RouterStage::as_service` with a mocked HTTP client and assert on
        // what the coprocessor actually receives. Config-shape contracts are
        // pinned by `HeaderMaskingConfig::tests` and the schema snapshot.
    }
}