/*
 * SPDX-FileCopyrightText: 2025 Copyright (c) Contributors to the Eclipse Foundation
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Apache License Version 2.0 which is available at
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * SPDX-License-Identifier: Apache-2.0
 */

use aide::{UseApi, transform::TransformOperation};
use axum::{
    Json,
    extract::{Query, State},
    http::StatusCode,
    response::{IntoResponse as _, Response},
};
use axum_extra::extract::WithRejection;
use cda_interfaces::{DynamicPlugin, SchemaProvider, UdsEcu, file_manager::FileManager};
use cda_plugin_security::Secured;
use sovd_interfaces::components::ecu::operations::OperationCollectionItem;

use crate::sovd::{
    WebserverEcuState, create_schema,
    error::{ApiError, ErrorWrapper},
    locks::require_ecu_access,
};

pub(crate) async fn get<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
    UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
    WithRejection(Query(query), _): WithRejection<
        Query<sovd_interfaces::IncludeSchemaQuery>,
        ApiError,
    >,
    State(WebserverEcuState {
        ecu_name,
        uds,
        locks,
        ..
    }): State<WebserverEcuState<T, U>>,
) -> Response {
    require_ecu_access!(
        read,
        security_plugin,
        &ecu_name,
        &locks,
        query.include_schema
    );
    let security_plugin: DynamicPlugin = security_plugin;
    match uds
        .get_components_operations_info(&ecu_name, &security_plugin)
        .await
    {
        Ok(items) => {
            let schema = if query.include_schema {
                Some(create_schema!(
                    sovd_interfaces::Items<OperationCollectionItem>
                ))
            } else {
                None
            };
            (
                StatusCode::OK,
                Json(sovd_interfaces::Items {
                    items: items
                        .into_iter()
                        .map(|info| OperationCollectionItem {
                            id: info.id,
                            name: info.name,
                            proximity_proof_required: false,
                            asynchronous_execution: info.has_stop || info.has_request_results,
                        })
                        .collect(),
                    schema,
                }),
            )
                .into_response()
        }
        Err(e) => ErrorWrapper {
            error: e.into(),
            include_schema: query.include_schema,
        }
        .into_response(),
    }
}

pub(crate) fn docs_get(op: TransformOperation) -> TransformOperation {
    op.description("Get all available operations for this ECU component")
        .response_with::<200, Json<sovd_interfaces::Items<OperationCollectionItem>>, _>(|res| {
            res.description("List of operations available on this ECU.")
        })
}

pub(crate) mod comparams {

    pub(crate) mod executions {
        use std::sync::Arc;

        use aide::{UseApi, transform::TransformOperation};
        use axum::{
            Json,
            extract::{OriginalUri, Path, Query, State},
            http::{StatusCode, header},
            response::{IntoResponse as _, Response},
        };
        use axum_extra::extract::WithRejection;
        use cda_interfaces::{
            HashMap, HashMapExtensions, UdsEcu, communication_control::CommunicationAccess,
            file_manager::FileManager,
        };
        use cda_plugin_security::Secured;
        use indexmap::IndexMap;
        use opensovd_axum_extra::ExtractHost;
        use sovd_interfaces::components::ecu::operations::comparams as sovd_comparams;
        use tokio::sync::RwLock;
        use uuid::Uuid;

        use crate::sovd::{
            ComparamExecution, IntoSovd, WebserverEcuState, acquire_communication_activity,
            create_schema,
            error::{ApiError, ErrorWrapper},
            locks::require_ecu_access,
        };

        fn parse_exec_uuid(id: &str, include_schema: bool) -> Result<Uuid, ErrorWrapper> {
            Uuid::parse_str(id).map_err(|e| ErrorWrapper {
                error: ApiError::BadRequest(format!("{e:?}")),
                include_schema,
            })
        }

        pub(crate) async fn get<T: UdsEcu + Clone, U: FileManager>(
            UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
            WithRejection(Query(query), _): WithRejection<
                Query<sovd_comparams::executions::get::Query>,
                ApiError,
            >,
            State(WebserverEcuState {
                ecu_name,
                locks,
                comparam_executions,
                ..
            }): State<WebserverEcuState<T, U>>,
        ) -> Response {
            require_ecu_access!(
                read,
                security_plugin,
                &ecu_name,
                &locks,
                query.include_schema
            );
            handler_read(comparam_executions, query.include_schema).await
        }

        pub(crate) fn docs_get(op: TransformOperation) -> TransformOperation {
            op.description("Get all comparam executions")
                .response_with::<200, Json<sovd_comparams::executions::get::Response>, _>(|res| {
                    res.description("Response with all comparam executions.")
                        .example(sovd_comparams::executions::get::Response {
                            items: vec![sovd_comparams::executions::Item {
                                id: "b7e2c1a2-3f4d-4e6a-9c8b-2a1d5e7f8c9b".to_string(),
                            }],
                            schema: None,
                        })
                })
        }

        pub(crate) async fn post<T: UdsEcu + Clone, U: FileManager>(
            UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
            WithRejection(Query(query), _): WithRejection<
                Query<sovd_comparams::executions::get::Query>,
                ApiError,
            >,
            State(WebserverEcuState {
                ecu_name,
                locks,
                comparam_executions,
                communication_access,
                ..
            }): State<WebserverEcuState<T, U>>,
            UseApi(ExtractHost(host), _): UseApi<ExtractHost, String>,
            OriginalUri(uri): OriginalUri,
            request_body: Option<Json<sovd_comparams::executions::update::Request>>,
        ) -> Response {
            require_ecu_access!(
                write,
                security_plugin,
                &ecu_name,
                &locks,
                query.include_schema
            );
            let path = format!("http://{host}{uri}");
            let body = if let Some(Json(body)) = request_body {
                Some(body)
            } else {
                None
            };
            handler_write(
                comparam_executions,
                communication_access,
                path,
                body,
                query.include_schema,
            )
            .await
        }

        pub(crate) fn docs_post(op: TransformOperation) -> TransformOperation {
            op.description("Create a new comparam execution")
                .response_with::<202, Json<sovd_comparams::executions::update::Response>, _>(
                    |res| {
                        res.description("Comparam execution created successfully.")
                            .example(sovd_comparams::executions::update::Response {
                                id: "b7e2c1a2-3f4d-4e6a-9c8b-2a1d5e7f8c9b".to_string(),
                                status: sovd_comparams::executions::Status::Running,
                                schema: None,
                            })
                    },
                )
        }

        pub(crate) async fn handler_read(
            executions: Arc<RwLock<IndexMap<Uuid, ComparamExecution>>>,
            include_schema: bool,
        ) -> Response {
            let schema = if include_schema {
                Some(create_schema!(sovd_comparams::executions::get::Response))
            } else {
                None
            };
            (
                StatusCode::OK,
                Json(sovd_comparams::executions::get::Response {
                    items: executions
                        .read()
                        .await
                        .keys()
                        .map(|id| sovd_comparams::executions::Item { id: id.to_string() })
                        .collect::<Vec<_>>(),
                    schema,
                }),
            )
                .into_response()
        }
        async fn handler_write(
            executions: Arc<RwLock<IndexMap<Uuid, ComparamExecution>>>,
            communication_access: Arc<dyn CommunicationAccess>,
            base_path: String,
            request: Option<sovd_comparams::executions::update::Request>,
            include_schema: bool,
        ) -> Response {
            // todo: not in scope for now: request can take body with
            // { timeout: INT, parameters: { ... }, proximity_response: STRING }
            let communication_activity =
                match acquire_communication_activity(&*communication_access) {
                    Ok(activity) => activity,
                    Err(error) => {
                        return ErrorWrapper {
                            error,
                            include_schema,
                        }
                        .into_response();
                    }
                };
            let id = Uuid::new_v4();
            let mut comparam_override: HashMap<String, sovd_comparams::ComParamValue> =
                HashMap::new();

            if let Some(sovd_comparams::executions::update::Request {
                parameters: Some(parameters),
                ..
            }) = request
            {
                for (k, v) in parameters {
                    comparam_override.insert(k, v);
                }
            }

            let schema = if include_schema {
                Some(create_schema!(sovd_comparams::executions::update::Response))
            } else {
                None
            };

            let create_execution_response = sovd_comparams::executions::update::Response {
                id: id.to_string(),
                status: sovd_comparams::executions::Status::Running,
                schema,
            };
            // Publish the lease before the execution becomes observable to DELETE.
            let mut executions = executions.write().await;
            executions.insert(
                id,
                ComparamExecution::new(
                    sovd_comparams::Execution {
                        capability: sovd_comparams::executions::Capability::Execute,
                        status: create_execution_response.status.clone(),
                        comparam_override,
                    },
                    communication_activity,
                ),
            );
            (
                StatusCode::ACCEPTED,
                [(header::LOCATION, format!("{base_path}/{id}"))],
                Json(create_execution_response),
            )
                .into_response()
        }

        pub(crate) mod id {
            use super::*;
            use crate::{openapi, sovd::components::IdPathParam};
            pub(crate) async fn get<T: UdsEcu + Clone, U: FileManager>(
                UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
                Path(id): Path<IdPathParam>,
                WithRejection(Query(query), _): WithRejection<
                    Query<sovd_comparams::executions::get::Query>,
                    ApiError,
                >,
                State(WebserverEcuState {
                    ecu_name,
                    uds,
                    locks,
                    comparam_executions,
                    ..
                }): State<WebserverEcuState<T, U>>,
            ) -> Response {
                let include_schema = query.include_schema;
                require_ecu_access!(read, security_plugin, &ecu_name, &locks, include_schema);
                let id = match parse_exec_uuid(&id, include_schema) {
                    Ok(v) => v,
                    Err(e) => return e.into_response(),
                };
                let mut executions: Vec<sovd_comparams::Execution> = Vec::new();

                let (idx, execution) = match comparam_executions
                    .read()
                    .await
                    .get_full(&id)
                    .ok_or_else(|| {
                        ApiError::NotFound(Some(format!("Execution with id {id} not found")))
                    }) {
                    Ok((idx, _, exec)) => (idx, exec.snapshot()),
                    Err(e) => {
                        return ErrorWrapper {
                            error: e,
                            include_schema,
                        }
                        .into_response();
                    }
                };
                let capability = execution.capability.clone();
                let status = execution.status.clone();

                // put in all executions with lower index than this one
                for (_, v) in &comparam_executions.read().await.as_slice()[..idx] {
                    executions.push(v.snapshot());
                }
                executions.push(execution);

                let mut parameters = match uds.get_comparams(&ecu_name).await {
                    Ok(v) => v.into_sovd(),
                    Err(e) => {
                        return ErrorWrapper {
                            error: e.into(),
                            include_schema,
                        }
                        .into_response();
                    }
                };

                for (k, v) in executions.into_iter().flat_map(|e| e.comparam_override) {
                    parameters.insert(k, v);
                }

                let schema = if include_schema {
                    Some(create_schema!(
                        sovd_comparams::executions::id::get::Response
                    ))
                } else {
                    None
                };

                (
                    StatusCode::OK,
                    Json(sovd_comparams::executions::id::get::Response {
                        capability,
                        parameters,
                        status,
                        schema,
                    }),
                )
                    .into_response()
            }

            pub(crate) fn docs_get(op: TransformOperation) -> TransformOperation {
                op.description("Get a specific comparam execution")
                    .response_with::<200, Json<sovd_comparams::executions::id::get::Response>, _>(
                        |res| {
                            res.description("Response with comparam execution details.")
                                .example(sovd_comparams::executions::id::get::Response {
                                    capability: sovd_comparams::executions::Capability::Execute,
                                    parameters: HashMap::new(),
                                    status: sovd_comparams::executions::Status::Running,
                                    schema: None,
                                })
                        },
                    )
                    .with(openapi::comparam_execution_errors)
            }

            pub(crate) async fn delete<T: UdsEcu + Clone, U: FileManager>(
                UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
                Path(id): Path<IdPathParam>,
                State(WebserverEcuState {
                    ecu_name,
                    locks,
                    comparam_executions,
                    ..
                }): State<WebserverEcuState<T, U>>,
            ) -> Response {
                require_ecu_access!(write, security_plugin, &ecu_name, &locks, false);
                let id = match parse_exec_uuid(&id, false) {
                    Ok(v) => v,
                    Err(e) => return e.into_response(),
                };
                let removed = {
                    let mut executions = comparam_executions.write().await;
                    executions.shift_remove(&id).is_some()
                };
                if !removed {
                    return ErrorWrapper {
                        error: ApiError::NotFound(Some(format!(
                            "Execution with id {id} not found"
                        ))),
                        include_schema: false,
                    }
                    .into_response();
                }
                StatusCode::NO_CONTENT.into_response()
            }

            pub(crate) fn docs_delete(op: TransformOperation) -> TransformOperation {
                op.description("Delete a specific comparam execution")
                    .response_with::<204, (), _>(|res| {
                        res.description("Comparam execution deleted successfully.")
                    })
                    .with(openapi::comparam_execution_errors)
            }

            pub(crate) async fn put<T: UdsEcu + Clone, U: FileManager>(
                UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
                Path(id): Path<IdPathParam>,
                WithRejection(Query(query), _): WithRejection<
                    Query<sovd_comparams::executions::update::Query>,
                    ApiError,
                >,
                State(WebserverEcuState {
                    ecu_name,
                    locks,
                    comparam_executions,
                    ..
                }): State<WebserverEcuState<T, U>>,
                UseApi(ExtractHost(host), _): UseApi<ExtractHost, String>,
                OriginalUri(uri): OriginalUri,
                WithRejection(Json(request), _): WithRejection<
                    Json<sovd_comparams::executions::update::Request>,
                    ApiError,
                >,
            ) -> Response {
                let include_schema = query.include_schema;
                require_ecu_access!(write, security_plugin, &ecu_name, &locks, include_schema);
                let id = match parse_exec_uuid(&id, include_schema) {
                    Ok(v) => v,
                    Err(e) => return e.into_response(),
                };
                let path = format!("http://{host}{uri}");
                // todo: (out of scope for now) handle timout and capability

                // todo: validate that the passed in CP is actually a valid CP for the ECU
                // let mut comparams = match uds.get_comparams(&ecu_name).await {
                //     Ok(v) => v,
                //     Err(e) => return ErrorWrapper(ApiError::BadRequest(e)).into_response(),
                // };

                let mut executions_lock = comparam_executions.write().await;
                let execution: &mut ComparamExecution =
                    match executions_lock.get_mut(&id).ok_or_else(|| {
                        ApiError::NotFound(Some(format!("Execution with id {id} not found")))
                    }) {
                        Ok(v) => v,
                        Err(e) => {
                            return ErrorWrapper {
                                error: e,
                                include_schema,
                            }
                            .into_response();
                        }
                    };

                if let Some(comparam_values) = request.parameters {
                    for (k, v) in comparam_values {
                        execution.comparam_override.insert(k, v);
                    }
                }

                let schema = if include_schema {
                    Some(create_schema!(sovd_comparams::executions::update::Response))
                } else {
                    None
                };

                (
                    StatusCode::ACCEPTED,
                    [(header::LOCATION, path)],
                    Json(sovd_comparams::executions::update::Response {
                        id: id.to_string(),
                        status: execution.status.clone(),
                        schema,
                    }),
                )
                    .into_response()
            }

            pub(crate) fn docs_put(op: TransformOperation) -> TransformOperation {
                op.description("Update a specific comparam execution")
                    .response_with::<202, Json<sovd_comparams::executions::update::Response>, _>(
                        |res| {
                            res.description("Comparam execution updated successfully.")
                                .example(sovd_comparams::executions::update::Response {
                                    id: "example_id".to_string(),
                                    status: sovd_comparams::executions::Status::Running,
                                    schema: None,
                                })
                        },
                    )
                    .with(openapi::comparam_execution_errors)
            }
        }
    }
}

pub(crate) mod service {
    use aide::UseApi;
    use axum::extract::{Path, Query, State};
    use axum_extra::extract::WithRejection;

    use crate::sovd::locks::require_ecu_access;

    /// `GET /operations/{service}` - get operation details or SDGs
    // [[ dimpl~sovd-api-component-operations-sdgsd, GET /operations/{service} SDG handler ]]
    pub(crate) async fn get<
        T: cda_interfaces::UdsEcu + cda_interfaces::SchemaProvider + Clone,
        U: cda_interfaces::file_manager::FileManager,
    >(
        UseApi(cda_plugin_security::Secured(security_plugin), _): UseApi<
            cda_plugin_security::Secured,
            (),
        >,
        Path(docs_endpoint::OperationNamePathParam { service }): Path<
            docs_endpoint::OperationNamePathParam,
        >,
        WithRejection(Query(query), _): WithRejection<
            Query<sovd_interfaces::components::ComponentQuery>,
            crate::sovd::error::ApiError,
        >,
        State(crate::sovd::WebserverEcuState {
            ecu_name,
            uds,
            locks,
            ..
        }): State<crate::sovd::WebserverEcuState<T, U>>,
    ) -> axum::response::Response {
        use axum::response::IntoResponse as _;

        let include_schema = query.include_schema;
        require_ecu_access!(read, security_plugin, &ecu_name, &locks, include_schema);
        if query.include_sdgs {
            return get_sdgs_handler::<T>(service, &ecu_name, &uds, include_schema).await;
        }

        // Normal response: return operation info for the named service
        let security_plugin: cda_interfaces::DynamicPlugin = security_plugin;
        let ops_info = match uds
            .get_components_operations_info(&ecu_name, &security_plugin)
            .await
        {
            Ok(info) => info,
            Err(e) => {
                return crate::sovd::error::ErrorWrapper {
                    error: e.into(),
                    include_schema,
                }
                .into_response();
            }
        };

        let Some(op_info) = ops_info
            .iter()
            .find(|o| o.id.eq_ignore_ascii_case(&service))
        else {
            return crate::sovd::error::ApiError::NotFound(Some(format!(
                "Operation '{service}' not found"
            )))
            .into_response();
        };

        let schema = if include_schema {
            Some(crate::sovd::create_schema!(
                sovd_interfaces::components::ecu::operations::OperationCollectionItem
            ))
        } else {
            None
        };

        (
            http::StatusCode::OK,
            axum::Json(sovd_interfaces::Items {
                items: vec![
                    sovd_interfaces::components::ecu::operations::OperationCollectionItem {
                        id: op_info.id.clone(),
                        name: op_info.name.clone(),
                        proximity_proof_required: false,
                        asynchronous_execution: op_info.has_stop || op_info.has_request_results,
                    },
                ],
                schema,
            }),
        )
            .into_response()
    }

    async fn get_sdgs_handler<T: cda_interfaces::UdsEcu + Clone>(
        service: String,
        ecu_name: &str,
        gateway: &T,
        include_schema: bool,
    ) -> axum::response::Response {
        use axum::response::IntoResponse as _;
        use cda_interfaces::{DiagComm, DiagCommType, HashMap, HashMapExtensions};

        use crate::sovd::IntoSovd;

        let service_ops = vec![
            DiagComm {
                name: service.clone(),
                type_: DiagCommType::Operations,
                lookup_name: None,
                subfunction_id: None,
            },
            DiagComm {
                name: service.clone(),
                type_: DiagCommType::Operations,
                lookup_name: None,
                subfunction_id: None,
            },
            DiagComm {
                name: service,
                type_: DiagCommType::Operations,
                lookup_name: None,
                subfunction_id: None,
            },
        ];
        let schema = if include_schema {
            Some(crate::sovd::create_schema!(
                sovd_interfaces::components::ecu::ServicesSdgs
            ))
        } else {
            None
        };
        let mut resp = sovd_interfaces::components::ecu::ServicesSdgs {
            items: HashMap::new(),
            schema,
        };
        for service in service_ops {
            match gateway.get_sdgs(ecu_name, Some(&service)).await {
                Ok(sdgs) => {
                    if sdgs.is_empty() {
                        continue;
                    }
                    resp.items.insert(
                        format!("{}_{:?}", service.name, service.action()).to_lowercase(),
                        sovd_interfaces::components::ecu::ServiceSdgs {
                            sdgs: sdgs.into_sovd(),
                        },
                    );
                }
                Err(e) => {
                    return crate::sovd::error::ErrorWrapper {
                        error: e.into(),
                        include_schema,
                    }
                    .into_response();
                }
            }
        }
        (http::StatusCode::OK, axum::Json(resp)).into_response()
    }

    pub(crate) fn docs_get(
        op: aide::transform::TransformOperation,
    ) -> aide::transform::TransformOperation {
        use aide::transform::TransformParameter;
        op.description("Get a specific operation or its SDG metadata.")
            .parameter("x-sovd2uds-includesdgs", |op: TransformParameter<bool>| {
                op.description("Set to true to include sdgs.")
            })
            .response_with::<200, axum::Json<
                sovd_interfaces::Items<
                    sovd_interfaces::components::ecu::operations::OperationCollectionItem,
                >,
            >, _>(|res| res.description("Operation details or SDG metadata."))
            .with(crate::openapi::error_not_found)
            .with(crate::openapi::error_internal_server)
    }

    /// `GET /operations/{service}/docs` - online capability description
    pub(crate) mod docs_endpoint {
        use aide::{UseApi, openapi::OpenApi, transform::TransformOperation};
        use axum::{
            Json,
            extract::{Path, State},
            http::StatusCode,
            response::{IntoResponse as _, Response},
        };
        use cda_interfaces::{
            DiagComm, DiagCommType, DynamicPlugin, SchemaProvider, UdsEcu,
            file_manager::FileManager, subfunction_ids,
        };
        use cda_plugin_security::Secured;

        use crate::{
            openapi,
            sovd::{
                WebserverEcuState,
                docs::{self, operations::OperationDocsMeta},
                error::ApiError,
                locks::require_ecu_access,
            },
        };

        openapi::aide_helper::gen_path_param!(OperationNamePathParam service String);

        pub(crate) async fn get<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
            UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
            Path(OperationNamePathParam { service }): Path<OperationNamePathParam>,
            State(WebserverEcuState {
                ecu_name,
                uds,
                locks,
                ..
            }): State<WebserverEcuState<T, U>>,
        ) -> Response {
            require_ecu_access!(read, security_plugin, &ecu_name, &locks, false);
            let security_plugin: DynamicPlugin = security_plugin;

            let ops_info = match uds
                .get_components_operations_info(&ecu_name, &security_plugin)
                .await
            {
                Ok(info) => info,
                Err(e) => return ApiError::from(e).into_response(),
            };

            let Some(op_info) = ops_info
                .iter()
                .find(|o| o.id.eq_ignore_ascii_case(&service))
            else {
                return ApiError::NotFound(Some(format!("Operation '{service}' not found")))
                    .into_response();
            };

            let is_async = op_info.has_stop || op_info.has_request_results;

            let diag_service = DiagComm {
                name: service.clone(),
                type_: DiagCommType::Operations,
                lookup_name: None,
                subfunction_id: Some(subfunction_ids::routine::START),
            };

            let request_params_schema = uds
                .schema_for_request(&ecu_name, &diag_service)
                .await
                .ok()
                .and_then(cda_interfaces::SchemaDescription::into_schema);

            let response_params_schema = uds
                .schema_for_responses(&ecu_name, &diag_service)
                .await
                .ok()
                .and_then(cda_interfaces::SchemaDescription::into_schema);

            let meta = OperationDocsMeta {
                name: service.clone(),
                is_async,
                request_params_schema,
                response_params_schema,
            };

            let base_path = format!("/components/{ecu_name}/operations/{service}");
            let path_items = docs::operations::build_path_items(&base_path, &meta);
            let doc = docs::build_openapi_doc(&format!("Operation: {service}"), path_items);

            (StatusCode::OK, Json(doc)).into_response()
        }

        pub(crate) fn docs_transform(op: TransformOperation) -> TransformOperation {
            op.description(
                "Online capability description for a specific operation on this ECU component \
                 (ISO 17978-3 Section 7.5). Returns a self-contained OpenAPI specification.",
            )
            .response_with::<200, Json<OpenApi>, _>(|res| {
                res.description("Self-contained OpenAPI 3.1 specification for this operation.")
            })
            .with(openapi::error_not_found)
        }
    }

    pub(crate) mod executions {
        use std::sync::{Arc, RwLock};

        use aide::{UseApi, transform::TransformOperation};
        use axum::{
            Json,
            body::Bytes,
            extract::{OriginalUri, Path, Query, State},
            http::{HeaderMap, StatusCode, header},
            response::{IntoResponse as _, Response},
        };
        use axum_extra::extract::WithRejection;
        use cda_interfaces::{
            DiagComm, DiagCommType, DiagServiceError, DynamicPlugin, SchemaProvider, UdsEcu,
            communication_control::CommunicationAccess,
            diagservices::{DiagServiceJsonResponse, DiagServiceResponse, DiagServiceResponseType},
            file_manager::FileManager,
            subfunction_ids,
            util::std_ext::{lock_read, lock_write},
        };
        use cda_plugin_security::{Secured, SecurityPlugin};
        use opensovd_axum_extra::ExtractHost;
        use sovd_interfaces::{
            common::operations::OperationIdItem,
            components::ecu::operations::{
                AsyncGetByIdResponse, AsyncPostResponse, ExecutionStatus, OperationDeleteQuery,
                OperationQuery, service::executions as sovd_executions,
            },
        };
        use uuid::Uuid;

        use crate::{
            openapi,
            sovd::{
                self, ExecutionReservation, ServiceExecution, WebserverEcuState,
                acquire_and_reserve_execution, api_error_from_diag_response,
                components::get_content_type_and_accept,
                create_response_schema, create_schema,
                error::{ApiError, ErrorWrapper, VendorErrorCode},
                field_parse_errors_to_json, guard_execution,
                locks::require_ecu_access,
            },
        };

        openapi::aide_helper::gen_path_param!(OperationServicePathParam service String);

        /// Options forwarded from the HTTP layer into `ecu_operation_write_handler`.
        pub(crate) struct WriteHandlerOptions {
            pub include_schema: bool,
            pub suppress_service: bool,
            pub base_path: String,
        }

        /// Request data forwarded from the HTTP layer into `ecu_operation_write_handler`.
        pub(crate) struct WriteHandlerRequest {
            pub service: String,
            pub headers: HeaderMap,
            pub body: Bytes,
        }

        pub(crate) async fn get<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
            UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
            Path(OperationServicePathParam { service }): Path<OperationServicePathParam>,
            WithRejection(Query(query), _): WithRejection<Query<sovd_executions::Query>, ApiError>,
            State(WebserverEcuState {
                ecu_name,
                locks,
                service_executions,
                ..
            }): State<WebserverEcuState<T, U>>,
        ) -> Response {
            require_ecu_access!(
                read,
                security_plugin,
                &ecu_name,
                &locks,
                query.include_schema
            );
            let schema = if query.include_schema {
                Some(create_schema!(sovd_interfaces::Items<OperationIdItem>))
            } else {
                None
            };
            let ids: Vec<OperationIdItem> = lock_read(&service_executions)
                .get(&service)
                .map(|op_map| {
                    op_map
                        .iter()
                        .filter(|(_, v)| v.is_created)
                        .map(|(k, _)| OperationIdItem { id: k.to_string() })
                        .collect()
                })
                .unwrap_or_default();
            (
                StatusCode::OK,
                Json(sovd_interfaces::Items { items: ids, schema }),
            )
                .into_response()
        }

        pub(crate) fn docs_get(op: TransformOperation) -> TransformOperation {
            op.description("List all active service operation executions")
                .response_with::<200, Json<sovd_interfaces::Items<String>>, _>(|res| {
                    res.description("List of active execution ids.")
                })
        }

        #[allow(
            clippy::too_many_arguments,
            reason = "Axum extractors cannot be combined without a new custom extractor"
        )]
        pub(crate) async fn post<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
            UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
            Path(OperationServicePathParam { service }): Path<OperationServicePathParam>,
            WithRejection(Query(query), _): WithRejection<Query<OperationQuery>, ApiError>,
            State(WebserverEcuState {
                ecu_name,
                uds,
                locks,
                service_executions,
                communication_access,
                ..
            }): State<WebserverEcuState<T, U>>,
            UseApi(ExtractHost(host), _): UseApi<ExtractHost, String>,
            OriginalUri(uri): OriginalUri,
            headers: HeaderMap,
            body: Bytes,
        ) -> Response {
            require_ecu_access!(
                write,
                security_plugin,
                &ecu_name,
                &locks,
                query.include_schema
            );
            let ctx = OperationWriteContext::new(service_executions, communication_access);
            ecu_operation_write_handler_with_activity::<T>(
                WriteHandlerRequest {
                    service,
                    headers,
                    body,
                },
                &ecu_name,
                &uds,
                ctx,
                security_plugin,
                WriteHandlerOptions {
                    include_schema: query.include_schema,
                    suppress_service: query.suppress_service,
                    base_path: format!("http://{host}{uri}"),
                },
            )
            .await
        }

        pub(crate) fn docs_post(op: TransformOperation) -> TransformOperation {
            openapi::request_json_and_octet::<sovd_executions::Request>(op)
                .description("Start a new operation execution (Start subfunction)")
                .response_with::<200, Json<sovd_executions::Response<VendorErrorCode>>, _>(|res| {
                    let mut res = res
                        .description("Execution started, synchronous result.")
                        .example(sovd_executions::Response {
                            parameters: Some(serde_json::Map::from_iter([(
                                "example_param".to_string(),
                                serde_json::Value::String("example_value".to_string()),
                            )])),
                            error: None,
                            schema: None,
                        });
                    res.inner().content.insert(
                        "application/octet-stream".to_owned(),
                        aide::openapi::MediaType {
                            example: Some(serde_json::json!([0xABu8, 0xCD, 0xEF, 0x00])),
                            ..Default::default()
                        },
                    );
                    res
                })
                .response_with::<
                    202,
                    Json<
                        AsyncPostResponse<
                            serde_json::Map<String, serde_json::Value>,
                            VendorErrorCode,
                        >,
                    >,
                    _,
                >(|res| {
                    res.description(
                        "Execution started asynchronously. Use the returned id for GET/DELETE on \
                         /executions/{id}.",
                    )
                })
                .with(openapi::error_bad_request)
                .with(openapi::error_not_found)
                .with(openapi::error_forbidden)
                .with(openapi::error_conflict)
                .with(openapi::error_internal_server)
                .with(openapi::error_bad_gateway)
        }

        fn validate_and_parse_write_request(
            headers: &HeaderMap,
            body: &Bytes,
        ) -> Result<(Option<cda_interfaces::diagservices::UdsPayloadData>, bool), ApiError>
        {
            let (Some(content_type), accept) = get_content_type_and_accept(headers)? else {
                return Err(ApiError::BadRequest("Missing Content-Type".to_owned()));
            };
            let data = sovd::get_payload_data::<sovd_executions::Request>(
                Some(&content_type),
                headers,
                body,
            )?;
            if accept != mime::APPLICATION_OCTET_STREAM && accept != mime::APPLICATION_JSON {
                return Err(ApiError::BadRequest(format!(
                    "Unsupported Accept header: {accept:?}"
                )));
            }
            Ok((data, accept == mime::APPLICATION_JSON))
        }

        /// Context for ECU operation write handlers, grouping execution-related state.
        pub(crate) struct OperationWriteContext {
            service_executions: Arc<
                RwLock<cda_interfaces::HashMap<String, indexmap::IndexMap<Uuid, ServiceExecution>>>,
            >,
            communication_access: Arc<dyn CommunicationAccess>,
        }

        impl OperationWriteContext {
            pub(crate) fn new(
                service_executions: Arc<
                    RwLock<
                        cda_interfaces::HashMap<String, indexmap::IndexMap<Uuid, ServiceExecution>>,
                    >,
                >,
                communication_access: Arc<dyn CommunicationAccess>,
            ) -> Self {
                Self {
                    service_executions,
                    communication_access,
                }
            }
        }

        pub(crate) async fn ecu_operation_write_handler_with_activity<
            T: UdsEcu + SchemaProvider + Clone,
        >(
            req: WriteHandlerRequest,
            ecu_name: &str,
            uds: &T,
            ctx: OperationWriteContext,
            security_plugin: Box<dyn SecurityPlugin>,
            opts: WriteHandlerOptions,
        ) -> Response {
            let OperationWriteContext {
                service_executions,
                communication_access,
            } = ctx;
            let WriteHandlerRequest {
                service,
                headers,
                body,
            } = req;
            let WriteHandlerOptions {
                include_schema,
                suppress_service,
                base_path,
            } = opts;
            let err_response = |error: ApiError| -> Response {
                ErrorWrapper {
                    error,
                    include_schema,
                }
                .into_response()
            };
            if service == "reset" {
                return ecu_reset_handler::<T>(
                    service,
                    ecu_name,
                    uds,
                    body,
                    security_plugin,
                    include_schema,
                )
                .await;
            }

            // Reserve an execution slot atomically: checks for a running
            // conflict and, if none, inserts a placeholder so that a second
            // concurrent POST for the same operation sees 409 Conflict.
            // The lease is the active-execution marker for update arbitration. It is
            // published with the reservation and retained until the entry is removed.
            let reservation = match acquire_and_reserve_execution(
                &*communication_access,
                Arc::clone(&service_executions),
                &service,
                &service,
                include_schema,
            ) {
                Ok(v) => v,
                Err(e) => return e.into_response(),
            };

            let security_plugin: DynamicPlugin = security_plugin;
            let ResolvedOperation {
                is_async,
                operation: diag_service,
            } = if suppress_service {
                ResolvedOperation {
                    is_async: true,
                    operation: DiagComm {
                        name: service.clone(),
                        type_: DiagCommType::Operations,
                        lookup_name: None,
                        subfunction_id: Some(subfunction_ids::routine::START),
                    },
                }
            } else {
                match resolve_operation(uds, ecu_name, &service, &security_plugin).await {
                    Ok(v) => v,
                    Err(e) => {
                        return err_response(e);
                    }
                }
            };

            let (data, map_to_json) = match validate_and_parse_write_request(&headers, &body) {
                Ok(v) => v,
                Err(e) => {
                    return err_response(e);
                }
            };

            let response = if suppress_service {
                None
            } else {
                match send_start_request(
                    uds,
                    ecu_name,
                    diag_service.clone(),
                    &security_plugin,
                    data,
                    map_to_json,
                    include_schema,
                )
                .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        return *e;
                    }
                }
            };
            finish_write_request(FinishWriteRequestArgs {
                is_async,
                response,
                map_to_json,
                include_schema,
                base_path,
                reservation,
                ecu_name,
                uds,
                diag_service,
            })
            .await
        }

        /// Arguments for [`finish_write_request`].
        struct FinishWriteRequestArgs<'a, T: UdsEcu + SchemaProvider + Clone> {
            is_async: bool,
            response: Option<T::Response>,
            map_to_json: bool,
            include_schema: bool,
            base_path: String,
            reservation: ExecutionReservation<ServiceExecution>,
            ecu_name: &'a str,
            uds: &'a T,
            diag_service: DiagComm,
        }

        /// Dispatch the tail of a write request to the async or sync
        /// post-processing path.
        async fn finish_write_request<T: UdsEcu + SchemaProvider + Clone>(
            args: FinishWriteRequestArgs<'_, T>,
        ) -> Response {
            let FinishWriteRequestArgs {
                is_async,
                response,
                map_to_json,
                include_schema,
                base_path,
                reservation,
                ecu_name,
                uds,
                diag_service,
            } = args;
            if is_async {
                handle_async_post::<T>(response, include_schema, &base_path, reservation)
            } else {
                drop(reservation);
                handle_sync_post::<T>(
                    response,
                    map_to_json,
                    include_schema,
                    ecu_name,
                    uds,
                    &diag_service,
                )
                .await
            }
        }

        struct ResolvedOperation {
            is_async: bool,
            operation: DiagComm,
        }

        /// Resolves an operation-id to its underlying UDS service type and addressing scheme:
        /// a `RoutineControl` group addressed through its Start subfunction, or an IO Control
        /// (and any other non-routine operation) addressed by its exact diagnostic service name.
        ///
        /// Each type is checked directly rather than inferring IO Control from the absence of a
        /// `RoutineControl` service, so an operation-id that matches neither fails immediately
        /// with `NotFound` instead of falling through into a doomed `send()` attempt.
        async fn resolve_operation<T: UdsEcu>(
            uds: &T,
            ecu_name: &str,
            service: &str,
            security_plugin: &DynamicPlugin,
        ) -> Result<ResolvedOperation, ApiError> {
            match uds
                .get_routine_subfunctions(ecu_name, service, security_plugin)
                .await
            {
                Ok(sf) => {
                    return Ok(ResolvedOperation {
                        is_async: sf.has_stop || sf.has_request_results,
                        operation: DiagComm {
                            name: service.to_owned(),
                            type_: DiagCommType::Operations,
                            lookup_name: None,
                            subfunction_id: Some(subfunction_ids::routine::START),
                        },
                    });
                }
                Err(DiagServiceError::NotFound(_)) => {}
                Err(error) => return Err(ApiError::from(error)),
            }

            match uds
                .get_io_control_service(ecu_name, service, security_plugin)
                .await
            {
                Ok(operation) => Ok(ResolvedOperation {
                    is_async: false,
                    operation,
                }),
                Err(DiagServiceError::NotFound(_)) => Err(ApiError::from(
                    DiagServiceError::NotFound(format!("Operation '{service}' not found")),
                )),
                Err(error) => Err(ApiError::from(error)),
            }
        }

        /// Sends the Start subfunction request and returns the positive response, or
        /// `Err(Response)` if the UDS call failed, returned a negative response, or
        /// was cancelled because the owning lock is closing.
        async fn send_start_request<T: UdsEcu>(
            uds: &T,
            ecu_name: &str,
            diag_service: DiagComm,
            security_plugin: &DynamicPlugin,
            data: Option<cda_interfaces::diagservices::UdsPayloadData>,
            map_to_json: bool,
            include_schema: bool,
        ) -> Result<Option<T::Response>, Box<Response>> {
            let response = match uds
                .send(ecu_name, diag_service, security_plugin, data, map_to_json)
                .await
            {
                Ok(v) => v,
                Err(e) => {
                    return Err(Box::new(
                        ErrorWrapper {
                            error: e.into(),
                            include_schema,
                        }
                        .into_response(),
                    ));
                }
            };
            if let DiagServiceResponseType::Negative = response.response_type() {
                return Err(Box::new(
                    api_error_from_diag_response(&response, include_schema).into_response(),
                ));
            }
            Ok(Some(response))
        }

        fn err_invalid_content(
            detail: String,
        ) -> sovd_interfaces::error::DataError<VendorErrorCode> {
            sovd_interfaces::error::DataError {
                path: "/parameters".to_owned(),
                error: sovd_interfaces::error::ApiErrorResponse {
                    message: detail,
                    error_code: sovd_interfaces::error::ErrorCode::InvalidResponseContent,
                    vendor_code: None,
                    parameters: None,
                    error_source: None,
                    schema: None,
                },
            }
        }

        /// Parses a positive, non-empty UDS response into `(parameters, errors)`.
        ///
        /// - `Object` -> parameters map + any field-parse errors
        /// - `Null`  -> empty parameters, no errors (ECU signalled no output)
        /// - anything else, or a parse failure -> empty parameters, one soft `DataError`
        fn parse_json_response_params<R: DiagServiceResponse>(
            response: R,
            context: &str,
        ) -> (
            serde_json::Map<String, serde_json::Value>,
            Vec<sovd_interfaces::error::DataError<VendorErrorCode>>,
        ) {
            match response.into_json() {
                Ok(DiagServiceJsonResponse {
                    data: serde_json::Value::Object(m),
                    errors,
                }) => (m, field_parse_errors_to_json(errors, "parameters")),
                Ok(DiagServiceJsonResponse {
                    data: serde_json::Value::Null,
                    ..
                }) => (serde_json::Map::new(), vec![]),
                Ok(v) => (
                    serde_json::Map::new(),
                    vec![err_invalid_content(format!(
                        "Expected JSON object but got: {}",
                        v.data
                    ))],
                ),
                Err(e) => (
                    serde_json::Map::new(),
                    vec![err_invalid_content(format!(
                        "Failed to parse {context} response: {e:?}"
                    ))],
                ),
            }
        }

        fn parse_exec_uuid(id: &str, include_schema: bool) -> Result<Uuid, ErrorWrapper> {
            Uuid::parse_str(id).map_err(|e| ErrorWrapper {
                error: ApiError::BadRequest(format!("{e:?}")),
                include_schema,
            })
        }

        /// Builds the `200 OK` `AsyncGetByIdResponse` body used by both the
        /// `RequestResults` success path and the `suppress_service` fallback path.
        /// An empty `parameters` map is serialised as `null` per the SOVD spec.
        fn get_by_id_response(
            status: ExecutionStatus,
            parameters: serde_json::Map<String, serde_json::Value>,
            error: Vec<sovd_interfaces::error::DataError<VendorErrorCode>>,
            include_schema: bool,
        ) -> Response {
            use sovd_interfaces::components::ecu::operations::GetByIdCapability;
            let schema = if include_schema {
                Some(create_schema!(AsyncGetByIdResponse<VendorErrorCode>))
            } else {
                None
            };
            let parameters = if parameters.is_empty() {
                None
            } else {
                Some(parameters)
            };
            (
                StatusCode::OK,
                Json(AsyncGetByIdResponse::<VendorErrorCode> {
                    status,
                    capability: GetByIdCapability::Execute,
                    parameters,
                    progress: None,
                    error,
                    schema,
                }),
            )
                .into_response()
        }

        /// Handles the async (Stop/RequestResults) POST path: finalises the
        /// previously reserved execution with the Start-response parameters,
        /// then returns them with the execution id and status.
        fn handle_async_post<T: UdsEcu>(
            response: Option<T::Response>,
            include_schema: bool,
            base_path: &str,
            reservation: ExecutionReservation<ServiceExecution>,
        ) -> Response {
            let (parameters, errors) = match response {
                Some(r) if !r.is_empty() => parse_json_response_params(r, "Start"),
                _ => (serde_json::Map::new(), vec![]),
            };
            let stored_parameters = parameters.clone();
            let exec_id = reservation.commit_async(|exec| {
                exec.parameters = stored_parameters;
            });
            let schema = if include_schema {
                Some(create_schema!(
                    AsyncPostResponse<serde_json::Map<String, serde_json::Value>, VendorErrorCode>
                ))
            } else {
                None
            };
            (
                StatusCode::ACCEPTED,
                [(header::LOCATION, format!("{base_path}/{exec_id}"))],
                Json(AsyncPostResponse {
                    id: exec_id.to_string(),
                    status: Some(ExecutionStatus::Running),
                    parameters,
                    errors,
                    schema,
                }),
            )
                .into_response()
        }

        /// Handles the synchronous (no Stop/RequestResults) POST path: returns 200 OK
        /// with the mapped response parameters, or raw bytes if the accept header
        /// requested octet-stream.
        async fn handle_sync_post<T: UdsEcu + SchemaProvider>(
            response: Option<T::Response>,
            map_to_json: bool,
            include_schema: bool,
            ecu_name: &str,
            uds: &T,
            diag_service: &DiagComm,
        ) -> Response {
            let schema = if map_to_json && include_schema {
                let subschema = get_subschema(ecu_name, uds, diag_service).await;
                Some(create_response_schema!(
                    sovd_executions::Response<VendorErrorCode>,
                    "parameters",
                    subschema
                ))
            } else {
                None
            };

            if map_to_json {
                let (mapped_data, parse_errors) =
                    match response.and_then(|r| (!r.is_empty()).then(|| r.into_json())) {
                        None => (serde_json::Map::new(), vec![]),
                        Some(Ok(DiagServiceJsonResponse {
                            data: serde_json::Value::Object(mapped_data),
                            errors,
                        })) => (mapped_data, errors),
                        Some(Ok(v)) => {
                            return ErrorWrapper {
                                error: ApiError::InternalServerError(Some(format!(
                                    "Expected JSON object but got: {}",
                                    v.data
                                ))),
                                include_schema,
                            }
                            .into_response();
                        }
                        Some(Err(e)) => {
                            return ErrorWrapper {
                                error: ApiError::InternalServerError(Some(format!("{e:?}"))),
                                include_schema,
                            }
                            .into_response();
                        }
                    };
                // Spec Table 183: `error` is singular (first parse error wins).
                let error = field_parse_errors_to_json(parse_errors, "parameters")
                    .into_iter()
                    .next();
                let parameters = if mapped_data.is_empty() {
                    None
                } else {
                    Some(mapped_data)
                };
                (
                    StatusCode::OK,
                    Json(sovd_executions::Response {
                        parameters,
                        error,
                        schema,
                    }),
                )
                    .into_response()
            } else {
                let data = response.map_or(vec![], |r| r.get_raw().to_vec());
                (StatusCode::OK, Bytes::from_owner(data)).into_response()
            }
        }

        #[allow(
            clippy::too_many_lines,
            reason = "Current implementation has little potential to extract smaller functions"
        )]
        async fn ecu_reset_handler<T: UdsEcu + SchemaProvider + Clone>(
            service: String,
            ecu_name: &str,
            uds: &T,
            body: Bytes,
            security_plugin: Box<dyn SecurityPlugin>,
            include_schema: bool,
        ) -> Response {
            // todo: in the future we have to handle possible parameters for the reset service
            let Some(request_parameters) =
                serde_json::from_slice::<sovd_executions::Request>(&body)
                    .ok()
                    .and_then(|v| v.parameters)
            else {
                return ErrorWrapper {
                    error: ApiError::BadRequest("Invalid request body".to_string()),
                    include_schema,
                }
                .into_response();
            };

            let Some(value) = request_parameters.get("value") else {
                return ErrorWrapper {
                    error: ApiError::BadRequest(
                        "Missing 'value' parameter in request body".to_owned(),
                    ),
                    include_schema,
                }
                .into_response();
            };

            let Some(value_str) = value.as_str() else {
                return ErrorWrapper {
                    error: ApiError::BadRequest(
                        "The 'value' parameter must be a string".to_owned(),
                    ),
                    include_schema,
                }
                .into_response();
            };

            let allowed_values = uds.get_ecu_reset_services(ecu_name).await;
            let allowed_values = match allowed_values {
                Ok(v) => v,
                Err(e) => {
                    return ErrorWrapper {
                        error: e.into(),
                        include_schema,
                    }
                    .into_response();
                }
            };

            if !allowed_values
                .iter()
                .any(|v| v.eq_ignore_ascii_case(value_str))
            {
                return ErrorWrapper {
                    error: ApiError::BadRequest(format!(
                        "Invalid value for reset service: {value_str}. Allowed values: [{}]",
                        allowed_values.join(", ")
                    )),
                    include_schema,
                }
                .into_response();
            }

            let diag_service = DiagComm {
                name: service.clone(),
                type_: DiagCommType::Modes, // ecureset is in modes
                lookup_name: Some(value_str.to_owned()),
                subfunction_id: None,
            };

            let schema = if include_schema {
                let subschema = get_subschema(ecu_name, uds, &diag_service).await;
                Some(create_response_schema!(
                    sovd_executions::Response<VendorErrorCode>,
                    "parameters",
                    subschema
                ))
            } else {
                None
            };

            let security_plugin: DynamicPlugin = security_plugin;
            let send_result = uds
                .send(ecu_name, diag_service, &security_plugin, None, true)
                .await;
            let response = match send_result {
                Ok(v) => v,
                Err(e) => {
                    return ErrorWrapper {
                        error: e.into(),
                        include_schema,
                    }
                    .into_response();
                }
            };

            match response.response_type() {
                DiagServiceResponseType::Negative => {
                    api_error_from_diag_response(&response, include_schema).into_response()
                }
                DiagServiceResponseType::Positive => {
                    if response.is_empty() {
                        StatusCode::NO_CONTENT.into_response()
                    } else {
                        let (response_data, parse_errors) = match response.into_json() {
                            Ok(DiagServiceJsonResponse {
                                data: serde_json::Value::Object(mapped_data),
                                errors,
                            }) => (mapped_data, errors),
                            Ok(DiagServiceJsonResponse {
                                data: serde_json::Value::Null,
                                errors,
                            }) => {
                                if errors.is_empty() {
                                    return StatusCode::NO_CONTENT.into_response();
                                }
                                (serde_json::Map::new(), errors)
                            }
                            Ok(v) => {
                                return ErrorWrapper {
                                    error: ApiError::InternalServerError(Some(format!(
                                        "Expected JSON object but got: {}",
                                        v.data
                                    ))),
                                    include_schema,
                                }
                                .into_response();
                            }
                            Err(e) => {
                                return ErrorWrapper {
                                    error: ApiError::InternalServerError(Some(format!("{e:?}"))),
                                    include_schema,
                                }
                                .into_response();
                            }
                        };
                        let error = field_parse_errors_to_json(parse_errors, "parameters")
                            .into_iter()
                            .next();
                        let parameters = if response_data.is_empty() {
                            None
                        } else {
                            Some(response_data)
                        };
                        (
                            StatusCode::OK,
                            Json(sovd_executions::Response {
                                parameters,
                                error,
                                schema,
                            }),
                        )
                            .into_response()
                    }
                }
            }
        }

        async fn get_subschema<T: SchemaProvider>(
            ecu_name: &str,
            uds: &T,
            diag_service: &DiagComm,
        ) -> Option<schemars::Schema> {
            match uds
                .schema_for_responses(ecu_name, diag_service)
                .await
                .map(cda_interfaces::SchemaDescription::into_schema)
            {
                Ok(v) => v,
                Err(e) => {
                    tracing::error!(
                        error = ?e,
                        diag_service = ?diag_service,
                        "Failed to get schema for diag service"
                    );
                    None
                }
            }
        }

        pub(crate) mod id {
            use super::*;
            use crate::sovd::locks::require_ecu_access;

            #[derive(serde::Deserialize, schemars::JsonSchema)]
            pub(crate) struct ServiceAndIdPathParam {
                pub service: String,
                pub id: String,
            }

            /// Process a successful UDS `RequestResults` response: parse it, update stored
            /// execution state, and return the `200 OK` response body.
            fn get_operations_response<T: UdsEcu>(
                response: T::Response,
                service: &str,
                exec_id: Uuid,
                service_executions: &std::sync::RwLock<
                    cda_interfaces::HashMap<String, indexmap::IndexMap<Uuid, ServiceExecution>>,
                >,
                include_schema: bool,
            ) -> Response {
                if let DiagServiceResponseType::Negative = response.response_type() {
                    tracing::warn!(
                        exec_id = %exec_id,
                        "RequestResults subfunction returned negative response"
                    );
                    return api_error_from_diag_response(&response, include_schema).into_response();
                }

                let (parameters, error_list) = if response.is_empty() {
                    (serde_json::Map::new(), vec![])
                } else {
                    parse_json_response_params::<T::Response>(response, "RequestResults")
                };

                // Copied before the lock is taken: the stored execution and the response body
                // both need the parameters, so the deep copy happens outside the critical
                // section rather than under the write lock.
                let stored_parameters = parameters.clone();
                if let Some(stored_mut) = lock_write(service_executions)
                    .get_mut(service)
                    .and_then(|m| m.get_mut(&exec_id))
                {
                    stored_mut.parameters = stored_parameters;
                    stored_mut.complete();
                }

                get_by_id_response(
                    ExecutionStatus::Completed,
                    parameters,
                    error_list,
                    include_schema,
                )
            }

            pub(crate) async fn get<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
                UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
                Path(ServiceAndIdPathParam { service, id }): Path<ServiceAndIdPathParam>,
                WithRejection(Query(query), _): WithRejection<Query<OperationQuery>, ApiError>,
                State(WebserverEcuState {
                    ecu_name,
                    uds,
                    locks,
                    service_executions,
                    ..
                }): State<WebserverEcuState<T, U>>,
            ) -> Response {
                let include_schema = query.include_schema;
                require_ecu_access!(write, security_plugin, &ecu_name, &locks, include_schema);
                let exec_id = match parse_exec_uuid(&id, include_schema) {
                    Ok(v) => v,
                    Err(e) => return e.into_response(),
                };

                let stored = match guard_execution(
                    &service_executions,
                    &service,
                    exec_id,
                    include_schema,
                    &format!("Execution {exec_id} is already in progress"),
                ) {
                    Ok(guard) => guard,
                    Err(error) => return error.into_response(),
                };

                // suppress_service: skip the UDS send, return stored state directly
                if query.suppress_service {
                    return get_by_id_response(
                        stored.snapshot().status.clone(),
                        stored.snapshot().parameters.clone(),
                        vec![],
                        include_schema,
                    );
                }

                // Try to send RequestResults (subfunction 0x03)
                let diag_service = DiagComm {
                    name: service.clone(),
                    type_: DiagCommType::Operations,
                    lookup_name: None,
                    subfunction_id: Some(subfunction_ids::routine::REQUEST_RESULTS),
                };
                let uds_result = uds
                    .send(
                        &ecu_name,
                        diag_service,
                        &(security_plugin as DynamicPlugin),
                        None,
                        true,
                    )
                    .await;

                match uds_result {
                    Err(e) => {
                        tracing::warn!(
                            error = ?e,
                            service = %service,
                            exec_id = %exec_id,
                            "RequestResults subfunction failed"
                        );
                        ErrorWrapper {
                            error: e.into(),
                            include_schema,
                        }
                        .into_response()
                    }
                    Ok(response) => get_operations_response::<T>(
                        response,
                        &service,
                        exec_id,
                        &service_executions,
                        include_schema,
                    ),
                }
            }

            pub(crate) fn docs_get(op: TransformOperation) -> TransformOperation {
                op.description(
                    "Get the result of an async operation execution (RequestResults subfunction)",
                )
                .response_with::<200, Json<AsyncGetByIdResponse<VendorErrorCode>>, _>(|res| {
                    res.description("Execution result retrieved successfully.")
                })
                .with(openapi::error_not_found)
                .with(openapi::error_bad_request)
                .with(openapi::error_bad_gateway)
            }

            #[allow(
                clippy::too_many_lines,
                reason = "Keeping validation, execution guarding, UDS Stop, and guard cleanup \
                          together makes cleanup on every response path visible. Splitting the \
                          flow would obscure that state invariant"
            )]
            pub(crate) async fn delete<T: UdsEcu + SchemaProvider + Clone, U: FileManager>(
                UseApi(Secured(security_plugin), _): UseApi<Secured, ()>,
                Path(ServiceAndIdPathParam { service, id }): Path<ServiceAndIdPathParam>,
                WithRejection(Query(query), _): WithRejection<
                    Query<OperationDeleteQuery>,
                    ApiError,
                >,
                State(WebserverEcuState {
                    ecu_name,
                    uds,
                    locks,
                    service_executions,
                    ..
                }): State<WebserverEcuState<T, U>>,
            ) -> Response {
                let include_schema = query.include_schema;
                require_ecu_access!(write, security_plugin, &ecu_name, &locks, include_schema);
                let exec_id = match parse_exec_uuid(&id, include_schema) {
                    Ok(v) => v,
                    Err(e) => return e.into_response(),
                };

                let _request_guard = match guard_execution(
                    &service_executions,
                    &service,
                    exec_id,
                    include_schema,
                    &format!("Execution {exec_id} is already being stopped"),
                ) {
                    Ok(guard) => guard,
                    Err(error) => return error.into_response(),
                };

                let diag_service = DiagComm {
                    name: service.clone(),
                    type_: DiagCommType::Operations,
                    lookup_name: None,
                    subfunction_id: Some(subfunction_ids::routine::STOP),
                };
                let uds_result = uds
                    .send(
                        &ecu_name,
                        diag_service,
                        &(security_plugin as DynamicPlugin),
                        None,
                        true,
                    )
                    .await;

                match uds_result {
                    Ok(r) if matches!(r.response_type(), DiagServiceResponseType::Positive) => {
                        if let Some(op_map) = lock_write(&service_executions).get_mut(&service) {
                            op_map.shift_remove(&exec_id);
                        }
                        if r.is_empty() {
                            StatusCode::NO_CONTENT.into_response()
                        } else {
                            let (parameters, error_list) =
                                parse_json_response_params::<T::Response>(r, "Stop");
                            get_by_id_response(
                                ExecutionStatus::Stopped,
                                parameters,
                                error_list,
                                include_schema,
                            )
                        }
                    }
                    Err(cda_interfaces::DiagServiceError::NotFound(_))
                        if query.suppress_service =>
                    {
                        tracing::warn!(
                            service = %service,
                            exec_id = %exec_id,
                            "Stop service not found (suppress_service=true), removing execution"
                        );
                        if let Some(op_map) = lock_write(&service_executions).get_mut(&service) {
                            op_map.shift_remove(&exec_id);
                        }
                        StatusCode::NO_CONTENT.into_response()
                    }
                    result => {
                        if let Err(e) = &result {
                            tracing::warn!(
                                error = ?e,
                                service = %service,
                                exec_id = %exec_id,
                                "Stop subfunction failed"
                            );
                        } else {
                            tracing::warn!(
                                service = %service,
                                exec_id = %exec_id,
                                "Stop subfunction returned negative response"
                            );
                        }
                        if query.force {
                            if let Some(op_map) = lock_write(&service_executions).get_mut(&service)
                            {
                                op_map.shift_remove(&exec_id);
                            }
                            return StatusCode::NO_CONTENT.into_response();
                        }

                        match result {
                            Err(e) => ErrorWrapper {
                                error: e.into(),
                                include_schema,
                            }
                            .into_response(),
                            Ok(r) => {
                                api_error_from_diag_response(&r, include_schema).into_response()
                            }
                        }
                    }
                }
            }

            pub(crate) fn docs_delete(op: TransformOperation) -> TransformOperation {
                op.description(
                    "Stop an async operation execution (Stop subfunction). Use \
                     ?x-sovd2uds-force=true to remove even on ECU error.",
                )
                .response_with::<204, (), _>(|res| {
                    res.description("Execution stopped and removed (no response data from ECU).")
                })
                .response_with::<200, Json<AsyncGetByIdResponse<VendorErrorCode>>, _>(|res| {
                    res.description(
                        "Execution stopped and removed. The ECU returned response data (non-spec \
                         extension).",
                    )
                })
                .with(openapi::error_not_found)
                .with(openapi::error_bad_request)
                .with(openapi::error_bad_gateway)
                .with(openapi::error_forbidden)
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use sovd_interfaces::components::ecu::operations::comparams::ComParamSimpleValue;

    mod comparam_execution_authorization {
        use std::sync::Arc;

        use aide::UseApi;
        use axum::extract::{OriginalUri, Path, Query, State};
        use axum_extra::extract::WithRejection;
        use cda_interfaces::{
            HashMap, HashMapExtensions, file_manager::mock::MockFileManager, mock::MockUdsEcu,
        };
        use cda_plugin_security::{
            Secured, SecurityPlugin,
            mock::{MockAuthApi, MockClaims, MockSecurityPlugin, TestSecurityPlugin},
        };
        use opensovd_axum_extra::ExtractHost;
        use sovd_interfaces::components::ecu::operations::comparams::{self, executions};
        use uuid::Uuid;

        use super::super::comparams::executions as handlers;
        use crate::sovd::{
            components::IdPathParam, locks::insert_test_ecu_lock,
            tests::create_test_webserver_state,
        };

        fn foreign_security_plugin() -> Box<dyn SecurityPlugin> {
            let mut claims = MockClaims::new();
            claims.expect_sub().return_const("foreign_user");
            let claims: &'static dyn cda_plugin_security::Claims = Box::leak(Box::new(claims));
            let mut auth = MockAuthApi::new();
            auth.expect_claims().return_const(Box::new(claims));
            let mut plugin = MockSecurityPlugin::new();
            plugin
                .expect_as_auth_plugin()
                .return_const(Box::new(auth) as Box<dyn cda_plugin_security::AuthApi>);
            Box::new(plugin)
        }

        fn query() -> WithRejection<Query<executions::get::Query>, crate::sovd::error::ApiError> {
            WithRejection(
                Query(sovd_interfaces::IncludeSchemaQuery {
                    include_schema: false,
                }),
                std::marker::PhantomData,
            )
        }

        fn request() -> executions::update::Request {
            executions::update::Request {
                capability: None,
                timeout: None,
                parameters: Some(HashMap::from_iter([(
                    "P2ServerMax".to_owned(),
                    comparams::ComParamValue::Simple(comparams::ComParamSimpleValue {
                        value: "100".to_owned(),
                        unit: None,
                    }),
                )])),
                proximity_response: None,
            }
        }

        fn execution(
            communication_access: &dyn cda_interfaces::communication_control::CommunicationAccess,
        ) -> crate::sovd::ComparamExecution {
            crate::sovd::ComparamExecution::new(
                comparams::Execution {
                    capability: executions::Capability::Execute,
                    status: executions::Status::Running,
                    comparam_override: HashMap::new(),
                },
                communication_access
                    .acquire()
                    .expect("Test communication must be enabled"),
            )
        }

        #[tokio::test]
        async fn get_foreign_exclusive_lock_skips_uds() {
            let mut uds = MockUdsEcu::new();
            uds.expect_get_comparams().times(0);
            let state =
                create_test_webserver_state("TestECU".to_owned(), uds, MockFileManager::new());
            insert_test_ecu_lock(&state.locks, "TestECU").await;
            let id = Uuid::new_v4();
            state
                .comparam_executions
                .write()
                .await
                .insert(id, execution(&*state.communication_access));
            let executions = Arc::clone(&state.comparam_executions);

            let response = handlers::id::get::<MockUdsEcu, MockFileManager>(
                UseApi(Secured(foreign_security_plugin()), std::marker::PhantomData),
                Path(IdPathParam { id: id.to_string() }),
                query(),
                State(state),
            )
            .await;

            assert_eq!(response.status(), http::StatusCode::LOCKED);
            assert!(executions.read().await.contains_key(&id));
        }

        #[tokio::test]
        async fn post_without_owned_write_lock_does_not_create_execution() {
            let state = create_test_webserver_state(
                "TestECU".to_owned(),
                MockUdsEcu::new(),
                MockFileManager::new(),
            );
            let executions = Arc::clone(&state.comparam_executions);

            let response = handlers::post::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                query(),
                State(state),
                UseApi(
                    ExtractHost("localhost".to_owned()),
                    std::marker::PhantomData,
                ),
                OriginalUri(
                    "/components/TestECU/operations/comparams/executions"
                        .parse()
                        .unwrap(),
                ),
                None,
            )
            .await;

            assert_eq!(response.status(), http::StatusCode::CONFLICT);
            assert!(executions.read().await.is_empty());
        }

        #[tokio::test]
        async fn put_without_owned_write_lock_does_not_mutate_execution() {
            let state = create_test_webserver_state(
                "TestECU".to_owned(),
                MockUdsEcu::new(),
                MockFileManager::new(),
            );
            let id = Uuid::new_v4();
            state
                .comparam_executions
                .write()
                .await
                .insert(id, execution(&*state.communication_access));
            let executions = Arc::clone(&state.comparam_executions);

            let response = handlers::id::put::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(IdPathParam { id: id.to_string() }),
                query(),
                State(state),
                UseApi(
                    ExtractHost("localhost".to_owned()),
                    std::marker::PhantomData,
                ),
                OriginalUri(
                    format!("/components/TestECU/operations/comparams/executions/{id}")
                        .parse()
                        .unwrap(),
                ),
                WithRejection(axum::Json(request()), std::marker::PhantomData),
            )
            .await;

            assert_eq!(response.status(), http::StatusCode::CONFLICT);
            let executions = executions.read().await;
            assert!(
                executions
                    .get(&id)
                    .is_some_and(|execution| execution.comparam_override.is_empty())
            );
        }

        #[tokio::test]
        async fn delete_without_owned_write_lock_does_not_remove_execution() {
            let state = create_test_webserver_state(
                "TestECU".to_owned(),
                MockUdsEcu::new(),
                MockFileManager::new(),
            );
            let id = Uuid::new_v4();
            state
                .comparam_executions
                .write()
                .await
                .insert(id, execution(&*state.communication_access));
            let executions = Arc::clone(&state.comparam_executions);

            let response = handlers::id::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(IdPathParam { id: id.to_string() }),
                State(state),
            )
            .await;

            assert_eq!(response.status(), http::StatusCode::CONFLICT);
            assert!(executions.read().await.contains_key(&id));
        }

        #[tokio::test]
        async fn post_with_owned_write_lock_creates_execution() {
            let state = create_test_webserver_state(
                "TestECU".to_owned(),
                MockUdsEcu::new(),
                MockFileManager::new(),
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;
            let executions = Arc::clone(&state.comparam_executions);

            let response = handlers::post::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                query(),
                State(state),
                UseApi(
                    ExtractHost("localhost".to_owned()),
                    std::marker::PhantomData,
                ),
                OriginalUri(
                    "/components/TestECU/operations/comparams/executions"
                        .parse()
                        .unwrap(),
                ),
                Some(axum::Json(request())),
            )
            .await;

            assert_eq!(response.status(), http::StatusCode::ACCEPTED);
            assert_eq!(executions.read().await.len(), 1);
        }
    }

    #[test]
    #[allow(clippy::float_cmp, reason = "Test verifies exact deserialized values")]
    fn com_param_simple_deserialization() {
        let json_data_string = "\"example_value\"";
        let deserialized_string: ComParamSimpleValue =
            serde_json::from_str(json_data_string).unwrap();
        assert_eq!(deserialized_string.value, "example_value");
        assert!(deserialized_string.unit.is_none());

        let json_data_struct = r#"{
        "value": "test",
        "unit": {
            "factor_to_si_unit": 1.0,
            "offset_to_si_unit": 0.0
        }
        }"#;
        let deserialized_struct: ComParamSimpleValue =
            serde_json::from_str(json_data_struct).unwrap();
        assert_eq!(deserialized_struct.value, "test");
        let unit = deserialized_struct.unit.unwrap();
        assert_eq!(unit.factor_to_si_unit.unwrap(), 1.0);
        assert_eq!(unit.offset_to_si_unit.unwrap(), 0.0);

        let json_data_struct = r#"{
        "value": "test",
        "unit": {
            "factor_to_si_unit": 1.0
        }
        }"#;
        let deserialized_struct: ComParamSimpleValue =
            serde_json::from_str(json_data_struct).unwrap();
        assert_eq!(deserialized_struct.value, "test");
        let unit = deserialized_struct.unit.unwrap();
        assert_eq!(unit.factor_to_si_unit.unwrap(), 1.0);
        assert!(unit.offset_to_si_unit.is_none());

        let json_data_struct = r#"{"value": "test"}"#;
        let deserialized_struct: ComParamSimpleValue =
            serde_json::from_str(json_data_struct).unwrap();
        assert_eq!(deserialized_struct.value, "test");
        assert!(deserialized_struct.unit.is_none());

        let json_data_struct = r#"{"unit": {"factor_to_si_unit": 1.0}}"#;
        assert!(serde_json::from_str::<ComParamSimpleValue>(json_data_struct).is_err());
    }

    mod ecu_operations_collection {
        use aide::UseApi;
        use axum::{extract::State, http::StatusCode};
        use axum_extra::extract::WithRejection;
        use cda_interfaces::{
            datatypes::ComponentOperationsInfo,
            file_manager::mock::MockFileManager,
            mock::{MockUdsEcu, mock_ecu_state_online_variant_detected},
        };
        use cda_plugin_security::{Secured, mock::TestSecurityPlugin};
        use sovd_interfaces::components::ecu::operations::OperationCollectionItem;

        use super::super::*;
        use crate::sovd::tests::create_test_webserver_state;

        #[tokio::test]
        async fn test_get_operations_returns_empty_list() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_ecu_state()
                .returning(|_| Ok(mock_ecu_state_online_variant_detected()));
            mock_uds
                .expect_get_components_operations_info()
                .withf(|ecu, _| ecu == "TestECU")
                .times(1)
                .returning(|_, _| Ok(vec![]));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );

            let response = get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                WithRejection(
                    Query(sovd_interfaces::IncludeSchemaQuery {
                        include_schema: false,
                    }),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: sovd_interfaces::Items<OperationCollectionItem> =
                serde_json::from_slice(&body).unwrap();
            assert!(result.items.is_empty());
            assert!(result.schema.is_none());
        }

        #[tokio::test]
        async fn test_get_operations_returns_items() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_ecu_state()
                .returning(|_| Ok(mock_ecu_state_online_variant_detected()));
            mock_uds
                .expect_get_components_operations_info()
                .withf(|ecu, _| ecu == "TestECU")
                .times(1)
                .returning(|_, _| {
                    Ok(vec![
                        ComponentOperationsInfo {
                            id: "CalibrateSensor".to_string(),
                            name: "Calibrate Sensor".to_string(),
                            has_stop: true,
                            has_request_results: true,
                        },
                        ComponentOperationsInfo {
                            id: "RunSelfTest".to_string(),
                            name: "Run Self Test".to_string(),
                            has_stop: false,
                            has_request_results: false,
                        },
                    ])
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );

            let response = get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                WithRejection(
                    Query(sovd_interfaces::IncludeSchemaQuery {
                        include_schema: false,
                    }),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: sovd_interfaces::Items<OperationCollectionItem> =
                serde_json::from_slice(&body).unwrap();
            assert_eq!(result.items.len(), 2);

            let first = result.items.first().expect("Expected at least one item");
            assert_eq!(first.id, "CalibrateSensor");
            assert_eq!(first.name, "Calibrate Sensor");
            assert!(!first.proximity_proof_required);
            assert!(first.asynchronous_execution);

            let second = result.items.get(1).expect("Expected a second item");
            assert_eq!(second.id, "RunSelfTest");
            assert!(!second.proximity_proof_required);
            assert!(!second.asynchronous_execution);
        }

        #[tokio::test]
        async fn test_get_operations_with_schema() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_ecu_state()
                .returning(|_| Ok(mock_ecu_state_online_variant_detected()));
            mock_uds
                .expect_get_components_operations_info()
                .times(1)
                .returning(|_, _| Ok(vec![]));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );

            let response = get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                WithRejection(
                    Query(sovd_interfaces::IncludeSchemaQuery {
                        include_schema: true,
                    }),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: sovd_interfaces::Items<OperationCollectionItem> =
                serde_json::from_slice(&body).unwrap();
            assert!(
                result.schema.is_some(),
                "Schema should be included when requested"
            );
        }
    }

    mod service_executions {
        use std::sync::Arc;

        use aide::UseApi;
        use axum::{
            extract::{Path, Query, State},
            http::StatusCode,
        };
        use axum_extra::extract::WithRejection;
        use cda_interfaces::{
            DiagComm, DiagCommType, DiagServiceError,
            diagservices::{
                DiagServiceJsonResponse, DiagServiceResponseType, mock::MockDiagServiceResponse,
            },
            file_manager::mock::MockFileManager,
            mock::MockUdsEcu,
            util::std_ext::{lock_read, lock_write},
        };
        use cda_plugin_communication_management::lifecycle::enabled_communication_access_for_test;
        use cda_plugin_security::{Secured, mock::TestSecurityPlugin};
        use indexmap::IndexMap;
        use sovd_interfaces::{
            common::operations::OperationIdItem, components::ecu::operations::ExecutionStatus,
        };

        use super::super::service::{executions as handlers, executions::id as id_handlers};
        use crate::sovd::{
            ServiceExecution, locks::insert_test_ecu_lock, tests::create_test_webserver_state,
        };

        fn make_json_response(data: serde_json::Value) -> MockDiagServiceResponse {
            let mut resp = MockDiagServiceResponse::new();
            resp.expect_response_type()
                .returning(|| DiagServiceResponseType::Positive);
            resp.expect_is_empty().returning(|| false);
            resp.expect_into_json().return_once(move || {
                Ok(DiagServiceJsonResponse {
                    data,
                    errors: vec![],
                })
            });
            resp
        }

        fn make_empty_positive_response() -> MockDiagServiceResponse {
            let mut resp = MockDiagServiceResponse::new();
            resp.expect_response_type()
                .returning(|| DiagServiceResponseType::Positive);
            resp.expect_is_empty().returning(|| true);
            resp
        }

        fn make_negative_response() -> MockDiagServiceResponse {
            let mut resp = MockDiagServiceResponse::new();
            resp.expect_response_type()
                .returning(|| DiagServiceResponseType::Negative);
            resp.expect_is_empty().returning(|| false);
            resp.expect_as_nrc().returning(|| {
                Ok(cda_interfaces::diagservices::MappedNRC {
                    code: Some(0x22),
                    description: Some("conditionsNotCorrect".to_string()),
                    sid: None,
                })
            });
            resp
        }

        async fn ecu_operation_write_handler<
            T: cda_interfaces::UdsEcu + cda_interfaces::SchemaProvider + Clone,
        >(
            req: handlers::WriteHandlerRequest,
            ecu_name: &str,
            uds: &T,
            service_executions: Arc<
                std::sync::RwLock<
                    cda_interfaces::HashMap<
                        String,
                        indexmap::IndexMap<uuid::Uuid, ServiceExecution>,
                    >,
                >,
            >,
            security_plugin: Box<dyn cda_plugin_security::SecurityPlugin>,
            opts: handlers::WriteHandlerOptions,
        ) -> axum::response::Response {
            let ctx = handlers::OperationWriteContext::new(
                service_executions,
                enabled_communication_access_for_test(),
            );
            handlers::ecu_operation_write_handler_with_activity::<T>(
                req,
                ecu_name,
                uds,
                ctx,
                security_plugin,
                opts,
            )
            .await
        }

        #[tokio::test]
        async fn test_list_executions_empty() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let response = handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(handlers::OperationServicePathParam {
                    service: "CalibrateSensor".to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::service::executions::Query {
                            include_schema: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: sovd_interfaces::Items<OperationIdItem> =
                serde_json::from_slice(&body).unwrap();
            assert!(result.items.is_empty());
        }

        #[tokio::test]
        async fn test_list_executions_shows_tracked_id() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            // Pre-populate an execution
            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(handlers::OperationServicePathParam {
                    service: "CalibrateSensor".to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::service::executions::Query {
                            include_schema: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: sovd_interfaces::Items<OperationIdItem> =
                serde_json::from_slice(&body).unwrap();
            assert_eq!(result.items.len(), 1);
            assert_eq!(
                result.items.first().expect("Expected at least one item").id,
                exec_id.to_string()
            );
        }

        #[tokio::test]
        async fn test_get_execution_by_id_not_found() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let unknown_id = uuid::Uuid::new_v4().to_string();
            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: unknown_id,
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::NOT_FOUND);
        }

        #[tokio::test]
        async fn test_get_execution_by_id_calls_request_results() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // Expect send with subfunction_id = REQUEST_RESULTS (0x03)
            mock_uds
                .expect_send()
                .withf(|ecu, service, _, payload, map_to_json| {
                    ecu == "TestECU"
                        && service.type_ == DiagCommType::Operations
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::REQUEST_RESULTS)
                        && service.lookup_name.is_none()
                        && payload.is_none()
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| {
                    Ok(make_json_response(serde_json::json!({
                        "result": "ok"
                    })))
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("completed")
            );
            assert_eq!(
                result.get("capability").expect("missing capability"),
                &serde_json::json!("execute")
            );
            assert_eq!(
                result
                    .get("parameters")
                    .expect("missing parameters")
                    .get("result")
                    .expect("missing result"),
                &serde_json::json!("ok")
            );
        }

        #[tokio::test]
        async fn test_get_execution_suppress_service_skips_send_returns_stored() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // suppress_service=true must skip the UDS send entirely
            mock_uds.expect_send().times(0);

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            let stored_params = {
                let mut m = serde_json::Map::new();
                m.insert("stored".to_string(), serde_json::json!("value"));
                m
            };
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: stored_params,
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: true,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // suppress_service=true -> should return 200 with stored params, no UDS send
            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("running")
            );
            assert_eq!(
                result
                    .get("parameters")
                    .expect("missing parameters")
                    .get("stored")
                    .expect("missing stored"),
                &serde_json::json!("value")
            );
        }

        #[tokio::test]
        async fn test_get_execution_not_found_without_suppress_returns_error() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                Err(DiagServiceError::NotFound(
                    "CalibrateSensor_RequestResults not found".to_string(),
                ))
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // suppress_service=false -> NotFound from UDS should propagate as error
            assert!(response.status().is_client_error() || response.status().is_server_error());
        }

        #[tokio::test]
        async fn test_delete_execution_not_found() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, &ecu_name).await;

            let unknown_id = uuid::Uuid::new_v4().to_string();
            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: unknown_id,
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::NOT_FOUND);
        }

        #[tokio::test]
        async fn test_delete_execution_calls_stop() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // Expect send with subfunction_id = STOP (0x02)
            mock_uds
                .expect_send()
                .withf(|ecu, service, _, _, _| {
                    ecu == "TestECU"
                        && service.type_ == DiagCommType::Operations
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::STOP)
                        && service.lookup_name.is_none()
                })
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_empty_positive_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            // Keep a reference to service_executions so we can verify after consuming state
            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::NO_CONTENT);
            // Verify execution was removed
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_stop_with_data_returns_200_stopped() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // ECU returns a non-empty positive response from Stop
            mock_uds
                .expect_send()
                .withf(|ecu, service, _, _, map_to_json| {
                    ecu == "TestECU"
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::STOP)
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| {
                    Ok(make_json_response(serde_json::json!({
                        "stop_result": "ok"
                    })))
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // ECU returned data -> 200 with status=stopped and the parameters
            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("stopped")
            );
            assert_eq!(
                result
                    .get("parameters")
                    .expect("missing parameters")
                    .get("stop_result")
                    .expect("missing stop_result"),
                &serde_json::json!("ok")
            );
            // Execution must be removed regardless
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_stop_with_null_json_returns_200_empty_parameters() {
            // Stop maps to JSON Null -> 200 with empty parameters (user-requested extension)
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_send()
                .withf(|ecu, service, _, _, map_to_json| {
                    ecu == "TestECU"
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::STOP)
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_json_response(serde_json::Value::Null)));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("stopped")
            );
            // parameters: None when empty -> field is omitted from JSON
            assert!(
                result.get("parameters").is_none(),
                "parameters should be absent when Stop returns Null"
            );
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_stop_non_object_json_returns_200_stopped_with_error() {
            // Stop maps to a non-object JSON value (e.g. a string) -> 200 stopped, error surfaced
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_send()
                .withf(|ecu, service, _, _, map_to_json| {
                    ecu == "TestECU"
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::STOP)
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| {
                    Ok(make_json_response(serde_json::json!("unexpected_string")))
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("stopped")
            );
            // error list (field name "error" per AsyncGetByIdResponse) must be non-empty
            let errors = result.get("error").expect("missing error field");
            assert!(errors.is_array() && !errors.as_array().unwrap().is_empty());
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_stop_into_json_error_returns_200_stopped_with_error() {
            // Stop response cannot be parsed (into_json fails) -> 200 stopped, error surfaced
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_send()
                .withf(|ecu, service, _, _, map_to_json| {
                    ecu == "TestECU"
                        && service.subfunction_id
                            == Some(cda_interfaces::subfunction_ids::routine::STOP)
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| {
                    let mut resp = MockDiagServiceResponse::new();
                    resp.expect_response_type()
                        .returning(|| DiagServiceResponseType::Positive);
                    resp.expect_is_empty().returning(|| false);
                    resp.expect_into_json().return_once(|| {
                        Err(DiagServiceError::BadPayload(
                            "simulated Stop parse failure".to_string(),
                        ))
                    });
                    Ok(resp)
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert_eq!(
                result.get("status").expect("missing status"),
                &serde_json::json!("stopped")
            );
            let errors = result.get("error").expect("missing error field");
            assert!(errors.is_array() && !errors.as_array().unwrap().is_empty());
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_force_removes_on_uds_error() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // UDS returns an error (non-NotFound)
            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                Err(DiagServiceError::SendFailed("timeout".to_string()))
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            // Keep a reference to service_executions so we can verify after consuming state
            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: true,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // force=true -> removes execution even on error
            assert_eq!(response.status(), StatusCode::NO_CONTENT);
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_force_removes_on_negative_response() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_send()
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_negative_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: true,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // force=true -> removes execution even on negative ECU response
            assert_eq!(response.status(), StatusCode::NO_CONTENT);
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_delete_execution_without_force_returns_error_on_uds_failure() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                Err(DiagServiceError::SendFailed("timeout".to_string()))
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            // Keep a reference to service_executions so we can verify after consuming state
            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // force=false -> error should be returned, execution should remain
            assert!(response.status().is_client_error() || response.status().is_server_error());
            assert_eq!(lock_read(&service_executions_ref).len(), 1);
        }

        #[tokio::test]
        async fn test_delete_execution_negative_response_resets_in_flight() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_send()
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_negative_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // Negative ECU response without force -> error returned, in_flight reset
            assert!(response.status().is_client_error() || response.status().is_server_error());
            let guard = lock_read(&service_executions_ref);
            let exec = guard
                .get("CalibrateSensor")
                .and_then(|m| m.get(&exec_id))
                .expect("execution should still exist");
            assert!(!exec.in_flight, "in_flight should be reset to false");
        }

        #[tokio::test]
        async fn test_delete_execution_suppress_service_removes_on_not_found() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                Err(DiagServiceError::NotFound(
                    "CalibrateSensor_Stop not found".to_string(),
                ))
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            // Keep a reference to service_executions so we can verify after consuming state
            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: true,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // suppress_service=true on NotFound -> removes execution, returns 204
            assert_eq!(response.status(), StatusCode::NO_CONTENT);
            assert!(
                lock_read(&service_executions_ref)
                    .values()
                    .all(IndexMap::is_empty)
            );
        }

        #[tokio::test]
        async fn test_get_execution_in_flight_returns_conflict() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: true,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::CONFLICT);
        }

        #[tokio::test]
        async fn test_delete_execution_in_flight_returns_conflict() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();
            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, &ecu_name).await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: true,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::delete::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationDeleteQuery {
                            include_schema: false,
                            suppress_service: false,
                            force: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            assert_eq!(response.status(), StatusCode::CONFLICT);
        }

        fn make_post_headers() -> axum::http::HeaderMap {
            let mut headers = axum::http::HeaderMap::new();
            headers.insert(
                axum::http::header::CONTENT_TYPE,
                axum::http::HeaderValue::from_static("application/json"),
            );
            headers.insert(
                axum::http::header::ACCEPT,
                axum::http::HeaderValue::from_static("application/json"),
            );
            headers
        }

        #[tokio::test]
        async fn test_post_operation_conflict_when_running_execution_exists() {
            let ecu_name = "TestECU".to_string();
            let mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            // Pre-populate a running execution for CalibrateSensor
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    uuid::Uuid::new_v4(),
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            assert_eq!(response.status(), StatusCode::CONFLICT);
        }

        #[tokio::test]
        async fn test_post_operation_no_conflict_for_different_service() {
            // An execution running for ServiceA must NOT block ServiceB
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Ok(cda_interfaces::datatypes::RoutineSubfunctions {
                        has_stop: false,
                        has_request_results: false,
                    })
                });
            mock_uds
                .expect_send()
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_empty_positive_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            // Pre-populate a running execution for a DIFFERENT service
            lock_write(&state.service_executions)
                .entry("OtherService".to_string())
                .or_default()
                .insert(
                    uuid::Uuid::new_v4(),
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            // Different service -> no conflict, should pass through to 200
            assert_eq!(response.status(), StatusCode::OK);
        }

        #[tokio::test]
        async fn test_post_operation_service_not_found_returns_404() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .withf(|ecu, svc, _p| ecu == "TestECU" && svc == "CalibrateSensor")
                .times(1)
                .returning(|_, _, _| {
                    Err(DiagServiceError::NotFound(
                        "Routine 'CalibrateSensor' not found in ECU description".to_string(),
                    ))
                });
            mock_uds
                .expect_get_io_control_service()
                .withf(|ecu, svc, _p| ecu == "TestECU" && svc == "CalibrateSensor")
                .times(1)
                .returning(|_, _, _| {
                    Err(DiagServiceError::NotFound(
                        "No InputOutputControlByIdentifier service found".to_owned(),
                    ))
                });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            assert_eq!(response.status(), StatusCode::NOT_FOUND);
        }

        #[tokio::test]
        async fn test_post_operation_sync_returns_200_on_empty_response() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Ok(cda_interfaces::datatypes::RoutineSubfunctions {
                        has_stop: false,
                        has_request_results: false,
                    })
                });
            mock_uds
                .expect_send()
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_empty_positive_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

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

        #[tokio::test]
        async fn test_post_named_io_control_uses_exact_operation_service() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Err(DiagServiceError::NotFound(
                        "Not a RoutineControl service".to_owned(),
                    ))
                });
            mock_uds
                .expect_get_io_control_service()
                .withf(|ecu, svc, _p| ecu == "TestECU" && svc == "TestOutput_Control")
                .times(1)
                .returning(|_, _, _| {
                    Ok(DiagComm {
                        name: "TestOutput_Control".to_owned(),
                        type_: DiagCommType::Operations,
                        lookup_name: Some("TestOutput_Control".to_owned()),
                        subfunction_id: None,
                    })
                });
            mock_uds
                .expect_send()
                .withf(|ecu, service, _, payload, map_to_json| {
                    ecu == "TestECU"
                        && service.type_ == DiagCommType::Operations
                        && service.name == "TestOutput_Control"
                        && service.lookup_name.as_deref() == Some("TestOutput_Control")
                        && service.subfunction_id.is_none()
                        && payload.is_some()
                        && *map_to_json
                })
                .times(1)
                .returning(|_, _, _, _, _| Ok(make_empty_positive_response()));

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "TestOutput_Control".to_owned(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(
                        br#"{"parameters":{"OutputValue":"ACTIVE"}}"#,
                    ),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/TestOutput_Control/executions"
                        .to_owned(),
                },
            )
            .await;

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

        #[tokio::test]
        async fn test_post_operation_async_returns_202_and_tracks_execution() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Ok(cda_interfaces::datatypes::RoutineSubfunctions {
                        has_stop: true,
                        has_request_results: true,
                    })
                });
            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                Ok(make_json_response(serde_json::json!({
                    "routineInfo": 1,
                    "result": 0
                })))
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            assert_eq!(response.status(), StatusCode::ACCEPTED);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert!(result.get("id").is_some(), "202 body must have id");
            assert_eq!(result.get("status"), Some(&serde_json::json!("running")));
            assert_eq!(
                result.get("parameters"),
                Some(&serde_json::json!({"routineInfo": 1, "result": 0}))
            );
            assert_eq!(lock_read(&service_executions_ref).len(), 1);
        }

        #[tokio::test]
        async fn test_post_operation_suppress_service_async_skips_send_returns_202_and_tracks() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds.expect_get_routine_subfunctions().times(0);
            // send must NOT be called when suppress_service=true
            mock_uds.expect_send().times(0);

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: true,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            assert_eq!(response.status(), StatusCode::ACCEPTED);
            // execution still tracked even though UDS was not called
            assert_eq!(lock_read(&service_executions_ref).len(), 1);
        }

        #[tokio::test]
        async fn test_post_operation_async_into_json_error_returns_empty_parameters() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Ok(cda_interfaces::datatypes::RoutineSubfunctions {
                        has_stop: true,
                        has_request_results: true,
                    })
                });
            // send returns a response whose into_json() fails
            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                let mut resp = MockDiagServiceResponse::new();
                resp.expect_response_type()
                    .returning(|| DiagServiceResponseType::Positive);
                resp.expect_is_empty().returning(|| false);
                resp.expect_into_json().return_once(|| {
                    Err(DiagServiceError::BadPayload(
                        "simulated parse failure".to_string(),
                    ))
                });
                Ok(resp)
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            // Decoding failures do not prevent the asynchronous execution from starting.
            assert_eq!(response.status(), StatusCode::ACCEPTED);
            // Execution must still be tracked
            assert_eq!(lock_read(&service_executions_ref).len(), 1);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert!(result.get("id").is_some(), "202 body must have id");
            assert!(result.get("status").is_some(), "202 body must have status");
            assert_eq!(result.get("parameters"), Some(&serde_json::json!({})));
            let errors = result
                .get("errors")
                .and_then(serde_json::Value::as_array)
                .expect("202 body must contain errors");
            assert_eq!(errors.len(), 1);
            assert_eq!(
                errors.first().and_then(|error| error.get("path")),
                Some(&serde_json::json!("/parameters"))
            );
            assert_eq!(
                errors
                    .first()
                    .and_then(|error| error.get("error"))
                    .and_then(|error| error.get("error_code")),
                Some(&serde_json::json!("invalid-response-content"))
            );
        }

        #[tokio::test]
        async fn test_post_operation_async_non_object_json_returns_empty_parameters() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            mock_uds
                .expect_get_routine_subfunctions()
                .times(1)
                .returning(|_, _, _| {
                    Ok(cda_interfaces::datatypes::RoutineSubfunctions {
                        has_stop: true,
                        has_request_results: true,
                    })
                });
            // send returns a response whose into_json() gives a non-Object JSON value
            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                let mut resp = MockDiagServiceResponse::new();
                resp.expect_response_type()
                    .returning(|| DiagServiceResponseType::Positive);
                resp.expect_is_empty().returning(|| false);
                resp.expect_into_json().return_once(|| {
                    Ok(DiagServiceJsonResponse {
                        data: serde_json::Value::String("unexpected_string".to_string()),
                        errors: vec![],
                    })
                });
                Ok(resp)
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name.clone(),
                mock_uds,
                mock_file_manager,
            );

            let service_executions_ref = Arc::clone(&state.service_executions);

            let response = ecu_operation_write_handler::<MockUdsEcu>(
                handlers::WriteHandlerRequest {
                    service: "CalibrateSensor".to_string(),
                    headers: make_post_headers(),
                    body: axum::body::Bytes::from_static(b"{\"parameters\":{}}"),
                },
                &ecu_name,
                &state.uds,
                Arc::clone(&state.service_executions),
                Box::new(cda_plugin_security::mock::TestSecurityPlugin),
                handlers::WriteHandlerOptions {
                    include_schema: false,
                    suppress_service: false,
                    base_path: "http://localhost/operations/CalibrateSensor/executions".to_string(),
                },
            )
            .await;

            // Non-object response data does not prevent the asynchronous execution from starting.
            assert_eq!(response.status(), StatusCode::ACCEPTED);
            assert_eq!(lock_read(&service_executions_ref).len(), 1);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            assert!(result.get("id").is_some(), "202 body must have id");
            assert!(result.get("status").is_some(), "202 body must have status");
            assert_eq!(result.get("parameters"), Some(&serde_json::json!({})));
            assert_eq!(
                result
                    .get("errors")
                    .and_then(serde_json::Value::as_array)
                    .map(Vec::len),
                Some(1)
            );
        }

        #[tokio::test]
        async fn test_request_results_into_json_error_surfaces_in_errors_field() {
            let ecu_name = "TestECU".to_string();
            let mut mock_uds = MockUdsEcu::new();
            let mock_file_manager = MockFileManager::new();

            // RequestResults returns a non-empty response whose into_json() fails
            mock_uds.expect_send().times(1).returning(|_, _, _, _, _| {
                let mut resp = MockDiagServiceResponse::new();
                resp.expect_response_type()
                    .returning(|| DiagServiceResponseType::Positive);
                resp.expect_is_empty().returning(|| false);
                resp.expect_into_json().return_once(|| {
                    Err(DiagServiceError::BadPayload(
                        "simulated parse failure".to_string(),
                    ))
                });
                Ok(resp)
            });

            let state = create_test_webserver_state::<MockUdsEcu, MockFileManager>(
                ecu_name,
                mock_uds,
                mock_file_manager,
            );
            insert_test_ecu_lock(&state.locks, "TestECU").await;

            let exec_id = uuid::Uuid::new_v4();
            lock_write(&state.service_executions)
                .entry("CalibrateSensor".to_string())
                .or_default()
                .insert(
                    exec_id,
                    ServiceExecution {
                        parameters: serde_json::Map::new(),
                        status: ExecutionStatus::Running,
                        in_flight: false,
                        is_created: true,
                        communication_lease: None,
                    },
                );

            let response = id_handlers::get::<MockUdsEcu, MockFileManager>(
                UseApi(
                    Secured(Box::new(TestSecurityPlugin)),
                    std::marker::PhantomData,
                ),
                Path(id_handlers::ServiceAndIdPathParam {
                    service: "CalibrateSensor".to_string(),
                    id: exec_id.to_string(),
                }),
                WithRejection(
                    Query(
                        sovd_interfaces::components::ecu::operations::OperationQuery {
                            include_schema: false,
                            suppress_service: false,
                        },
                    ),
                    std::marker::PhantomData,
                ),
                State(state),
            )
            .await;

            // Must be 200, not 500
            assert_eq!(response.status(), StatusCode::OK);
            let body = axum::body::to_bytes(response.into_body(), usize::MAX)
                .await
                .unwrap();
            let result: serde_json::Value = serde_json::from_slice(&body).unwrap();
            // Spec Table 189: field is named `error` (singular key, array value)
            let errors = result.get("error").expect("missing error field");
            assert!(
                errors.as_array().is_some_and(|a| !a.is_empty()),
                "error should be non-empty when RequestResults into_json fails"
            );
        }
    }
}