hl7v2-server 1.5.0

//! gRPC service implementation for HL7v2.

use crate::models::{
    EvidenceBundleSummary as ServerEvidenceBundleSummary,
    QuarantineOutputSummary as ServerQuarantineOutputSummary,
    QuarantineReason as ServerQuarantineReason, RedactionAction as ServerRedactionAction,
    RedactionActionReceipt as ServerRedactionActionReceipt,
    RedactionActionStatus as ServerRedactionActionStatus,
    RedactionReceipt as ServerRedactionReceipt,
};
use crate::redaction::redact_message;
use crate::server::AppState;
use hl7v2::{
    Comp as RustComp, Field as RustField, Message as RustMessage, Rep as RustRep,
    Segment as RustSegment, parse as rust_parse,
};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::sync::Arc;
use tonic::{Request, Response, Status};

// Include the generated gRPC code
/// Generated gRPC protocol code (protobuf messages and service traits).
#[expect(
    missing_docs,
    reason = "protobuf bindings are generated by tonic/prost"
)]
pub mod proto {
    tonic::include_proto!("hl7v2.v1");
}

use proto::hl7_service_server::Hl7Service;
use proto::*;

/// Implementation of the HL7Service gRPC trait
pub struct Hl7ServiceImpl {
    state: Arc<AppState>,
}

impl Hl7ServiceImpl {
    /// Create a new gRPC service instance
    pub fn new(state: Arc<AppState>) -> Self {
        Self { state }
    }
}

#[tonic::async_trait]
impl Hl7Service for Hl7ServiceImpl {
    async fn parse(
        &self,
        request: Request<ParseRequest>,
    ) -> Result<Response<ParseResponse>, Status> {
        let req = request.into_inner();
        Ok(Response::new(parse_grpc_message_response(
            &req.message,
            req.mllp_framed,
        )))
    }

    type ParseStreamStream =
        tokio_stream::wrappers::ReceiverStream<Result<ParseStreamResponse, Status>>;

    async fn parse_stream(
        &self,
        request: Request<tonic::Streaming<ParseStreamRequest>>,
    ) -> Result<Response<Self::ParseStreamStream>, Status> {
        let mut stream = request.into_inner();
        let (sender, receiver) = tokio::sync::mpsc::channel(8);

        tokio::spawn(async move {
            loop {
                match stream.message().await {
                    Ok(Some(request)) => {
                        let response = parse_stream_response_from_parse_response(
                            parse_grpc_message_response(&request.message, request.mllp_framed),
                        );
                        if sender.send(Ok(response)).await.is_err() {
                            break;
                        }
                    }
                    Ok(None) => break,
                    Err(status) => {
                        if sender.send(Err(status)).await.is_err() {
                            break;
                        }
                        break;
                    }
                }
            }
        });

        Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(
            receiver,
        )))
    }

    async fn validate(
        &self,
        request: Request<ValidateRequest>,
    ) -> Result<Response<ValidateResponse>, Status> {
        let req = request.into_inner();

        let message_bytes = if req.mllp_framed {
            hl7v2::unwrap_mllp(&req.message)
                .map_err(|e| Status::invalid_argument(format!("Failed to unwrap MLLP: {}", e)))?
        } else {
            req.message.as_slice()
        };

        let message = rust_parse(message_bytes)
            .map_err(|e| Status::invalid_argument(format!("Failed to parse HL7: {}", e)))?;

        let profile = hl7v2::load_profile(&req.profile)
            .map_err(|_error| Status::invalid_argument(crate::PROFILE_LOAD_SAFE_MESSAGE))?;

        let issues = hl7v2::validate(&message, &profile);
        let report = hl7v2::ValidationReport::from_issues(
            &message,
            Some(profile.message_structure.clone()),
            issues.clone(),
        );
        let valid = report.valid;

        let mut errors = Vec::new();
        let mut warnings = Vec::new();

        for issue in issues {
            let location = issue.path.map(|p| {
                let mut loc = Location::default();
                let parts: Vec<&str> = p.split('.').collect();
                if !parts.is_empty() {
                    loc.segment = parts[0].to_string();
                }
                if parts.len() > 1 {
                    loc.field = parts[1].parse().unwrap_or(0);
                }
                if parts.len() > 2 {
                    loc.component = parts[2].parse().unwrap_or(0);
                }
                loc
            });

            let proto_issue = ValidationIssue {
                code: issue.code,
                message: issue.detail,
                severity: match issue.severity {
                    hl7v2::Severity::Error => validation_issue::Severity::Error as i32,
                    hl7v2::Severity::Warning => validation_issue::Severity::Warning as i32,
                },
                location,
                advice: String::new(),
            };

            if issue.severity == hl7v2::Severity::Error {
                errors.push(proto_issue);
            } else {
                warnings.push(proto_issue);
            }
        }

        let summary = Some(ValidationSummary {
            error_count: errors.len() as i32,
            warning_count: warnings.len() as i32,
            info_count: 0,
        });

        Ok(Response::new(ValidateResponse {
            valid,
            errors,
            warnings,
            summary,
            validation_report: Some(proto_validation_report_from_rust(&report)),
            validation_report_v2: (req.report_schema_version == 2).then(|| {
                let profile_identity = Some(hl7v2::ValidationReportProfileIdentity {
                    label: profile.message_structure.clone(),
                    message_structure: Some(profile.message_structure.clone()),
                    version: Some(profile.version.clone()),
                    sha256: None,
                });
                proto_validation_report_v2_from_rust(&report.to_v2(
                    "hl7v2-server-grpc",
                    env!("CARGO_PKG_VERSION"),
                    profile_identity,
                ))
            }),
        }))
    }

    async fn profile_lint(
        &self,
        request: Request<ProfileLintRequest>,
    ) -> Result<Response<ProfileLintResponse>, Status> {
        let req = request.into_inner();
        let report_schema_version =
            grpc_requested_schema_version(req.report_schema_version, "profile lint report")
                .map_err(Status::invalid_argument)?;

        let report = hl7v2::lint_profile_yaml(&req.profile);
        let profile_lint_report_v2 = (report_schema_version == 2).then(|| {
            proto_profile_lint_report_v2_from_rust(
                &report.to_v2("hl7v2-server-grpc", env!("CARGO_PKG_VERSION")),
            )
        });

        Ok(Response::new(ProfileLintResponse {
            profile_lint_report: Some(proto_profile_lint_report_from_rust(&report)),
            profile_lint_report_v2,
        }))
    }

    async fn profile_explain(
        &self,
        request: Request<ProfileExplainRequest>,
    ) -> Result<Response<ProfileExplainResponse>, Status> {
        let req = request.into_inner();
        let report_schema_version =
            grpc_requested_schema_version(req.report_schema_version, "profile explain report")
                .map_err(Status::invalid_argument)?;

        let profile = hl7v2::load_profile_checked(&req.profile)
            .map_err(|_error| Status::invalid_argument(crate::PROFILE_LOAD_SAFE_MESSAGE))?;
        let lint_report = hl7v2::lint_profile_yaml(&req.profile);
        let report =
            hl7v2::explain_profile("<inline-profile>", &req.profile, &profile, &lint_report);
        let profile_explain_report_v2 = (report_schema_version == 2).then(|| {
            proto_profile_explain_report_v2_from_rust(
                &report.to_v2("hl7v2-server-grpc", env!("CARGO_PKG_VERSION")),
            )
        });

        Ok(Response::new(ProfileExplainResponse {
            profile_explain_report: Some(proto_profile_explain_report_from_rust(&report)),
            profile_explain_report_v2,
        }))
    }

    async fn profile_test(
        &self,
        request: Request<ProfileTestRequest>,
    ) -> Result<Response<ProfileTestResponse>, Status> {
        let req = request.into_inner();
        let report_schema_version =
            grpc_requested_schema_version(req.report_schema_version, "profile test report")
                .map_err(Status::invalid_argument)?;

        let profile = hl7v2::load_profile_checked(&req.profile)
            .map_err(|_error| Status::invalid_argument(crate::PROFILE_LOAD_SAFE_MESSAGE))?;
        let report = grpc_profile_test_report_from_inline_fixtures(&req.fixtures, &profile)
            .map_err(Status::invalid_argument)?;
        let profile_test_report_v2 = (report_schema_version == 2).then(|| {
            proto_profile_test_report_v2_from_rust(
                &report.to_v2("hl7v2-server-grpc", env!("CARGO_PKG_VERSION")),
            )
        });

        Ok(Response::new(ProfileTestResponse {
            profile_test_report: Some(proto_profile_test_report_from_rust(&report)),
            profile_test_report_v2,
        }))
    }

    async fn validate_redacted(
        &self,
        request: Request<ValidateRedactedRequest>,
    ) -> Result<Response<ValidateRedactedResponse>, Status> {
        let req = request.into_inner();
        let report_schema_version =
            grpc_requested_schema_version(req.report_schema_version, "validation report")
                .map_err(Status::invalid_argument)?;
        let redaction_receipt_schema_version = grpc_requested_schema_version(
            req.redaction_receipt_schema_version,
            "redaction receipt",
        )
        .map_err(Status::invalid_argument)?;
        let quarantine_schema_version =
            grpc_requested_schema_version(req.quarantine_schema_version, "quarantine output")
                .map_err(Status::invalid_argument)?;

        let raw_input = req.message;
        let message_bytes = if req.mllp_framed {
            hl7v2::unwrap_mllp(&raw_input)
                .map_err(|e| Status::invalid_argument(format!("Failed to unwrap MLLP: {e}")))?
        } else {
            raw_input.as_slice()
        };

        let mut message = rust_parse(message_bytes)
            .map_err(|e| Status::invalid_argument(format!("Failed to parse HL7: {e}")))?;

        let receipt = redact_message(&mut message, &req.redaction_policy)
            .map_err(|e| Status::invalid_argument(format!("Failed to redact HL7: {e}")))?;
        let redacted_hl7 = String::from_utf8(hl7v2::write(&message)).map_err(|error| {
            Status::internal(format!(
                "redacted message could not be encoded as UTF-8: {error}"
            ))
        })?;

        let profile = hl7v2::load_profile_checked(&req.profile)
            .map_err(|_error| Status::invalid_argument(crate::PROFILE_LOAD_SAFE_MESSAGE))?;

        let issues = hl7v2::validate(&message, &profile);
        let report = hl7v2::ValidationReport::from_issues(
            &message,
            Some(profile.message_structure.clone()),
            issues,
        );

        let validation_report_v2 = (report_schema_version == 2).then(|| {
            let profile_identity = Some(hl7v2::ValidationReportProfileIdentity {
                label: profile.message_structure.clone(),
                message_structure: Some(profile.message_structure.clone()),
                version: Some(profile.version.clone()),
                sha256: Some(compute_sha256(&req.profile)),
            });
            proto_validation_report_v2_from_rust(&report.to_v2(
                "hl7v2-server-grpc",
                env!("CARGO_PKG_VERSION"),
                profile_identity,
            ))
        });

        let redaction_receipt_v2 = (redaction_receipt_schema_version == 2)
            .then(|| proto_redaction_receipt_v2_from_server(&receipt, "hl7v2-server-grpc"));
        let quarantine = maybe_write_grpc_redacted_quarantine(GrpcRedactedQuarantineContext {
            state: &self.state,
            raw_input: &raw_input,
            profile_yaml: &req.profile,
            policy_text: &req.redaction_policy,
            redacted_message: &message,
            redacted_hl7: &redacted_hl7,
            redaction_receipt: &receipt,
            validation_report: &report,
        })
        .map_err(grpc_quarantine_output_status)?;
        let quarantine_v2 = if quarantine_schema_version == 2 {
            quarantine.as_ref().map(|summary| {
                proto_quarantine_output_summary_v2_from_server(
                    &summary.to_v2("hl7v2-server-grpc", env!("CARGO_PKG_VERSION")),
                )
            })
        } else {
            None
        };

        Ok(Response::new(ValidateRedactedResponse {
            validation_report: Some(proto_validation_report_from_rust(&report)),
            validation_report_v2,
            redaction_receipt: Some(proto_redaction_receipt_from_server(&receipt)),
            redaction_receipt_v2,
            redacted_hl7: req.include_redacted_hl7.then(|| redacted_hl7.into_bytes()),
            quarantine: quarantine
                .as_ref()
                .map(proto_quarantine_output_summary_from_server),
            quarantine_v2,
        }))
    }

    async fn create_evidence_bundle(
        &self,
        request: Request<CreateEvidenceBundleRequest>,
    ) -> Result<Response<CreateEvidenceBundleResponse>, Status> {
        let req = request.into_inner();
        let artifact_schema_version: u8 =
            if grpc_requested_schema_version(req.bundle_artifact_schema_version, "bundle artifact")
                .map_err(Status::invalid_argument)?
                == 2
            {
                2
            } else {
                1
            };
        let bundle_output_root =
            self.state.bundle_output_root.as_deref().ok_or_else(|| {
                Status::failed_precondition("bundle output root is not configured")
            })?;

        let raw_input = req.message;
        let message_bytes = if req.mllp_framed {
            hl7v2::unwrap_mllp(&raw_input)
                .map_err(|e| Status::invalid_argument(format!("Failed to unwrap MLLP: {e}")))?
        } else {
            raw_input.as_slice()
        };
        let mut message = rust_parse(message_bytes)
            .map_err(|e| Status::invalid_argument(format!("Failed to parse HL7: {e}")))?;

        let receipt = redact_message(&mut message, &req.redaction_policy)
            .map_err(|e| Status::invalid_argument(format!("Failed to redact HL7: {e}")))?;
        let redacted_hl7 = String::from_utf8(hl7v2::write(&message)).map_err(|error| {
            Status::internal(format!(
                "redacted message could not be encoded as UTF-8: {error}"
            ))
        })?;

        let profile = hl7v2::load_profile_checked(&req.profile)
            .map_err(|_error| Status::invalid_argument(crate::PROFILE_LOAD_SAFE_MESSAGE))?;
        let issues = hl7v2::validate(&message, &profile);
        let validation_report = hl7v2::ValidationReport::from_issues(
            &message,
            Some("profile.yaml".to_string()),
            issues,
        );

        let summary =
            crate::evidence::write_evidence_bundle(crate::evidence::EvidenceBundleWriteRequest {
                root: bundle_output_root,
                bundle_id: &req.bundle_id,
                public_output_dir: Some(&crate::audit::hash_identifier(&req.bundle_id)),
                raw_input: &raw_input,
                profile_yaml: &req.profile,
                policy_text: &req.redaction_policy,
                redacted_message: &message,
                redacted_hl7: &redacted_hl7,
                redaction_receipt: &receipt,
                validation_report: &validation_report,
                artifact_schema_version,
            })
            .map_err(grpc_evidence_bundle_error)?;

        Ok(Response::new(CreateEvidenceBundleResponse {
            summary: Some(proto_evidence_bundle_summary_from_server(&summary)),
        }))
    }

    async fn replay_evidence_bundle(
        &self,
        request: Request<ReplayEvidenceBundleRequest>,
    ) -> Result<Response<ReplayEvidenceBundleResponse>, Status> {
        let req = request.into_inner();
        let report_schema_version =
            grpc_requested_schema_version(req.replay_report_schema_version, "replay report")
                .map_err(Status::invalid_argument)?;
        let bundle_output_root =
            self.state.bundle_output_root.as_deref().ok_or_else(|| {
                Status::failed_precondition("bundle output root is not configured")
            })?;
        let bundle_dir = crate::evidence::bundle_path_for_id(bundle_output_root, &req.bundle_id)
            .map_err(grpc_evidence_bundle_error)?;

        if !bundle_dir.is_dir() {
            crate::metrics::record_replay_result(false);
            return Err(Status::not_found("bundle id was not found"));
        }

        let report = hl7v2::evidence::replay_evidence_bundle(&bundle_dir, "hl7v2-server-grpc");
        crate::metrics::record_replay_result(report.reproduced);
        let replay_report_v2 = (report_schema_version == 2)
            .then(|| proto_evidence_replay_report_v2_from_rust(&report.to_v2()));

        Ok(Response::new(ReplayEvidenceBundleResponse {
            replay_report: Some(proto_evidence_replay_report_from_rust(&report)),
            replay_report_v2,
        }))
    }

    async fn corpus_summarize(
        &self,
        request: Request<CorpusSummarizeRequest>,
    ) -> Result<Response<CorpusSummarizeResponse>, Status> {
        let req = request.into_inner();
        let summary_schema_version =
            grpc_requested_schema_version(req.summary_schema_version, "corpus summary")
                .map_err(Status::invalid_argument)?;
        let ids =
            validated_grpc_corpus_message_ids(&req.messages).map_err(Status::invalid_argument)?;
        let messages = grpc_corpus_message_refs(&req.messages, &ids);
        let summary =
            hl7v2::synthetic::corpus::summarize_corpus_messages("<inline-corpus>", &messages);
        let summary_v2 = (summary_schema_version == 2).then(|| {
            proto_corpus_summary_v2_from_rust(
                &summary.to_v2("hl7v2-server-grpc", env!("CARGO_PKG_VERSION")),
            )
        });

        Ok(Response::new(CorpusSummarizeResponse {
            summary: Some(proto_corpus_summary_from_rust(&summary)),
            summary_v2,
        }))
    }

    async fn corpus_fingerprint(
        &self,
        request: Request<CorpusFingerprintRequest>,
    ) -> Result<Response<CorpusFingerprintResponse>, Status> {
        let req = request.into_inner();
        let fingerprint_schema_version =
            grpc_requested_schema_version(req.fingerprint_schema_version, "corpus fingerprint")
                .map_err(Status::invalid_argument)?;
        let ids =
            validated_grpc_corpus_message_ids(&req.messages).map_err(Status::invalid_argument)?;
        let messages = grpc_corpus_message_refs(&req.messages, &ids);
        let mut fingerprint =
            hl7v2::synthetic::corpus::fingerprint_corpus_messages("<inline-corpus>", &messages);

        if let Some(profile_yaml) = req.profile.as_deref() {
            attach_grpc_profile_to_fingerprint(&mut fingerprint, profile_yaml, &req.messages)
                .map_err(Status::invalid_argument)?;
        }

        let fingerprint_v2 = (fingerprint_schema_version == 2).then(|| {
            proto_corpus_fingerprint_v2_from_rust(&fingerprint.to_v2("hl7v2-server-grpc"))
        });

        Ok(Response::new(CorpusFingerprintResponse {
            fingerprint: Some(proto_corpus_fingerprint_from_rust(&fingerprint)),
            fingerprint_v2,
        }))
    }

    async fn corpus_diff(
        &self,
        request: Request<CorpusDiffRequest>,
    ) -> Result<Response<CorpusDiffResponse>, Status> {
        let req = request.into_inner();
        let diff_schema_version =
            grpc_requested_schema_version(req.diff_schema_version, "corpus diff")
                .map_err(Status::invalid_argument)?;
        let before_ids =
            validated_grpc_corpus_message_ids(&req.before).map_err(Status::invalid_argument)?;
        let after_ids =
            validated_grpc_corpus_message_ids(&req.after).map_err(Status::invalid_argument)?;
        let before_messages = grpc_corpus_message_refs(&req.before, &before_ids);
        let after_messages = grpc_corpus_message_refs(&req.after, &after_ids);
        let mut before_fingerprint = hl7v2::synthetic::corpus::fingerprint_corpus_messages(
            "<inline-before>",
            &before_messages,
        );
        let mut after_fingerprint = hl7v2::synthetic::corpus::fingerprint_corpus_messages(
            "<inline-after>",
            &after_messages,
        );

        if let Some(profile_yaml) = req.profile.as_deref() {
            let (profile, profile_metadata) =
                load_grpc_fingerprint_profile(profile_yaml).map_err(Status::invalid_argument)?;
            before_fingerprint.profile = Some(profile_metadata.clone());
            after_fingerprint.profile = Some(profile_metadata);
            before_fingerprint.validation_issue_code_counts =
                grpc_validation_issue_counts_for_loaded_profile(&req.before, &profile);
            after_fingerprint.validation_issue_code_counts =
                grpc_validation_issue_counts_for_loaded_profile(&req.after, &profile);
        }

        let diff = hl7v2::synthetic::corpus::diff_corpus_fingerprints(
            &before_fingerprint,
            &after_fingerprint,
        );
        let diff_v2 = (diff_schema_version == 2)
            .then(|| proto_corpus_diff_report_v2_from_rust(&diff.to_v2("hl7v2-server-grpc")));

        Ok(Response::new(CorpusDiffResponse {
            diff: Some(proto_corpus_diff_report_from_rust(&diff)),
            diff_v2,
        }))
    }

    async fn generate_ack(
        &self,
        request: Request<GenerateAckRequest>,
    ) -> Result<Response<GenerateAckResponse>, Status> {
        let req = request.into_inner();

        let message = rust_parse(&req.message)
            .map_err(|e| Status::invalid_argument(format!("Failed to parse HL7: {}", e)))?;

        let ack_code = match req.code() {
            generate_ack_request::AckCode::Aa => hl7v2::AckCode::AA,
            generate_ack_request::AckCode::Ae => hl7v2::AckCode::AE,
            generate_ack_request::AckCode::Ar => hl7v2::AckCode::AR,
            _ => hl7v2::AckCode::AA,
        };

        let ack_msg = hl7v2::ack(&message, ack_code)
            .map_err(|e| Status::internal(format!("Failed to generate ACK: {}", e)))?;
        let ack_bytes = hl7v2::write(&ack_msg);
        let proto_ack = proto::Message::from(ack_msg);

        Ok(Response::new(GenerateAckResponse {
            ack_message: ack_bytes,
            parsed_ack: Some(proto_ack),
        }))
    }

    async fn normalize(
        &self,
        request: Request<NormalizeRequest>,
    ) -> Result<Response<NormalizeResponse>, Status> {
        let req = request.into_inner();
        let canonical = req
            .options
            .as_ref()
            .map(|o| o.canonical_delimiters)
            .unwrap_or(true);

        let normalized_bytes = hl7v2::normalize(&req.message, canonical)
            .map_err(|e| Status::invalid_argument(format!("Failed to normalize HL7: {}", e)))?;

        let mut final_bytes = normalized_bytes;
        if let Some(options) = req.options
            && options.mllp_frame
        {
            final_bytes = hl7v2::wrap_mllp(&final_bytes);
        }

        Ok(Response::new(NormalizeResponse {
            normalized: final_bytes,
        }))
    }

    async fn health_check(
        &self,
        _request: Request<HealthCheckRequest>,
    ) -> Result<Response<HealthCheckResponse>, Status> {
        let response = HealthCheckResponse {
            status: health_check_response::ServingStatus::Serving as i32,
            version: env!("CARGO_PKG_VERSION").to_string(),
            uptime_seconds: 0,
        };
        Ok(Response::new(response))
    }
}

// ============================================================================
// Conversions
// ============================================================================

fn parse_grpc_message_response(message: &[u8], mllp_framed: bool) -> ParseResponse {
    let parse_result = if mllp_framed {
        match hl7v2::unwrap_mllp(message) {
            Ok(hl7) => rust_parse(hl7),
            Err(e) => {
                return ParseResponse {
                    success: false,
                    message: None,
                    errors: vec![Error {
                        code: "MLLP_ERROR".to_string(),
                        message: format!("Failed to unwrap MLLP: {}", e),
                        details: std::collections::HashMap::new(),
                        trace_id: String::new(),
                    }],
                    metadata: None,
                };
            }
        }
    } else {
        rust_parse(message)
    };

    match parse_result {
        Ok(msg) => {
            let metadata = extract_grpc_metadata(&msg);
            let proto_msg = proto::Message::from(msg);

            ParseResponse {
                success: true,
                message: Some(proto_msg),
                errors: Vec::new(),
                metadata: Some(metadata),
            }
        }
        Err(e) => ParseResponse {
            success: false,
            message: None,
            errors: vec![Error {
                code: "PARSE_ERROR".to_string(),
                message: format!("Failed to parse HL7: {}", e),
                details: std::collections::HashMap::new(),
                trace_id: String::new(),
            }],
            metadata: None,
        },
    }
}

fn parse_stream_response_from_parse_response(response: ParseResponse) -> ParseStreamResponse {
    ParseStreamResponse {
        success: response.success,
        message: response.message,
        errors: response.errors,
        metadata: response.metadata,
    }
}

fn grpc_requested_schema_version(version: i32, artifact: &str) -> Result<i32, String> {
    match version {
        0 | 1 => Ok(1),
        2 => Ok(2),
        other => Err(format!(
            "unsupported {artifact} schema version {other}; expected 1 or 2"
        )),
    }
}

fn compute_sha256(input: &str) -> String {
    let mut hasher = Sha256::new();
    hasher.update(input.as_bytes());
    format!("{:x}", hasher.finalize())
}

struct GrpcRedactedQuarantineContext<'a> {
    state: &'a AppState,
    raw_input: &'a [u8],
    profile_yaml: &'a str,
    policy_text: &'a str,
    redacted_message: &'a hl7v2::Message,
    redacted_hl7: &'a str,
    redaction_receipt: &'a ServerRedactionReceipt,
    validation_report: &'a hl7v2::ValidationReport,
}

enum GrpcQuarantineOutputError {
    MissingRoot,
    Write(crate::evidence::EvidenceBundleError),
}

fn maybe_write_grpc_redacted_quarantine(
    context: GrpcRedactedQuarantineContext<'_>,
) -> Result<Option<ServerQuarantineOutputSummary>, GrpcQuarantineOutputError> {
    let GrpcRedactedQuarantineContext {
        state,
        raw_input,
        profile_yaml,
        policy_text,
        redacted_message,
        redacted_hl7,
        redaction_receipt,
        validation_report,
    } = context;

    if validation_report.valid || !state.quarantine.enabled {
        return Ok(None);
    }

    let root = state
        .quarantine
        .path
        .as_deref()
        .ok_or(GrpcQuarantineOutputError::MissingRoot)?;
    let output_id = generated_quarantine_id();
    let summary =
        crate::evidence::write_quarantine_output(crate::evidence::QuarantineOutputWriteRequest {
            root,
            output_id: &output_id,
            config: &state.quarantine,
            raw_input,
            profile_yaml,
            policy_text,
            redacted_message,
            redacted_hl7,
            redaction_receipt,
            validation_report,
        })
        .map_err(GrpcQuarantineOutputError::Write)?;

    Ok(Some(summary))
}

fn generated_quarantine_id() -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |duration| duration.as_nanos());
    format!("quarantine-{}-{nanos}", std::process::id())
}

fn validated_grpc_corpus_message_ids(
    messages: &[CorpusMessageInput],
) -> Result<Vec<String>, &'static str> {
    if messages.is_empty() {
        return Err("messages must contain at least one message");
    }

    messages
        .iter()
        .enumerate()
        .map(|(index, message)| {
            let label = message
                .id
                .clone()
                .unwrap_or_else(|| format!("message-{}", index.saturating_add(1)));
            validate_grpc_corpus_message_id(&label)?;
            Ok(label)
        })
        .collect()
}

fn validate_grpc_corpus_message_id(label: &str) -> Result<(), &'static str> {
    if label.is_empty() || label == "." || label == ".." || label.len() > 128 {
        return Err("corpus message id must be 1-128 characters and cannot be '.' or '..'");
    }

    if !label
        .chars()
        .all(|character| character.is_ascii_alphanumeric() || matches!(character, '.' | '_' | '-'))
    {
        return Err("corpus message id must use only ASCII letters, numbers, '.', '_' or '-'");
    }

    Ok(())
}

fn grpc_corpus_message_refs<'a>(
    messages: &'a [CorpusMessageInput],
    ids: &'a [String],
) -> Vec<hl7v2::synthetic::corpus::CorpusMessageRef<'a>> {
    messages
        .iter()
        .zip(ids.iter())
        .map(|(message, id)| {
            hl7v2::synthetic::corpus::CorpusMessageRef::new(id.as_str(), message.message.as_slice())
        })
        .collect()
}

fn attach_grpc_profile_to_fingerprint(
    fingerprint: &mut hl7v2::synthetic::corpus::CorpusFingerprint,
    profile_yaml: &str,
    messages: &[CorpusMessageInput],
) -> Result<hl7v2::synthetic::corpus::CorpusFingerprintProfile, &'static str> {
    let (profile, metadata) = load_grpc_fingerprint_profile(profile_yaml)?;
    fingerprint.profile = Some(metadata.clone());
    fingerprint.validation_issue_code_counts =
        grpc_validation_issue_counts_for_loaded_profile(messages, &profile);
    Ok(metadata)
}

fn load_grpc_fingerprint_profile(
    profile_yaml: &str,
) -> Result<
    (
        hl7v2::Profile,
        hl7v2::synthetic::corpus::CorpusFingerprintProfile,
    ),
    &'static str,
> {
    let profile = hl7v2::load_profile_checked(profile_yaml)
        .map_err(|_error| crate::PROFILE_LOAD_SAFE_MESSAGE)?;
    let metadata = hl7v2::synthetic::corpus::CorpusFingerprintProfile {
        path: "<inline-profile>".to_string(),
        sha256: compute_sha256(profile_yaml),
        version: profile.version.clone(),
        message_structure: profile.message_structure.clone(),
    };
    Ok((profile, metadata))
}

fn grpc_validation_issue_counts_for_loaded_profile(
    messages: &[CorpusMessageInput],
    profile: &hl7v2::Profile,
) -> Vec<hl7v2::synthetic::corpus::CorpusCount> {
    let mut counts: BTreeMap<String, usize> = BTreeMap::new();

    for message in messages {
        let parsed = if hl7v2::is_mllp_framed(message.message.as_slice()) {
            hl7v2::parse_mllp(message.message.as_slice())
        } else {
            hl7v2::parse(message.message.as_slice())
        };
        let Ok(parsed) = parsed else {
            continue;
        };

        let issues = hl7v2::validate(&parsed, profile);
        let report = hl7v2::ValidationReport::from_issues(
            &parsed,
            Some(profile.message_structure.clone()),
            issues,
        );
        for issue in report.issues {
            let count = counts.entry(issue.code).or_insert(0);
            *count = count.saturating_add(1);
        }
    }

    counts
        .into_iter()
        .map(|(value, count)| hl7v2::synthetic::corpus::CorpusCount { value, count })
        .collect()
}

fn extract_grpc_metadata(msg: &RustMessage) -> MessageMetadata {
    MessageMetadata {
        message_type: joined_components(msg, "MSH.9").unwrap_or_else(|| "UNKNOWN".to_string()),
        version: hl7v2::get(msg, "MSH.12").unwrap_or("UNKNOWN").to_string(),
        control_id: hl7v2::get(msg, "MSH.10").unwrap_or("UNKNOWN").to_string(),
        sending_facility: hl7v2::get(msg, "MSH.4").unwrap_or("").to_string(),
        receiving_facility: hl7v2::get(msg, "MSH.6").unwrap_or("").to_string(),
    }
}

fn proto_corpus_summary_from_rust(
    summary: &hl7v2::synthetic::corpus::CorpusSummary,
) -> CorpusSummary {
    CorpusSummary {
        root: summary.root.clone(),
        file_count: usize_to_u32(summary.file_count),
        message_count: usize_to_u32(summary.message_count),
        parse_error_count: usize_to_u32(summary.parse_error_count),
        total_bytes: usize_to_u64(summary.total_bytes),
        message_types: summary
            .message_types
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        segments: summary
            .segments
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        field_presence: summary
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_from_rust)
            .collect(),
        parse_errors: summary
            .parse_errors
            .iter()
            .map(proto_corpus_parse_failure_from_rust)
            .collect(),
    }
}

fn proto_corpus_summary_v2_from_rust(
    summary: &hl7v2::synthetic::corpus::CorpusSummaryV2,
) -> CorpusSummaryV2 {
    CorpusSummaryV2 {
        schema_version: summary.schema_version.clone(),
        tool_name: summary.tool_name.clone(),
        tool_version: summary.tool_version.clone(),
        root: summary.summary.root.clone(),
        file_count: usize_to_u32(summary.summary.file_count),
        message_count: usize_to_u32(summary.summary.message_count),
        parse_error_count: usize_to_u32(summary.summary.parse_error_count),
        total_bytes: usize_to_u64(summary.summary.total_bytes),
        message_types: summary
            .summary
            .message_types
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        segments: summary
            .summary
            .segments
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        field_presence: summary
            .summary
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_from_rust)
            .collect(),
        parse_errors: summary
            .summary
            .parse_errors
            .iter()
            .map(proto_corpus_parse_failure_from_rust)
            .collect(),
    }
}

fn proto_corpus_count_from_rust(count: &hl7v2::synthetic::corpus::CorpusCount) -> CorpusCount {
    CorpusCount {
        value: count.value.clone(),
        count: usize_to_u32(count.count),
    }
}

fn proto_corpus_field_presence_from_rust(
    field: &hl7v2::synthetic::corpus::CorpusFieldPresence,
) -> CorpusFieldPresence {
    CorpusFieldPresence {
        path: field.path.clone(),
        message_count: usize_to_u32(field.message_count),
        occurrence_count: usize_to_u32(field.occurrence_count),
    }
}

fn proto_corpus_parse_failure_from_rust(
    failure: &hl7v2::synthetic::corpus::CorpusParseFailure,
) -> CorpusParseFailure {
    CorpusParseFailure {
        path: failure.path.clone(),
        error: failure.error.clone(),
    }
}

fn proto_corpus_fingerprint_from_rust(
    fingerprint: &hl7v2::synthetic::corpus::CorpusFingerprint,
) -> CorpusFingerprint {
    CorpusFingerprint {
        fingerprint_version: fingerprint.fingerprint_version.clone(),
        tool_version: fingerprint.tool_version.clone(),
        root: fingerprint.root.clone(),
        profile: fingerprint
            .profile
            .as_ref()
            .map(proto_corpus_fingerprint_profile_from_rust),
        file_count: usize_to_u32(fingerprint.file_count),
        message_count: usize_to_u32(fingerprint.message_count),
        parse_error_count: usize_to_u32(fingerprint.parse_error_count),
        message_type_counts: fingerprint
            .message_type_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        segment_counts: fingerprint
            .segment_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        field_presence: fingerprint
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_from_rust)
            .collect(),
        field_cardinality: fingerprint
            .field_cardinality
            .iter()
            .map(proto_corpus_field_cardinality_from_rust)
            .collect(),
        value_shape_stats: fingerprint
            .value_shape_stats
            .iter()
            .map(proto_corpus_value_shape_stats_from_rust)
            .collect(),
        validation_issue_code_counts: fingerprint
            .validation_issue_code_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
    }
}

fn proto_corpus_fingerprint_v2_from_rust(
    fingerprint: &hl7v2::synthetic::corpus::CorpusFingerprintV2,
) -> CorpusFingerprintV2 {
    let report = &fingerprint.fingerprint;
    CorpusFingerprintV2 {
        schema_version: fingerprint.schema_version.clone(),
        tool_name: fingerprint.tool_name.clone(),
        fingerprint_version: report.fingerprint_version.clone(),
        tool_version: report.tool_version.clone(),
        root: report.root.clone(),
        profile: report
            .profile
            .as_ref()
            .map(proto_corpus_fingerprint_profile_from_rust),
        file_count: usize_to_u32(report.file_count),
        message_count: usize_to_u32(report.message_count),
        parse_error_count: usize_to_u32(report.parse_error_count),
        message_type_counts: report
            .message_type_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        segment_counts: report
            .segment_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
        field_presence: report
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_from_rust)
            .collect(),
        field_cardinality: report
            .field_cardinality
            .iter()
            .map(proto_corpus_field_cardinality_from_rust)
            .collect(),
        value_shape_stats: report
            .value_shape_stats
            .iter()
            .map(proto_corpus_value_shape_stats_from_rust)
            .collect(),
        validation_issue_code_counts: report
            .validation_issue_code_counts
            .iter()
            .map(proto_corpus_count_from_rust)
            .collect(),
    }
}

fn proto_corpus_diff_report_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusDiffReport,
) -> CorpusDiffReport {
    CorpusDiffReport {
        diff_version: diff.diff_version.clone(),
        tool_version: diff.tool_version.clone(),
        before_root: diff.before_root.clone(),
        after_root: diff.after_root.clone(),
        profile: diff
            .profile
            .as_ref()
            .map(proto_corpus_fingerprint_profile_from_rust),
        file_count: Some(proto_corpus_total_diff_from_rust(&diff.file_count)),
        message_count: Some(proto_corpus_total_diff_from_rust(&diff.message_count)),
        parse_error_count: Some(proto_corpus_total_diff_from_rust(&diff.parse_error_count)),
        new_message_types: diff.new_message_types.clone(),
        removed_message_types: diff.removed_message_types.clone(),
        new_segments: diff.new_segments.clone(),
        removed_segments: diff.removed_segments.clone(),
        message_type_counts: diff
            .message_type_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
        segment_counts: diff
            .segment_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
        field_presence: diff
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_diff_from_rust)
            .collect(),
        field_cardinality: diff
            .field_cardinality
            .iter()
            .map(proto_corpus_field_cardinality_diff_from_rust)
            .collect(),
        value_shape_stats: diff
            .value_shape_stats
            .iter()
            .map(proto_corpus_value_shape_stats_diff_from_rust)
            .collect(),
        validation_issue_code_counts: diff
            .validation_issue_code_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
    }
}

fn proto_corpus_diff_report_v2_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusDiffReportV2,
) -> CorpusDiffReportV2 {
    let report = &diff.report;
    CorpusDiffReportV2 {
        schema_version: diff.schema_version.clone(),
        tool_name: diff.tool_name.clone(),
        diff_version: report.diff_version.clone(),
        tool_version: report.tool_version.clone(),
        before_root: report.before_root.clone(),
        after_root: report.after_root.clone(),
        profile: report
            .profile
            .as_ref()
            .map(proto_corpus_fingerprint_profile_from_rust),
        file_count: Some(proto_corpus_total_diff_from_rust(&report.file_count)),
        message_count: Some(proto_corpus_total_diff_from_rust(&report.message_count)),
        parse_error_count: Some(proto_corpus_total_diff_from_rust(&report.parse_error_count)),
        new_message_types: report.new_message_types.clone(),
        removed_message_types: report.removed_message_types.clone(),
        new_segments: report.new_segments.clone(),
        removed_segments: report.removed_segments.clone(),
        message_type_counts: report
            .message_type_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
        segment_counts: report
            .segment_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
        field_presence: report
            .field_presence
            .iter()
            .map(proto_corpus_field_presence_diff_from_rust)
            .collect(),
        field_cardinality: report
            .field_cardinality
            .iter()
            .map(proto_corpus_field_cardinality_diff_from_rust)
            .collect(),
        value_shape_stats: report
            .value_shape_stats
            .iter()
            .map(proto_corpus_value_shape_stats_diff_from_rust)
            .collect(),
        validation_issue_code_counts: report
            .validation_issue_code_counts
            .iter()
            .map(proto_corpus_count_diff_from_rust)
            .collect(),
    }
}

fn proto_corpus_fingerprint_profile_from_rust(
    profile: &hl7v2::synthetic::corpus::CorpusFingerprintProfile,
) -> CorpusFingerprintProfile {
    CorpusFingerprintProfile {
        path: profile.path.clone(),
        sha256: profile.sha256.clone(),
        version: profile.version.clone(),
        message_structure: profile.message_structure.clone(),
    }
}

fn proto_corpus_total_diff_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusTotalDiff,
) -> CorpusTotalDiff {
    CorpusTotalDiff {
        before: usize_to_u64(diff.before),
        after: usize_to_u64(diff.after),
        delta: i128_to_i64(diff.delta),
    }
}

fn proto_corpus_count_diff_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusCountDiff,
) -> CorpusCountDiff {
    CorpusCountDiff {
        value: diff.value.clone(),
        before: usize_to_u64(diff.before),
        after: usize_to_u64(diff.after),
        delta: i128_to_i64(diff.delta),
    }
}

fn proto_corpus_field_presence_diff_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusFieldPresenceDiff,
) -> CorpusFieldPresenceDiff {
    CorpusFieldPresenceDiff {
        path: diff.path.clone(),
        before_message_count: usize_to_u64(diff.before_message_count),
        after_message_count: usize_to_u64(diff.after_message_count),
        message_count_delta: i128_to_i64(diff.message_count_delta),
        before_occurrence_count: usize_to_u64(diff.before_occurrence_count),
        after_occurrence_count: usize_to_u64(diff.after_occurrence_count),
        occurrence_count_delta: i128_to_i64(diff.occurrence_count_delta),
    }
}

fn proto_corpus_field_cardinality_diff_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusFieldCardinalityDiff,
) -> CorpusFieldCardinalityDiff {
    CorpusFieldCardinalityDiff {
        path: diff.path.clone(),
        before_min_per_message: usize_to_u64(diff.before_min_per_message),
        after_min_per_message: usize_to_u64(diff.after_min_per_message),
        min_per_message_delta: i128_to_i64(diff.min_per_message_delta),
        before_max_per_message: usize_to_u64(diff.before_max_per_message),
        after_max_per_message: usize_to_u64(diff.after_max_per_message),
        max_per_message_delta: i128_to_i64(diff.max_per_message_delta),
        before_total_occurrences: usize_to_u64(diff.before_total_occurrences),
        after_total_occurrences: usize_to_u64(diff.after_total_occurrences),
        total_occurrences_delta: i128_to_i64(diff.total_occurrences_delta),
        before_message_count: usize_to_u64(diff.before_message_count),
        after_message_count: usize_to_u64(diff.after_message_count),
        message_count_delta: i128_to_i64(diff.message_count_delta),
    }
}

fn proto_corpus_value_shape_stats_diff_from_rust(
    diff: &hl7v2::synthetic::corpus::CorpusValueShapeStatsDiff,
) -> CorpusValueShapeStatsDiff {
    CorpusValueShapeStatsDiff {
        path: diff.path.clone(),
        coded_count: Some(proto_corpus_total_diff_from_rust(&diff.coded_count)),
        timestamp_count: Some(proto_corpus_total_diff_from_rust(&diff.timestamp_count)),
        numeric_count: Some(proto_corpus_total_diff_from_rust(&diff.numeric_count)),
        null_count: Some(proto_corpus_total_diff_from_rust(&diff.null_count)),
        text_count: Some(proto_corpus_total_diff_from_rust(&diff.text_count)),
    }
}

fn proto_corpus_field_cardinality_from_rust(
    field: &hl7v2::synthetic::corpus::CorpusFieldCardinality,
) -> CorpusFieldCardinality {
    CorpusFieldCardinality {
        path: field.path.clone(),
        min_per_message: usize_to_u32(field.min_per_message),
        max_per_message: usize_to_u32(field.max_per_message),
        total_occurrences: usize_to_u32(field.total_occurrences),
        message_count: usize_to_u32(field.message_count),
    }
}

fn proto_corpus_value_shape_stats_from_rust(
    stats: &hl7v2::synthetic::corpus::CorpusValueShapeStats,
) -> CorpusValueShapeStats {
    CorpusValueShapeStats {
        path: stats.path.clone(),
        coded_count: usize_to_u32(stats.coded_count),
        timestamp_count: usize_to_u32(stats.timestamp_count),
        numeric_count: usize_to_u32(stats.numeric_count),
        null_count: usize_to_u32(stats.null_count),
        text_count: usize_to_u32(stats.text_count),
    }
}

fn usize_to_u32(value: usize) -> u32 {
    u32::try_from(value).unwrap_or(u32::MAX)
}

fn usize_to_u64(value: usize) -> u64 {
    u64::try_from(value).unwrap_or(u64::MAX)
}

fn i128_to_i64(value: i128) -> i64 {
    i64::try_from(value).unwrap_or(if value.is_negative() {
        i64::MIN
    } else {
        i64::MAX
    })
}

fn proto_validation_report_from_rust(report: &hl7v2::ValidationReport) -> ValidationReport {
    ValidationReport {
        valid: report.valid,
        message_type: report.message_type.clone(),
        profile: report.profile.clone(),
        segment_count: report.segment_count as i32,
        issue_count: report.issue_count as i32,
        issues: report
            .issues
            .iter()
            .map(proto_validation_report_issue_from_rust)
            .collect(),
    }
}

fn proto_validation_report_v2_from_rust(report: &hl7v2::ValidationReportV2) -> ValidationReportV2 {
    ValidationReportV2 {
        schema_version: report.schema_version.clone(),
        tool_name: report.tool_name.clone(),
        tool_version: report.tool_version.clone(),
        valid: report.valid,
        message_type: report.message_type.clone(),
        profile: report.profile.clone(),
        profile_identity: report
            .profile_identity
            .as_ref()
            .map(proto_validation_report_profile_identity_from_rust),
        segment_count: report.segment_count as i32,
        issue_count: report.issue_count as i32,
        issues: report
            .issues
            .iter()
            .map(proto_validation_report_issue_from_rust)
            .collect(),
    }
}

fn proto_validation_report_profile_identity_from_rust(
    profile: &hl7v2::ValidationReportProfileIdentity,
) -> ValidationReportProfileIdentity {
    ValidationReportProfileIdentity {
        label: profile.label.clone(),
        message_structure: profile.message_structure.clone(),
        version: profile.version.clone(),
        sha256: profile.sha256.clone(),
    }
}

fn proto_validation_report_issue_from_rust(
    issue: &hl7v2::ValidationReportIssue,
) -> ValidationReportIssue {
    ValidationReportIssue {
        code: issue.code.clone(),
        severity: issue.severity.as_str().to_string(),
        path: issue.path.clone(),
        rule_id: issue.rule_id.clone(),
        message: issue.message.clone(),
        segment_index: issue.segment_index.map(|value| value as u32),
        field_index: issue.field_index.map(|value| value as u32),
    }
}

fn proto_profile_lint_report_from_rust(report: &hl7v2::ProfileLintReport) -> ProfileLintReport {
    ProfileLintReport {
        valid: report.valid,
        error_count: usize_to_u32(report.error_count),
        warning_count: usize_to_u32(report.warning_count),
        issue_count: usize_to_u32(report.issue_count),
        issues: report
            .issues
            .iter()
            .map(proto_profile_lint_issue_from_rust)
            .collect(),
    }
}

fn proto_profile_lint_report_v2_from_rust(
    report: &hl7v2::ProfileLintReportV2,
) -> ProfileLintReportV2 {
    ProfileLintReportV2 {
        schema_version: report.schema_version.clone(),
        tool_name: report.tool_name.clone(),
        tool_version: report.tool_version.clone(),
        valid: report.report.valid,
        error_count: usize_to_u32(report.report.error_count),
        warning_count: usize_to_u32(report.report.warning_count),
        issue_count: usize_to_u32(report.report.issue_count),
        issues: report
            .report
            .issues
            .iter()
            .map(proto_profile_lint_issue_from_rust)
            .collect(),
    }
}

fn proto_profile_lint_issue_from_rust(issue: &hl7v2::ProfileLintIssue) -> ProfileLintIssue {
    ProfileLintIssue {
        code: issue.code.clone(),
        severity: issue.severity.as_str().to_string(),
        path: issue.path.clone(),
        message: issue.message.clone(),
    }
}

fn proto_profile_explain_report_from_rust(
    report: &hl7v2::ProfileExplainReport,
) -> ProfileExplainReport {
    ProfileExplainReport {
        profile: report.profile.clone(),
        profile_sha256: report.profile_sha256.clone(),
        message_structure: report.message_structure.clone(),
        version: report.version.clone(),
        message_type: report.message_type.clone(),
        parent: report.parent.clone(),
        summary: Some(proto_profile_explain_summary_from_rust(&report.summary)),
        segments: report
            .segments
            .iter()
            .map(proto_profile_explain_segment_from_rust)
            .collect(),
        required_fields: report
            .required_fields
            .iter()
            .map(proto_profile_explain_required_field_from_rust)
            .collect(),
        field_constraints: report
            .field_constraints
            .iter()
            .map(proto_profile_explain_constraint_from_rust)
            .collect(),
        length_rules: report
            .length_rules
            .iter()
            .map(proto_profile_explain_length_rule_from_rust)
            .collect(),
        datatype_rules: report
            .datatype_rules
            .iter()
            .map(proto_profile_explain_datatype_rule_from_rust)
            .collect(),
        value_sets: report
            .value_sets
            .iter()
            .map(proto_profile_explain_value_set_from_rust)
            .collect(),
        rules: Some(proto_profile_explain_rules_from_rust(&report.rules)),
        hl7_tables: report
            .hl7_tables
            .iter()
            .map(proto_profile_explain_table_from_rust)
            .collect(),
        table_precedence: report.table_precedence.clone(),
        expression_guardrails: Some(proto_profile_explain_guardrails_from_rust(
            &report.expression_guardrails,
        )),
        lint: Some(proto_profile_explain_lint_summary_from_rust(&report.lint)),
    }
}

fn proto_profile_explain_report_v2_from_rust(
    report: &hl7v2::ProfileExplainReportV2,
) -> ProfileExplainReportV2 {
    ProfileExplainReportV2 {
        schema_version: report.schema_version.clone(),
        tool_name: report.tool_name.clone(),
        tool_version: report.tool_version.clone(),
        report: Some(proto_profile_explain_report_from_rust(&report.report)),
    }
}

fn proto_profile_explain_summary_from_rust(
    summary: &hl7v2::ProfileExplainSummary,
) -> ProfileExplainSummary {
    ProfileExplainSummary {
        segment_count: usize_to_u32(summary.segment_count),
        required_field_count: usize_to_u32(summary.required_field_count),
        field_constraint_count: usize_to_u32(summary.field_constraint_count),
        length_rule_count: usize_to_u32(summary.length_rule_count),
        datatype_rule_count: usize_to_u32(summary.datatype_rule_count),
        advanced_datatype_rule_count: usize_to_u32(summary.advanced_datatype_rule_count),
        value_set_count: usize_to_u32(summary.value_set_count),
        cross_field_rule_count: usize_to_u32(summary.cross_field_rule_count),
        temporal_rule_count: usize_to_u32(summary.temporal_rule_count),
        contextual_rule_count: usize_to_u32(summary.contextual_rule_count),
        custom_rule_count: usize_to_u32(summary.custom_rule_count),
        hl7_table_count: usize_to_u32(summary.hl7_table_count),
    }
}

fn proto_profile_explain_segment_from_rust(
    segment: &hl7v2::ProfileExplainSegment,
) -> ProfileExplainSegment {
    ProfileExplainSegment {
        id: segment.id.clone(),
    }
}

fn proto_profile_explain_required_field_from_rust(
    field: &hl7v2::ProfileExplainRequiredField,
) -> ProfileExplainRequiredField {
    ProfileExplainRequiredField {
        path: field.path.clone(),
        conditional: field.conditional,
    }
}

fn proto_profile_explain_constraint_from_rust(
    constraint: &hl7v2::ProfileExplainConstraint,
) -> ProfileExplainConstraint {
    ProfileExplainConstraint {
        path: constraint.path.clone(),
        required: constraint.required,
        conditional: constraint.conditional,
        component_min: constraint.component_min.map(usize_to_u32),
        component_max: constraint.component_max.map(usize_to_u32),
        allowed_value_count: usize_to_u32(constraint.allowed_value_count),
        allowed_values: constraint.allowed_values.clone(),
        pattern: constraint.pattern.clone(),
    }
}

fn proto_profile_explain_length_rule_from_rust(
    rule: &hl7v2::ProfileExplainLengthRule,
) -> ProfileExplainLengthRule {
    ProfileExplainLengthRule {
        path: rule.path.clone(),
        max: rule.max.map(usize_to_u32),
        policy: rule.policy.clone(),
    }
}

fn proto_profile_explain_datatype_rule_from_rust(
    rule: &hl7v2::ProfileExplainDatatypeRule,
) -> ProfileExplainDatatypeRule {
    ProfileExplainDatatypeRule {
        path: rule.path.clone(),
        datatype: rule.datatype.clone(),
        kind: rule.kind.clone(),
        pattern: rule.pattern.clone(),
        min_length: rule.min_length.map(usize_to_u32),
        max_length: rule.max_length.map(usize_to_u32),
        format: rule.format.clone(),
        checksum: rule.checksum.clone(),
    }
}

fn proto_profile_explain_value_set_from_rust(
    value_set: &hl7v2::ProfileExplainValueSet,
) -> ProfileExplainValueSet {
    ProfileExplainValueSet {
        name: value_set.name.clone(),
        path: value_set.path.clone(),
        source: value_set.source.clone(),
        inline_code_count: usize_to_u32(value_set.inline_code_count),
        table_code_count: usize_to_u32(value_set.table_code_count),
    }
}

fn proto_profile_explain_rules_from_rust(
    rules: &hl7v2::ProfileExplainRules,
) -> ProfileExplainRules {
    ProfileExplainRules {
        cross_field: rules
            .cross_field
            .iter()
            .map(proto_profile_explain_rule_from_rust)
            .collect(),
        temporal: rules
            .temporal
            .iter()
            .map(proto_profile_explain_rule_from_rust)
            .collect(),
        contextual: rules
            .contextual
            .iter()
            .map(proto_profile_explain_rule_from_rust)
            .collect(),
        custom: rules
            .custom
            .iter()
            .map(proto_profile_explain_rule_from_rust)
            .collect(),
    }
}

fn proto_profile_explain_rule_from_rust(rule: &hl7v2::ProfileExplainRule) -> ProfileExplainRule {
    ProfileExplainRule {
        id: rule.id.clone(),
        description: rule.description.clone(),
    }
}

fn proto_profile_explain_table_from_rust(
    table: &hl7v2::ProfileExplainTable,
) -> ProfileExplainTable {
    ProfileExplainTable {
        id: table.id.clone(),
        name: table.name.clone(),
        version: table.version.clone(),
        code_count: usize_to_u32(table.code_count),
    }
}

fn proto_profile_explain_guardrails_from_rust(
    guardrails: &hl7v2::ProfileExplainExpressionGuardrails,
) -> ProfileExplainExpressionGuardrails {
    ProfileExplainExpressionGuardrails {
        max_depth: guardrails.max_depth.map(usize_to_u32),
        max_length: guardrails.max_length.map(usize_to_u32),
        allow_custom_scripts: guardrails.allow_custom_scripts,
    }
}

fn proto_profile_explain_lint_summary_from_rust(
    lint: &hl7v2::ProfileExplainLintSummary,
) -> ProfileExplainLintSummary {
    ProfileExplainLintSummary {
        valid: lint.valid,
        error_count: usize_to_u32(lint.error_count),
        warning_count: usize_to_u32(lint.warning_count),
        issue_count: usize_to_u32(lint.issue_count),
        ignored_or_unsupported: lint
            .ignored_or_unsupported
            .iter()
            .map(proto_profile_lint_issue_from_rust)
            .collect(),
    }
}

fn grpc_profile_test_report_from_inline_fixtures(
    fixtures: &[ProfileTestFixture],
    profile: &hl7v2::Profile,
) -> Result<hl7v2::ProfileTestReport, String> {
    if fixtures.is_empty() {
        return Err("fixtures must contain at least one fixture".to_string());
    }

    let cases = fixtures
        .iter()
        .enumerate()
        .map(|(index, fixture)| grpc_profile_test_case_from_inline_fixture(index, fixture, profile))
        .collect::<Result<Vec<_>, _>>()?;
    let passed_count = cases.iter().filter(|case| case.passed).count();
    let case_count = cases.len();
    let failed_count = case_count.saturating_sub(passed_count);

    Ok(hl7v2::ProfileTestReport {
        profile: "<inline-profile>".to_string(),
        fixtures: "<inline-fixtures>".to_string(),
        valid: failed_count == 0,
        case_count,
        passed_count,
        failed_count,
        cases,
    })
}

fn grpc_profile_test_case_from_inline_fixture(
    index: usize,
    fixture: &ProfileTestFixture,
    profile: &hl7v2::Profile,
) -> Result<hl7v2::ProfileTestCaseReport, String> {
    let name = grpc_profile_test_fixture_label(index, fixture);
    let expectation = grpc_profile_test_fixture_expectation(fixture.expectation)?;
    let message_bytes = if fixture.mllp_framed {
        match hl7v2::unwrap_mllp(&fixture.message) {
            Ok(message) => message,
            Err(err) => {
                return Ok(hl7v2::ProfileTestCaseReport {
                    name: name.clone(),
                    path: name,
                    expectation,
                    passed: false,
                    message: format!("fixture did not unwrap as MLLP: {err}"),
                    validation_report: None,
                    expected_report: None,
                });
            }
        }
    } else {
        fixture.message.as_slice()
    };

    let message = match rust_parse(message_bytes) {
        Ok(message) => message,
        Err(err) => {
            return Ok(hl7v2::ProfileTestCaseReport {
                name: name.clone(),
                path: name,
                expectation,
                passed: false,
                message: format!("fixture did not parse as HL7: {err}"),
                validation_report: None,
                expected_report: None,
            });
        }
    };

    let issues = hl7v2::validate(&message, profile);
    let validation_report = hl7v2::ValidationReport::from_issues(
        &message,
        Some("<inline-profile>".to_string()),
        issues,
    );
    let expected_valid = expectation == hl7v2::ProfileFixtureExpectation::Valid;
    let mut passed = validation_report.valid == expected_valid;
    let mut case_message = if passed {
        format!(
            "expected {} and report was {}",
            expectation.as_str(),
            if validation_report.valid {
                "valid"
            } else {
                "invalid"
            }
        )
    } else {
        format!(
            "expected {} but report was {}",
            expectation.as_str(),
            if validation_report.valid {
                "valid"
            } else {
                "invalid"
            }
        )
    };

    let expected_report = fixture.expected_report_json.as_deref().map(|expected| {
        grpc_compare_inline_expected_report_json(&name, expected, &validation_report)
    });
    if let Some(comparison) = &expected_report {
        if comparison.matched {
            case_message.push_str("; expected report matched");
        } else {
            passed = false;
            let detail = comparison
                .message
                .as_deref()
                .unwrap_or("expected report did not match");
            case_message.push_str(&format!("; {detail}"));
        }
    }

    Ok(hl7v2::ProfileTestCaseReport {
        name: name.clone(),
        path: name,
        expectation,
        passed,
        message: case_message,
        validation_report: Some(validation_report),
        expected_report,
    })
}

fn grpc_profile_test_fixture_label(index: usize, fixture: &ProfileTestFixture) -> String {
    let trimmed = fixture.name.trim();
    if trimmed.is_empty() {
        format!("fixture-{}", index.saturating_add(1))
    } else {
        trimmed.to_string()
    }
}

fn grpc_profile_test_fixture_expectation(
    value: i32,
) -> Result<hl7v2::ProfileFixtureExpectation, String> {
    match profile_test_fixture::Expectation::try_from(value)
        .unwrap_or(profile_test_fixture::Expectation::Unspecified)
    {
        profile_test_fixture::Expectation::Valid => Ok(hl7v2::ProfileFixtureExpectation::Valid),
        profile_test_fixture::Expectation::Invalid => Ok(hl7v2::ProfileFixtureExpectation::Invalid),
        profile_test_fixture::Expectation::Unspecified => {
            Err("fixture expectation must be valid or invalid".to_string())
        }
    }
}

fn grpc_compare_inline_expected_report_json(
    fixture_name: &str,
    expected_json: &str,
    actual_report: &hl7v2::ValidationReport,
) -> hl7v2::ExpectedReportComparison {
    let path = format!("{fixture_name}.expected-report.json");
    let expected = match serde_json::from_str::<serde_json::Value>(expected_json) {
        Ok(expected) => expected,
        Err(err) => {
            return hl7v2::ExpectedReportComparison {
                path,
                matched: false,
                message: Some(format!("expected report is not valid JSON: {err}")),
            };
        }
    };
    let actual = match serde_json::to_value(actual_report) {
        Ok(actual) => actual,
        Err(err) => {
            return hl7v2::ExpectedReportComparison {
                path,
                matched: false,
                message: Some(format!("actual report could not be serialized: {err}")),
            };
        }
    };

    match grpc_json_subset_matches(&expected, &actual, "$") {
        Ok(()) => hl7v2::ExpectedReportComparison {
            path,
            matched: true,
            message: None,
        },
        Err(message) => hl7v2::ExpectedReportComparison {
            path,
            matched: false,
            message: Some(message),
        },
    }
}

fn grpc_json_subset_matches(
    expected: &serde_json::Value,
    actual: &serde_json::Value,
    path: &str,
) -> Result<(), String> {
    match (expected, actual) {
        (serde_json::Value::Object(expected), serde_json::Value::Object(actual)) => {
            for (key, expected_value) in expected {
                let actual_value = actual
                    .get(key)
                    .ok_or_else(|| format!("{path}.{key} was missing from actual report"))?;
                grpc_json_subset_matches(expected_value, actual_value, &format!("{path}.{key}"))?;
            }
            Ok(())
        }
        (serde_json::Value::Array(expected), serde_json::Value::Array(actual)) => {
            for (index, expected_value) in expected.iter().enumerate() {
                let matched = actual.iter().any(|actual_value| {
                    grpc_json_subset_matches(
                        expected_value,
                        actual_value,
                        &format!("{path}[{index}]"),
                    )
                    .is_ok()
                });
                if !matched {
                    return Err(format!(
                        "{path}[{index}] did not match any actual report item"
                    ));
                }
            }
            Ok(())
        }
        _ if expected == actual => Ok(()),
        _ => Err(format!(
            "{path} expected {expected} but actual report had {actual}"
        )),
    }
}

fn proto_profile_test_report_from_rust(report: &hl7v2::ProfileTestReport) -> ProfileTestReport {
    ProfileTestReport {
        profile: report.profile.clone(),
        fixtures: report.fixtures.clone(),
        valid: report.valid,
        case_count: usize_to_u32(report.case_count),
        passed_count: usize_to_u32(report.passed_count),
        failed_count: usize_to_u32(report.failed_count),
        cases: report
            .cases
            .iter()
            .map(proto_profile_test_case_report_from_rust)
            .collect(),
    }
}

fn proto_profile_test_report_v2_from_rust(
    report: &hl7v2::ProfileTestReportV2,
) -> ProfileTestReportV2 {
    ProfileTestReportV2 {
        schema_version: report.schema_version.clone(),
        tool_name: report.tool_name.clone(),
        tool_version: report.tool_version.clone(),
        report: Some(proto_profile_test_report_from_rust(&report.report)),
    }
}

fn proto_profile_test_case_report_from_rust(
    case: &hl7v2::ProfileTestCaseReport,
) -> ProfileTestCaseReport {
    ProfileTestCaseReport {
        name: case.name.clone(),
        path: case.path.clone(),
        expectation: case.expectation.as_str().to_string(),
        passed: case.passed,
        message: case.message.clone(),
        validation_report: case
            .validation_report
            .as_ref()
            .map(proto_validation_report_from_rust),
        expected_report: case
            .expected_report
            .as_ref()
            .map(proto_expected_report_comparison_from_rust),
    }
}

fn proto_expected_report_comparison_from_rust(
    comparison: &hl7v2::ExpectedReportComparison,
) -> ExpectedReportComparison {
    ExpectedReportComparison {
        path: comparison.path.clone(),
        matched: comparison.matched,
        message: comparison.message.clone(),
    }
}

fn proto_redaction_receipt_from_server(receipt: &ServerRedactionReceipt) -> RedactionReceipt {
    RedactionReceipt {
        phi_removed: receipt.phi_removed,
        hash_algorithm: receipt.hash_algorithm.clone(),
        actions: receipt
            .actions
            .iter()
            .map(proto_redaction_action_receipt_from_server)
            .collect(),
    }
}

fn proto_redaction_receipt_v2_from_server(
    receipt: &ServerRedactionReceipt,
    tool_name: &str,
) -> RedactionReceiptV2 {
    RedactionReceiptV2 {
        schema_version: "2".to_string(),
        tool_name: tool_name.to_string(),
        tool_version: env!("CARGO_PKG_VERSION").to_string(),
        phi_removed: receipt.phi_removed,
        hash_algorithm: receipt.hash_algorithm.clone(),
        actions: receipt
            .actions
            .iter()
            .map(proto_redaction_action_receipt_from_server)
            .collect(),
    }
}

fn proto_quarantine_output_summary_from_server(
    summary: &ServerQuarantineOutputSummary,
) -> QuarantineOutputSummary {
    QuarantineOutputSummary {
        quarantine_version: summary.quarantine_version.clone(),
        output_dir: summary.output_dir.clone(),
        reason: server_quarantine_reason_name(summary.reason).to_string(),
        validation_issue_count: usize_to_u32(summary.validation_issue_count),
        artifacts: summary.artifacts.clone(),
    }
}

fn proto_quarantine_output_summary_v2_from_server(
    summary: &crate::models::QuarantineOutputSummaryV2,
) -> QuarantineOutputSummaryV2 {
    QuarantineOutputSummaryV2 {
        schema_version: summary.schema_version.clone(),
        tool_name: summary.tool_name.clone(),
        tool_version: summary.tool_version.clone(),
        quarantine_version: summary.summary.quarantine_version.clone(),
        output_dir: summary.summary.output_dir.clone(),
        reason: server_quarantine_reason_name(summary.summary.reason).to_string(),
        validation_issue_count: usize_to_u32(summary.summary.validation_issue_count),
        artifacts: summary.summary.artifacts.clone(),
    }
}

fn proto_evidence_bundle_summary_from_server(
    summary: &ServerEvidenceBundleSummary,
) -> EvidenceBundleSummary {
    EvidenceBundleSummary {
        bundle_version: summary.bundle_version.clone(),
        output_dir: summary.output_dir.clone(),
        message_type: summary.message_type.clone(),
        validation_valid: summary.validation_valid,
        validation_issue_count: usize_to_u32(summary.validation_issue_count),
        redaction_phi_removed: summary.redaction_phi_removed,
        artifacts: summary.artifacts.clone(),
    }
}

fn proto_evidence_replay_report_from_rust(
    report: &hl7v2::evidence::EvidenceReplayReport,
) -> EvidenceReplayReport {
    EvidenceReplayReport {
        replay_version: report.replay_version.clone(),
        bundle_version: report.bundle_version.clone(),
        tool_name: report.tool_name.clone(),
        tool_version: report.tool_version.clone(),
        message_type: report.message_type.clone(),
        reproduced: report.reproduced,
        validation_valid: report.validation_valid,
        validation_issue_count: report.validation_issue_count.map(usize_to_u32),
        checks: report
            .checks
            .iter()
            .map(proto_evidence_replay_check_from_rust)
            .collect(),
        validation_report: report
            .validation_report
            .as_ref()
            .map(proto_validation_report_from_rust),
    }
}

fn proto_evidence_replay_report_v2_from_rust(
    report: &hl7v2::evidence::EvidenceReplayReportV2,
) -> EvidenceReplayReportV2 {
    EvidenceReplayReportV2 {
        schema_version: report.schema_version.clone(),
        report: Some(proto_evidence_replay_report_from_rust(&report.report)),
    }
}

fn proto_evidence_replay_check_from_rust(
    check: &hl7v2::evidence::EvidenceReplayCheck,
) -> EvidenceReplayCheck {
    EvidenceReplayCheck {
        name: check.name.clone(),
        status: match check.status {
            hl7v2::evidence::EvidenceReplayCheckStatus::Pass => {
                evidence_replay_check::Status::Pass as i32
            }
            hl7v2::evidence::EvidenceReplayCheckStatus::Fail => {
                evidence_replay_check::Status::Fail as i32
            }
        },
        message: check.message.clone(),
    }
}

fn grpc_evidence_bundle_error(error: crate::evidence::EvidenceBundleError) -> Status {
    match error {
        crate::evidence::EvidenceBundleError::InvalidRequest(message) => {
            Status::invalid_argument(message)
        }
        crate::evidence::EvidenceBundleError::Conflict(message) => Status::already_exists(message),
        crate::evidence::EvidenceBundleError::Io(message) => Status::unavailable(message),
    }
}

fn grpc_quarantine_output_status(error: GrpcQuarantineOutputError) -> Status {
    match error {
        GrpcQuarantineOutputError::MissingRoot => Status::failed_precondition(
            "server quarantine output is enabled but no path is configured",
        ),
        GrpcQuarantineOutputError::Write(error) => grpc_quarantine_evidence_error(error),
    }
}

fn grpc_quarantine_evidence_error(error: crate::evidence::EvidenceBundleError) -> Status {
    match error {
        crate::evidence::EvidenceBundleError::InvalidRequest(message) => {
            Status::invalid_argument(message)
        }
        crate::evidence::EvidenceBundleError::Conflict(message) => Status::already_exists(message),
        crate::evidence::EvidenceBundleError::Io(message) => Status::unavailable(message),
    }
}

fn proto_redaction_action_receipt_from_server(
    action: &ServerRedactionActionReceipt,
) -> RedactionActionReceipt {
    RedactionActionReceipt {
        path: action.path.clone(),
        action: server_redaction_action_name(action.action).to_string(),
        reason: action.reason.clone(),
        matched_count: u32::try_from(action.matched_count).unwrap_or(u32::MAX),
        optional: action.optional,
        status: server_redaction_action_status_name(action.status).to_string(),
    }
}

fn server_redaction_action_name(action: ServerRedactionAction) -> &'static str {
    match action {
        ServerRedactionAction::Hash => "hash",
        ServerRedactionAction::Drop => "drop",
        ServerRedactionAction::Retain => "retain",
    }
}

fn server_redaction_action_status_name(status: ServerRedactionActionStatus) -> &'static str {
    match status {
        ServerRedactionActionStatus::Applied => "applied",
        ServerRedactionActionStatus::Retained => "retained",
        ServerRedactionActionStatus::NotFound => "not_found",
    }
}

fn server_quarantine_reason_name(reason: ServerQuarantineReason) -> &'static str {
    match reason {
        ServerQuarantineReason::ValidationError => "validation_error",
    }
}

fn joined_components(msg: &RustMessage, field_path: &str) -> Option<String> {
    let mut components = Vec::new();

    for component in 1.. {
        let path = format!("{field_path}.{component}");
        match hl7v2::get(msg, &path) {
            Some(value) if !value.is_empty() => components.push(value.to_string()),
            _ if component == 1 => return None,
            _ => break,
        }
    }

    Some(components.join("^"))
}

impl From<RustMessage> for proto::Message {
    fn from(msg: RustMessage) -> Self {
        proto::Message {
            delimiters: Some(proto::Delimiters {
                field: msg.delims.field.to_string(),
                component: msg.delims.comp.to_string(),
                repetition: msg.delims.rep.to_string(),
                escape: msg.delims.esc.to_string(),
                subcomponent: msg.delims.sub.to_string(),
            }),
            segments: msg.segments.into_iter().map(Into::into).collect(),
        }
    }
}

impl From<RustSegment> for proto::Segment {
    fn from(seg: RustSegment) -> Self {
        proto::Segment {
            id: String::from_utf8_lossy(&seg.id).to_string(),
            fields: seg.fields.into_iter().map(Into::into).collect(),
            sequence: 0,
        }
    }
}

impl From<RustField> for proto::Field {
    fn from(f: RustField) -> Self {
        let presence = if f.reps.is_empty() {
            proto::field::Presence::Missing as i32
        } else {
            proto::field::Presence::Value as i32
        };

        proto::Field {
            presence,
            value: f.first_text().map(String::from),
            repetitions: f.reps.into_iter().map(Into::into).collect(),
        }
    }
}

impl From<RustRep> for proto::Repetition {
    fn from(r: RustRep) -> Self {
        proto::Repetition {
            value: r.first_text().map(String::from),
            components: r.comps.into_iter().map(Into::into).collect(),
        }
    }
}

impl From<RustComp> for proto::Component {
    fn from(c: RustComp) -> Self {
        proto::Component {
            value: c.first_text().map(String::from),
            subcomponents: c
                .subs
                .into_iter()
                .filter_map(|a| match a {
                    hl7v2::Atom::Text(t) => Some(t),
                    _ => None,
                })
                .collect(),
        }
    }
}