notification/

plugin.rs

use std::{cell::RefCell, fmt, rc::Rc, time::Duration as StdDuration};

use chrono::{DateTime, Datelike, SecondsFormat, Utc};
use lenso::prelude::*;
use lenso_capability_email_dispatch as email;
use lenso_capability_email_dispatch::{
    DispatchRequest, DispatchResponseOutcome, EmailDispatchInvocationError,
};
use lenso_capability_notification_admin as admin;
use lenso_capability_notification_delivery as delivery;
use lenso_capability_notification_template as notification_template;
use lenso_capability_notification_transactional as transactional;
use lenso_capability_secrets::{ResolveRequest, SecretsInvocationError};
use lenso_postgres_kit::OwnedPostgres;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use zeroize::Zeroizing;

use crate::contracts::{
    DispatchOutcome, EMAIL_DISPATCH_OBSERVED_EVENT, EMAIL_RECEIPT_OBSERVED_EVENT,
    EmailDispatchObserved, EmailReceiptObserved, ORGANIZATION_INVITATION_ACCEPTED_EVENT,
    ORGANIZATION_INVITATION_EXPIRED_EVENT, ORGANIZATION_INVITATION_REVOKED_EVENT,
    OrganizationInvitationLifecycle, ReceiptKind, RemoteReceiptSummary, SanitizedFailure,
};
use crate::domain::MAX_SAFE_WIRE_INTEGER;
use crate::error::{ErrorCode, NotificationError};
use crate::events::{NotificationEventApplier, ObservationEnvelope};
use crate::migrations::schema_plan;
use crate::operator::verify_managed_catalog;
use crate::public::{
    AccessRequestNotificationEvent, AccessRequestNotificationTemplateV1, AccessRequestRoleV1,
    AccessRequestScopeV1, CreateAccessRequestNotificationIntent, CreateTransactionalEmailIntent,
    EmailRecipient, IntentSource, OrganizationInvitationTemplateV1, RenderedTemplate,
    access_request_template_id, create_access_request_notification_in_tx,
    create_transactional_email_intent_in_tx, find_access_request_notification_replay,
    find_transactional_email_intent_replay,
};
use crate::repository::{
    ADMIN_ATTEMPT_LIMIT, ADMIN_RECEIPT_LIMIT, ADMIN_RETRY_REQUEST_LIMIT, AttemptRecord,
    DeliveryDetail, DeliverySummary, PostgresNotificationRepository, ReceiptRecord, RetryRecord,
    RetryResult,
};
use crate::runtime::{DispatchWork, claim_one_due};
use crate::snapshot::AeadSnapshotProtector;

const DEPENDENCY_TIMEOUT: StdDuration = StdDuration::from_secs(10);
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct NotificationConfig {
    schema: String,
    database_url_secret: String,
    snapshot_key_secret: String,
    transactional_callers: Vec<String>,
    dispatch_callers: Vec<String>,
    receipt_callers: Vec<String>,
    admin_callers: Vec<String>,
}

impl NotificationConfig {
    pub fn new(
        database_url_secret: impl Into<String>,
        snapshot_key_secret: impl Into<String>,
        transactional_callers: Vec<String>,
        dispatch_callers: Vec<String>,
        receipt_callers: Vec<String>,
        admin_callers: Vec<String>,
    ) -> Result<Self, NotificationConfigError> {
        let config = Self {
            schema: "notification".to_owned(),
            database_url_secret: database_url_secret.into(),
            snapshot_key_secret: snapshot_key_secret.into(),
            transactional_callers,
            dispatch_callers,
            receipt_callers,
            admin_callers,
        };
        config.validate()?;
        Ok(config)
    }

    fn validate(&self) -> Result<(), NotificationConfigError> {
        if self.schema != "notification" {
            return Err(NotificationConfigError::InvalidSchema);
        }
        if !valid_secret_reference(&self.database_url_secret)
            || !valid_secret_reference(&self.snapshot_key_secret)
            || self.database_url_secret == self.snapshot_key_secret
        {
            return Err(NotificationConfigError::InvalidSecretReference);
        }
        for callers in [
            &self.transactional_callers,
            &self.dispatch_callers,
            &self.receipt_callers,
            &self.admin_callers,
        ] {
            if callers.is_empty()
                || callers.len() > 64
                || callers.iter().any(|caller| !valid_instance(caller))
                || callers
                    .iter()
                    .enumerate()
                    .any(|(index, caller)| callers[..index].contains(caller))
            {
                return Err(NotificationConfigError::InvalidCallers);
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Error, Eq, PartialEq)]
pub enum NotificationConfigError {
    #[error("the legacy Notification schema identity must remain `notification`")]
    InvalidSchema,
    #[error("database and snapshot keys require distinct valid secret references")]
    InvalidSecretReference,
    #[error("each authority role requires unique valid caller Instance keys")]
    InvalidCallers,
}

fn validate_config(config: &NotificationConfig) -> Result<(), RuntimeFailure> {
    config
        .validate()
        .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
            detail: error.to_string(),
        })
}

#[lenso::plugin(
    lifecycle,
    configuration_schema = "configuration.schema.json",
    validate = validate_config
)]
#[derive(Clone)]
struct NotificationPlugin {
    #[config]
    config: NotificationConfig,
    secrets: Port<lenso_capability_secrets::SecretsClient>,
    email: Port<email::EmailDispatchClient>,
    templates: Port<notification_template::NotificationTemplateClient>,
    state: Rc<RefCell<Option<PreparedNotification>>>,
}

#[derive(Clone)]
struct PreparedNotification {
    postgres: OwnedPostgres,
    protector: AeadSnapshotProtector,
    email_provider_instance: String,
}

impl fmt::Debug for PreparedNotification {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("PreparedNotification")
            .field("schema", &self.postgres.schema())
            .field("email_provider_instance", &self.email_provider_instance)
            .finish_non_exhaustive()
    }
}

impl fmt::Debug for NotificationPlugin {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("NotificationPlugin")
            .field("prepared", &self.state.borrow().is_some())
            .field(
                "transactional_caller_count",
                &self.config.transactional_callers.len(),
            )
            .field("dispatch_caller_count", &self.config.dispatch_callers.len())
            .field("receipt_caller_count", &self.config.receipt_callers.len())
            .field("admin_caller_count", &self.config.admin_callers.len())
            .finish_non_exhaustive()
    }
}

#[lenso::provides(transactional::Transactional, delivery::Delivery, admin::Admin)]
impl NotificationPlugin {
    async fn create_organization_invitation(
        &self,
        context: Ctx,
        request: transactional::CreateOrganizationInvitationRequest,
    ) -> PluginResult<
        transactional::CreateOrganizationInvitationResponse,
        transactional::CreateOrganizationInvitationError,
    > {
        let Some(caller) =
            authorized_caller(&context, &self.config.transactional_callers).map(str::to_owned)
        else {
            return Err(PluginError::domain(
                transactional::CreateOrganizationInvitationError::Unauthorized,
            ));
        };
        let now = Utc::now();
        if !valid_create_request(&request, now) {
            return Err(PluginError::domain(
                transactional::CreateOrganizationInvitationError::InvalidIntent,
            ));
        }
        let expires_at = parse_time(&request.template.expires_at).map_err(|_| {
            PluginError::domain(transactional::CreateOrganizationInvitationError::InvalidIntent)
        })?;
        let locale = match request.recipient.locale {
            transactional::CreateOrganizationInvitationRequestRecipientLocale::En => "en",
            transactional::CreateOrganizationInvitationRequestRecipientLocale::EnUS => "en-US",
        };
        let intent = CreateTransactionalEmailIntent {
            source: intent_source(caller, request.source),
            recipient: EmailRecipient {
                address: request.recipient.address,
                display_name: request.recipient.display_name,
                locale: locale.to_owned(),
            },
            template: OrganizationInvitationTemplateV1 {
                organization_id: request.template.organization_id,
                organization_name: request.template.organization_name,
                invitation_id: request.template.invitation_id,
                invitation_url: request.template.invitation_url,
                inviter_display_name: request.template.inviter_display_name,
                role_name: request.template.role_name,
                expires_at,
            },
            idempotency_key: request.idempotency_key,
            correlation_id: request.correlation_id,
            causation_id: request.causation_id,
            requested_by: request.requested_by,
        };
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        if let Some(receipt) =
            find_transactional_email_intent_replay(prepared.postgres.pool(), &intent, now)
                .await
                .map_err(map_create_error)?
        {
            return Ok(transactional::CreateOrganizationInvitationResponse {
                delivery_id: receipt.delivery_id,
                idempotent_replay: true,
                intent_id: receipt.intent_id,
                status: transactional::CreateOrganizationInvitationResponseStatus::Queued,
            });
        }
        let rendered = self
            .render_organization_invitation(context, &intent)
            .await
            .map_err(PluginError::runtime)?;
        let mut transaction = prepared
            .postgres
            .pool()
            .begin()
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        let receipt = create_transactional_email_intent_in_tx(
            &mut transaction,
            &intent,
            &rendered,
            now,
            &prepared.protector,
        )
        .await
        .map_err(map_create_error)?;
        transaction
            .commit()
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        Ok(transactional::CreateOrganizationInvitationResponse {
            delivery_id: receipt.delivery_id,
            idempotent_replay: receipt.idempotent_replay,
            intent_id: receipt.intent_id,
            status: transactional::CreateOrganizationInvitationResponseStatus::Queued,
        })
    }

    async fn create_access_request_notification(
        &self,
        context: Ctx,
        request: transactional::CreateAccessRequestNotificationRequest,
    ) -> PluginResult<
        transactional::CreateAccessRequestNotificationResponse,
        transactional::CreateAccessRequestNotificationError,
    > {
        let Some(caller) =
            authorized_caller(&context, &self.config.transactional_callers).map(str::to_owned)
        else {
            return Err(PluginError::domain(
                transactional::CreateAccessRequestNotificationError::Unauthorized,
            ));
        };
        let now = Utc::now();
        if !valid_access_request_notification_request(&request, now) {
            return Err(PluginError::domain(
                transactional::CreateAccessRequestNotificationError::InvalidIntent,
            ));
        }
        let event = match request.event {
            transactional::CreateAccessRequestNotificationRequestEvent::Submitted => {
                AccessRequestNotificationEvent::Submitted
            }
            transactional::CreateAccessRequestNotificationRequestEvent::Approved => {
                AccessRequestNotificationEvent::Approved
            }
            transactional::CreateAccessRequestNotificationRequestEvent::Denied => {
                AccessRequestNotificationEvent::Denied
            }
            transactional::CreateAccessRequestNotificationRequestEvent::Expiring => {
                AccessRequestNotificationEvent::Expiring
            }
        };
        let expires_at = request
            .expires_at
            .as_deref()
            .map(parse_time)
            .transpose()
            .map_err(|_| {
                PluginError::domain(
                    transactional::CreateAccessRequestNotificationError::InvalidIntent,
                )
            })?;
        let locale = match request.recipient.locale {
            transactional::CreateAccessRequestNotificationRequestRecipientLocale::En => "en",
            transactional::CreateAccessRequestNotificationRequestRecipientLocale::EnUS => "en-US",
        };
        let intent = CreateAccessRequestNotificationIntent {
            source: IntentSource {
                module_id: caller,
                entity_type: "access_request".to_owned(),
                entity_id: request.request_id.clone(),
            },
            recipient: EmailRecipient {
                address: request.recipient.address,
                display_name: request.recipient.display_name,
                locale: locale.to_owned(),
            },
            template: AccessRequestNotificationTemplateV1 {
                request_id: request.request_id,
                organization_id: request.organization_id,
                event,
                role: AccessRequestRoleV1 {
                    role_id: request.role.role_id,
                    display_name: request.role.display_name,
                },
                scope: AccessRequestScopeV1 {
                    kind: request.scope.kind,
                    id: request.scope.id,
                    display_name: request.scope.display_name,
                },
                expires_at,
            },
            idempotency_key: request.idempotency_key,
            correlation_id: request.correlation_id,
            causation_id: request.causation_id,
            requested_by: request.requested_by,
        };
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        if let Some(receipt) =
            find_access_request_notification_replay(prepared.postgres.pool(), &intent, now)
                .await
                .map_err(map_access_request_create_error)?
        {
            return Ok(transactional::CreateAccessRequestNotificationResponse {
                delivery_id: receipt.delivery_id,
                idempotent_replay: true,
                intent_id: receipt.intent_id,
                status: transactional::CreateAccessRequestNotificationResponseStatus::Queued,
            });
        }
        let rendered = self
            .render_access_request_notification(context, &intent)
            .await
            .map_err(PluginError::runtime)?;
        let mut transaction = prepared
            .postgres
            .pool()
            .begin()
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        let receipt = create_access_request_notification_in_tx(
            &mut transaction,
            &intent,
            &rendered,
            now,
            &prepared.protector,
        )
        .await
        .map_err(map_access_request_create_error)?;
        transaction
            .commit()
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        Ok(transactional::CreateAccessRequestNotificationResponse {
            delivery_id: receipt.delivery_id,
            idempotent_replay: receipt.idempotent_replay,
            intent_id: receipt.intent_id,
            status: transactional::CreateAccessRequestNotificationResponseStatus::Queued,
        })
    }

    async fn observe_invitation_lifecycle(
        &self,
        context: Ctx,
        request: transactional::ObserveInvitationLifecycleRequest,
    ) -> PluginResult<
        transactional::ObserveInvitationLifecycleResponse,
        transactional::ObserveInvitationLifecycleError,
    > {
        let Some(caller) =
            authorized_caller(&context, &self.config.transactional_callers).map(str::to_owned)
        else {
            return Err(PluginError::domain(
                transactional::ObserveInvitationLifecycleError::Unauthorized,
            ));
        };
        if !valid_lifecycle_request(&request) {
            return Err(PluginError::domain(
                transactional::ObserveInvitationLifecycleError::InvalidObservation,
            ));
        }
        let observed_at = parse_time(&request.observed_at).map_err(|_| {
            PluginError::domain(transactional::ObserveInvitationLifecycleError::InvalidObservation)
        })?;
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let event_name = match request.lifecycle {
            transactional::ObserveInvitationLifecycleRequestLifecycle::Accepted => {
                ORGANIZATION_INVITATION_ACCEPTED_EVENT
            }
            transactional::ObserveInvitationLifecycleRequestLifecycle::Expired => {
                ORGANIZATION_INVITATION_EXPIRED_EVENT
            }
            transactional::ObserveInvitationLifecycleRequestLifecycle::Revoked => {
                ORGANIZATION_INVITATION_REVOKED_EVENT
            }
        };
        let payload = OrganizationInvitationLifecycle {
            invitation_id: request.invitation_id.clone(),
            organization_id: request.organization_id,
            observed_at,
        };
        apply_observation(
            prepared.postgres.pool(),
            ObservationEnvelope {
                id: request.observation_id,
                event_name: event_name.to_owned(),
                event_version: 1,
                source_module: caller,
                aggregate_id: request.invitation_id,
                occurred_at: observed_at,
                payload: serde_json::to_value(payload)
                    .map_err(|error| PluginError::runtime(runtime(error)))?,
            },
        )
        .await
        .map_err(map_lifecycle_error)?;
        Ok(transactional::ObserveInvitationLifecycleResponse { recorded: true })
    }

    async fn dispatch_due(
        &self,
        context: Ctx,
        _request: delivery::DispatchDueRequest,
    ) -> PluginResult<delivery::DispatchDueResponse, delivery::DispatchDueError> {
        if !authorized(&context, &self.config.dispatch_callers) {
            return Err(PluginError::domain(
                delivery::DispatchDueError::Unauthorized,
            ));
        }
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let now = Utc::now();
        let Some(work) = claim_one_due(prepared.postgres.pool(), &prepared.protector, now)
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?
        else {
            return Err(PluginError::domain(
                delivery::DispatchDueError::NoDeliveryDue,
            ));
        };
        let dispatch = dispatch_request(&work);
        let response = match self.email.dispatch(dispatch).await {
            Ok(response) => response,
            Err(EmailDispatchInvocationError::Domain(email::DispatchError::InvalidDispatch)) => {
                let observed = permanent_rejection(
                    &work,
                    &prepared.email_provider_instance,
                    "email_invalid_dispatch",
                    now,
                );
                apply_dispatch(prepared.postgres.pool(), &work, &observed)
                    .await
                    .map_err(|error| PluginError::runtime(runtime(error)))?;
                return Err(PluginError::domain(
                    delivery::DispatchDueError::DispatchRejected,
                ));
            }
            Err(EmailDispatchInvocationError::Domain(email::DispatchError::UnsupportedMessage)) => {
                let observed = permanent_rejection(
                    &work,
                    &prepared.email_provider_instance,
                    "email_unsupported_message",
                    now,
                );
                apply_dispatch(prepared.postgres.pool(), &work, &observed)
                    .await
                    .map_err(|error| PluginError::runtime(runtime(error)))?;
                return Err(PluginError::domain(
                    delivery::DispatchDueError::DispatchRejected,
                ));
            }
            Err(EmailDispatchInvocationError::Domain(email::DispatchError::Unknown(_))) => {
                let observed = unknown_dispatch(
                    &work,
                    &prepared.email_provider_instance,
                    now,
                    "email_dispatch_unknown_domain_error",
                );
                apply_dispatch(prepared.postgres.pool(), &work, &observed)
                    .await
                    .map_err(|error| PluginError::runtime(runtime(error)))?;
                return Err(PluginError::runtime(email_protocol_violation()));
            }
            Err(EmailDispatchInvocationError::Runtime(error)) => {
                let observed = unknown_dispatch(
                    &work,
                    &prepared.email_provider_instance,
                    now,
                    "email_dispatch_runtime_failure",
                );
                apply_dispatch(prepared.postgres.pool(), &work, &observed)
                    .await
                    .map_err(|storage| PluginError::runtime(runtime(storage)))?;
                return Err(PluginError::runtime(error));
            }
        };
        let observed =
            match dispatch_observation(&work, &prepared.email_provider_instance, response) {
                Ok(observed) => observed,
                Err(error) => {
                    let observed = unknown_dispatch(
                        &work,
                        &prepared.email_provider_instance,
                        now,
                        "email_dispatch_protocol_failure",
                    );
                    apply_dispatch(prepared.postgres.pool(), &work, &observed)
                        .await
                        .map_err(|storage| PluginError::runtime(runtime(storage)))?;
                    return Err(PluginError::runtime(error));
                }
            };
        apply_dispatch(prepared.postgres.pool(), &work, &observed)
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        let detail = PostgresNotificationRepository::from_pool(prepared.postgres.pool().clone())
            .get_delivery(&work.claim.delivery_id)
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?
            .ok_or_else(|| PluginError::runtime(runtime("claimed delivery disappeared")))?;
        let next_attempt_at = detail
            .delivery
            .next_attempt_at
            .map(format_time)
            .transpose()
            .map_err(PluginError::runtime)?;
        let observed_at = format_time(observed.observed_at).map_err(PluginError::runtime)?;
        Ok(delivery::DispatchDueResponse {
            attempt_id: work.claim.attempt_id,
            delivery_id: work.claim.delivery_id,
            next_attempt_at,
            observed_at,
            run_id: work.claim.run_id,
            status: delivery_status(&detail.delivery.status).map_err(PluginError::runtime)?,
        })
    }

    async fn observe_receipt(
        &self,
        context: Ctx,
        request: delivery::ObserveReceiptRequest,
    ) -> PluginResult<delivery::ObserveReceiptResponse, delivery::ObserveReceiptError> {
        let Some(caller) =
            authorized_caller(&context, &self.config.receipt_callers).map(str::to_owned)
        else {
            return Err(PluginError::domain(
                delivery::ObserveReceiptError::Unauthorized,
            ));
        };
        if !valid_receipt_request(&request) {
            return Err(PluginError::domain(
                delivery::ObserveReceiptError::InvalidReceipt,
            ));
        }
        let observed_at = parse_time(&request.observed_at)
            .map_err(|_| PluginError::domain(delivery::ObserveReceiptError::InvalidReceipt))?;
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let kind = match request.kind {
            delivery::ObserveReceiptRequestKind::Delivered => ReceiptKind::Delivered,
            delivery::ObserveReceiptRequestKind::Bounced => ReceiptKind::Bounced,
            delivery::ObserveReceiptRequestKind::Rejected => ReceiptKind::Rejected,
        };
        let aggregate_id = request.delivery_id.clone();
        let payload = receipt_observation(caller.clone(), &request, kind, observed_at);
        apply_observation(
            prepared.postgres.pool(),
            ObservationEnvelope {
                id: request.observation_id,
                event_name: EMAIL_RECEIPT_OBSERVED_EVENT.to_owned(),
                event_version: 1,
                source_module: caller,
                aggregate_id,
                occurred_at: observed_at,
                payload: serde_json::to_value(payload)
                    .map_err(|error| PluginError::runtime(runtime(error)))?,
            },
        )
        .await
        .map_err(map_receipt_error)?;
        Ok(delivery::ObserveReceiptResponse { recorded: true })
    }

    async fn list_deliveries(
        &self,
        context: Ctx,
        request: admin::ListDeliveriesRequest,
    ) -> PluginResult<admin::ListDeliveriesResponse, admin::ListDeliveriesError> {
        if !authorized(&context, &self.config.admin_callers) {
            return Err(PluginError::domain(
                admin::ListDeliveriesError::Unauthorized,
            ));
        }
        if !valid_list_request(&request) {
            return Err(PluginError::domain(
                admin::ListDeliveriesError::InvalidFilter,
            ));
        }
        let limit = request.limit.unwrap_or(100);
        let status = request.status.as_ref().map(admin_status);
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let rows = PostgresNotificationRepository::from_pool(prepared.postgres.pool().clone())
            .list_deliveries(limit + 1, request.cursor.as_deref(), status)
            .await
            .map_err(|error| PluginError::runtime(runtime(error)))?;
        let has_more = i64::try_from(rows.len()).is_ok_and(|count| count > limit);
        let records = rows
            .into_iter()
            .take(usize::try_from(limit).unwrap_or_default())
            .map(admin_delivery)
            .collect::<Result<Vec<_>, _>>()
            .map_err(PluginError::runtime)?;
        let next_cursor = has_more
            .then(|| records.last().map(|record| record.id.clone()))
            .flatten();
        Ok(admin::ListDeliveriesResponse {
            next_cursor,
            records,
        })
    }

    async fn get_delivery(
        &self,
        context: Ctx,
        request: admin::GetDeliveryRequest,
    ) -> PluginResult<admin::GetDeliveryResponse, admin::GetDeliveryError> {
        if !authorized(&context, &self.config.admin_callers) {
            return Err(PluginError::domain(admin::GetDeliveryError::Unauthorized));
        }
        if !required_bounded(&request.delivery_id, 1, 160) {
            return Err(PluginError::domain(admin::GetDeliveryError::InvalidRequest));
        }
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let detail = PostgresNotificationRepository::from_pool(prepared.postgres.pool().clone())
            .get_delivery(&request.delivery_id)
            .await
            .map_err(map_get_error)?
            .ok_or_else(|| PluginError::domain(admin::GetDeliveryError::DeliveryNotFound))?;
        admin_detail(detail).map_err(PluginError::runtime)
    }

    async fn retry_delivery(
        &self,
        context: Ctx,
        request: admin::RetryDeliveryRequest,
    ) -> PluginResult<admin::RetryDeliveryResponse, admin::RetryDeliveryError> {
        let Some(caller) =
            authorized_caller(&context, &self.config.admin_callers).map(str::to_owned)
        else {
            return Err(PluginError::domain(admin::RetryDeliveryError::Unauthorized));
        };
        if !valid_retry_request(&request) {
            return Err(PluginError::domain(
                admin::RetryDeliveryError::InvalidRequest,
            ));
        }
        if request.revision == MAX_SAFE_WIRE_INTEGER {
            return Err(PluginError::domain(
                admin::RetryDeliveryError::RetryNotAllowed,
            ));
        }
        let prepared = self.prepared().map_err(PluginError::runtime)?;
        let result = PostgresNotificationRepository::from_pool(prepared.postgres.pool().clone())
            .request_manual_retry(
                &request.delivery_id,
                request.revision,
                &request.idempotency_key,
                &caller,
                Utc::now(),
            )
            .await
            .map_err(map_retry_error)?;
        admin_retry(result).map_err(PluginError::runtime)
    }
}

impl NotificationPlugin {
    fn prepared(&self) -> Result<PreparedNotification, RuntimeFailure> {
        self.state
            .borrow()
            .clone()
            .ok_or_else(|| RuntimeFailure::PluginFailure {
                detail: "Notification Plugin is not prepared".to_owned(),
            })
    }

    async fn render_organization_invitation(
        &self,
        context: Ctx,
        intent: &CreateTransactionalEmailIntent,
    ) -> Result<RenderedTemplate, RuntimeFailure> {
        self.templates
            .render_with_context(context, organization_invitation_render_request(intent))
            .await
            .map(rendered_template)
            .map_err(map_template_render_error)
    }

    async fn render_access_request_notification(
        &self,
        context: Ctx,
        intent: &CreateAccessRequestNotificationIntent,
    ) -> Result<RenderedTemplate, RuntimeFailure> {
        self.templates
            .render_with_context(context, access_request_render_request(intent))
            .await
            .map(rendered_template)
            .map_err(map_template_render_error)
    }
}

fn organization_invitation_render_request(
    intent: &CreateTransactionalEmailIntent,
) -> notification_template::RenderRequest {
    notification_template::RenderRequest {
        template_id: crate::public::ORGANIZATION_INVITATION_TEMPLATE_ID.to_owned(),
        version: Some(crate::public::ORGANIZATION_INVITATION_TEMPLATE_VERSION.to_owned()),
        locale: intent.recipient.locale.clone(),
        variables: render_variables([
            (
                "expires_at",
                format_template_time(intent.template.expires_at),
            ),
            ("invitation_url", intent.template.invitation_url.clone()),
            (
                "inviter_display_name",
                trimmed_optional(intent.template.inviter_display_name.as_deref()),
            ),
            ("locale", intent.recipient.locale.clone()),
            (
                "organization_name",
                intent.template.organization_name.trim().to_owned(),
            ),
            (
                "recipient_display_name",
                trimmed_optional(intent.recipient.display_name.as_deref()),
            ),
            (
                "role_name",
                trimmed_optional(intent.template.role_name.as_deref()),
            ),
        ]),
    }
}

fn access_request_render_request(
    intent: &CreateAccessRequestNotificationIntent,
) -> notification_template::RenderRequest {
    let role = intent
        .template
        .role
        .display_name
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or(&intent.template.role.role_id)
        .to_owned();
    let scope = intent
        .template
        .scope
        .display_name
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or(&intent.template.scope.id)
        .to_owned();
    let mut variables = vec![
        render_variable("locale", intent.recipient.locale.clone()),
        render_variable("organization_id", intent.template.organization_id.clone()),
        render_variable(
            "recipient_display_name",
            trimmed_optional(intent.recipient.display_name.as_deref()),
        ),
        render_variable("request_id", intent.template.request_id.clone()),
        render_variable("role", role),
        render_variable("scope", scope),
        render_variable("scope_kind", intent.template.scope.kind.clone()),
    ];
    if intent.template.event != AccessRequestNotificationEvent::Denied {
        variables.push(render_variable(
            "expires_at",
            intent
                .template
                .expires_at
                .map(format_template_time)
                .unwrap_or_default(),
        ));
    }
    variables.sort_by(|left, right| left.name.cmp(&right.name));
    notification_template::RenderRequest {
        template_id: access_request_template_id(intent.template.event).to_owned(),
        version: Some(crate::public::ACCESS_REQUEST_TEMPLATE_VERSION.to_owned()),
        locale: intent.recipient.locale.clone(),
        variables,
    }
}

fn render_variables<const N: usize>(
    variables: [(&str, String); N],
) -> Vec<notification_template::RenderRequestVariablesItem> {
    variables
        .into_iter()
        .map(|(name, value)| render_variable(name, value))
        .collect()
}

fn render_variable(name: &str, value: String) -> notification_template::RenderRequestVariablesItem {
    notification_template::RenderRequestVariablesItem {
        name: name.to_owned(),
        value,
    }
}

fn trimmed_optional(value: Option<&str>) -> String {
    value.map(str::trim).unwrap_or_default().to_owned()
}

fn format_template_time(value: DateTime<Utc>) -> String {
    value.to_rfc3339_opts(SecondsFormat::Secs, true)
}

fn rendered_template(response: notification_template::RenderResponse) -> RenderedTemplate {
    RenderedTemplate {
        template_id: response.template_id,
        template_version: response.version,
        requested_locale: response.requested_locale,
        resolved_locale: response.resolved_locale,
        fallback_used: response.fallback_used,
        renderer_identity: response.renderer_identity,
        template_digest: response.template_digest,
        content_digest: response.content_digest,
        subject: response.subject,
        text: response.text,
        html: response.html,
    }
}

fn map_template_render_error(
    error: notification_template::NotificationTemplateRenderInvocationError,
) -> RuntimeFailure {
    match error {
        notification_template::NotificationTemplateRenderInvocationError::Runtime(error) => error,
        notification_template::NotificationTemplateRenderInvocationError::Domain(
            notification_template::RenderError::NotFound,
        ) => RuntimeFailure::PluginFailure {
            detail: "required Notification Template version is unavailable".to_owned(),
        },
        notification_template::NotificationTemplateRenderInvocationError::Domain(
            notification_template::RenderError::Unauthorized,
        ) => RuntimeFailure::PluginFailure {
            detail:
                "Notification is not authorized to render through its configured Template Provider"
                    .to_owned(),
        },
        notification_template::NotificationTemplateRenderInvocationError::Domain(_) => {
            RuntimeFailure::ProtocolViolation {
                capability: notification_template::CAPABILITY_ID,
            }
        }
    }
}

fn authorized(context: &Ctx, allowed: &[String]) -> bool {
    authorized_caller(context, allowed).is_some()
}

fn authorized_caller<'a>(context: &'a Ctx, allowed: &[String]) -> Option<&'a str> {
    context
        .caller_instance()
        .filter(|caller| allowed.iter().any(|allowed| allowed == caller))
}

fn required_bounded(value: &str, minimum: usize, maximum: usize) -> bool {
    let length = value.chars().count();
    !value.trim().is_empty() && (minimum..=maximum).contains(&length)
}

fn optional_bounded(value: Option<&str>, maximum: usize) -> bool {
    value.is_none_or(|value| value.chars().count() <= maximum)
}

fn valid_digest(value: &str) -> bool {
    value.strip_prefix("sha256:").is_some_and(|digest| {
        digest.len() == 64
            && digest
                .bytes()
                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    })
}

fn valid_create_request(
    request: &transactional::CreateOrganizationInvitationRequest,
    now: DateTime<Utc>,
) -> bool {
    required_bounded(&request.source.entity_type, 1, 160)
        && request.source.entity_type == "organization_invitation"
        && required_bounded(&request.source.entity_id, 1, 240)
        && required_bounded(&request.recipient.address, 3, 320)
        && request.recipient.address.contains('@')
        && optional_bounded(request.recipient.display_name.as_deref(), 240)
        && required_bounded(&request.template.organization_id, 1, 240)
        && required_bounded(&request.template.organization_name, 1, 240)
        && required_bounded(&request.template.invitation_id, 1, 240)
        && request.source.entity_id == request.template.invitation_id
        && required_bounded(&request.template.invitation_url, 1, 4_096)
        && (request.template.invitation_url.starts_with("https://")
            || request
                .template
                .invitation_url
                .starts_with("http://localhost"))
        && optional_bounded(request.template.inviter_display_name.as_deref(), 240)
        && optional_bounded(request.template.role_name.as_deref(), 160)
        && valid_time(&request.template.expires_at)
        && parse_time(&request.template.expires_at).is_ok_and(|expires_at| expires_at > now)
        && required_bounded(&request.idempotency_key, 1, 240)
        && required_bounded(&request.correlation_id, 1, 240)
        && optional_bounded(request.causation_id.as_deref(), 240)
        && optional_bounded(request.requested_by.as_deref(), 240)
}

fn valid_access_request_notification_request(
    request: &transactional::CreateAccessRequestNotificationRequest,
    now: DateTime<Utc>,
) -> bool {
    let event = match &request.event {
        transactional::CreateAccessRequestNotificationRequestEvent::Submitted => "submitted",
        transactional::CreateAccessRequestNotificationRequestEvent::Approved => "approved",
        transactional::CreateAccessRequestNotificationRequestEvent::Denied => "denied",
        transactional::CreateAccessRequestNotificationRequestEvent::Expiring => "expiring",
    };
    let expected_idempotency_key = format!("access-request:{}:{event}", request.request_id);
    let expiry = request
        .expires_at
        .as_deref()
        .and_then(|value| parse_time(value).ok());
    let expiry_valid = match &request.event {
        transactional::CreateAccessRequestNotificationRequestEvent::Expiring => {
            expiry.is_some_and(|value| value > now)
        }
        transactional::CreateAccessRequestNotificationRequestEvent::Denied => {
            request.expires_at.is_none()
        }
        _ => request.expires_at.is_none() || expiry.is_some_and(|value| value > now),
    };
    required_bounded(&request.request_id, 1, 160)
        && required_bounded(&request.organization_id, 1, 240)
        && required_bounded(&request.recipient.address, 3, 320)
        && request.recipient.address.contains('@')
        && optional_bounded(request.recipient.display_name.as_deref(), 240)
        && required_bounded(&request.role.role_id, 1, 160)
        && optional_bounded(request.role.display_name.as_deref(), 160)
        && required_bounded(&request.scope.kind, 1, 160)
        && required_bounded(&request.scope.id, 1, 240)
        && optional_bounded(request.scope.display_name.as_deref(), 240)
        && request.expires_at.as_deref().is_none_or(valid_time)
        && expiry_valid
        && request.idempotency_key == expected_idempotency_key
        && required_bounded(&request.correlation_id, 1, 240)
        && optional_bounded(request.causation_id.as_deref(), 240)
        && optional_bounded(request.requested_by.as_deref(), 240)
}

fn valid_lifecycle_request(request: &transactional::ObserveInvitationLifecycleRequest) -> bool {
    required_bounded(&request.observation_id, 1, 240)
        && required_bounded(&request.organization_id, 1, 240)
        && required_bounded(&request.invitation_id, 1, 240)
        && valid_time(&request.observed_at)
}

fn valid_receipt_request(request: &delivery::ObserveReceiptRequest) -> bool {
    required_bounded(&request.observation_id, 1, 240)
        && required_bounded(&request.delivery_id, 1, 160)
        && required_bounded(&request.attempt_id, 1, 160)
        && required_bounded(&request.run_id, 1, 160)
        && valid_time(&request.observed_at)
        && required_bounded(&request.remote_id, 1, 320)
        && valid_digest(&request.digest)
}

fn valid_list_request(request: &admin::ListDeliveriesRequest) -> bool {
    request.limit.is_none_or(|limit| (1..=200).contains(&limit))
        && request
            .cursor
            .as_deref()
            .is_none_or(|cursor| required_bounded(cursor, 1, 160))
}

fn valid_retry_request(request: &admin::RetryDeliveryRequest) -> bool {
    required_bounded(&request.delivery_id, 1, 160)
        && (1..=MAX_SAFE_WIRE_INTEGER).contains(&request.revision)
        && required_bounded(&request.idempotency_key, 1, 240)
}

fn intent_source(
    caller: String,
    source: transactional::CreateOrganizationInvitationRequestSource,
) -> IntentSource {
    IntentSource {
        module_id: caller,
        entity_type: source.entity_type,
        entity_id: source.entity_id,
    }
}

fn receipt_observation(
    caller: String,
    request: &delivery::ObserveReceiptRequest,
    kind: ReceiptKind,
    observed_at: DateTime<Utc>,
) -> EmailReceiptObserved {
    EmailReceiptObserved {
        delivery_id: request.delivery_id.clone(),
        attempt_id: request.attempt_id.clone(),
        function_run_id: request.run_id.clone(),
        kind,
        source: caller,
        observed_at,
        remote_id: request.remote_id.clone(),
        digest: request.digest.clone(),
    }
}

impl Lifecycle for NotificationPlugin {
    async fn prepare(&self, context: PrepareContext) -> Result<(), RuntimeFailure> {
        let dependencies = context.dependencies().clone();
        let cancellation = context.cancellation();
        let email_providers = email::EmailDispatchClient::many_from_dependencies(&dependencies)?;
        let [email_provider] = email_providers.as_slice() else {
            return Err(RuntimeFailure::InvalidResolvedPlan {
                detail: "Notification requires exactly one lenso.email-dispatch@1 Provider"
                    .to_owned(),
            });
        };
        let email_provider_instance = email_provider.provider_instance().to_owned();
        let secrets = lenso_capability_secrets::SecretsClient::from_dependencies(&dependencies)?;
        let database_context =
            dependencies.invocation_context_after(DEPENDENCY_TIMEOUT, cancellation.clone())?;
        let database_url =
            resolve_secret(&secrets, database_context, &self.config.database_url_secret).await?;
        let snapshot_context =
            dependencies.invocation_context_after(DEPENDENCY_TIMEOUT, cancellation)?;
        let snapshot_key =
            resolve_secret(&secrets, snapshot_context, &self.config.snapshot_key_secret).await?;
        let protector = AeadSnapshotProtector::from_base64_key(
            &snapshot_key,
            self.config.snapshot_key_secret.clone(),
        )
        .map_err(runtime)?;
        let postgres = OwnedPostgres::prepare(
            &database_url,
            schema_plan(self.config.schema.clone()).map_err(|error| {
                RuntimeFailure::InvalidResolvedPlan {
                    detail: error.to_string(),
                }
            })?,
        )
        .await
        .map_err(runtime)?;
        if let Err(error) = verify_managed_catalog(postgres.pool()).await {
            postgres.pool().close().await;
            return Err(runtime(error));
        }
        self.state.replace(Some(PreparedNotification {
            postgres,
            protector,
            email_provider_instance,
        }));
        Ok(())
    }

    async fn deactivate(&self, _context: DeactivateContext) -> Result<(), RuntimeFailure> {
        let prepared = self.state.borrow_mut().take();
        if let Some(prepared) = prepared {
            prepared.postgres.pool().close().await;
        }
        Ok(())
    }
}

fn dispatch_request(work: &DispatchWork) -> DispatchRequest {
    work.request.clone()
}

fn dispatch_observation(
    work: &DispatchWork,
    email_provider_instance: &str,
    response: email::DispatchResponse,
) -> Result<EmailDispatchObserved, RuntimeFailure> {
    if !valid_dispatch_response(&response, email_provider_instance) {
        return Err(email_protocol_violation());
    }
    let observed_at = parse_time(&response.observed_at).map_err(|_| email_protocol_violation())?;
    let failure = response
        .failure
        .map(|failure| {
            Ok(SanitizedFailure {
                code: failure.code,
                classification: failure.classification,
                retry_after_ms: failure
                    .retry_after_ms
                    .map(u64::try_from)
                    .transpose()
                    .map_err(|_| email_protocol_violation())?,
            })
        })
        .transpose()?;
    Ok(EmailDispatchObserved {
        delivery_id: work.claim.delivery_id.clone(),
        attempt_id: work.claim.attempt_id.clone(),
        function_run_id: work.claim.run_id.clone(),
        outcome: match response.outcome {
            DispatchResponseOutcome::Accepted => DispatchOutcome::Accepted,
            DispatchResponseOutcome::TemporaryFailure => DispatchOutcome::TemporaryFailure,
            DispatchResponseOutcome::PermanentFailure => DispatchOutcome::PermanentFailure,
            DispatchResponseOutcome::DeliveryUnknown => DispatchOutcome::DeliveryUnknown,
        },
        provider: email_provider_instance.to_owned(),
        observed_at,
        remote_receipt: response.remote_receipt.map(|receipt| RemoteReceiptSummary {
            remote_id: receipt.remote_id,
            source: email_provider_instance.to_owned(),
            digest: receipt.digest,
        }),
        failure,
    })
}

fn valid_dispatch_response(
    response: &email::DispatchResponse,
    email_provider_instance: &str,
) -> bool {
    if response.provider != email_provider_instance || !valid_time(&response.observed_at) {
        return false;
    }
    let remote_receipt_valid = response.remote_receipt.as_ref().is_none_or(|receipt| {
        required_bounded(&receipt.remote_id, 1, 320)
            && receipt.source == email_provider_instance
            && valid_digest(&receipt.digest)
    });
    if !remote_receipt_valid {
        return false;
    }
    match &response.outcome {
        DispatchResponseOutcome::Accepted => response.failure.is_none(),
        DispatchResponseOutcome::TemporaryFailure => {
            response.remote_receipt.is_none()
                && response.failure.as_ref().is_some_and(|failure| {
                    valid_dispatch_failure(failure, "temporary_failure")
                        && failure
                            .retry_after_ms
                            .is_none_or(|delay| (0..=86_400_000).contains(&delay))
                })
        }
        DispatchResponseOutcome::PermanentFailure => {
            response.remote_receipt.is_none()
                && response.failure.as_ref().is_some_and(|failure| {
                    valid_dispatch_failure(failure, "permanent_failure")
                        && failure.retry_after_ms.is_none()
                })
        }
        DispatchResponseOutcome::DeliveryUnknown => {
            response.remote_receipt.is_none()
                && response.failure.as_ref().is_some_and(|failure| {
                    valid_dispatch_failure(failure, "delivery_unknown")
                        && failure.retry_after_ms.is_none()
                })
        }
    }
}

fn valid_dispatch_failure(failure: &email::DispatchResponseFailure, classification: &str) -> bool {
    required_bounded(&failure.code, 1, 160)
        && required_bounded(&failure.classification, 1, 160)
        && failure.classification == classification
}

fn email_protocol_violation() -> RuntimeFailure {
    RuntimeFailure::ProtocolViolation {
        capability: email::CAPABILITY_ID,
    }
}

fn permanent_rejection(
    work: &DispatchWork,
    email_provider_instance: &str,
    code: &str,
    observed_at: DateTime<Utc>,
) -> EmailDispatchObserved {
    EmailDispatchObserved {
        delivery_id: work.claim.delivery_id.clone(),
        attempt_id: work.claim.attempt_id.clone(),
        function_run_id: work.claim.run_id.clone(),
        outcome: DispatchOutcome::PermanentFailure,
        provider: email_provider_instance.to_owned(),
        observed_at,
        remote_receipt: None,
        failure: Some(SanitizedFailure {
            code: code.to_owned(),
            classification: "permanent_failure".to_owned(),
            retry_after_ms: None,
        }),
    }
}

fn unknown_dispatch(
    work: &DispatchWork,
    email_provider_instance: &str,
    observed_at: DateTime<Utc>,
    code: &str,
) -> EmailDispatchObserved {
    EmailDispatchObserved {
        delivery_id: work.claim.delivery_id.clone(),
        attempt_id: work.claim.attempt_id.clone(),
        function_run_id: work.claim.run_id.clone(),
        outcome: DispatchOutcome::DeliveryUnknown,
        provider: email_provider_instance.to_owned(),
        observed_at,
        remote_receipt: None,
        failure: Some(SanitizedFailure {
            code: code.to_owned(),
            classification: "delivery_unknown".to_owned(),
            retry_after_ms: None,
        }),
    }
}

async fn apply_dispatch(
    pool: &sqlx::PgPool,
    work: &DispatchWork,
    observed: &EmailDispatchObserved,
) -> Result<(), NotificationError> {
    apply_observation(
        pool,
        ObservationEnvelope {
            id: format!("dispatch-observation:{}", work.claim.attempt_id),
            event_name: EMAIL_DISPATCH_OBSERVED_EVENT.to_owned(),
            event_version: 1,
            source_module: observed.provider.clone(),
            aggregate_id: work.claim.delivery_id.clone(),
            occurred_at: observed.observed_at,
            payload: serde_json::to_value(observed).map_err(|error| {
                NotificationError::new(
                    ErrorCode::Internal,
                    "Notification dispatch observation encoding failed",
                )
                .with_source(error)
            })?,
        },
    )
    .await
}

async fn apply_observation(
    pool: &sqlx::PgPool,
    envelope: ObservationEnvelope,
) -> Result<(), NotificationError> {
    NotificationEventApplier::new(pool.clone())
        .apply(&envelope)
        .await
}

fn delivery_status(status: &str) -> Result<delivery::DispatchDueResponseStatus, RuntimeFailure> {
    match status {
        "accepted" => Ok(delivery::DispatchDueResponseStatus::Accepted),
        "retry_scheduled" => Ok(delivery::DispatchDueResponseStatus::RetryScheduled),
        "failed" => Ok(delivery::DispatchDueResponseStatus::Failed),
        "delivery_unknown" => Ok(delivery::DispatchDueResponseStatus::DeliveryUnknown),
        other => Err(runtime(format!(
            "stored delivery status `{other}` is invalid after dispatch"
        ))),
    }
}

fn admin_detail(value: DeliveryDetail) -> Result<admin::GetDeliveryResponse, RuntimeFailure> {
    if value.attempts.len() > ADMIN_ATTEMPT_LIMIT
        || value.receipts.len() > ADMIN_RECEIPT_LIMIT
        || value.retry_requests.len() > ADMIN_RETRY_REQUEST_LIMIT
    {
        return Err(invalid_admin_projection());
    }
    let correlation_id = value.delivery.correlation_id.clone();
    Ok(admin::GetDeliveryResponse {
        attempts: value
            .attempts
            .into_iter()
            .map(admin_attempt)
            .collect::<Result<Vec<_>, _>>()?,
        delivery: admin_delivery(value.delivery)?,
        open_in_story_correlation_id: correlation_id,
        receipts: value
            .receipts
            .into_iter()
            .map(admin_receipt)
            .collect::<Result<Vec<_>, _>>()?,
        retry_requests: value
            .retry_requests
            .into_iter()
            .map(admin_retry_record)
            .collect::<Result<Vec<_>, _>>()?,
    })
}

fn admin_delivery(value: DeliverySummary) -> Result<admin::Delivery, RuntimeFailure> {
    if !required_bounded(&value.id, 1, 160)
        || !required_bounded(&value.recipient_mask, 3, 320)
        || !required_bounded(&value.template_id, 1, 160)
        || !required_bounded(&value.template_version, 1, 80)
        || !required_bounded(&value.locale, 2, 32)
        || !(1..=MAX_SAFE_WIRE_INTEGER).contains(&value.revision)
        || !(0..=10).contains(&value.attempt_count)
        || !(1..=10).contains(&value.max_attempts)
        || value.attempt_count > value.max_attempts
        || value.redacted_preview.chars().count() > 160
        || !valid_digest(&value.content_digest)
        || !required_bounded(&value.correlation_id, 1, 240)
        || !optional_bounded(value.final_reason.as_deref(), 160)
    {
        return Err(invalid_admin_projection());
    }
    let status = match value.status.as_str() {
        "queued" => admin::DeliveryStatus::Queued,
        "attempting" => admin::DeliveryStatus::Attempting,
        "accepted" => admin::DeliveryStatus::Accepted,
        "retry_scheduled" => admin::DeliveryStatus::RetryScheduled,
        "delivered" => admin::DeliveryStatus::Delivered,
        "failed" => admin::DeliveryStatus::Failed,
        "delivery_unknown" => admin::DeliveryStatus::DeliveryUnknown,
        other => {
            let _ = other;
            return Err(invalid_admin_projection());
        }
    };
    Ok(admin::Delivery {
        attempt_count: i64::from(value.attempt_count),
        content_digest: value.content_digest,
        correlation_id: value.correlation_id,
        created_at: format_time(value.created_at)?,
        final_reason: value.final_reason,
        id: value.id,
        locale: value.locale,
        max_attempts: i64::from(value.max_attempts),
        next_attempt_at: value.next_attempt_at.map(format_time).transpose()?,
        recipient_mask: value.recipient_mask,
        redacted_preview: value.redacted_preview,
        retry_now_eligible: value.status == "retry_scheduled"
            && value.attempt_count < value.max_attempts,
        revision: value.revision,
        status,
        template_id: value.template_id,
        template_version: value.template_version,
        updated_at: format_time(value.updated_at)?,
    })
}

fn admin_attempt(value: AttemptRecord) -> Result<admin::Attempt, RuntimeFailure> {
    if !required_bounded(&value.id, 1, 160)
        || !(1..=10).contains(&value.sequence)
        || !required_bounded(&value.function_run_id, 1, 160)
        || !optional_bounded(value.provider.as_deref(), 160)
        || !optional_bounded(value.remote_receipt_id.as_deref(), 320)
        || !optional_bounded(value.failure_code.as_deref(), 160)
        || !optional_bounded(value.failure_classification.as_deref(), 160)
    {
        return Err(invalid_admin_projection());
    }
    let status = match value.status.as_str() {
        "dispatching" => admin::AttemptStatus::Dispatching,
        "accepted" => admin::AttemptStatus::Accepted,
        "temporary_failure" => admin::AttemptStatus::TemporaryFailure,
        "permanent_failure" => admin::AttemptStatus::PermanentFailure,
        "delivery_unknown" => admin::AttemptStatus::DeliveryUnknown,
        _ => return Err(invalid_admin_projection()),
    };
    Ok(admin::Attempt {
        completed_at: value.completed_at.map(format_time).transpose()?,
        failure_classification: value.failure_classification,
        failure_code: value.failure_code,
        id: value.id,
        provider: value.provider,
        remote_receipt_id: value.remote_receipt_id,
        run_id: value.function_run_id,
        sequence: i64::from(value.sequence),
        started_at: format_time(value.started_at)?,
        status,
    })
}

fn admin_receipt(value: ReceiptRecord) -> Result<admin::Receipt, RuntimeFailure> {
    if !required_bounded(&value.id, 1, 160)
        || !required_bounded(&value.attempt_id, 1, 160)
        || !required_bounded(&value.source, 1, 160)
        || !required_bounded(&value.remote_id, 1, 320)
        || !valid_digest(&value.digest)
    {
        return Err(invalid_admin_projection());
    }
    let kind = match value.kind.as_str() {
        "accepted" => admin::ReceiptKind::Accepted,
        "delivered" => admin::ReceiptKind::Delivered,
        "bounced" => admin::ReceiptKind::Bounced,
        "rejected" => admin::ReceiptKind::Rejected,
        _ => return Err(invalid_admin_projection()),
    };
    Ok(admin::Receipt {
        attempt_id: value.attempt_id,
        digest: value.digest,
        id: value.id,
        kind,
        observed_at: format_time(value.observed_at)?,
        remote_id: value.remote_id,
        source: value.source,
    })
}

fn admin_retry_record(value: RetryRecord) -> Result<admin::RetryRecord, RuntimeFailure> {
    if !required_bounded(&value.id, 1, 160)
        || !optional_bounded(value.requested_by.as_deref(), 240)
        || !(1..=MAX_SAFE_WIRE_INTEGER).contains(&value.source_revision)
        || !optional_bounded(value.reason.as_deref(), 160)
    {
        return Err(invalid_admin_projection());
    }
    let kind = match value.kind.as_str() {
        "automatic" => admin::RetryRecordKind::Automatic,
        "manual" => admin::RetryRecordKind::Manual,
        _ => return Err(invalid_admin_projection()),
    };
    let decision = match value.decision.as_str() {
        "scheduled" => admin::RetryRecordDecision::Scheduled,
        "rejected" => admin::RetryRecordDecision::Rejected,
        _ => return Err(invalid_admin_projection()),
    };
    Ok(admin::RetryRecord {
        created_at: format_time(value.created_at)?,
        decision,
        id: value.id,
        kind,
        reason: value.reason,
        requested_by: value.requested_by,
        scheduled_at: value.scheduled_at.map(format_time).transpose()?,
        source_revision: value.source_revision,
    })
}

fn admin_retry(value: RetryResult) -> Result<admin::RetryDeliveryResponse, RuntimeFailure> {
    if !required_bounded(&value.delivery_id, 1, 160)
        || !(1..=MAX_SAFE_WIRE_INTEGER).contains(&value.revision)
        || value.status != "retry_scheduled"
    {
        return Err(invalid_admin_projection());
    }
    Ok(admin::RetryDeliveryResponse {
        delivery_id: value.delivery_id,
        idempotent_replay: value.idempotent_replay,
        revision: value.revision,
        scheduled_at: format_time(value.scheduled_at)?,
        status: admin::RetryDeliveryResponseStatus::RetryScheduled,
    })
}

fn invalid_admin_projection() -> RuntimeFailure {
    runtime("stored Notification Admin projection violates its bounded Capability Contract")
}

fn admin_status(status: &admin::DeliveryStatus) -> &'static str {
    match status {
        admin::DeliveryStatus::Queued => "queued",
        admin::DeliveryStatus::Attempting => "attempting",
        admin::DeliveryStatus::Accepted => "accepted",
        admin::DeliveryStatus::RetryScheduled => "retry_scheduled",
        admin::DeliveryStatus::Delivered => "delivered",
        admin::DeliveryStatus::Failed => "failed",
        admin::DeliveryStatus::DeliveryUnknown => "delivery_unknown",
    }
}

fn map_create_error(
    error: NotificationError,
) -> PluginError<transactional::CreateOrganizationInvitationError> {
    match error.code {
        ErrorCode::Validation => {
            PluginError::domain(transactional::CreateOrganizationInvitationError::InvalidIntent)
        }
        ErrorCode::Conflict => PluginError::domain(
            transactional::CreateOrganizationInvitationError::IdempotencyConflict,
        ),
        ErrorCode::NotFound | ErrorCode::EvidenceOverflow | ErrorCode::Internal => {
            PluginError::runtime(runtime(error))
        }
    }
}

fn map_access_request_create_error(
    error: NotificationError,
) -> PluginError<transactional::CreateAccessRequestNotificationError> {
    match error.code {
        ErrorCode::Validation => {
            PluginError::domain(transactional::CreateAccessRequestNotificationError::InvalidIntent)
        }
        ErrorCode::Conflict => PluginError::domain(
            transactional::CreateAccessRequestNotificationError::IdempotencyConflict,
        ),
        ErrorCode::NotFound | ErrorCode::EvidenceOverflow | ErrorCode::Internal => {
            PluginError::runtime(runtime(error))
        }
    }
}

fn map_lifecycle_error(
    error: NotificationError,
) -> PluginError<transactional::ObserveInvitationLifecycleError> {
    match error.code {
        ErrorCode::Validation => {
            PluginError::domain(transactional::ObserveInvitationLifecycleError::InvalidObservation)
        }
        ErrorCode::Conflict => {
            PluginError::domain(transactional::ObserveInvitationLifecycleError::ObservationConflict)
        }
        ErrorCode::NotFound | ErrorCode::EvidenceOverflow | ErrorCode::Internal => {
            PluginError::runtime(runtime(error))
        }
    }
}

fn map_receipt_error(error: NotificationError) -> PluginError<delivery::ObserveReceiptError> {
    match error.code {
        ErrorCode::Validation => PluginError::domain(delivery::ObserveReceiptError::InvalidReceipt),
        ErrorCode::NotFound => PluginError::domain(delivery::ObserveReceiptError::DeliveryNotFound),
        ErrorCode::Conflict => PluginError::domain(delivery::ObserveReceiptError::ReceiptConflict),
        ErrorCode::EvidenceOverflow | ErrorCode::Internal => PluginError::runtime(runtime(error)),
    }
}

fn map_get_error(error: NotificationError) -> PluginError<admin::GetDeliveryError> {
    match error.code {
        ErrorCode::EvidenceOverflow => {
            PluginError::domain(admin::GetDeliveryError::EvidenceOverflow)
        }
        ErrorCode::Validation | ErrorCode::NotFound | ErrorCode::Conflict | ErrorCode::Internal => {
            PluginError::runtime(runtime(error))
        }
    }
}

fn map_retry_error(error: NotificationError) -> PluginError<admin::RetryDeliveryError> {
    match error.code {
        ErrorCode::NotFound => PluginError::domain(admin::RetryDeliveryError::DeliveryNotFound),
        ErrorCode::Conflict if error.message().contains("revision is stale") => {
            PluginError::domain(admin::RetryDeliveryError::StaleRevision)
        }
        ErrorCode::Conflict if error.message().contains("not eligible") => {
            PluginError::domain(admin::RetryDeliveryError::RetryNotAllowed)
        }
        ErrorCode::Conflict => PluginError::domain(admin::RetryDeliveryError::IdempotencyConflict),
        ErrorCode::Validation | ErrorCode::EvidenceOverflow | ErrorCode::Internal => {
            PluginError::runtime(runtime(error))
        }
    }
}

async fn resolve_secret(
    secrets: &lenso_capability_secrets::SecretsClient,
    context: Ctx,
    reference: &str,
) -> Result<Zeroizing<String>, RuntimeFailure> {
    secrets
        .resolve_with_context(
            context,
            ResolveRequest {
                reference: reference.to_owned(),
            },
        )
        .await
        .map(|response| Zeroizing::new(response.value))
        .map_err(|error| match error {
            SecretsInvocationError::Domain(_) => RuntimeFailure::PluginFailure {
                detail: format!("secret `{reference}` was rejected"),
            },
            SecretsInvocationError::Runtime(error) => error,
        })
}

fn valid_secret_reference(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 256
        && !value.starts_with('/')
        && !value.ends_with('/')
        && value.split('/').all(|segment| {
            !segment.is_empty()
                && segment != "."
                && segment != ".."
                && segment
                    .bytes()
                    .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
        })
}

fn valid_instance(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 160
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b'/'))
}

fn parse_time(value: &str) -> Result<DateTime<Utc>, chrono::ParseError> {
    DateTime::parse_from_rfc3339(value).map(|value| value.with_timezone(&Utc))
}

fn valid_time(value: &str) -> bool {
    required_bounded(value, 1, 64) && parse_time(value).is_ok()
}

pub(crate) fn format_time(value: DateTime<Utc>) -> Result<String, RuntimeFailure> {
    if !(0..=9_999).contains(&value.year()) {
        return Err(runtime(
            "stored Notification timestamp is outside the RFC 3339 four-digit year range",
        ));
    }
    let formatted = value.to_rfc3339_opts(SecondsFormat::Millis, true);
    if !valid_time(&formatted) {
        return Err(runtime(
            "stored Notification timestamp violates its bounded Capability Contract",
        ));
    }
    Ok(formatted)
}

fn runtime(error: impl fmt::Display) -> RuntimeFailure {
    RuntimeFailure::PluginFailure {
        detail: error.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use std::{cell::RefCell, rc::Rc, time::Duration};

    use lenso_app_plan::{
        AppComposition, CapabilityBinding, CapabilityEndpointPlan, CapabilityRequirementPlan,
        PluginInstancePlan, ResolvedAppPlan,
    };
    use lenso_kernel::{
        ActivateContext, CancellationToken, DeterministicDriver, InvocationContext, Kernel,
        NativeRequestEndpoint, NativeRequestFuture, PluginFuture, PluginLifecycle, RuntimeFailure,
        ShutdownOutcome,
    };
    use lenso_native_adapter::{
        NativePluginFactory, NativePluginFactoryContext, NativePluginInstance, NativePluginRegistry,
    };

    use super::*;
    use crate::domain::DeliveryStatus as StoredDeliveryStatus;

    const CONSUMER_PACKAGE_ID: &str = "test.notification-consumer";
    const FIXTURE_PROVIDER_PACKAGE_ID: &str = "test.notification-transactional-provider";
    const EMPTY_PACKAGE_ID: &str = "test.without-notification";
    const SECRETS_PACKAGE_ID: &str = "test.notification-secrets";
    const EMAIL_PACKAGE_ID: &str = "test.notification-email-dispatch";
    const TEMPLATE_PACKAGE_ID: &str = "test.notification-template";

    #[derive(Clone, Copy, Debug)]
    enum FixtureOutcome {
        Success,
        Domain,
        Runtime,
    }

    type InvocationResult = Result<
        transactional::CreateOrganizationInvitationResponse,
        transactional::TransactionalCreateOrganizationInvitationInvocationError,
    >;

    #[derive(Clone, Debug)]
    struct FixtureTransactionalProvider {
        outcome: FixtureOutcome,
    }

    impl transactional::TransactionalProvider for FixtureTransactionalProvider {
        fn create_access_request_notification(
            &self,
            _context: InvocationContext,
            _request: transactional::CreateAccessRequestNotificationRequest,
        ) -> NativeRequestFuture<transactional::TransactionalCreateAccessRequestNotification>
        {
            Box::pin(std::future::ready(Ok(Ok(
                transactional::CreateAccessRequestNotificationResponse {
                    delivery_id: "ntf_dlv_access_fixture".to_owned(),
                    idempotent_replay: false,
                    intent_id: "ntf_int_access_fixture".to_owned(),
                    status: transactional::CreateAccessRequestNotificationResponseStatus::Queued,
                },
            ))))
        }

        fn create_organization_invitation(
            &self,
            _context: InvocationContext,
            _request: transactional::CreateOrganizationInvitationRequest,
        ) -> NativeRequestFuture<transactional::TransactionalCreateOrganizationInvitation> {
            let result = match self.outcome {
                FixtureOutcome::Success => {
                    Ok(Ok(transactional::CreateOrganizationInvitationResponse {
                        delivery_id: "ntf_dlv_fixture".to_owned(),
                        idempotent_replay: false,
                        intent_id: "ntf_int_fixture".to_owned(),
                        status: transactional::CreateOrganizationInvitationResponseStatus::Queued,
                    }))
                }
                FixtureOutcome::Domain => Ok(Err(
                    transactional::CreateOrganizationInvitationError::Unauthorized,
                )),
                FixtureOutcome::Runtime => Err(RuntimeFailure::PluginFailure {
                    detail: "fixture notification storage unavailable".to_owned(),
                }),
            };
            Box::pin(std::future::ready(result))
        }

        fn observe_invitation_lifecycle(
            &self,
            _context: InvocationContext,
            _request: transactional::ObserveInvitationLifecycleRequest,
        ) -> NativeRequestFuture<transactional::TransactionalObserveInvitationLifecycle> {
            Box::pin(std::future::ready(Ok(Ok(
                transactional::ObserveInvitationLifecycleResponse { recorded: true },
            ))))
        }
    }

    #[derive(Clone, Debug)]
    struct FixtureProviderFactory {
        outcome: FixtureOutcome,
    }

    impl NativePluginFactory for FixtureProviderFactory {
        fn package_id(&self) -> &'static str {
            FIXTURE_PROVIDER_PACKAGE_ID
        }

        fn instantiate(
            &self,
            _context: NativePluginFactoryContext<'_>,
        ) -> Result<NativePluginInstance, RuntimeFailure> {
            let endpoint = Rc::new(transactional::TransactionalEndpoint::new(
                FixtureTransactionalProvider {
                    outcome: self.outcome,
                },
            )) as Rc<dyn NativeRequestEndpoint>;
            Ok(NativePluginInstance::new(vec![endpoint]))
        }
    }

    #[derive(Clone, Debug)]
    struct ConsumerFactory {
        observed: Rc<RefCell<Option<InvocationResult>>>,
    }

    impl NativePluginFactory for ConsumerFactory {
        fn package_id(&self) -> &'static str {
            CONSUMER_PACKAGE_ID
        }

        fn instantiate(
            &self,
            _context: NativePluginFactoryContext<'_>,
        ) -> Result<NativePluginInstance, RuntimeFailure> {
            Ok(NativePluginInstance::with_lifecycle(
                Vec::new(),
                ConsumerLifecycle {
                    observed: self.observed.clone(),
                },
            ))
        }
    }

    #[derive(Clone, Debug)]
    struct ConsumerLifecycle {
        observed: Rc<RefCell<Option<InvocationResult>>>,
    }

    impl PluginLifecycle for ConsumerLifecycle {
        fn activate(&self, context: ActivateContext) -> PluginFuture {
            let client =
                transactional::TransactionalClient::from_dependencies(context.dependencies());
            let observed = self.observed.clone();
            Box::pin(async move {
                let client = client?;
                observed.replace(Some(
                    client
                        .create_organization_invitation(generated_request())
                        .await,
                ));
                Ok(())
            })
        }
    }

    #[derive(Clone, Copy, Debug)]
    struct EmptyFactory(&'static str);

    impl NativePluginFactory for EmptyFactory {
        fn package_id(&self) -> &'static str {
            self.0
        }

        fn instantiate(
            &self,
            _context: NativePluginFactoryContext<'_>,
        ) -> Result<NativePluginInstance, RuntimeFailure> {
            Ok(NativePluginInstance::default())
        }
    }

    #[test]
    fn generated_descriptor_declares_business_roles_and_exact_dependencies() {
        let descriptor: serde_json::Value = serde_json::from_str(PLUGIN_DESCRIPTOR_JSON)
            .expect("generated Plugin descriptor must be JSON");
        assert_eq!(PACKAGE_ID, "lenso.notification");
        assert_eq!(descriptor["plugin_id"], PACKAGE_ID);
        assert_eq!(descriptor["root_slot"], "notifications");
        let provided = descriptor["provided_capabilities"]
            .as_array()
            .expect("provided capabilities")
            .iter()
            .map(|entry| entry["capability_id"].as_str().expect("capability id"))
            .collect::<Vec<_>>();
        assert_eq!(
            provided,
            vec![
                transactional::CAPABILITY_ID,
                delivery::CAPABILITY_ID,
                admin::CAPABILITY_ID,
            ]
        );
        let required = descriptor["required_capabilities"]
            .as_array()
            .expect("required capabilities")
            .iter()
            .map(|entry| entry["capability_id"].as_str().expect("capability id"))
            .collect::<Vec<_>>();
        assert_eq!(
            required,
            vec![
                lenso_capability_secrets::CAPABILITY_ID,
                email::CAPABILITY_ID,
                notification_template::CAPABILITY_ID,
            ]
        );
        let linked = NativePluginRegistry::new()
            .with_linked_factories()
            .factories()
            .filter(|factory| factory.package_id() == PACKAGE_ID)
            .count();
        assert_eq!(linked, 1);
    }

    #[test]
    fn template_requests_are_exact_versioned_and_event_typed() {
        let expires_at = parse_time("2026-09-01T00:00:00Z").expect("timestamp");
        let invitation = CreateTransactionalEmailIntent {
            source: IntentSource {
                module_id: "organization-blue".to_owned(),
                entity_type: "organization_invitation".to_owned(),
                entity_id: "invite_1".to_owned(),
            },
            recipient: EmailRecipient {
                address: "member@example.com".to_owned(),
                display_name: Some(" Member ".to_owned()),
                locale: "en-US".to_owned(),
            },
            template: OrganizationInvitationTemplateV1 {
                organization_id: "org_1".to_owned(),
                organization_name: " Acme ".to_owned(),
                invitation_id: "invite_1".to_owned(),
                invitation_url: "https://example.test/invite".to_owned(),
                inviter_display_name: None,
                role_name: Some(" Member ".to_owned()),
                expires_at,
            },
            idempotency_key: "organization-invitation:invite_1".to_owned(),
            correlation_id: "corr_1".to_owned(),
            causation_id: None,
            requested_by: None,
        };
        let request = organization_invitation_render_request(&invitation);
        assert_eq!(request.template_id, "organization-invitation");
        assert_eq!(request.version.as_deref(), Some("v1"));
        assert_eq!(request.locale, "en-US");
        let variables = request
            .variables
            .into_iter()
            .map(|item| (item.name, item.value))
            .collect::<std::collections::BTreeMap<_, _>>();
        assert_eq!(variables.len(), 7);
        assert_eq!(variables["organization_name"], "Acme");
        assert_eq!(variables["recipient_display_name"], "Member");
        assert_eq!(variables["inviter_display_name"], "");
        assert_eq!(variables["expires_at"], "2026-09-01T00:00:00Z");

        let mut access = CreateAccessRequestNotificationIntent {
            source: IntentSource {
                module_id: "access-request-blue".to_owned(),
                entity_type: "access_request".to_owned(),
                entity_id: "ar_1".to_owned(),
            },
            recipient: EmailRecipient {
                address: "member@example.com".to_owned(),
                display_name: None,
                locale: "en".to_owned(),
            },
            template: AccessRequestNotificationTemplateV1 {
                request_id: "ar_1".to_owned(),
                organization_id: "org_1".to_owned(),
                event: AccessRequestNotificationEvent::Submitted,
                role: AccessRequestRoleV1 {
                    role_id: "role_member".to_owned(),
                    display_name: None,
                },
                scope: AccessRequestScopeV1 {
                    kind: "organization".to_owned(),
                    id: "org_1".to_owned(),
                    display_name: None,
                },
                expires_at: Some(expires_at),
            },
            idempotency_key: "access-request:ar_1:submitted".to_owned(),
            correlation_id: "corr_2".to_owned(),
            causation_id: None,
            requested_by: None,
        };
        let submitted = access_request_render_request(&access);
        assert_eq!(submitted.template_id, "access-request-submitted");
        assert!(
            submitted
                .variables
                .iter()
                .any(|item| item.name == "expires_at")
        );

        access.template.event = AccessRequestNotificationEvent::Denied;
        access.template.expires_at = None;
        let denied = access_request_render_request(&access);
        assert_eq!(denied.template_id, "access-request-denied");
        assert!(
            denied
                .variables
                .iter()
                .all(|item| item.name != "expires_at")
        );
    }

    #[test]
    fn template_render_failures_preserve_runtime_and_fail_closed_on_domain_results() {
        let runtime = RuntimeFailure::Unavailable {
            capability: notification_template::CAPABILITY_ID,
        };
        assert!(matches!(
            map_template_render_error(
                notification_template::NotificationTemplateRenderInvocationError::Runtime(runtime)
            ),
            RuntimeFailure::Unavailable { capability }
                if capability == notification_template::CAPABILITY_ID
        ));
        assert!(matches!(
            map_template_render_error(
                notification_template::NotificationTemplateRenderInvocationError::Domain(
                    notification_template::RenderError::NotFound,
                )
            ),
            RuntimeFailure::PluginFailure { .. }
        ));
        assert!(matches!(
            map_template_render_error(
                notification_template::NotificationTemplateRenderInvocationError::Domain(
                    notification_template::RenderError::UnexpectedVariable,
                )
            ),
            RuntimeFailure::ProtocolViolation { capability }
                if capability == notification_template::CAPABILITY_ID
        ));
    }

    #[test]
    fn generated_client_provider_and_plan_preserve_success_domain_and_runtime_lanes() {
        let success = run_generated_fixture(FixtureOutcome::Success);
        assert!(matches!(
            success,
            Ok(transactional::CreateOrganizationInvitationResponse {
                status: transactional::CreateOrganizationInvitationResponseStatus::Queued,
                ..
            })
        ));
        let domain = run_generated_fixture(FixtureOutcome::Domain);
        assert!(matches!(
            domain,
            Err(
                transactional::TransactionalCreateOrganizationInvitationInvocationError::Domain(
                    transactional::CreateOrganizationInvitationError::Unauthorized
                )
            )
        ));
        let runtime = run_generated_fixture(FixtureOutcome::Runtime);
        assert!(matches!(
            runtime,
            Err(
                transactional::TransactionalCreateOrganizationInvitationInvocationError::Runtime(
                    RuntimeFailure::PluginFailure { detail }
                )
            ) if detail == "fixture notification storage unavailable"
        ));
    }

    #[test]
    fn caller_identity_is_derived_and_cannot_be_spoofed_by_payload() {
        let request = generated_request();
        let source = intent_source("organization-blue".to_owned(), request.source);
        assert_eq!(source.module_id, "organization-blue");

        let transactional_schema: serde_json::Value =
            serde_json::from_str(transactional::CREATE_ORGANIZATION_INVITATION_REQUEST_SCHEMA_JSON)
                .expect("transactional request schema");
        assert!(
            transactional_schema["properties"]["source"]["properties"]
                .get("plugin_id")
                .is_none()
        );

        let receipt_request = delivery::ObserveReceiptRequest {
            attempt_id: "attempt".to_owned(),
            delivery_id: "delivery".to_owned(),
            digest: format!("sha256:{}", "a".repeat(64)),
            kind: delivery::ObserveReceiptRequestKind::Delivered,
            observation_id: "observation".to_owned(),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            remote_id: "remote".to_owned(),
            run_id: "run".to_owned(),
        };
        let receipt = receipt_observation(
            "email-provider-blue".to_owned(),
            &receipt_request,
            ReceiptKind::Delivered,
            Utc::now(),
        );
        assert_eq!(receipt.source, "email-provider-blue");
        let delivery_schema: serde_json::Value =
            serde_json::from_str(delivery::OBSERVE_RECEIPT_REQUEST_SCHEMA_JSON)
                .expect("receipt request schema");
        assert!(delivery_schema["properties"].get("source").is_none());
    }

    #[test]
    fn observation_authorities_are_disjoint_per_operation() {
        let plugin = unprepared_plugin();
        assert!(matches!(
            futures::executor::block_on(plugin.dispatch_due(
                caller_context("email-provider-blue"),
                delivery::DispatchDueRequest {},
            )),
            Err(PluginError::Domain(
                delivery::DispatchDueError::Unauthorized
            ))
        ));

        let request = delivery::ObserveReceiptRequest {
            attempt_id: "attempt".to_owned(),
            delivery_id: "delivery".to_owned(),
            digest: format!("sha256:{}", "a".repeat(64)),
            kind: delivery::ObserveReceiptRequestKind::Delivered,
            observation_id: "observation".to_owned(),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            remote_id: "remote".to_owned(),
            run_id: "run".to_owned(),
        };
        assert!(matches!(
            futures::executor::block_on(
                plugin.observe_receipt(caller_context("notification-worker-blue"), request,)
            ),
            Err(PluginError::Domain(
                delivery::ObserveReceiptError::Unauthorized
            ))
        ));

        let lifecycle = transactional::ObserveInvitationLifecycleRequest {
            invitation_id: "invite".to_owned(),
            lifecycle: transactional::ObserveInvitationLifecycleRequestLifecycle::Revoked,
            observation_id: "observation".to_owned(),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            organization_id: "organization".to_owned(),
        };
        assert!(matches!(
            futures::executor::block_on(plugin.observe_invitation_lifecycle(
                caller_context("notification-worker-blue"),
                lifecycle,
            )),
            Err(PluginError::Domain(
                transactional::ObserveInvitationLifecycleError::Unauthorized
            ))
        ));
    }

    #[test]
    fn native_typed_requests_revalidate_every_bound_before_state_or_dependencies() {
        let now = parse_time("2026-08-30T00:00:00Z").expect("fixed validation time");
        for request in invalid_create_requests() {
            assert!(!valid_create_request(&request, now));
            assert!(matches!(
                futures::executor::block_on(
                    unprepared_plugin().create_organization_invitation(
                        caller_context("organization-blue"),
                        request,
                    )
                ),
                Err(PluginError::Domain(
                    transactional::CreateOrganizationInvitationError::InvalidIntent
                ))
            ));
        }

        let mut invalid_access_request = generated_access_request();
        invalid_access_request.idempotency_key = "caller-selected".to_owned();
        assert!(!valid_access_request_notification_request(
            &invalid_access_request,
            now
        ));
        assert!(matches!(
            futures::executor::block_on(unprepared_plugin().create_access_request_notification(
                caller_context("organization-blue"),
                invalid_access_request,
            )),
            Err(PluginError::Domain(
                transactional::CreateAccessRequestNotificationError::InvalidIntent
            ))
        ));

        let lifecycle = transactional::ObserveInvitationLifecycleRequest {
            invitation_id: "invite".to_owned(),
            lifecycle: transactional::ObserveInvitationLifecycleRequestLifecycle::Revoked,
            observation_id: "observation".to_owned(),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            organization_id: "organization".to_owned(),
        };
        assert!(valid_lifecycle_request(&lifecycle));
        for request in [
            transactional::ObserveInvitationLifecycleRequest {
                observation_id: "x".repeat(241),
                ..lifecycle.clone()
            },
            transactional::ObserveInvitationLifecycleRequest {
                organization_id: "x".repeat(241),
                ..lifecycle.clone()
            },
            transactional::ObserveInvitationLifecycleRequest {
                invitation_id: "x".repeat(241),
                ..lifecycle.clone()
            },
            transactional::ObserveInvitationLifecycleRequest {
                observed_at: "not-a-time".to_owned(),
                ..lifecycle
            },
        ] {
            assert!(!valid_lifecycle_request(&request));
            assert!(matches!(
                futures::executor::block_on(
                    unprepared_plugin().observe_invitation_lifecycle(
                        caller_context("organization-blue"),
                        request,
                    )
                ),
                Err(PluginError::Domain(
                    transactional::ObserveInvitationLifecycleError::InvalidObservation
                ))
            ));
        }

        let receipt = delivery::ObserveReceiptRequest {
            attempt_id: "attempt".to_owned(),
            delivery_id: "delivery".to_owned(),
            digest: format!("sha256:{}", "a".repeat(64)),
            kind: delivery::ObserveReceiptRequestKind::Delivered,
            observation_id: "observation".to_owned(),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            remote_id: "remote".to_owned(),
            run_id: "run".to_owned(),
        };
        assert!(valid_receipt_request(&receipt));
        for request in [
            delivery::ObserveReceiptRequest {
                observation_id: "x".repeat(241),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                delivery_id: "x".repeat(161),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                attempt_id: "x".repeat(161),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                run_id: "x".repeat(161),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                observed_at: "not-a-time".to_owned(),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                remote_id: "x".repeat(321),
                ..receipt.clone()
            },
            delivery::ObserveReceiptRequest {
                digest: "sha256:not-a-digest".to_owned(),
                ..receipt
            },
        ] {
            assert!(!valid_receipt_request(&request));
            assert!(matches!(
                futures::executor::block_on(
                    unprepared_plugin()
                        .observe_receipt(caller_context("email-provider-blue"), request,)
                ),
                Err(PluginError::Domain(
                    delivery::ObserveReceiptError::InvalidReceipt
                ))
            ));
        }

        for request in [
            admin::ListDeliveriesRequest {
                cursor: None,
                limit: Some(0),
                status: None,
            },
            admin::ListDeliveriesRequest {
                cursor: None,
                limit: Some(201),
                status: None,
            },
            admin::ListDeliveriesRequest {
                cursor: Some("x".repeat(161)),
                limit: None,
                status: None,
            },
            admin::ListDeliveriesRequest {
                cursor: Some(String::new()),
                limit: None,
                status: None,
            },
        ] {
            assert!(!valid_list_request(&request));
            assert!(matches!(
                futures::executor::block_on(
                    unprepared_plugin().list_deliveries(caller_context("console-blue"), request,)
                ),
                Err(PluginError::Domain(
                    admin::ListDeliveriesError::InvalidFilter
                ))
            ));
        }

        for delivery_id in [String::new(), "x".repeat(161)] {
            assert!(!required_bounded(&delivery_id, 1, 160));
            assert!(matches!(
                futures::executor::block_on(unprepared_plugin().get_delivery(
                    caller_context("console-blue"),
                    admin::GetDeliveryRequest { delivery_id },
                )),
                Err(PluginError::Domain(admin::GetDeliveryError::InvalidRequest))
            ));
        }

        for request in [
            admin::RetryDeliveryRequest {
                delivery_id: "x".repeat(161),
                idempotency_key: "retry".to_owned(),
                revision: 1,
            },
            admin::RetryDeliveryRequest {
                delivery_id: "delivery".to_owned(),
                idempotency_key: "x".repeat(241),
                revision: 1,
            },
            admin::RetryDeliveryRequest {
                delivery_id: "delivery".to_owned(),
                idempotency_key: "retry".to_owned(),
                revision: 0,
            },
            admin::RetryDeliveryRequest {
                delivery_id: "delivery".to_owned(),
                idempotency_key: "retry".to_owned(),
                revision: MAX_SAFE_WIRE_INTEGER + 1,
            },
        ] {
            assert!(!valid_retry_request(&request));
            assert!(matches!(
                futures::executor::block_on(
                    unprepared_plugin().retry_delivery(caller_context("console-blue"), request,)
                ),
                Err(PluginError::Domain(
                    admin::RetryDeliveryError::InvalidRequest
                ))
            ));
        }

        let maximum_revision = admin::RetryDeliveryRequest {
            delivery_id: "delivery".to_owned(),
            idempotency_key: "retry".to_owned(),
            revision: MAX_SAFE_WIRE_INTEGER,
        };
        assert!(valid_retry_request(&maximum_revision));
        assert!(matches!(
            futures::executor::block_on(
                unprepared_plugin()
                    .retry_delivery(caller_context("console-blue"), maximum_revision,)
            ),
            Err(PluginError::Domain(
                admin::RetryDeliveryError::RetryNotAllowed
            ))
        ));
    }

    #[test]
    fn native_admin_outputs_reject_unbounded_or_nonportable_storage_values() {
        let now = Utc::now();
        let summary = DeliverySummary {
            id: "delivery".to_owned(),
            recipient_mask: "a***@example.test".to_owned(),
            template_id: "organization-invitation".to_owned(),
            template_version: "v1".to_owned(),
            locale: "en".to_owned(),
            status: "queued".to_owned(),
            revision: 1,
            attempt_count: 0,
            max_attempts: 4,
            redacted_preview: "Invitation".to_owned(),
            content_digest: format!("sha256:{}", "a".repeat(64)),
            correlation_id: "story".to_owned(),
            next_attempt_at: Some(now),
            final_reason: None,
            created_at: now,
            updated_at: now,
        };
        assert!(admin_delivery(summary.clone()).is_ok());
        assert!(
            admin_delivery(DeliverySummary {
                revision: MAX_SAFE_WIRE_INTEGER + 1,
                ..summary.clone()
            })
            .is_err()
        );
        assert!(
            admin_delivery(DeliverySummary {
                id: "x".repeat(161),
                ..summary.clone()
            })
            .is_err()
        );

        let attempt = AttemptRecord {
            id: "attempt".to_owned(),
            sequence: 1,
            function_run_id: "run".to_owned(),
            status: "dispatching".to_owned(),
            provider: None,
            remote_receipt_id: None,
            failure_code: None,
            failure_classification: None,
            started_at: now,
            completed_at: None,
        };
        assert!(admin_attempt(attempt.clone()).is_ok());
        assert!(
            admin_attempt(AttemptRecord {
                function_run_id: "x".repeat(161),
                ..attempt.clone()
            })
            .is_err()
        );

        let detail = DeliveryDetail {
            delivery: summary,
            attempts: vec![attempt; ADMIN_ATTEMPT_LIMIT + 1],
            receipts: Vec::new(),
            retry_requests: Vec::new(),
        };
        assert!(admin_detail(detail).is_err());

        let mapped = map_get_error(NotificationError::new(
            ErrorCode::EvidenceOverflow,
            "bounded evidence overflow fixture",
        ));
        assert!(matches!(
            mapped,
            PluginError::Domain(admin::GetDeliveryError::EvidenceOverflow)
        ));

        assert!(
            admin_retry(RetryResult {
                delivery_id: "delivery".to_owned(),
                revision: MAX_SAFE_WIRE_INTEGER + 1,
                status: "retry_scheduled".to_owned(),
                scheduled_at: now,
                idempotent_replay: false,
            })
            .is_err()
        );
    }

    #[test]
    fn native_email_response_validation_rejects_invalid_bounds_and_cross_fields() {
        let work = dispatch_work();
        let email_provider_instance = "email-provider-blue";
        let accepted = email::DispatchResponse {
            failure: None,
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            outcome: DispatchResponseOutcome::Accepted,
            provider: "email-provider-blue".to_owned(),
            remote_receipt: Some(email::DispatchResponseRemoteReceipt {
                digest: format!("sha256:{}", "a".repeat(64)),
                remote_id: "remote".to_owned(),
                source: "email-provider-blue".to_owned(),
            }),
        };
        let accepted_observation =
            dispatch_observation(&work, email_provider_instance, accepted.clone())
                .expect("bound Provider response is valid");
        assert_eq!(accepted_observation.provider, email_provider_instance);
        assert_eq!(
            accepted_observation
                .remote_receipt
                .expect("accepted receipt")
                .source,
            email_provider_instance
        );

        let temporary = email::DispatchResponse {
            failure: Some(email::DispatchResponseFailure {
                classification: "temporary_failure".to_owned(),
                code: "rate_limited".to_owned(),
                retry_after_ms: Some(1_000),
            }),
            observed_at: "2026-08-30T00:00:00Z".to_owned(),
            outcome: DispatchResponseOutcome::TemporaryFailure,
            provider: "email-provider-blue".to_owned(),
            remote_receipt: None,
        };
        assert!(dispatch_observation(&work, email_provider_instance, temporary.clone()).is_ok());

        let invalid = [
            email::DispatchResponse {
                provider: String::new(),
                ..accepted.clone()
            },
            email::DispatchResponse {
                provider: "different-provider".to_owned(),
                ..accepted.clone()
            },
            email::DispatchResponse {
                observed_at: "not-a-time".to_owned(),
                ..accepted.clone()
            },
            email::DispatchResponse {
                observed_at: format!("2026-08-30T00:00:00.{}Z", "1".repeat(50)),
                ..accepted.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    classification: "permanent_failure".to_owned(),
                    code: "should-not-exist".to_owned(),
                    retry_after_ms: None,
                }),
                ..accepted.clone()
            },
            email::DispatchResponse {
                outcome: DispatchResponseOutcome::PermanentFailure,
                remote_receipt: None,
                failure: None,
                ..accepted.clone()
            },
            email::DispatchResponse {
                outcome: DispatchResponseOutcome::DeliveryUnknown,
                remote_receipt: None,
                failure: None,
                ..accepted.clone()
            },
            email::DispatchResponse {
                remote_receipt: Some(email::DispatchResponseRemoteReceipt {
                    remote_id: String::new(),
                    ..accepted.remote_receipt.clone().expect("accepted receipt")
                }),
                ..accepted.clone()
            },
            email::DispatchResponse {
                remote_receipt: Some(email::DispatchResponseRemoteReceipt {
                    remote_id: "x".repeat(321),
                    ..accepted.remote_receipt.clone().expect("accepted receipt")
                }),
                ..accepted.clone()
            },
            email::DispatchResponse {
                remote_receipt: Some(email::DispatchResponseRemoteReceipt {
                    source: "different-provider".to_owned(),
                    ..accepted.remote_receipt.clone().expect("accepted receipt")
                }),
                ..accepted.clone()
            },
            email::DispatchResponse {
                remote_receipt: Some(email::DispatchResponseRemoteReceipt {
                    digest: "sha256:not-a-digest".to_owned(),
                    ..accepted.remote_receipt.clone().expect("accepted receipt")
                }),
                ..accepted.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    retry_after_ms: Some(-1),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    retry_after_ms: Some(86_400_001),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    code: String::new(),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    code: "x".repeat(161),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    classification: String::new(),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    classification: "x".repeat(161),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                failure: Some(email::DispatchResponseFailure {
                    classification: "permanent_failure".to_owned(),
                    ..temporary.failure.clone().expect("temporary failure")
                }),
                ..temporary.clone()
            },
            email::DispatchResponse {
                remote_receipt: accepted.remote_receipt,
                ..temporary
            },
        ];
        for response in invalid {
            assert!(matches!(
                dispatch_observation(&work, email_provider_instance, response),
                Err(RuntimeFailure::ProtocolViolation {
                    capability: email::CAPABILITY_ID
                })
            ));
        }
    }

    #[test]
    fn linked_factory_rejects_invalid_configuration_before_startup() {
        let driver = DeterministicDriver::new();
        let error = driver
            .run(Kernel::start_native(
                invalid_actual_plan(),
                driver.clone(),
                NativePluginRegistry::new()
                    .with_linked_factories()
                    .with_factory(EmptyFactory(SECRETS_PACKAGE_ID))
                    .with_factory(EmptyFactory(EMAIL_PACKAGE_ID))
                    .with_factory(EmptyFactory(TEMPLATE_PACKAGE_ID)),
            ))
            .expect_err("invalid Notification authority must reject startup");
        assert!(matches!(error, RuntimeFailure::InvalidResolvedPlan { .. }));
    }

    #[test]
    fn removing_notification_selection_removes_runtime_behavior() {
        let plan = AppComposition::new(
            vec![PluginInstancePlan::new("empty", EMPTY_PACKAGE_ID)],
            Vec::new(),
        )
        .resolve()
        .expect("App without Notification must resolve");
        let driver = DeterministicDriver::new();
        let app = driver
            .run(Kernel::start_native(
                plan,
                driver.clone(),
                NativePluginRegistry::new()
                    .with_linked_factories()
                    .with_factory(EmptyFactory(EMPTY_PACKAGE_ID)),
            ))
            .expect("unselected Notification Plugin must be inert");
        assert_eq!(
            driver.run(app.shutdown(Duration::from_secs(1))),
            ShutdownOutcome::Clean
        );
    }

    #[test]
    fn configuration_rejects_ambient_or_ambiguous_authority() {
        let config = NotificationConfig::new(
            "notification/database",
            "notification/snapshot-key",
            vec!["organization".to_owned()],
            vec!["notification-worker".to_owned()],
            vec!["email-provider".to_owned()],
            vec!["notification-console-http".to_owned()],
        )
        .expect("valid config");
        assert_eq!(config.schema, "notification");

        let mut duplicate_secret = config.clone();
        duplicate_secret.snapshot_key_secret = duplicate_secret.database_url_secret.clone();
        assert_eq!(
            duplicate_secret.validate(),
            Err(NotificationConfigError::InvalidSecretReference)
        );

        let mut ambient = config;
        ambient.admin_callers.clear();
        assert_eq!(
            ambient.validate(),
            Err(NotificationConfigError::InvalidCallers)
        );
    }

    #[test]
    fn removing_notification_needs_no_kernel_branch() {
        let remaining = lenso_app_plan::AppComposition::new(
            vec![lenso_app_plan::PluginInstancePlan::new(
                "organization",
                "test.organization",
            )],
            vec![],
        )
        .resolve()
        .expect("App without Notification should resolve");
        assert_eq!(remaining.plugin_instances().len(), 1);
        assert!(remaining.capability_bindings().is_empty());
    }

    #[test]
    fn unknown_dispatch_is_terminal_and_redaction_safe() {
        let work = DispatchWork {
            claim: crate::runtime::DispatchClaim {
                delivery_id: "delivery".to_owned(),
                attempt_id: "attempt".to_owned(),
                run_id: "run".to_owned(),
            },
            request: DispatchRequest {
                delivery_id: "delivery".to_owned(),
                attempt_id: "attempt".to_owned(),
                run_id: "run".to_owned(),
                idempotency_key: "attempt".to_owned(),
                recipient: email::DispatchRequestRecipient {
                    address: "member@example.com".to_owned(),
                },
                message: email::DispatchRequestMessage {
                    template_id: "organization-invitation".to_owned(),
                    template_version: "v1".to_owned(),
                    locale: "en".to_owned(),
                    subject: "secret".to_owned(),
                    text: "secret".to_owned(),
                    html: "secret".to_owned(),
                    content_digest: format!("sha256:{}", "a".repeat(64)),
                },
                correlation_id: "correlation".to_owned(),
            },
        };
        let observation =
            unknown_dispatch(&work, "email-provider-blue", Utc::now(), "runtime_failure");
        assert_eq!(observation.outcome, DispatchOutcome::DeliveryUnknown);
        assert!(observation.failure.is_some());
        assert!(!format!("{observation:?}").contains("member@example.com"));
    }

    #[test]
    fn stored_delivery_unknown_never_maps_to_a_retry_state() {
        assert_eq!(
            delivery_status("delivery_unknown").unwrap(),
            delivery::DispatchDueResponseStatus::DeliveryUnknown
        );
        assert!(crate::domain::can_transition(
            StoredDeliveryStatus::Attempting,
            StoredDeliveryStatus::DeliveryUnknown
        ));
        assert!(!crate::domain::can_transition(
            StoredDeliveryStatus::DeliveryUnknown,
            StoredDeliveryStatus::RetryScheduled
        ));
    }

    fn run_generated_fixture(outcome: FixtureOutcome) -> InvocationResult {
        let observed = Rc::new(RefCell::new(None));
        let driver = DeterministicDriver::new();
        let app = driver
            .run(Kernel::start_native(
                generated_fixture_plan(),
                driver.clone(),
                NativePluginRegistry::new()
                    .with_factory(FixtureProviderFactory { outcome })
                    .with_factory(ConsumerFactory {
                        observed: observed.clone(),
                    }),
            ))
            .expect("generated Notification fixture must start");
        let outcome = observed
            .borrow_mut()
            .take()
            .expect("generated Client consumer must observe an invocation");
        assert_eq!(
            driver.run(app.shutdown(Duration::from_secs(1))),
            ShutdownOutcome::Clean
        );
        outcome
    }

    fn generated_fixture_plan() -> ResolvedAppPlan {
        let consumer = PluginInstancePlan::new("consumer", CONSUMER_PACKAGE_ID).with_requirement(
            CapabilityRequirementPlan::one(
                transactional::CAPABILITY_ID,
                transactional::DESCRIPTOR_VERSION,
            ),
        );
        let provider = PluginInstancePlan::new("notification", FIXTURE_PROVIDER_PACKAGE_ID)
            .with_capability(CapabilityEndpointPlan::new(
                transactional::CAPABILITY_ID,
                transactional::DESCRIPTOR_VERSION,
                [
                    transactional::CREATE_ACCESS_REQUEST_NOTIFICATION_OPERATION,
                    transactional::CREATE_ORGANIZATION_INVITATION_OPERATION,
                    transactional::OBSERVE_INVITATION_LIFECYCLE_OPERATION,
                ],
            ));
        AppComposition::new(
            vec![consumer, provider],
            vec![CapabilityBinding::new(
                "consumer",
                transactional::CAPABILITY_ID,
                transactional::DESCRIPTOR_VERSION,
                "notification",
            )],
        )
        .resolve()
        .expect("generated Client/Provider Composition must resolve")
    }

    fn invalid_actual_plan() -> ResolvedAppPlan {
        let invalid = NotificationConfig {
            schema: "notification".to_owned(),
            database_url_secret: "notification/database".to_owned(),
            snapshot_key_secret: "notification/snapshot-key".to_owned(),
            transactional_callers: Vec::new(),
            dispatch_callers: vec!["worker".to_owned()],
            receipt_callers: vec!["email-provider".to_owned()],
            admin_callers: vec!["admin".to_owned()],
        };
        let notification = PluginInstancePlan::new("notification", PACKAGE_ID)
            .with_configuration(serde_json::to_string(&invalid).expect("serialize invalid config"))
            .with_requirement(CapabilityRequirementPlan::one(
                lenso_capability_secrets::CAPABILITY_ID,
                lenso_capability_secrets::DESCRIPTOR_VERSION,
            ))
            .with_requirement(CapabilityRequirementPlan::one(
                email::CAPABILITY_ID,
                email::DESCRIPTOR_VERSION,
            ))
            .with_requirement(CapabilityRequirementPlan::one(
                notification_template::CAPABILITY_ID,
                notification_template::DESCRIPTOR_VERSION,
            ));
        let secrets = PluginInstancePlan::new("secrets", SECRETS_PACKAGE_ID).with_capability(
            CapabilityEndpointPlan::new(
                lenso_capability_secrets::CAPABILITY_ID,
                lenso_capability_secrets::DESCRIPTOR_VERSION,
                [lenso_capability_secrets::RESOLVE_OPERATION],
            ),
        );
        let email_provider = PluginInstancePlan::new("email", EMAIL_PACKAGE_ID).with_capability(
            CapabilityEndpointPlan::new(
                email::CAPABILITY_ID,
                email::DESCRIPTOR_VERSION,
                [email::DISPATCH_OPERATION],
            ),
        );
        let template_provider = PluginInstancePlan::new("templates", TEMPLATE_PACKAGE_ID)
            .with_capability(CapabilityEndpointPlan::new(
                notification_template::CAPABILITY_ID,
                notification_template::DESCRIPTOR_VERSION,
                [notification_template::RENDER_OPERATION],
            ));
        AppComposition::new(
            vec![notification, secrets, email_provider, template_provider],
            vec![
                CapabilityBinding::new(
                    "notification",
                    lenso_capability_secrets::CAPABILITY_ID,
                    lenso_capability_secrets::DESCRIPTOR_VERSION,
                    "secrets",
                ),
                CapabilityBinding::new(
                    "notification",
                    email::CAPABILITY_ID,
                    email::DESCRIPTOR_VERSION,
                    "email",
                ),
                CapabilityBinding::new(
                    "notification",
                    notification_template::CAPABILITY_ID,
                    notification_template::DESCRIPTOR_VERSION,
                    "templates",
                ),
            ],
        )
        .resolve()
        .expect("invalid configuration should pass structural Plan resolution")
    }

    fn caller_context(caller: &str) -> InvocationContext {
        InvocationContext::new(1, None, CancellationToken::new())
            .with_caller_instance(caller.to_owned())
    }

    fn unprepared_plugin() -> NotificationPlugin {
        NotificationPlugin {
            config: NotificationConfig::new(
                "notification/database",
                "notification/snapshot-key",
                vec!["organization-blue".to_owned()],
                vec!["notification-worker-blue".to_owned()],
                vec!["email-provider-blue".to_owned()],
                vec!["console-blue".to_owned()],
            )
            .expect("valid unprepared test Plugin"),
            secrets: Port::new(),
            email: Port::new(),
            templates: Port::new(),
            state: Rc::new(RefCell::new(None)),
        }
    }

    fn invalid_create_requests() -> Vec<transactional::CreateOrganizationInvitationRequest> {
        let base = generated_request();
        vec![
            transactional::CreateOrganizationInvitationRequest {
                source: transactional::CreateOrganizationInvitationRequestSource {
                    entity_type: "x".repeat(161),
                    ..base.source.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                source: transactional::CreateOrganizationInvitationRequestSource {
                    entity_id: "x".repeat(241),
                    ..base.source.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                recipient: transactional::CreateOrganizationInvitationRequestRecipient {
                    address: format!("a@{}", "x".repeat(319)),
                    ..base.recipient.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                recipient: transactional::CreateOrganizationInvitationRequestRecipient {
                    display_name: Some("x".repeat(241)),
                    ..base.recipient.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    organization_id: "x".repeat(241),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    organization_name: "x".repeat(241),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    invitation_id: "x".repeat(241),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    invitation_url: format!("https://{}", "x".repeat(4_089)),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    inviter_display_name: Some("x".repeat(241)),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    role_name: Some("x".repeat(161)),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                template: transactional::CreateOrganizationInvitationRequestTemplate {
                    expires_at: "not-a-time".to_owned(),
                    ..base.template.clone()
                },
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                idempotency_key: "x".repeat(241),
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                correlation_id: "x".repeat(241),
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                causation_id: Some("x".repeat(241)),
                ..base.clone()
            },
            transactional::CreateOrganizationInvitationRequest {
                requested_by: Some("x".repeat(241)),
                ..base
            },
        ]
    }

    fn dispatch_work() -> DispatchWork {
        DispatchWork {
            claim: crate::runtime::DispatchClaim {
                delivery_id: "delivery".to_owned(),
                attempt_id: "attempt".to_owned(),
                run_id: "run".to_owned(),
            },
            request: DispatchRequest {
                delivery_id: "delivery".to_owned(),
                attempt_id: "attempt".to_owned(),
                run_id: "run".to_owned(),
                idempotency_key: "attempt".to_owned(),
                recipient: email::DispatchRequestRecipient {
                    address: "member@example.com".to_owned(),
                },
                message: email::DispatchRequestMessage {
                    template_id: "organization-invitation".to_owned(),
                    template_version: "v1".to_owned(),
                    locale: "en".to_owned(),
                    subject: "secret".to_owned(),
                    text: "secret".to_owned(),
                    html: "secret".to_owned(),
                    content_digest: format!("sha256:{}", "a".repeat(64)),
                },
                correlation_id: "correlation".to_owned(),
            },
        }
    }

    fn generated_request() -> transactional::CreateOrganizationInvitationRequest {
        transactional::CreateOrganizationInvitationRequest {
            causation_id: Some("obs_invitation_fixture".to_owned()),
            correlation_id: "corr_notification_fixture".to_owned(),
            idempotency_key: "organization-invitation:fixture".to_owned(),
            recipient: transactional::CreateOrganizationInvitationRequestRecipient {
                address: "member@example.com".to_owned(),
                display_name: Some("Member".to_owned()),
                locale: transactional::CreateOrganizationInvitationRequestRecipientLocale::En,
            },
            requested_by: Some("usr_fixture".to_owned()),
            source: transactional::CreateOrganizationInvitationRequestSource {
                entity_id: "invite_fixture".to_owned(),
                entity_type: "organization_invitation".to_owned(),
            },
            template: transactional::CreateOrganizationInvitationRequestTemplate {
                expires_at: "2026-09-01T00:00:00Z".to_owned(),
                invitation_id: "invite_fixture".to_owned(),
                invitation_url: "https://example.test/invitations/secret".to_owned(),
                inviter_display_name: Some("Operator".to_owned()),
                organization_id: "org_fixture".to_owned(),
                organization_name: "Fixture Organization".to_owned(),
                role_name: Some("Member".to_owned()),
            },
        }
    }

    fn generated_access_request() -> transactional::CreateAccessRequestNotificationRequest {
        transactional::CreateAccessRequestNotificationRequest {
            causation_id: Some("access_request_fixture:submitted".to_owned()),
            correlation_id: "corr_access_request_fixture".to_owned(),
            event: transactional::CreateAccessRequestNotificationRequestEvent::Submitted,
            expires_at: Some("2026-09-01T00:00:00Z".to_owned()),
            idempotency_key: "access-request:ar_fixture:submitted".to_owned(),
            organization_id: "org_fixture".to_owned(),
            recipient: transactional::CreateAccessRequestNotificationRequestRecipient {
                address: "requester@example.com".to_owned(),
                display_name: Some("Requester".to_owned()),
                locale: transactional::CreateAccessRequestNotificationRequestRecipientLocale::En,
            },
            request_id: "ar_fixture".to_owned(),
            requested_by: Some("subject_fixture".to_owned()),
            role: transactional::CreateAccessRequestNotificationRequestRole {
                display_name: Some("Member".to_owned()),
                role_id: "role_member".to_owned(),
            },
            scope: transactional::CreateAccessRequestNotificationRequestScope {
                display_name: Some("Fixture Organization".to_owned()),
                id: "org_fixture".to_owned(),
                kind: "organization".to_owned(),
            },
        }
    }
}