ahri_tre_app/

projections.rs

mod semantic;
pub use semantic::*;
mod lifecycle;
pub use lifecycle::*;
mod governance;
use ahri_tre_protocol::{
    datastore::{
        ComponentStatus, DatastoreMigrationAction, DatastoreMigrationApplyResult,
        DatastoreMigrationPlanResult, DatastoreMigrationScope, DatastoreMigrationStepResult,
        DatastorePingResult, DatastoreSchemaCompatibilityStatus, DatastoreSchemaStatusResult,
        DatastoreSummary, DuckLakeCatalogMigrationOwnership,
    },
    public_error::{ProtocolError, ProtocolErrorCode},
    refs::{
        DatastoreRef, DomainRef, StudyRef, TagRef, TagScope, TagSummary, encode_scoped_integer_ref,
    },
    study::{StudyRegistrationPayload, StudySummary},
    warning::{ProtocolWarning, ProtocolWarningCode},
};
pub use governance::*;
use serde_json::json;

use crate::{AppError, RegistrationResource};

/// Maps a configured Session-open failure to its stable, redacted protocol
/// contract.
pub fn configured_datastore_open_protocol_error(
    error: crate::ConfiguredDatastoreOpenError,
) -> ProtocolError {
    match error {
        crate::ConfiguredDatastoreOpenError::AuthenticationRejected => ProtocolError::new(
            ProtocolErrorCode::Unauthorized,
            "Authenticated TRE user is not admitted or is no longer authorized",
        ),
        _ => ProtocolError::new(
            ProtocolErrorCode::OperationFailed,
            "Authenticated Session could not be opened",
        ),
    }
}

/// Canonical Domain identity shared by analyst and Browser projections.
pub fn domain_ref(
    datastore_id: ahri_tre_protocol::PublicUuid,
    id: ahri_tre_types::DomainId,
) -> DomainRef {
    DomainRef {
        datastore_id,
        kind: ahri_tre_protocol::refs::ObjectKind::Domain,
        id: encode_scoped_integer_ref("domain", id.0),
    }
}

/// Projects one app-owned Domain record into its safe public shape.
pub fn domain_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    domain: ahri_tre_types::DomainRecord,
) -> ahri_tre_protocol::domain::DomainSummary {
    ahri_tre_protocol::domain::DomainSummary {
        domain: domain_ref(datastore_id, domain.domain_id),
        name: domain.name.as_str().to_string(),
        description: domain.description,
        uri: domain.uri,
    }
}

/// Projects one app-owned Domain list outcome into the stable public result.
pub fn domain_list_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::DomainListResult,
) -> ahri_tre_protocol::domain::DomainListResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::domain::DomainListResponse {
        session,
        domains: result
            .domains
            .into_iter()
            .map(|domain| domain_summary(datastore_id, domain))
            .collect(),
    }
}

/// Projects one app-owned Domain add outcome into the stable public result.
pub fn domain_add_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::DomainAddResult,
) -> ahri_tre_protocol::domain::DomainAddResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::domain::DomainAddResponse {
        session,
        domain: ahri_tre_protocol::domain::DomainDetail {
            summary: domain_summary(datastore_id, result.domain),
        },
    }
}

/// Maps Study-context resolution failures without disclosing metadata details.
pub fn study_context_protocol_error(error: AppError) -> ProtocolError {
    let code = match error {
        AppError::NotFound(_) => ProtocolErrorCode::NotFound,
        AppError::Conflict(_) => ProtocolErrorCode::Conflict,
        AppError::Validation(_) => ProtocolErrorCode::ValidationFailed,
        _ => ProtocolErrorCode::OperationFailed,
    };
    ProtocolError::new(code, "Current study could not be resolved or authorized")
        .with_target("current_study")
}

/// Maps a Domain catalogue failure to one safe stable protocol error.
pub fn domain_list_protocol_error(error: AppError) -> ProtocolError {
    catalogue_list_protocol_error(error)
}

fn catalogue_list_protocol_error(error: AppError) -> ProtocolError {
    match error {
        AppError::NotFound(_) => ProtocolError::new(
            ProtocolErrorCode::NotFound,
            "Catalogue scope was not found or is not authorized",
        ),
        _ => ProtocolError::new(
            ProtocolErrorCode::OperationFailed,
            "Catalogue metadata could not be listed",
        ),
    }
}

/// Maps a Domain registration failure to one safe stable protocol error.
pub fn domain_registration_protocol_error(error: AppError) -> ProtocolError {
    registration_protocol_error(error, RegistrationResource::Domain)
}

fn registration_protocol_error(error: AppError, resource: RegistrationResource) -> ProtocolError {
    match error {
        AppError::Conflict(_) => ProtocolError::new(
            ProtocolErrorCode::Conflict,
            format!("{resource} registration conflicts with existing catalogue metadata"),
        ),
        AppError::NotFound(_) => ProtocolError::new(
            ProtocolErrorCode::NotFound,
            format!("{resource} registration scope was not found or is not authorized"),
        ),
        AppError::Validation(_) => ProtocolError::new(
            ProtocolErrorCode::ValidationFailed,
            format!("{resource} registration request is invalid"),
        ),
        AppError::NotImplemented(_) => ProtocolError::new(
            ProtocolErrorCode::UnsupportedOperation,
            format!("{resource} registration is not available"),
        ),
        AppError::Infrastructure(_)
        | AppError::DatasetAdmissionOutcomeUnknown(_)
        | AppError::RegistrationRollback { .. } => ProtocolError::new(
            ProtocolErrorCode::OperationFailed,
            format!("{resource} could not be registered"),
        ),
    }
}

/// Projects one app-owned Study registration into its safe public shape.
pub fn study_registration_payload(
    datastore_id: ahri_tre_protocol::PublicUuid,
    registration: crate::StudyRegistration,
) -> StudyRegistrationPayload {
    let domains = registration
        .domains
        .into_iter()
        .map(|domain| ahri_tre_protocol::domain::DomainSummary {
            domain: DomainRef {
                datastore_id,
                kind: ahri_tre_protocol::refs::ObjectKind::Domain,
                id: encode_scoped_integer_ref("domain", domain.domain_id.0),
            },
            name: domain.name.as_str().to_string(),
            description: domain.description,
            uri: domain.uri,
        })
        .collect::<Vec<_>>();
    let tags = registration
        .tags
        .into_iter()
        .map(|tag| TagSummary {
            tag: TagRef {
                datastore_id,
                kind: ahri_tre_protocol::refs::ObjectKind::Tag,
                id: encode_scoped_integer_ref("tag", tag.tag_id.0),
            },
            scope: TagScope::Global,
            label: tag.name,
        })
        .collect();

    StudyRegistrationPayload {
        study: StudySummary {
            study_type: registration.study.study_type_id.map(|id| {
                ahri_tre_protocol::refs::ObjectRef {
                    datastore_id,
                    kind: ahri_tre_protocol::refs::ObjectKind::StudyType,
                    id: ahri_tre_protocol::refs::encode_scoped_integer_ref("study_type", id.0),
                }
            }),
            study: StudyRef {
                datastore_id,
                kind: ahri_tre_protocol::refs::ObjectKind::Study,
                id: ahri_tre_protocol::PublicUuid::from_uuid(registration.study.study_id.0),
            },
            name: registration.study.name.as_str().to_string(),
            external_id: registration.study.external_id,
            title: None,
            description: registration.study.description,
            domains: domains.iter().map(|domain| domain.domain).collect(),
            tags,
        },
        domains,
    }
}

/// Projects one app-owned Study list outcome into the stable public result.
pub fn study_list_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::StudyListResult,
) -> ahri_tre_protocol::study::StudyListResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::study::StudyListResponse {
        session,
        studies: result
            .studies
            .into_iter()
            .map(|registration| study_registration_payload(datastore_id, registration))
            .collect(),
    }
}

/// Projects one app-owned Study add outcome into the stable public result.
pub fn study_add_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::StudyAddResult,
) -> ahri_tre_protocol::study::StudyAddResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::study::StudyAddResponse {
        session,
        registration: study_registration_payload(datastore_id, result.registration),
    }
}

/// Maps a Study catalogue failure to one safe stable protocol error.
pub fn study_list_protocol_error(error: AppError) -> ProtocolError {
    catalogue_list_protocol_error(error)
}

/// Maps a Study registration failure to one safe stable protocol error.
pub fn study_registration_protocol_error(error: AppError) -> ProtocolError {
    registration_protocol_error(error, RegistrationResource::Study)
}

/// Projects one app-owned Datafile list outcome into the stable public result.
pub fn datafile_list_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::DataFileListResult,
) -> ahri_tre_protocol::datafile::DataFileListResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::datafile::DataFileListResponse {
        session,
        study: result.study_name.as_str().to_string(),
        study_id: object_ref(
            datastore_id,
            ahri_tre_protocol::refs::ObjectKind::Study,
            result.study_id.0,
        ),
        datafiles: result
            .datafiles
            .into_iter()
            .map(|entry| ahri_tre_protocol::datafile::DataFileEntryPayload {
                catalog: datafile_list_asset_catalog(datastore_id, entry.catalog),
                datafiles: entry
                    .datafiles
                    .into_iter()
                    .map(
                        |datafile| ahri_tre_protocol::datafile::SafeDataFileVersionSummary {
                            datafile_id: object_ref(
                                datastore_id,
                                ahri_tre_protocol::refs::ObjectKind::AssetVersion,
                                datafile.datafile_id.0,
                            ),
                            compressed: datafile.compressed,
                            encrypted: datafile.encrypted,
                            compression_algorithm: datafile.compression_algorithm,
                            encryption_algorithm: datafile.encryption_algorithm,
                            edam_format: datafile.edam_format,
                        },
                    )
                    .collect(),
            })
            .collect(),
    }
}

fn datafile_list_asset_catalog(
    datastore_id: ahri_tre_protocol::PublicUuid,
    catalog: crate::DataFileListAssetCatalog,
) -> ahri_tre_protocol::asset::SafeAssetCatalog {
    ahri_tre_protocol::asset::SafeAssetCatalog {
        asset: ahri_tre_protocol::asset::SafeAssetSummary {
            asset_id: object_ref(
                datastore_id,
                ahri_tre_protocol::refs::ObjectKind::Asset,
                catalog.asset.asset_id.0,
            ),
            study_id: object_ref(
                datastore_id,
                ahri_tre_protocol::refs::ObjectKind::Study,
                catalog.asset.study_id.0,
            ),
            name: catalog.asset.name.as_str().to_string(),
            description: catalog.asset.description,
            asset_type: catalog.asset.asset_type,
            date_created: catalog.asset.date_created,
            created_by: catalog.asset.created_by,
        },
        versions: catalog
            .versions
            .into_iter()
            .map(|value| safe_asset_version_summary(datastore_id, value))
            .collect(),
        latest_version: catalog
            .latest_version
            .map(|value| safe_asset_version_summary(datastore_id, value)),
    }
}

/// Projects an app-owned asset catalogue into its safe public shape.
pub fn safe_asset_catalog(
    datastore_id: ahri_tre_protocol::PublicUuid,
    catalog: crate::AssetVersionCatalog,
) -> ahri_tre_protocol::asset::SafeAssetCatalog {
    datafile_list_asset_catalog(
        datastore_id,
        crate::DataFileListAssetCatalog::from_catalogue(catalog),
    )
}

pub fn safe_asset_version_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    version: ahri_tre_types::AssetVersionRecord,
) -> ahri_tre_protocol::asset::SafeAssetVersionSummary {
    ahri_tre_protocol::asset::SafeAssetVersionSummary {
        version_id: object_ref(
            datastore_id,
            ahri_tre_protocol::refs::ObjectKind::AssetVersion,
            version.version_id.0,
        ),
        asset_id: object_ref(
            datastore_id,
            ahri_tre_protocol::refs::ObjectKind::Asset,
            version.asset_id.0,
        ),
        major: version.major,
        minor: version.minor,
        patch: version.patch,
        version_label: version.version_label,
        version_note: version.version_note,
        is_latest: version.is_latest,
        doi: version.doi,
        created_at: version.created_at,
        created_by: version.created_by,
    }
}

/// Maps a Datafile catalogue failure to one safe stable protocol error.
pub fn datafile_list_protocol_error(error: AppError) -> ProtocolError {
    catalogue_list_protocol_error(error)
}

/// Projects one app-owned Datafile ingest outcome into the stable public result.
pub fn datafile_ingest_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::DataFileIngestResult,
    source: ahri_tre_protocol::ingest::IngestSourceSummary,
) -> ahri_tre_protocol::ingest::IngestFileResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    let response_version = result.catalog.latest_version.clone();
    let catalog = datafile_list_asset_catalog(datastore_id, result.catalog);
    let asset = catalog.asset;
    let study = StudyRef {
        datastore_id,
        kind: ahri_tre_protocol::refs::ObjectKind::Study,
        id: ahri_tre_protocol::PublicUuid::from_uuid(result.study_id.0),
    };
    let version = response_version
        .as_ref()
        .map(|version| ahri_tre_protocol::ingest::version_ref(datastore_id, version.version_id));
    ahri_tre_protocol::ingest::IngestFileResponse {
        session,
        study: result.study_name.as_str().to_string(),
        asset: ahri_tre_protocol::ingest::IngestedAssetSummary {
            name: asset.name.clone(),
            asset_type: asset.asset_type,
            study,
            description: asset.description.clone(),
            version,
        },
        datafile: ahri_tre_protocol::datafile::DataFileSummary {
            version,
            datafile: asset.asset_id,
            study,
            name: asset.name,
            format: result.datafile.edam_format,
            description: asset.description,
            size_bytes: Some(result.stored_size_bytes),
            digest_ref: Some(result.datafile.digest),
            tags: Vec::new(),
        },
        version: response_version.map(|version| safe_asset_version_summary(datastore_id, version)),
        transformation_id: result.transformation_id,
        transformation: Some(ahri_tre_protocol::refs::ObjectRef {
            datastore_id,
            kind: ahri_tre_protocol::refs::ObjectKind::Transformation,
            id: ahri_tre_protocol::refs::encode_scoped_integer_ref(
                "transformation",
                result.transformation_id.0,
            ),
        }),
        source_size_bytes: result.source_size_bytes,
        stored_size_bytes: result.stored_size_bytes,
        source,
    }
}

/// Maps a Datafile ingest failure to one safe stable protocol error.
pub fn datafile_ingest_protocol_error(error: AppError) -> ProtocolError {
    match error {
        AppError::Validation(_) => ProtocolError::new(
            ProtocolErrorCode::ValidationFailed,
            "Remote CSV upload request is invalid",
        ),
        AppError::NotFound(_) => ProtocolError::new(
            ProtocolErrorCode::NotFound,
            "Remote CSV upload Study was not found or is not authorized",
        ),
        AppError::Conflict(_) => ProtocolError::new(
            ProtocolErrorCode::Conflict,
            "Remote CSV upload conflicts with existing Datafile metadata",
        ),
        AppError::NotImplemented(_) => ProtocolError::new(
            ProtocolErrorCode::UnsupportedOperation,
            "Remote CSV upload is not available",
        ),
        AppError::Infrastructure(_)
        | AppError::DatasetAdmissionOutcomeUnknown(_)
        | AppError::RegistrationRollback { .. } => ProtocolError::new(
            ProtocolErrorCode::OperationFailed,
            "Remote CSV upload could not be completed",
        ),
    }
}

/// Projects a managed Datafile-to-Dataset workflow into its stable public
/// result without exposing the Lake relation or any storage path.
pub fn dataset_materialization_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    study: String,
    result: crate::ManagedDataFileDatasetMaterialization,
) -> ahri_tre_protocol::ingest::DatasetMaterializationResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    let source_study = StudyRef {
        datastore_id,
        kind: ahri_tre_protocol::refs::ObjectKind::Study,
        id: ahri_tre_protocol::PublicUuid::from_uuid(result.source.catalog.asset.study_id.0),
    };
    let source_name = result.source.catalog.asset.name.as_str().to_string();
    let source_description = result.source.catalog.asset.description.clone();
    let source_datafile = ahri_tre_protocol::datafile::DataFileSummary {
        version: Some(ahri_tre_protocol::ingest::version_ref(
            datastore_id,
            result.source.version.version_id,
        )),
        datafile: ahri_tre_protocol::ingest::datafile_ref(
            datastore_id,
            result.source.catalog.asset.asset_id,
        ),
        study: source_study,
        name: source_name,
        format: result.source.datafile.edam_format.clone(),
        description: source_description,
        size_bytes: None,
        digest_ref: Some(result.source.datafile.digest.as_str().to_string()),
        tags: Vec::new(),
    };
    let source_catalog = public_asset_catalog(result.source.catalog);
    let dataset_catalog = public_asset_catalog(result.dataset.catalog);
    ahri_tre_protocol::ingest::DatasetMaterializationResponse {
        session,
        study,
        catalog: dataset_catalog,
        dataset: result.dataset.dataset,
        lineage: workflow_lineage(datastore_id, result.dataset.lineage),
        row_count: result.dataset.row_count,
        column_count: result.dataset.column_count,
        variables: result
            .dataset
            .variables
            .into_iter()
            .map(
                |variable| ahri_tre_protocol::dataset::RegisteredDatasetVariablePayload {
                    variable: variable_summary(datastore_id, variable.variable),
                    vocabulary_items_withheld: variable.vocabulary.is_some(),
                    vocabulary: variable
                        .vocabulary
                        .map(|value| vocabulary_summary(datastore_id, value)),
                    vocabulary_items: Vec::new(),
                    link: dataset_variable_link_summary(datastore_id, variable.link),
                },
            )
            .collect(),
        metadata_warnings: result.dataset.metadata_warnings,
        source: ahri_tre_protocol::ingest::DatasetMaterializationSourceSummary {
            upload: None,
            kind: ahri_tre_protocol::ingest::DatasetMaterializationSourceKind::ManagedDataFile,
            request_only: false,
        },
        source_catalog: Some(source_catalog),
        source_version: Some(result.source.version),
        source_datafile: Some(source_datafile),
    }
}

fn public_asset_catalog(
    catalog: crate::AssetVersionCatalog,
) -> ahri_tre_protocol::asset::AssetCatalog {
    let mut asset = catalog.asset;
    asset.agent_instructions = None;
    ahri_tre_protocol::asset::AssetCatalog {
        asset,
        versions: catalog.versions,
        latest_version: catalog.latest_version,
    }
}

/// Maps a managed Datafile-to-Dataset failure to a stable redacted result.
pub fn dataset_materialization_protocol_error(error: AppError) -> ProtocolError {
    match error {
        AppError::Validation(_) => ProtocolError::new(
            ProtocolErrorCode::ValidationFailed,
            "Dataset derivation request is invalid",
        ),
        AppError::NotFound(_) => ProtocolError::new(
            ProtocolErrorCode::NotFound,
            "Dataset derivation scope was not found or is not authorized",
        ),
        AppError::Conflict(_) => ProtocolError::new(
            ProtocolErrorCode::Conflict,
            "Dataset derivation conflicts with existing catalogue metadata",
        ),
        AppError::NotImplemented(_) => ProtocolError::new(
            ProtocolErrorCode::UnsupportedOperation,
            "Dataset derivation is not available",
        ),
        AppError::Infrastructure(_)
        | AppError::DatasetAdmissionOutcomeUnknown(_)
        | AppError::RegistrationRollback { .. } => ProtocolError::new(
            ProtocolErrorCode::OperationFailed,
            "Dataset derivation could not be completed",
        ),
    }
}

/// Projects one app-owned Dataset metadata read into the stable public result.
pub fn dataset_metadata_response(
    session: ahri_tre_protocol::session::SessionStatusPayload,
    result: crate::DatasetMetadataResult,
) -> ahri_tre_protocol::dataset::DatasetMetadataResponse {
    let datastore_id = session
        .datastore_id
        .expect("catalogue Session retains its Datastore binding");
    ahri_tre_protocol::dataset::DatasetMetadataResponse {
        session,
        study: result.study.name.as_str().to_string(),
        metadata: ahri_tre_protocol::dataset::DatasetMetadataPayload {
            catalog: safe_asset_catalog(datastore_id, result.metadata.catalog),
            dataset: dataset_summary(datastore_id, result.metadata.dataset),
            variables_included: result.metadata.variables_included,
            variable_count: result.metadata.variable_count,
            variables: result
                .metadata
                .variables
                .into_iter()
                .map(
                    |variable| ahri_tre_protocol::dataset::DatasetVariableMetadataPayload {
                        link: dataset_variable_link_summary(datastore_id, variable.link),
                        variable: variable_summary(datastore_id, variable.variable),
                        value_type: value_type_summary(datastore_id, variable.value_type),
                        vocabulary_items_withheld: variable.vocabulary.is_some(),
                        vocabulary: variable
                            .vocabulary
                            .map(|value| vocabulary_summary(datastore_id, value)),
                        vocabulary_items: Vec::new(),
                    },
                )
                .collect(),
        },
    }
}

/// Maps Dataset metadata lookup failures without exposing catalogue internals.
pub fn dataset_metadata_protocol_error(error: AppError) -> ProtocolError {
    catalogue_protocol_error(error)
}

pub fn catalogue_protocol_error(error: AppError) -> ProtocolError {
    let (code, message) = match error {
        AppError::Validation(_) => (
            ProtocolErrorCode::ValidationFailed,
            "Catalogue selector is invalid",
        ),
        AppError::Conflict(_) => (
            ProtocolErrorCode::Conflict,
            "Catalogue selectors are ambiguous or disagree",
        ),
        AppError::NotFound(_) => (
            ProtocolErrorCode::NotFound,
            "Catalogue scope was not found or is not authorized",
        ),
        AppError::NotImplemented(_) => (
            ProtocolErrorCode::UnsupportedOperation,
            "Catalogue workflow is unavailable",
        ),
        _ => (
            ProtocolErrorCode::OperationFailed,
            "Catalogue metadata could not be read",
        ),
    };
    ProtocolError::new(code, message)
}

pub fn app_error_to_protocol_error(error: &AppError) -> ProtocolError {
    match error {
        AppError::Validation(message) => {
            ProtocolError::new(ProtocolErrorCode::ValidationFailed, safe_message(message))
        }
        AppError::NotFound(message) => {
            ProtocolError::new(ProtocolErrorCode::NotFound, safe_message(message))
        }
        AppError::Conflict(message) => {
            ProtocolError::new(ProtocolErrorCode::Conflict, safe_message(message))
        }
        AppError::NotImplemented(message) => ProtocolError::new(
            ProtocolErrorCode::UnsupportedOperation,
            format!("unsupported operation: {}", safe_message(message)),
        ),
        _ => ProtocolError::new(
            ProtocolErrorCode::Internal,
            "The operation failed before a public result was available.",
        ),
    }
}

pub fn safe_warning(code: ProtocolWarningCode, message: impl AsRef<str>) -> ProtocolWarning {
    ProtocolWarning::new(code, safe_message(message.as_ref()))
}

pub fn datastore_summary(
    datastore: ahri_tre_protocol::refs::DatastoreRef,
    name: impl Into<String>,
    ready: bool,
) -> DatastoreSummary {
    DatastoreSummary {
        datastore,
        name: name.into(),
        ready,
    }
}

pub fn datastore_ping_result(
    datastore: ahri_tre_protocol::refs::DatastoreRef,
    metadata_ready: bool,
    lake_ready: bool,
) -> DatastorePingResult {
    DatastorePingResult {
        datastore,
        metadata: ComponentStatus {
            ready: metadata_ready,
            message: None,
        },
        lake: ComponentStatus {
            ready: lake_ready,
            message: None,
        },
    }
}

pub fn datastore_schema_status_result(
    datastore: DatastoreRef,
    runtime: &ahri_tre_runtime::DataStoreRuntime,
    status: ahri_tre_pgmeta::DatastoreSchemaStatus,
) -> DatastoreSchemaStatusResult {
    DatastoreSchemaStatusResult {
        datastore,
        database: status.database,
        status: datastore_schema_compatibility_status(status.status),
        current_version: status.current_version,
        target_version: status.target_version,
        migration_history_present: status.migration_history_present,
        applied_versions: status.applied_versions,
        missing_tables: status.missing_tables,
        ducklake_catalog_migrations: ducklake_catalog_migration_ownership(runtime),
    }
}

pub fn datastore_migration_plan_result(
    runtime: &ahri_tre_runtime::DataStoreRuntime,
    plan: ahri_tre_pgmeta::DatastoreMigrationPlan,
) -> DatastoreMigrationPlanResult {
    let blocked_reason = plan.blocked_reason;
    let blocked = blocked_reason
        .as_deref()
        .is_some_and(|reason| !reason.trim().is_empty());
    let target_version = plan.target_version;
    DatastoreMigrationPlanResult {
        status: datastore_schema_compatibility_status(plan.status),
        current_version: plan.current_version,
        target_version: target_version.clone(),
        steps: plan
            .steps
            .into_iter()
            .map(datastore_migration_step_result)
            .collect(),
        blocked,
        blocked_reason: blocked_reason.clone(),
        dependency_summary: blocked.then(|| {
            json!({
                "status": "blocked",
                "items": [{
                    "kind": "datastore_schema",
                    "identifier": target_version,
                    "action": "block"
                }]
            })
        }),
        next_step: blocked.then(|| {
            "resolve the datastore schema compatibility issue, then rerun schema-plan or schema-migrate"
                .to_string()
        }),
        ducklake_catalog_migrations: ducklake_catalog_migration_ownership(runtime),
    }
}

pub fn datastore_migration_apply_result(
    datastore: DatastoreRef,
    runtime: &ahri_tre_runtime::DataStoreRuntime,
    report: ahri_tre_pgmeta::DatastoreMigrationApplyReport,
) -> DatastoreMigrationApplyResult {
    let blocked_reason = report.blocked_reason;
    let blocked = blocked_reason
        .as_deref()
        .is_some_and(|reason| !reason.trim().is_empty());
    let before = datastore_migration_plan_result(runtime, report.before);
    let target_version = before.target_version.clone();
    DatastoreMigrationApplyResult {
        before,
        after: datastore_schema_status_result(datastore, runtime, report.after),
        applied_steps: report
            .applied_steps
            .into_iter()
            .map(datastore_migration_step_result)
            .collect(),
        blocked,
        blocked_reason: blocked_reason.clone(),
        dependency_summary: blocked.then(|| {
            json!({
                "status": "blocked",
                "items": [{
                    "kind": "datastore_schema",
                    "identifier": target_version,
                    "action": "block"
                }]
            })
        }),
        next_step: blocked.then(|| {
            "resolve the datastore schema compatibility issue, then rerun schema-plan or schema-migrate"
                .to_string()
        }),
    }
}

pub fn datastore_schema_compatibility_status(
    status: ahri_tre_pgmeta::DatastoreSchemaCompatibility,
) -> DatastoreSchemaCompatibilityStatus {
    match status {
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::Current => {
            DatastoreSchemaCompatibilityStatus::Current
        }
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::Pending => {
            DatastoreSchemaCompatibilityStatus::Pending
        }
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::MissingHistory => {
            DatastoreSchemaCompatibilityStatus::MissingHistory
        }
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::Dirty => {
            DatastoreSchemaCompatibilityStatus::Dirty
        }
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::TooNew => {
            DatastoreSchemaCompatibilityStatus::TooNew
        }
        ahri_tre_pgmeta::DatastoreSchemaCompatibility::Unsupported => {
            DatastoreSchemaCompatibilityStatus::Unsupported
        }
    }
}

fn ducklake_catalog_migration_ownership(
    runtime: &ahri_tre_runtime::DataStoreRuntime,
) -> DuckLakeCatalogMigrationOwnership {
    DuckLakeCatalogMigrationOwnership {
        database: runtime.lake.catalog_database.clone(),
        schema: runtime.lake.catalog_schema.clone(),
        managed_by: "ducklake".to_string(),
        included_in_datastore_schema_migration: false,
    }
}

fn datastore_migration_step_result(
    step: ahri_tre_pgmeta::DatastoreMigrationStep,
) -> DatastoreMigrationStepResult {
    DatastoreMigrationStepResult {
        version: step.version,
        description: step.description,
        scope: match step.scope {
            ahri_tre_pgmeta::DatastoreMigrationScope::Metadata => DatastoreMigrationScope::Metadata,
        },
        transactional: step.transactional,
        action: match step.action {
            ahri_tre_pgmeta::DatastoreMigrationAction::Apply => DatastoreMigrationAction::Apply,
            ahri_tre_pgmeta::DatastoreMigrationAction::Skipped => DatastoreMigrationAction::Skipped,
            ahri_tre_pgmeta::DatastoreMigrationAction::Failed => DatastoreMigrationAction::Failed,
            ahri_tre_pgmeta::DatastoreMigrationAction::Blocked => DatastoreMigrationAction::Blocked,
        },
    }
}

fn safe_message(message: &str) -> String {
    let lower = message.to_ascii_lowercase();
    if lower.contains("password")
        || lower.contains("token")
        || lower.contains("secret")
        || lower.contains("postgres://")
        || lower.contains("select ")
        || lower.contains("/tmp/")
        || lower.contains("/workspaces/")
    {
        "The operation failed with details omitted from the public protocol response.".to_string()
    } else {
        message.to_string()
    }
}

pub fn object_ref(
    datastore_id: ahri_tre_protocol::PublicUuid,
    kind: ahri_tre_protocol::refs::ObjectKind,
    id: uuid::Uuid,
) -> ahri_tre_protocol::refs::ObjectRef {
    ahri_tre_protocol::refs::ObjectRef {
        datastore_id,
        kind,
        id: ahri_tre_protocol::PublicUuid::from_uuid(id),
    }
}

pub fn safe_datafile_version_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    datafile: ahri_tre_types::DataFileRecord,
) -> ahri_tre_protocol::datafile::SafeDataFileVersionSummary {
    ahri_tre_protocol::datafile::SafeDataFileVersionSummary {
        datafile_id: object_ref(
            datastore_id,
            ahri_tre_protocol::refs::ObjectKind::AssetVersion,
            datafile.datafile_id.0,
        ),
        compressed: datafile.compressed,
        encrypted: datafile.encrypted,
        compression_algorithm: datafile.compression_algorithm,
        encryption_algorithm: datafile.encryption_algorithm,
        edam_format: datafile.edam_format,
    }
}

pub fn dataset_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::DatasetRecord,
) -> ahri_tre_protocol::catalogue::DatasetSummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::DatasetSummary {
        dataset_id: object_ref(datastore_id, ObjectKind::AssetVersion, value.dataset_id.0),
    }
}

pub fn dataset_variable_link_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::DatasetVariableLinkRecord,
) -> ahri_tre_protocol::catalogue::DatasetVariableLinkSummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::DatasetVariableLinkSummary {
        dataset_id: object_ref(datastore_id, ObjectKind::AssetVersion, value.dataset_id.0),
        variable_id: object_ref(
            datastore_id,
            ObjectKind::Variable,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("variable", value.variable_id.0)
                .as_uuid(),
        ),
        row_role: value.row_role,
    }
}

pub fn variable_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::VariableRecord,
) -> ahri_tre_protocol::catalogue::VariableSummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::VariableSummary {
        variable_id: object_ref(
            datastore_id,
            ObjectKind::Variable,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("variable", value.variable_id.0)
                .as_uuid(),
        ),
        domain_id: object_ref(
            datastore_id,
            ObjectKind::Domain,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("domain", value.domain_id.0)
                .as_uuid(),
        ),
        name: value.name,
        value_type_id: object_ref(
            datastore_id,
            ObjectKind::ValueType,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("value_type", value.value_type_id.0)
                .as_uuid(),
        ),
        value_format: value.value_format,
        vocabulary_id: value.vocabulary_id.map(|id| {
            object_ref(
                datastore_id,
                ObjectKind::Vocabulary,
                ahri_tre_protocol::refs::encode_scoped_integer_ref("vocabulary", id.0).as_uuid(),
            )
        }),
        key_role: value.key_role,
        description: value.description,
        note: value.note,
        ontology_namespace: value.ontology_namespace,
        ontology_class: value.ontology_class,
    }
}

pub fn value_type_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::ValueTypeRecord,
) -> ahri_tre_protocol::catalogue::ValueTypeSummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::ValueTypeSummary {
        value_type_id: object_ref(
            datastore_id,
            ObjectKind::ValueType,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("value_type", value.value_type_id.0)
                .as_uuid(),
        ),
        value_type: value.value_type,
        description: value.description,
    }
}

pub fn vocabulary_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::VocabularyRecord,
) -> ahri_tre_protocol::catalogue::VocabularySummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::VocabularySummary {
        vocabulary_id: object_ref(
            datastore_id,
            ObjectKind::Vocabulary,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("vocabulary", value.vocabulary_id.0)
                .as_uuid(),
        ),
        domain_id: object_ref(
            datastore_id,
            ObjectKind::Domain,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("domain", value.domain_id.0)
                .as_uuid(),
        ),
        name: value.name,
        description: value.description,
    }
}

pub fn vocabulary_item_summary(
    datastore_id: ahri_tre_protocol::PublicUuid,
    value: ahri_tre_types::VocabularyItemRecord,
) -> ahri_tre_protocol::catalogue::VocabularyItemSummary {
    use ahri_tre_protocol::refs::ObjectKind;
    ahri_tre_protocol::catalogue::VocabularyItemSummary {
        vocabulary_item_id: object_ref(
            datastore_id,
            ObjectKind::VocabularyItem,
            ahri_tre_protocol::refs::encode_scoped_integer_ref(
                "vocabulary_item",
                value.vocabulary_item_id.0,
            )
            .as_uuid(),
        ),
        vocabulary_id: object_ref(
            datastore_id,
            ObjectKind::Vocabulary,
            ahri_tre_protocol::refs::encode_scoped_integer_ref("vocabulary", value.vocabulary_id.0)
                .as_uuid(),
        ),
        value: value.value,
        code: value.code,
        description: value.description,
    }
}

/// Projects the shared Dataset listing without adapter-owned selection policy.
pub fn dataset_list_response(
    datastore_id: ahri_tre_protocol::PublicUuid,
    session: ahri_tre_protocol::session::SessionStatusPayload,
    study: ahri_tre_types::StudyRecord,
    listing: crate::StudyDatasetList,
) -> ahri_tre_protocol::dataset::DatasetListResponse {
    ahri_tre_protocol::dataset::DatasetListResponse {
        session,
        study: study.name.as_str().to_string(),
        study_id: object_ref(
            datastore_id,
            ahri_tre_protocol::refs::ObjectKind::Study,
            study.study_id.0,
        ),
        datasets: listing
            .datasets
            .into_iter()
            .map(
                |entry| ahri_tre_protocol::dataset::DatasetListEntryPayload {
                    catalog: safe_asset_catalog(datastore_id, entry.catalog),
                    latest_dataset: entry
                        .latest_dataset
                        .map(|value| dataset_summary(datastore_id, value)),
                },
            )
            .collect(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use ahri_tre_protocol::{
        ProtocolErrorCode, PublicName,
        refs::SessionSelector,
        session::{SessionAvailability, SessionStatusPayload},
    };
    use uuid::Uuid;

    fn test_session_status() -> SessionStatusPayload {
        SessionStatusPayload {
            name: (SessionSelector::Name {
                name: PublicName::new("analysis").unwrap(),
            })
            .name()
            .map(|name| name.as_str().to_owned())
            .unwrap_or_default(),
            session: None,
            adapter_observations: Vec::new(),
            execution_profile: Some("research".to_string()),
            datastore_key: Some("test-datastore".to_string()),
            datastore_id: Some(ahri_tre_protocol::PublicUuid::from_uuid(
                uuid::Uuid::from_u128(5),
            )),
            availability: SessionAvailability::Live,
            authenticated_tre_user: Some("0000-0001-2345-6789".to_string()),
            current_study: None,
            current_study_name: None,
            created_at: None,
            updated_at: None,
            unavailable_reason: None,
        }
    }

    #[test]
    fn domain_list_workflow_has_one_safe_public_result() {
        let result = crate::DomainListResult {
            domains: vec![ahri_tre_types::DomainRecord {
                domain_id: ahri_tre_types::DomainId(10),
                name: ahri_tre_types::NcName::parse("public_health").unwrap(),
                description: Some("Public health".to_string()),
                uri: Some("https://example.test/public-health".to_string()),
            }],
        };

        let projected = domain_list_response(test_session_status(), result);

        assert_eq!(projected.session.name, "analysis");
        assert_eq!(projected.domains.len(), 1);
        assert_eq!(projected.domains[0].name, "public_health");
        assert_eq!(
            projected.domains[0].domain.id,
            encode_scoped_integer_ref("domain", 10)
        );
    }

    #[test]
    fn domain_add_workflow_has_one_safe_public_result_and_error() {
        let result = crate::DomainAddResult {
            domain: ahri_tre_types::DomainRecord {
                domain_id: ahri_tre_types::DomainId(11),
                name: ahri_tre_types::NcName::parse("population_health").unwrap(),
                description: Some("Population health".to_string()),
                uri: None,
            },
        };

        let projected = domain_add_response(test_session_status(), result);
        let private_error =
            crate::AppError::Infrastructure("database password and /tmp/private-path".to_string());
        let error = domain_registration_protocol_error(private_error);

        assert_eq!(projected.domain.summary.name, "population_health");
        assert_eq!(error.code, ProtocolErrorCode::OperationFailed);
        assert_eq!(error.message, "Domain could not be registered");
    }

    #[test]
    fn domain_workflow_errors_have_one_safe_public_meaning() {
        let list_missing = domain_list_protocol_error(crate::AppError::NotFound(
            "private catalogue detail".to_string(),
        ));
        let list_failed = domain_list_protocol_error(crate::AppError::Infrastructure(
            "postgres://private.example/tre".to_string(),
        ));
        let registration_cases = [
            (
                crate::AppError::Validation("private bad name".to_string()),
                ProtocolErrorCode::ValidationFailed,
                "Domain registration request is invalid",
            ),
            (
                crate::AppError::NotFound("private scope".to_string()),
                ProtocolErrorCode::NotFound,
                "Domain registration scope was not found or is not authorized",
            ),
            (
                crate::AppError::Conflict("private metadata".to_string()),
                ProtocolErrorCode::Conflict,
                "Domain registration conflicts with existing catalogue metadata",
            ),
        ];

        assert_eq!(list_missing.code, ProtocolErrorCode::NotFound);
        assert_eq!(
            list_missing.message,
            "Catalogue scope was not found or is not authorized"
        );
        assert_eq!(list_failed.code, ProtocolErrorCode::OperationFailed);
        assert_eq!(
            list_failed.message,
            "Catalogue metadata could not be listed"
        );
        for (app_error, code, message) in registration_cases {
            let projected = domain_registration_protocol_error(app_error);
            assert_eq!(projected.code, code);
            assert_eq!(projected.message, message);
            assert!(!projected.message.contains("private"));
        }
    }

    #[test]
    fn study_registration_projection_keeps_public_domains_and_tags() {
        let registration = crate::StudyRegistration {
            study: ahri_tre_types::StudyRecord {
                study_id: ahri_tre_types::StudyId(
                    Uuid::parse_str("33333333-3333-4333-8333-333333333333").unwrap(),
                ),
                name: ahri_tre_types::NcName::parse("cohort").unwrap(),
                description: Some("Safe catalogue metadata".to_string()),
                documentation: None,
                agent_instructions: None,
                external_id: Some("EXT-001".to_string()),
                study_type_id: None,
                date_created: None,
                created_by: Some("0000-0000-0000-0001".to_string()),
            },
            domains: vec![ahri_tre_types::DomainRecord {
                domain_id: ahri_tre_types::DomainId(10),
                name: ahri_tre_types::NcName::parse("public_health").unwrap(),
                description: Some("Public health".to_string()),
                uri: None,
            }],
            tags: vec![ahri_tre_types::TagRecord {
                tag_id: ahri_tre_types::TagId(1),
                name: "curated".to_string(),
            }],
        };

        let projected = study_registration_payload(
            ahri_tre_protocol::PublicUuid::from_uuid(uuid::Uuid::from_u128(5)),
            registration,
        );

        assert_eq!(projected.domains.len(), 1);
        assert_eq!(projected.study.domains, vec![projected.domains[0].domain]);
        assert_eq!(projected.study.tags.len(), 1);
        assert_eq!(projected.study.tags[0].label, "curated");
    }

    #[test]
    fn study_workflows_have_one_safe_public_result_path() {
        let registration = crate::StudyRegistration {
            study: ahri_tre_types::StudyRecord {
                study_id: ahri_tre_types::StudyId(
                    Uuid::parse_str("44444444-4444-4444-8444-444444444444").unwrap(),
                ),
                name: ahri_tre_types::NcName::parse("cohort").unwrap(),
                description: Some("Safe catalogue metadata".to_string()),
                documentation: None,
                agent_instructions: None,
                external_id: Some("EXT-002".to_string()),
                study_type_id: Some(ahri_tre_types::StudyTypeId(1)),
                date_created: None,
                created_by: Some("0000-0000-0000-0002".to_string()),
            },
            domains: vec![ahri_tre_types::DomainRecord {
                domain_id: ahri_tre_types::DomainId(12),
                name: ahri_tre_types::NcName::parse("population_health").unwrap(),
                description: Some("Population health".to_string()),
                uri: None,
            }],
            tags: vec![ahri_tre_types::TagRecord {
                tag_id: ahri_tre_types::TagId(2),
                name: "approved".to_string(),
            }],
        };

        let listed = study_list_response(
            test_session_status(),
            crate::StudyListResult {
                studies: vec![registration.clone()],
            },
        );
        let added = study_add_response(
            test_session_status(),
            crate::StudyAddResult { registration },
        );

        assert_eq!(listed.studies.len(), 1);
        assert_eq!(listed.studies[0].study.name, "cohort");
        assert_eq!(listed.studies[0].domains.len(), 1);
        assert_eq!(listed.studies[0].study.tags[0].label, "approved");
        assert_eq!(added.registration.study.name, "cohort");
        assert_eq!(added.registration.domains.len(), 1);
    }

    #[test]
    fn study_workflow_errors_have_one_safe_public_meaning() {
        let list_missing =
            study_list_protocol_error(crate::AppError::NotFound("private Study scope".to_string()));
        let list_failed = study_list_protocol_error(crate::AppError::Infrastructure(
            "postgres://private.example/tre".to_string(),
        ));
        let registration_cases = [
            (
                crate::AppError::Validation("private bad Study".to_string()),
                ProtocolErrorCode::ValidationFailed,
                "Study registration request is invalid",
            ),
            (
                crate::AppError::NotFound("private Domain scope".to_string()),
                ProtocolErrorCode::NotFound,
                "Study registration scope was not found or is not authorized",
            ),
            (
                crate::AppError::Conflict("private metadata".to_string()),
                ProtocolErrorCode::Conflict,
                "Study registration conflicts with existing catalogue metadata",
            ),
        ];

        assert_eq!(list_missing.code, ProtocolErrorCode::NotFound);
        assert_eq!(
            list_missing.message,
            "Catalogue scope was not found or is not authorized"
        );
        assert_eq!(list_failed.code, ProtocolErrorCode::OperationFailed);
        assert_eq!(
            list_failed.message,
            "Catalogue metadata could not be listed"
        );
        for (app_error, code, message) in registration_cases {
            let projected = study_registration_protocol_error(app_error);
            assert_eq!(projected.code, code);
            assert_eq!(projected.message, message);
            assert!(!projected.message.contains("private"));
        }
    }

    #[test]
    fn datafile_list_workflow_has_one_safe_public_result() {
        let study_id = ahri_tre_types::StudyId(
            Uuid::parse_str("55555555-5555-4555-8555-555555555555").unwrap(),
        );
        let asset_id = ahri_tre_types::AssetId(
            Uuid::parse_str("66666666-6666-4666-8666-666666666666").unwrap(),
        );
        let version_id = ahri_tre_types::VersionId(
            Uuid::parse_str("77777777-7777-4777-8777-777777777777").unwrap(),
        );
        let result = crate::DataFileListResult::from_catalogue(
            ahri_tre_types::StudyRecord {
                study_id,
                name: ahri_tre_types::NcName::parse("cohort").unwrap(),
                description: None,
                documentation: None,
                agent_instructions: None,
                external_id: Some("EXT-003".to_string()),
                study_type_id: Some(ahri_tre_types::StudyTypeId(1)),
                date_created: None,
                created_by: None,
            },
            vec![crate::StudyDataFileEntry {
                catalog: crate::AssetVersionCatalog {
                    asset: ahri_tre_types::AssetRecord {
                        asset_id,
                        study_id,
                        name: ahri_tre_types::NcName::parse("baseline_csv").unwrap(),
                        description: Some("Baseline data".to_string()),
                        agent_instructions: Some("private analyst note".to_string()),
                        asset_type: ahri_tre_types::AssetType::File,
                        date_created: None,
                        created_by: None,
                    },
                    versions: vec![ahri_tre_types::AssetVersionRecord {
                        version_id,
                        asset_id,
                        major: 1,
                        minor: 0,
                        patch: 0,
                        version_label: Some("v1.0.0".to_string()),
                        version_note: None,
                        is_latest: Some(true),
                        doi: None,
                        created_at: None,
                        created_by: None,
                    }],
                    latest_version: None,
                },
                datafiles: vec![ahri_tre_types::DataFileRecord {
                    datafile_id: version_id,
                    size_bytes: Some(1_024),
                    storage_uri: "lake://private/baseline.csv".to_string(),
                    digest: ahri_tre_types::Sha256Hex::parse(
                        "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                    )
                    .unwrap(),
                    encryption_key: Some(vec![1, 2, 3, 4]),
                    compressed: Some(true),
                    encrypted: Some(true),
                    compression_algorithm: Some("zstd".to_string()),
                    encryption_algorithm: Some("aes-256-gcm".to_string()),
                    edam_format: "EDAM:format_3752".to_string(),
                }],
            }],
        );
        let app_json = serde_json::to_string(&result).unwrap();

        let projected = datafile_list_response(test_session_status(), result);
        let json = serde_json::to_string(&projected).unwrap();

        assert_eq!(projected.study, "cohort");
        assert_eq!(projected.study_id.id.as_uuid(), study_id.0);
        assert_eq!(projected.datafiles.len(), 1);
        assert_eq!(
            projected.datafiles[0].datafiles[0].edam_format,
            "EDAM:format_3752"
        );
        for forbidden in [
            "lake://",
            "private/baseline.csv",
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
            "storage_uri",
            "digest",
            "private analyst note",
            "encryption_key",
        ] {
            assert!(
                !app_json.contains(forbidden),
                "app result contains {forbidden}"
            );
            assert!(
                !json.contains(forbidden),
                "public result contains {forbidden}"
            );
        }
    }

    #[test]
    fn datafile_list_errors_hide_missing_and_unauthorized_scope() {
        let missing =
            datafile_list_protocol_error(crate::AppError::NotFound("missing Study".to_string()));
        let unauthorized = datafile_list_protocol_error(crate::AppError::NotFound(
            "Study hidden by PostgreSQL policy".to_string(),
        ));

        assert_eq!(missing, unauthorized);
        assert_eq!(missing.code, ProtocolErrorCode::NotFound);
        assert_eq!(
            missing.message,
            "Catalogue scope was not found or is not authorized"
        );
    }

    #[test]
    fn dataset_materialization_errors_are_stable_and_redacted() {
        let cases = [
            (
                crate::AppError::Validation("private source path".to_string()),
                ProtocolErrorCode::ValidationFailed,
            ),
            (
                crate::AppError::NotFound("private Study identity".to_string()),
                ProtocolErrorCode::NotFound,
            ),
            (
                crate::AppError::Conflict("private version identity".to_string()),
                ProtocolErrorCode::Conflict,
            ),
            (
                crate::AppError::Infrastructure("postgres password".to_string()),
                ProtocolErrorCode::OperationFailed,
            ),
        ];

        for (error, expected_code) in cases {
            let projected = dataset_materialization_protocol_error(error);
            assert_eq!(projected.code, expected_code);
            for forbidden in ["private", "path", "postgres", "password"] {
                assert!(!projected.message.to_ascii_lowercase().contains(forbidden));
            }
        }
    }

    #[test]
    fn datafile_ingest_projection_has_one_safe_meaning_for_all_adapters() {
        let study_id = ahri_tre_types::StudyId(uuid::Uuid::new_v4());
        let asset_id = ahri_tre_types::AssetId(uuid::Uuid::new_v4());
        let version_id = ahri_tre_types::VersionId(uuid::Uuid::new_v4());
        let result = crate::DataFileIngestResult {
            study_id,
            study_name: ahri_tre_types::NcName::parse("cohort").unwrap(),
            catalog: crate::DataFileListAssetCatalog {
                asset: crate::DataFileListAssetSummary {
                    asset_id,
                    study_id,
                    name: ahri_tre_types::NcName::parse("visits").unwrap(),
                    description: Some("Visit rows".to_string()),
                    asset_type: ahri_tre_types::AssetType::File,
                    date_created: None,
                    created_by: None,
                },
                versions: Vec::new(),
                latest_version: Some(ahri_tre_types::AssetVersionRecord {
                    version_id,
                    asset_id,
                    major: 1,
                    minor: 0,
                    patch: 0,
                    version_label: Some("v1.0.0".to_string()),
                    version_note: None,
                    is_latest: Some(true),
                    doi: None,
                    created_at: None,
                    created_by: None,
                }),
            },
            datafile: crate::DataFileIngestVersion {
                datafile_id: version_id,
                edam_format: "EDAM:format_3752".to_string(),
                digest: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                    .to_string(),
            },
            transformation_id: ahri_tre_types::TransformationId(7),
            source_size_bytes: 18,
            stored_size_bytes: 18,
        };

        let projected = datafile_ingest_response(
            test_session_status(),
            result,
            ahri_tre_protocol::ingest::IngestSourceSummary {
                kind: ahri_tre_protocol::ingest::IngestSourceKind::RestrictedLocalPath,
                request_only: true,
                upload: None,
            },
        );
        let json = serde_json::to_string(&projected).unwrap();

        assert_eq!(projected.study, "cohort");
        assert_eq!(projected.asset.name, "visits");
        assert_eq!(projected.datafile.format, "EDAM:format_3752");
        assert_eq!(
            projected.datafile.digest_ref.as_deref(),
            Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
        );
        for forbidden in ["lake://", "storage_uri", "encryption_key", "/workspaces/"] {
            assert!(!json.contains(forbidden));
        }
    }

    #[test]
    fn datafile_ingest_errors_have_one_safe_public_meaning() {
        let missing = datafile_ingest_protocol_error(crate::AppError::NotFound(
            "missing private Study".to_string(),
        ));
        let unauthorized = datafile_ingest_protocol_error(crate::AppError::NotFound(
            "PostgreSQL policy denied private Study".to_string(),
        ));
        let failed = datafile_ingest_protocol_error(crate::AppError::Infrastructure(
            "Lake location /private/lake and password leaked".to_string(),
        ));

        assert_eq!(missing, unauthorized);
        assert_eq!(missing.code, ProtocolErrorCode::NotFound);
        assert_eq!(failed.code, ProtocolErrorCode::OperationFailed);
        for forbidden in ["private", "PostgreSQL", "/private/lake", "password"] {
            assert!(!missing.message.contains(forbidden));
            assert!(!failed.message.contains(forbidden));
        }
    }

    #[test]
    fn projection_redacts_sensitive_public_messages() {
        let warning = safe_warning(
            ProtocolWarningCode::UnsupportedDetailOmitted,
            "token leaked at /workspaces/secret.sql select * from users",
        );
        assert!(!warning.message.contains("token"));
        assert!(!warning.message.contains("/workspaces"));
        assert!(!warning.message.contains("select"));
    }

    #[test]
    fn app_error_projection_uses_public_error_codes() {
        let error = app_error_to_protocol_error(&AppError::Validation("bad request".to_string()));
        assert_eq!(error.code, ProtocolErrorCode::ValidationFailed);
        assert_eq!(error.message, "bad request");

        let error =
            app_error_to_protocol_error(&AppError::NotImplemented("datastore.lake.move.resume"));
        assert_eq!(error.code, ProtocolErrorCode::UnsupportedOperation);
        assert_eq!(
            error.message,
            "unsupported operation: datastore.lake.move.resume"
        );
    }
}