ahri_tre_app/service/

operations.rs

//! Application-owned operation lifecycle and authorization-aware projection.
mod history;
use super::datafile_materialization::{
    DatafileLifecycle, DatafileStage, ResolvedDatafileMaterialization,
};
use super::*;
use ahri_tre_core::{OperationRepository, OperationResourceScope, StoredOperation};
use ahri_tre_pgmeta::PgMetadataRepository;
use ahri_tre_protocol::operation::*;
use ahri_tre_protocol::pagination::Page;
use ahri_tre_protocol::session::SessionStatusPayload;
use ahri_tre_protocol::{ProtocolError, ProtocolErrorCode, RequestId};
use chrono::{Duration, Utc};
use uuid::Uuid;

/// Cancellation policy is separate from the current execution window.
#[derive(Debug, Clone, Copy)]
pub enum OperationCancellationPolicy {
    Cooperative,
    Unsupported,
}
impl OperationCancellationPolicy {
    pub fn request(
        self,
        status: OperationStatus,
        cancellable: bool,
    ) -> Result<OperationStatus, Box<ProtocolError>> {
        if matches!(self, Self::Unsupported) {
            return Err(Box::new(ProtocolError::new(
                ProtocolErrorCode::UnsupportedOperation,
                "Operation does not support cancellation",
            )));
        }
        if status == OperationStatus::CancelRequested {
            return Ok(status);
        }
        if status.is_terminal() || !cancellable {
            return Err(Box::new(cancellation_conflict()));
        }
        Ok(OperationStatus::CancelRequested)
    }
}

/// Live Session authority is checked at each governed stage; cleanup needs no user capability.
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum OperationSessionAuthority {
    Active,
    Closed,
    Lost,
}

/// Newly authenticated context. Only safe correlation fields are persisted.
pub struct OperationContext {
    pub request_id: RequestId,
    pub observation: ahri_tre_observability::CorrelationContext,
    pub owner: String,
    pub session: SessionStatusPayload,
    pub authority: Box<dyn Fn() -> OperationSessionAuthority + Send>,
}

/// Independent lifecycle capability exposing only authorized inspection and cancellation.
pub struct OperationControl {
    repository: PgMetadataRepository<'static>,
    deployment_id: Uuid,
    datastore_id: Uuid,
}

impl OperationControl {
    pub(crate) fn new(
        repository: PgMetadataRepository<'static>,
        deployment_id: Uuid,
        datastore_id: Uuid,
    ) -> Self {
        Self {
            repository,
            deployment_id,
            datastore_id,
        }
    }

    fn service(&self) -> AppService {
        let mut service = AppService::new(
            crate::session::ScopedSessionMetadataRepositories::from_repository(
                std::sync::Arc::new(self.repository.clone()),
            )
            .into(),
        );
        service.datastore_id = Some(self.datastore_id);
        service
    }

    pub async fn get(
        &self,
        owner: &str,
        operation: &OperationRef,
        events: Option<&ahri_tre_protocol::pagination::PageRequest>,
    ) -> Result<OperationDetail, ProtocolError> {
        self.service()
            .get_operation(self, owner, operation, events)
            .await
    }

    pub async fn cancel(
        &self,
        owner: &str,
        operation: &OperationRef,
    ) -> Result<OperationSummary, ProtocolError> {
        self.service()
            .cancel_operation(self, owner, operation)
            .await
    }

    pub async fn result(
        &self,
        owner: &str,
        operation: &OperationRef,
    ) -> Result<OperationResult, ProtocolError> {
        self.service()
            .get_operation_result(self, owner, operation)
            .await
    }
}

/// Newly authorized admission or a retained result, resolved before worker and
/// output ownership acquisition. The private key guard lives only to acceptance.
pub enum OperationAdmission {
    Retained(Box<OperationSummary>),
    New(OperationSubmission),
}

pub struct OperationSubmission {
    key: Option<ahri_tre_pgmeta::PgOperationKey>,
    preconditions: Vec<ahri_tre_protocol::mutation::Precondition>,
}

impl OperationControl {
    pub async fn admit(
        &self,
        owner: &str,
        request: &mut ahri_tre_protocol::ingest::DatasetCreateFromDatafileRequest,
        current_study: Option<ahri_tre_protocol::refs::ObjectRef>,
    ) -> Result<OperationAdmission, ProtocolError> {
        use ahri_tre_protocol::mutation::Precondition;
        // Resolve on the independent operation-control connection, so retries
        // and polling do not wait for a producer's live datastore lock.
        let service = self.service();
        let domain = service
            .get_domain(GetDomainRequest {
                domain: service
                    .catalogue_domain_selector(request.domain.clone())
                    .map_err(crate::projections::catalogue_protocol_error)?,
            })
            .await
            .map_err(crate::projections::catalogue_protocol_error)?
            .ok_or_else(|| {
                ProtocolError::new(ProtocolErrorCode::NotFound, "Domain was not found")
            })?;
        request.domain = ahri_tre_protocol::domain::DomainSelector::Id {
            domain: crate::projections::domain_ref(
                ahri_tre_protocol::PublicUuid::from_uuid(self.datastore_id),
                domain.domain_id,
            ),
        };
        let output = service
            .resolve_session_study(self.datastore_id, request.study.take(), current_study)
            .await?;
        let canonical = |reference: ahri_tre_protocol::refs::ObjectRef| {
            ahri_tre_protocol::study::StudySelector::Id { study: reference }
        };
        request.study = Some(canonical(output));
        // Resolve identity and every supplied scope before checking agreement.
        // Keep an omitted/latest selection moving in the idempotency intent;
        // the producer pins it once before reservation and source preparation.
        let (source_study, catalog, pinned) = service
            .resolve_catalogue_asset(request.source.asset.clone())
            .await
            .map_err(crate::projections::catalogue_protocol_error)?;
        if source_study.study_id.0 != output.id.as_uuid() {
            return Err(ProtocolError::new(
                ProtocolErrorCode::ValidationFailed,
                "This materialization requires source and Output in the same Study",
            ));
        }
        if catalog.asset.asset_type != ahri_tre_types::AssetType::File {
            return Err(ProtocolError::new(
                ProtocolErrorCode::ValidationFailed,
                "A Datafile source is required",
            ));
        }
        // A pinned reference stays immutable across the worker handoff. Its
        // semantic version is used only to compare equivalent retry intents.
        let pinned_latest = pinned.is_some()
            && request
                .source
                .version
                .as_deref()
                .is_some_and(|value| value.trim().eq_ignore_ascii_case("latest"));
        let pinned_version = if pinned.is_some() {
            Some(
                service
                    .catalogue_version(
                        &catalog,
                        pinned,
                        if pinned_latest {
                            None
                        } else {
                            request.source.version.as_deref()
                        },
                    )
                    .map_err(crate::projections::catalogue_protocol_error)?,
            )
        } else {
            None
        };
        let source_asset = ahri_tre_protocol::ingest::datafile_ref(
            ahri_tre_protocol::PublicUuid::from_uuid(self.datastore_id),
            catalog.asset.asset_id,
        );
        request.source.asset = ahri_tre_protocol::asset::AssetSelector::Id {
            asset: pinned
                .map(|version| {
                    ahri_tre_protocol::ingest::version_ref(
                        ahri_tre_protocol::PublicUuid::from_uuid(self.datastore_id),
                        version,
                    )
                })
                .unwrap_or(source_asset),
            study: Some(canonical(output)),
            asset_type: Some(ahri_tre_types::AssetType::File),
        };
        let mut preconditions = request.mutation.preconditions.clone();
        for condition in &mut preconditions {
            let supported = match condition {
                Precondition::Exists { target } => matches!(target.as_str(), "source" | "dataset"),
                Precondition::DoesNotExist { target } => target == "dataset",
                Precondition::ResourceVersion { target, version } => {
                    if let Some((major, minor, patch)) = parse_semver_selector(version) {
                        *version = format!("{major}.{minor}.{patch}");
                        target == "source"
                    } else {
                        false
                    }
                }
                _ => false,
            };
            if !supported {
                return Err(ProtocolError::new(
                    ProtocolErrorCode::ValidationFailed,
                    "Unsupported materialization precondition",
                ));
            }
        }
        let key = if let Some(key) = &request.mutation.idempotency_key {
            use sha2::{Digest, Sha256};
            // Value serialization sorts object keys and applies DTO defaults;
            // arrays (including preconditions) preserve order and duplicates.
            let mut intent = request.clone();
            intent.session = None;
            intent.mutation.idempotency_key = None;
            intent.mutation.preconditions = preconditions.clone();
            // These source-selector spellings resolve identically on admission.
            // Normalize the selector, never resolve latest again for a retry.
            if let Some(version) = &pinned_version {
                intent.source.asset = ahri_tre_protocol::asset::AssetSelector::Id {
                    asset: source_asset,
                    study: Some(canonical(output)),
                    asset_type: Some(ahri_tre_types::AssetType::File),
                };
                intent.source.version = Some(format!(
                    "{}.{}.{}",
                    version.major, version.minor, version.patch
                ));
            }
            intent.source.version = match intent.source.version.as_deref().map(str::trim) {
                None | Some("") => None,
                Some(value) if value.eq_ignore_ascii_case("latest") => None,
                Some(value) => Some(
                    parse_semver_selector(value)
                        .map(|(major, minor, patch)| format!("{major}.{minor}.{patch}"))
                        .unwrap_or_else(|| value.to_string()),
                ),
            };
            let canonical = serde_json::to_value(intent).map_err(|_| unavailable())?;
            let canonical = serde_json::to_vec(&canonical).map_err(|_| unavailable())?;
            let scope = serde_json::to_vec(&(
                owner,
                self.deployment_id,
                self.datastore_id,
                ahri_tre_protocol::request::kind::INGEST_DATASET_FROM_DATAFILE,
                key.as_str(),
            ))
            .map_err(|_| unavailable())?;
            let key = self
                .repository
                .lock_operation_key(
                    Sha256::digest(scope)
                        .iter()
                        .map(|byte| format!("{byte:02x}"))
                        .collect(),
                    Sha256::digest(canonical)
                        .iter()
                        .map(|byte| format!("{byte:02x}"))
                        .collect(),
                )
                .map_err(|_| unavailable())?;
            if let Some((record, same_intent)) = key.retained().map_err(|_| unavailable())? {
                // Recheck pinned resources before disclosing either replay or conflict.
                let record = self
                    .service()
                    .authorized_operation(self, owner, &record.detail.summary.operation)
                    .await?;
                if !same_intent {
                    let mut error = ProtocolError::new(
                        ProtocolErrorCode::Conflict,
                        "Idempotency key was accepted with different intent",
                    );
                    error.details = Some(
                        ahri_tre_protocol::public_error::ProtocolErrorDetails::IdempotencyConflict,
                    );
                    return Err(error);
                }
                return Ok(OperationAdmission::Retained(Box::new(
                    record.detail.summary,
                )));
            }
            Some(key)
        } else {
            None
        };
        // A retained retry uses its original pinned resources. Only a new
        // admission must establish agreement with a supplied moving latest.
        if pinned_latest
            && pinned
                != catalog
                    .latest_version
                    .as_ref()
                    .map(|version| version.version_id)
        {
            return Err(ProtocolError::new(
                ProtocolErrorCode::Conflict,
                "Version selectors disagree",
            ));
        }
        if pinned.is_some() {
            request.source.version = None;
        }
        Ok(OperationAdmission::New(OperationSubmission {
            key,
            preconditions,
        }))
    }
}

struct TrackedDatafileLifecycle {
    submission: OperationSubmission,
    rejection: Option<ProtocolError>,
    context: OperationContext,
    repository: PgMetadataRepository<'static>,
    id: Uuid,
    datastore_id: Uuid,
    cleanup_succeeded: Option<bool>,
    cancellation_observed: bool,
    authority_lost: bool,
    source: Option<(ahri_tre_types::StudyRecord, DataFileMetadata)>,
    acceptance: Option<OperationSummary>,
    notify: Option<Box<dyn FnOnce(OperationSummary) + Send>>,
}

impl TrackedDatafileLifecycle {
    fn check_session_authority(&mut self) -> Result<(), AppError> {
        match (self.context.authority)() {
            OperationSessionAuthority::Active => {
                if self
                    .repository
                    .operation_stage_authorized(self.id)
                    .unwrap_or(false)
                {
                    return Ok(());
                }
                self.authority_lost = true;
                Err(AppError::Conflict("Session authority has ended".into()))
            }
            OperationSessionAuthority::Lost => {
                self.authority_lost = true;
                Err(AppError::Conflict("Session authority has ended".into()))
            }
            OperationSessionAuthority::Closed => {
                self.cancellation_observed = true;
                update_operation(&self.repository, self.id, |record| {
                    if record.detail.summary.status.is_terminal()
                        || record.detail.summary.status == OperationStatus::CancelRequested
                    {
                        return Ok(false);
                    }
                    record.detail.summary.status = OperationStatus::CancelRequested;
                    record.detail.summary.cancellable = false;
                    let mut entry = event(
                        record.detail.events.items.len() as u64 + 1,
                        Utc::now().to_rfc3339(),
                        "Session close requested cancellation",
                    );
                    entry.kind = OperationEventKind::CancellationRequested;
                    record.detail.events.items.push(entry);
                    Ok(true)
                })?;
                Err(AppError::Conflict(
                    "Operation cancellation requested".into(),
                ))
            }
        }
    }
}

impl AppService {
    /// Executes in the caller's worker, notifying only after durable acceptance.
    pub async fn derive_dataset_from_managed_datafile_tracked(
        &self,
        session: &mut DataStoreSession,
        request: DeriveDatasetFromManagedDataFileRequest,
        mut context: OperationContext,
        submission: OperationSubmission,
        notify: impl FnOnce(OperationSummary) + Send + 'static,
    ) -> Result<StartResult<ahri_tre_protocol::ingest::DatasetMaterializationResponse>, ProtocolError>
    {
        let id = Uuid::new_v4();
        let span = context
            .observation
            .with_operation_id(id)
            .span(ahri_tre_observability::Stage::Application);
        context.observation = span.context();
        span.record_start();
        let result = async {
            session
                .require_operation_recovery()
                .map_err(crate::projections::dataset_materialization_protocol_error)?;
            if (context.authority)() != OperationSessionAuthority::Active {
                return Err(ProtocolError::new(
                    ProtocolErrorCode::Forbidden,
                    "Session authority has ended",
                ));
            }
            let repository = session
                .dataset_admission_repository()
                .ok_or_else(unavailable)?;
            let mut lifecycle = TrackedDatafileLifecycle {
                submission,
                rejection: None,
                context,
                repository,
                id,
                datastore_id: session
                    .runtime()
                    .identity_binding
                    .as_ref()
                    .ok_or_else(unavailable)?
                    .datastore_id,
                source: None,
                cleanup_succeeded: None,
                cancellation_observed: false,
                authority_lost: false,
                acceptance: None,
                notify: Some(Box::new(notify)),
            };
            let result = self
                .derive_dataset_with_lifecycle(session, request, &mut lifecycle)
                .await;
            if let Some(error) = lifecycle.rejection.take() {
                return Err(error);
            }
            let Some(mut record) = lifecycle
                .repository
                .get_operation(lifecycle.id)
                .map_err(|_| unavailable())?
            else {
                return result
                    .and_then(|_| Err(persistence_error()))
                    .map_err(crate::projections::dataset_materialization_protocol_error);
            };
            if let Err(error) = result {
                // Unknown commit cannot be terminalized or compensated until resolved.
                if matches!(error, AppError::DatasetAdmissionOutcomeUnknown(_))
                    && record.detail.summary.status != OperationStatus::Completed
                {
                    return Err(crate::projections::dataset_materialization_protocol_error(
                        error,
                    ));
                }
                let cancelled =
                    lifecycle.cancellation_observed && lifecycle.cleanup_succeeded == Some(true);
                let mut failure = crate::projections::dataset_materialization_protocol_error(error);
                if lifecycle.authority_lost {
                    failure = ProtocolError::new(
                        ProtocolErrorCode::OperationFailed,
                        "Operation interrupted after Session authority ended",
                    )
                    .with_target("interrupted");
                } else if lifecycle.cleanup_succeeded == Some(false) {
                    failure = ProtocolError::new(
                        ProtocolErrorCode::OperationFailed,
                        "Dataset cleanup requires reconciliation",
                    );
                    failure.target = Some("cleanup_failed".into());
                }
                // Existing authorized failure readback retains the acquisition times.
                failure.adapter_observations = session.observations().snapshot(None);
                record = update_operation(&lifecycle.repository, lifecycle.id, |record| {
                    if record.detail.summary.status.is_terminal() {
                        return Ok(false);
                    }
                    terminalize(
                        record,
                        if cancelled {
                            OperationStatus::Cancelled
                        } else {
                            OperationStatus::Failed
                        },
                        if cancelled {
                            None
                        } else {
                            Some(failure.clone())
                        },
                    );
                    Ok(true)
                })
                .map_err(crate::projections::dataset_materialization_protocol_error)?;
            }
            Ok(StartResult::Operation {
                operation: Box::new(record.detail.summary),
            })
        }
        .await;
        use ahri_tre_observability::{FailureCategory, Outcome};
        let (outcome, category) = match &result {
            Ok(StartResult::Operation { operation }) => match operation.status {
                OperationStatus::Completed => (Outcome::Success, None),
                OperationStatus::Cancelled => (Outcome::Cancelled, None),
                OperationStatus::Failed => (
                    Outcome::Unavailable,
                    Some(
                        match operation
                            .final_error
                            .as_ref()
                            .and_then(|error| error.target.as_deref())
                        {
                            Some("cleanup_failed") => FailureCategory::Cleanup,
                            Some("interrupted") => FailureCategory::Authorization,
                            _ => FailureCategory::Dependency,
                        },
                    ),
                ),
                _ => (Outcome::Unavailable, Some(FailureCategory::CommitUnknown)),
            },
            Err(error) => match error.code {
                ProtocolErrorCode::Internal => {
                    (Outcome::InternalFailure, Some(FailureCategory::Internal))
                }
                ProtocolErrorCode::OperationFailed => {
                    (Outcome::Unavailable, Some(FailureCategory::Dependency))
                }
                ProtocolErrorCode::Forbidden => {
                    (Outcome::Rejected, Some(FailureCategory::Authorization))
                }
                _ => (Outcome::Rejected, None),
            },
            Ok(StartResult::Completed { .. }) => (Outcome::Success, None),
        };
        span.finish(outcome, category);
        result
    }

    async fn authorized_operation(
        &self,
        inspection: &OperationControl,
        owner: &str,
        id: &OperationRef,
    ) -> Result<StoredOperation, ProtocolError> {
        if id.kind != ahri_tre_protocol::refs::ObjectKind::Operation {
            return Err(ProtocolError::new(
                ProtocolErrorCode::ValidationFailed,
                "Expected an Operation reference",
            ));
        }
        if id.datastore_id.as_uuid() != inspection.datastore_id {
            return Err(ProtocolError::new(
                ProtocolErrorCode::Conflict,
                "Operation reference belongs to another Datastore",
            ));
        }
        let id = id.id.as_uuid();
        let mut record = inspection
            .repository
            .get_operation(id)
            .map_err(|_| unavailable())?
            .ok_or_else(not_found)?;
        let now = inspection
            .repository
            .operation_now()
            .map_err(|_| unavailable())?;
        if operation_expired(&record.detail.summary, now).map_err(|_| unavailable())? {
            return Err(not_found());
        }
        self.authorize_operation_scope(inspection, owner, &record.scope)
            .await?;
        let dataset_id = record.result.as_ref().map(|result| match result {
            OperationResult::DatasetFromDatafile { data, .. } => data.dataset.dataset_id,
        });
        self.project_result_availability(&mut record.detail.summary, dataset_id, now)
            .await?;
        if record.detail.summary.status.is_terminal()
            && deadline_reached(
                record.detail.summary.retention.events_expires_at.as_deref(),
                now,
            )
            .map_err(|_| unavailable())?
        {
            record.detail.events = Page {
                items: vec![],
                next_cursor: None,
            };
        }
        Ok(record)
    }

    async fn authorize_operation_scope(
        &self,
        inspection: &OperationControl,
        owner: &str,
        scope: &OperationResourceScope,
    ) -> Result<(), ProtocolError> {
        if scope.owner != owner
            || scope.datastore_id != inspection.datastore_id
            || scope.deployment_id != inspection.deployment_id
        {
            return Err(not_found());
        }
        // Study reads enforce current grants through the authenticated metadata adapter.
        if self
            .studies
            .get_study_by_id(scope.study_id)
            .await
            .map_err(|_| unavailable())?
            .is_none()
            || self
                .assets
                .get_datafile(scope.source_version_id)
                .await
                .map_err(|_| unavailable())?
                .is_none()
        {
            return Err(not_found());
        }
        if !self
            .study_domains
            .list_domains_for_study(scope.study_id)
            .await
            .map_err(|_| unavailable())?
            .iter()
            .any(|domain| domain.domain_id == scope.metadata_domain_id)
        {
            return Err(not_found());
        }
        Ok(())
    }

    pub async fn get_operation(
        &self,
        inspection: &OperationControl,
        owner: &str,
        operation: &OperationRef,
        events: Option<&ahri_tre_protocol::pagination::PageRequest>,
    ) -> Result<OperationDetail, ProtocolError> {
        let mut detail = self
            .authorized_operation(inspection, owner, operation)
            .await?
            .detail;
        history::page_events(inspection, owner, &mut detail, events)?;
        Ok(detail)
    }

    async fn cancel_operation(
        &self,
        control: &OperationControl,
        owner: &str,
        operation: &OperationRef,
    ) -> Result<OperationSummary, ProtocolError> {
        loop {
            // Reauthorize on every competing transition; IDs confer no authority.
            let mut record = self.authorized_operation(control, owner, operation).await?;
            let summary = &record.detail.summary;
            if matches!(&record.detail.progress, Some(OperationProgress::Stage { name, .. }) if name == "committing")
            {
                return Err(cancellation_conflict());
            }
            let policy = if summary.operation_kind.is_supported() {
                OperationCancellationPolicy::Cooperative
            } else {
                OperationCancellationPolicy::Unsupported
            };
            let next = policy
                .request(summary.status, summary.cancellable)
                .map_err(|e| *e)?;
            if summary.status == next {
                return Ok(summary.clone());
            }
            let expected = summary.status;
            let sequence = record.detail.events.items.len() as u64;
            record.detail.summary.status = next;
            record.detail.summary.cancellable = false;
            let mut requested = event(
                sequence + 1,
                Utc::now().to_rfc3339(),
                "cancellation requested",
            );
            requested.kind = OperationEventKind::CancellationRequested;
            record.detail.events.items.push(requested);
            match control
                .repository
                .transition_operation(expected, sequence, &record)
            {
                Ok(()) => return Ok(record.detail.summary),
                Err(CoreError::Conflict(_)) => continue,
                Err(_) => return Err(unavailable()),
            }
        }
    }

    pub async fn get_operation_result(
        &self,
        inspection: &OperationControl,
        owner: &str,
        operation: &OperationRef,
    ) -> Result<OperationResult, ProtocolError> {
        let record = self
            .authorized_operation(inspection, owner, operation)
            .await?;
        match record.detail.summary.status {
            OperationStatus::Pending
            | OperationStatus::Running
            | OperationStatus::CancelRequested => Err(ProtocolError::new(
                ProtocolErrorCode::OperationNotReady,
                "Operation result is not ready",
            )),
            OperationStatus::Failed => {
                let mut error = record.detail.summary.final_error.unwrap_or_else(|| {
                    ProtocolError::new(
                        ProtocolErrorCode::OperationFailed,
                        "Dataset materialization failed",
                    )
                });
                error.code = ProtocolErrorCode::OperationFailed;
                Err(error)
            }
            OperationStatus::Cancelled => Err(ProtocolError::new(
                ProtocolErrorCode::Conflict,
                "Operation was cancelled",
            )),
            OperationStatus::Completed => {
                let availability = record.detail.summary.result.as_ref();
                if let Some(reason) = unavailable_result_kind(availability) {
                    return Err(result_unavailable(reason));
                }
                match record.result {
                    Some(OperationResult::DatasetFromDatafile { mut data, .. }) => {
                        // Stored operation JSON is never a retained disclosure
                        // capability. Historical inferred dictionaries stay withheld.
                        for variable in &mut data.variables {
                            variable.vocabulary_items.clear();
                            variable.vocabulary_items_withheld = variable.vocabulary.is_some();
                        }
                        Ok(OperationResult::DatasetFromDatafile {
                            data,
                            retention: record.detail.summary.retention,
                            availability: availability
                                .cloned()
                                .ok_or_else(|| result_unavailable("unavailable"))?,
                        })
                    }
                    None => Err(result_unavailable("unavailable")),
                }
            }
        }
    }
}
impl DatafileLifecycle for TrackedDatafileLifecycle {
    fn observation(&self) -> Option<ahri_tre_observability::CorrelationContext> {
        Some(self.context.observation)
    }

    fn accepted(&mut self) {
        // Release only after the acceptance transaction has committed.
        self.submission.key.take();
        if let (Some(summary), Some(notify)) = (self.acceptance.take(), self.notify.take()) {
            notify(summary);
        }
    }
    fn cleanup_finished(&mut self, succeeded: bool) {
        self.cleanup_succeeded = Some(succeeded);
    }
    fn resolved_source(&mut self, study: &ahri_tre_types::StudyRecord, source: &DataFileMetadata) {
        self.source = Some((study.clone(), source.clone()));
    }
    fn record_acceptance(
        &mut self,
        repository: &PgMetadataRepository<'_>,
        attempt: &ahri_tre_pgmeta::RecoverableDatasetAttempt,
        resolved: &ResolvedDatafileMaterialization,
    ) -> Result<(), CoreError> {
        // Source resolution can block before this transaction. A close or
        // revocation in that interval must roll back acceptance itself.
        if (self.context.authority)() != OperationSessionAuthority::Active {
            return Err(CoreError::Conflict("Session authority has ended".into()));
        }
        let fail = || CoreError::Validation("Operation context is unavailable".into());
        let status = &self.context.session;
        let scope = OperationResourceScope {
            deployment_id: status
                .session
                .as_ref()
                .ok_or_else(fail)?
                .deployment_id
                .as_uuid(),
            datastore_id: self.datastore_id,
            owner: self.context.owner.clone(),
            session_id: status
                .session
                .as_ref()
                .ok_or_else(fail)?
                .session_id
                .as_uuid(),
            study_id: resolved.request.study_id,
            metadata_domain_id: resolved.request.metadata_domain_id.ok_or_else(fail)?,
            source_version_id: resolved.source_version.version_id,
            dataset_name: resolved.request.dataset_name.as_str().into(),
        };
        for condition in &self.submission.preconditions {
            if !repository.operation_precondition_satisfied(&scope, condition)? {
                let mut error = ProtocolError::new(
                    ProtocolErrorCode::PreconditionFailed,
                    "Materialization precondition failed",
                );
                error.details = Some(
                    ahri_tre_protocol::public_error::ProtocolErrorDetails::PreconditionFailed {
                        precondition: serde_json::to_string(condition).map_err(|_| fail())?,
                    },
                );
                self.rejection = Some(error);
                return Err(CoreError::Validation(
                    "Materialization precondition failed".into(),
                ));
            }
        }
        let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Micros, true);
        let summary = OperationSummary {
            operation: OperationRef {
                datastore_id: ahri_tre_protocol::PublicUuid::from_uuid(self.datastore_id),
                kind: ahri_tre_protocol::refs::ObjectKind::Operation,
                id: ahri_tre_protocol::PublicUuid::from_uuid(self.id),
            },
            operation_kind: OperationKind::new(
                ahri_tre_protocol::request::kind::INGEST_DATASET_FROM_DATAFILE,
            ),
            operation_scope: OperationScope::Datastore,
            status: OperationStatus::Pending,
            cancellable: true,
            started_by_request_id: self.context.request_id,
            timestamps: OperationTimestamps {
                created_at: now.clone(),
                started_at: None,
                finished_at: None,
            },
            retention: OperationRetention::default(),
            result: None,
            final_error: None,
            warnings: vec![],
        };
        let record = StoredOperation {
            scope: scope.clone(),
            detail: OperationDetail {
                summary,
                progress: Some(Stage::Queued.progress()),
                events: Page {
                    items: vec![Stage::Queued.event(1, now, OperationEventKind::StatusChanged)],
                    next_cursor: None,
                },
            },
            result: None,
        };
        repository.create_operation(attempt.attempt_id, &record)?;
        if let Some(key) = &self.submission.key {
            key.bind(repository, self.id)?;
        }
        self.acceptance = Some(record.detail.summary);
        Ok(())
    }
    fn checkpoint(
        &mut self,
        stage: DatafileStage,
        _: &ResolvedDatafileMaterialization,
    ) -> Result<(), AppError> {
        let record = self
            .repository
            .get_operation(self.id)
            .map_err(|_| persistence_error())?
            .ok_or_else(persistence_error)?;
        self.check_session_authority()?;
        if record.detail.summary.status == OperationStatus::CancelRequested {
            self.cancellation_observed = true;
            return Err(AppError::Conflict(
                "Operation cancellation requested".into(),
            ));
        }
        if matches!(
            stage,
            DatafileStage::BeforeLoad | DatafileStage::AfterPreparation
        ) {
            update_operation(&self.repository, self.id, |record| {
                // Progress must not overwrite a cancellation accepted at this checkpoint.
                if record.detail.summary.status == OperationStatus::CancelRequested {
                    self.cancellation_observed = true;
                    return Err(AppError::Conflict(
                        "Operation cancellation requested".into(),
                    ));
                }
                stage_event(
                    record,
                    if stage == DatafileStage::BeforeLoad {
                        Stage::Materializing
                    } else {
                        Stage::Preparing
                    },
                    OperationEventKind::ProgressUpdated,
                );
                Ok(true)
            })?;
        }
        if stage == DatafileStage::BeforeMetadataCommit {
            // The last cooperative checkpoint has passed. A cancellation accepted
            // concurrently may lose to authoritative admission. Close the window
            // in a short conditional update before starting the admission transaction.
            update_operation(&self.repository, self.id, |record| {
                if record.detail.summary.status.is_terminal() {
                    return Err(persistence_error());
                }
                record.detail.summary.cancellable = false;
                stage_event(
                    record,
                    Stage::Committing,
                    OperationEventKind::ProgressUpdated,
                );
                Ok(true)
            })?;
        }
        Ok(())
    }
    fn claim_execution(&mut self) -> Result<(), AppError> {
        let repository = self.repository.clone();
        update_operation(&repository, self.id, |record| {
            self.check_session_authority()?;
            if record.detail.summary.status == OperationStatus::CancelRequested {
                self.cancellation_observed = true;
                return Err(AppError::Conflict(
                    "Operation cancellation requested".into(),
                ));
            }
            if record.detail.summary.status != OperationStatus::Pending {
                return Err(persistence_error());
            }
            let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Micros, true);
            record.detail.summary.status = OperationStatus::Running;
            record.detail.summary.timestamps.started_at = Some(now.clone());
            record.detail.progress = Some(Stage::Preparing.progress());
            record.detail.events.items.push(Stage::Preparing.event(
                2,
                now,
                OperationEventKind::StatusChanged,
            ));
            Ok(true)
        })
        .map(|_| ())
    }
    fn cleanup_started(&mut self) {
        // Failure to publish progress never interrupts compensation or releases ownership.
        let _ = update_operation(&self.repository, self.id, |record| {
            if record.detail.summary.status.is_terminal() {
                return Ok(false);
            }
            record.detail.summary.cancellable = false;
            stage_event(
                record,
                Stage::CleaningUp,
                OperationEventKind::CleanupStarted,
            );
            Ok(true)
        });
    }
    fn record_completion(
        &mut self,
        repository: &PgMetadataRepository<'_>,
        result: &DatasetMaterialization,
    ) -> Result<(), AppError> {
        let mut record = repository
            .get_operation(self.id)
            .map_err(|_| persistence_error())?
            .ok_or_else(persistence_error)?;
        let (study, source) = self.source.clone().ok_or_else(persistence_error)?;
        let mut response = crate::projections::dataset_materialization_response(
            self.context.session.clone(),
            study.name.as_str().into(),
            ManagedDataFileDatasetMaterialization {
                source,
                dataset: result.clone(),
            },
        );
        response.lineage.transformation.description =
            "materialize Dataset from managed Datafile".into();
        response.metadata_warnings = response
            .metadata_warnings
            .iter()
            .take(100)
            .map(|_| "Dataset metadata inference notice".to_string())
            .collect();
        let expected = record.detail.summary.status;
        let sequence = record.detail.events.items.len() as u64;
        terminalize(&mut record, OperationStatus::Completed, None);
        record.detail.summary.result = Some(OperationResultRef::Durable {
            ref_id: self.id.to_string(),
        });
        record.result = Some(OperationResult::DatasetFromDatafile {
            data: Box::new(response),
            retention: record.detail.summary.retention.clone(),
            availability: record
                .detail
                .summary
                .result
                .clone()
                .ok_or_else(persistence_error)?,
        });
        repository
            .transition_operation(expected, sequence, &record)
            .map_err(|_| persistence_error())
    }
}
fn update_operation(
    repository: &PgMetadataRepository<'_>,
    id: Uuid,
    mut update: impl FnMut(&mut StoredOperation) -> Result<bool, AppError>,
) -> Result<StoredOperation, AppError> {
    loop {
        let mut record = repository
            .get_operation(id)
            .map_err(|_| persistence_error())?
            .ok_or_else(persistence_error)?;
        let expected = record.detail.summary.status;
        let sequence = record.detail.events.items.len() as u64;
        if !update(&mut record)? {
            return Ok(record);
        }
        match repository.transition_operation(expected, sequence, &record) {
            Ok(()) => return Ok(record),
            Err(CoreError::Conflict(_)) => continue,
            Err(_) => return Err(persistence_error()),
        }
    }
}
#[derive(Clone, Copy)]
enum Stage {
    Queued,
    Preparing,
    Materializing,
    Committing,
    CleaningUp,
}
impl Stage {
    fn name(self) -> &'static str {
        match self {
            Self::Queued => "queued",
            Self::Preparing => "preparing",
            Self::Materializing => "materializing",
            Self::Committing => "committing",
            Self::CleaningUp => "cleaning up",
        }
    }
    fn progress(self) -> OperationProgress {
        OperationProgress::Stage {
            name: self.name().into(),
            message: None,
        }
    }
    fn event(self, sequence: u64, at: String, kind: OperationEventKind) -> OperationEvent {
        let mut entry = event(sequence, at, self.name());
        entry.kind = kind;
        entry.progress = Some(self.progress());
        entry
    }
}
fn stage_event(record: &mut StoredOperation, stage: Stage, kind: OperationEventKind) {
    record.detail.progress = Some(stage.progress());
    record.detail.events.items.push(stage.event(
        record.detail.events.items.len() as u64 + 1,
        Utc::now().to_rfc3339(),
        kind,
    ));
}
fn cancellation_conflict() -> ProtocolError {
    ProtocolError::new(
        ProtocolErrorCode::Conflict,
        "Operation cannot be cancelled in its current state",
    )
}
fn terminalize(
    record: &mut StoredOperation,
    status: OperationStatus,
    error: Option<ProtocolError>,
) {
    let now = Utc::now();
    let expiry = (now + Duration::days(30)).to_rfc3339();
    record.detail.summary.status = status;
    record.detail.summary.cancellable = false;
    record.detail.progress = None;
    record.detail.summary.final_error = error;
    record.detail.summary.timestamps.finished_at = Some(now.to_rfc3339());
    record.detail.summary.retention = OperationRetention {
        operation_expires_at: Some(expiry.clone()),
        events_expires_at: Some(expiry.clone()),
        result_expires_at: Some(expiry),
    };
    let sequence = record.detail.events.items.len() as u64 + 1;
    record.detail.events.items.push(event(
        sequence,
        now.to_rfc3339(),
        if status == OperationStatus::Completed {
            "completed"
        } else if status == OperationStatus::Cancelled {
            "cancelled"
        } else {
            "failed"
        },
    ));
}
fn event(sequence: u64, at: String, message: &str) -> OperationEvent {
    OperationEvent {
        event_sequence: sequence,
        event_at: at,
        kind: OperationEventKind::StatusChanged,
        message: Some(message.into()),
        target: None,
        progress: None,
        warning: None,
    }
}
fn persistence_error() -> AppError {
    AppError::Infrastructure("Operation persistence is unavailable".into())
}
fn unavailable() -> ProtocolError {
    ProtocolError::new(
        ProtocolErrorCode::OperationFailed,
        "Operation persistence is unavailable",
    )
}
fn not_found() -> ProtocolError {
    ProtocolError::new(ProtocolErrorCode::NotFound, "Operation not found")
}
fn result_unavailable(availability: &str) -> ProtocolError {
    let mut error = ProtocolError::new(
        ProtocolErrorCode::OperationResultUnavailable,
        format!("Operation result is {availability}"),
    );
    error.details = Some(
        ahri_tre_protocol::public_error::ProtocolErrorDetails::OperationResultUnavailable {
            availability: availability.into(),
        },
    );
    error
}

/// Projection policy supplied to the adapter's stopped-attempt recovery boundary.
pub(crate) fn interrupt_operation(record: &mut StoredOperation) {
    terminalize(
        record,
        OperationStatus::Failed,
        Some(
            ProtocolError::new(
                ProtocolErrorCode::OperationFailed,
                "Operation interrupted after Session authority ended",
            )
            .with_target("interrupted"),
        ),
    );
}

fn deadline_reached(
    deadline: Option<&str>,
    now: chrono::DateTime<Utc>,
) -> Result<bool, chrono::ParseError> {
    deadline
        .map(|value| chrono::DateTime::parse_from_rfc3339(value).map(|value| now >= value))
        .transpose()
        .map(|expired| expired.unwrap_or(false))
}
fn operation_expired(
    summary: &OperationSummary,
    now: chrono::DateTime<Utc>,
) -> Result<bool, chrono::ParseError> {
    Ok(summary.status.is_terminal()
        && deadline_reached(summary.retention.operation_expires_at.as_deref(), now)?)
}

fn unavailable_result_kind(reference: Option<&OperationResultRef>) -> Option<&'static str> {
    match reference {
        Some(OperationResultRef::Durable { .. } | OperationResultRef::Temporary { .. }) => None,
        Some(OperationResultRef::Expired) => Some("expired"),
        Some(OperationResultRef::Removed) => Some("removed"),
        Some(OperationResultRef::Unavailable) | None => Some("unavailable"),
    }
}
impl AppService {
    async fn project_result_availability(
        &self,
        summary: &mut OperationSummary,
        dataset_id: Option<ahri_tre_types::VersionId>,
        now: chrono::DateTime<Utc>,
    ) -> Result<(), ProtocolError> {
        if summary.status != OperationStatus::Completed {
            return Ok(());
        }
        if deadline_reached(summary.retention.result_expires_at.as_deref(), now)
            .map_err(|_| unavailable())?
        {
            summary.result = Some(OperationResultRef::Expired);
        } else if unavailable_result_kind(summary.result.as_ref()).is_some() {
            summary
                .result
                .get_or_insert(OperationResultRef::Unavailable);
        } else if let Some(id) = dataset_id {
            if self
                .assets
                .get_dataset(id)
                .await
                .map_err(|_| unavailable())?
                .is_none()
                || self
                    .assets
                    .get_dataset_version_withdrawal(id)
                    .await
                    .map_err(|_| unavailable())?
                    .is_some()
            {
                summary.result = Some(OperationResultRef::Unavailable);
            }
        } else {
            summary.result = Some(OperationResultRef::Unavailable);
        }
        Ok(())
    }
}