teaql_runtime/

context.rs

use std::any::{Any, TypeId};
use std::collections::{BTreeMap, HashMap};
use std::future::Future;

use std::pin::Pin;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::SystemTime;

use crate::EntityRuntimeState;
use crate::business_id::BusinessIdSchemaService;
use crate::{
    BusinessIdSchemaContributor, BusinessIdService, CheckObjectStatus, CheckResult, CheckResults,
    CheckerRegistry, ContextError, EntityDataServiceBehavior, EntityDataServiceBehaviorRegistry,
    EntityGraphBuilder, EntityRegistry, GraphNode, InMemoryEntityGraphDecoderRegistry,
    InternalIdGenerator, Language, MetadataStore, ObjectLocation, RawAuditEvent, RawAuditEventSink,
    RequestPolicy, RuntimeError, local_id_generator,
};
use teaql_core::business_id::BusinessIdAllocator;
use teaql_core::{EntityDescriptor, Value};

mod locking;
mod logging;
mod pagination;
mod transaction;
pub use locking::RemoteLockProvider;
use locking::next_local_lock_owner;
pub use logging::{
    InfoLogEntry, LogPayload, SqlLogEntry, SqlLogOperation, SqlLogOptions, UnifiedLogBuffer,
    UnifiedLogEntry,
};
pub use pagination::{
    ContinuousPageCursor, ContinuousPageCursorStore, IdSetStore, InMemoryContinuousPageCursorStore,
    InMemoryIdSetStore, RetainedIdSet,
};
pub use transaction::TransactionScope;

tokio::task_local! {
    static GENERATED_SCHEMA_BOOTSTRAP_MODE: ();
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ContextEntityRef {
    pub entity_type: String,
    pub id: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ContextRootError {
    pub expected_entity_type: String,
    pub actual_root: Option<ContextEntityRef>,
}

impl std::fmt::Display for ContextRootError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.actual_root {
            None => write!(
                formatter,
                "active root {} is missing from UserContext",
                self.expected_entity_type
            ),
            Some(actual) => write!(
                formatter,
                "active root type is {}, expected {}",
                actual.entity_type, self.expected_entity_type
            ),
        }
    }
}

impl std::error::Error for ContextRootError {}

#[cfg(test)]
mod active_root_tests {
    use super::UserContext;

    #[test]
    fn active_root_is_typed_and_fails_closed() {
        let context = UserContext::new().with_active_root("Tenant", 42);
        assert_eq!(context.require_active_root("Tenant").unwrap().id, 42);
        assert!(context.require_active_root("Organization").is_err());
        assert!(UserContext::new().require_active_root("Tenant").is_err());
    }
}

/// Context-owned proof required by the provider SPI. Its private field prevents
/// application crates from invoking a schema provider directly.
///
/// ```compile_fail
/// let _ = teaql_runtime::SchemaInvocation { _context_owned: () };
/// ```
pub struct SchemaInvocation {
    _context_owned: (),
}

pub trait SchemaProvider: Send + Sync {
    fn ensure_schema<'a>(
        &'a self,
        context: &'a UserContext,
        invocation: &'a SchemaInvocation,
    ) -> Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>>;
}

pub type GeneratedSchemaBootstrapFuture<'a> =
    Pin<Box<dyn Future<Output = Result<(), RuntimeError>> + Send + 'a>>;
pub type GeneratedSchemaBootstrap =
    for<'a> fn(&'a UserContext) -> GeneratedSchemaBootstrapFuture<'a>;

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FixEvidenceSource {
    Clock,
    Context,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FixEvidence {
    pub entity_type: String,
    pub model_path: String,
    pub source: FixEvidenceSource,
    pub source_label: String,
}

impl FixEvidence {
    pub fn new(
        entity_type: &str,
        model_path: &str,
        source: FixEvidenceSource,
        source_label: &str,
    ) -> Self {
        assert!(
            !entity_type.trim().is_empty(),
            "entity_type must not be blank"
        );
        assert!(
            !model_path.trim().is_empty(),
            "model_path must not be blank"
        );
        assert!(
            !source_label.trim().is_empty(),
            "source_label must not be blank"
        );
        let normalized = source_label.to_ascii_lowercase();
        assert!(
            !normalized.contains("authorization")
                && !normalized.contains("cookie")
                && !normalized.contains("token="),
            "source_label must be a safe framework label"
        );
        Self {
            entity_type: entity_type.to_owned(),
            model_path: model_path.to_owned(),
            source,
            source_label: source_label.to_owned(),
        }
    }
}

pub struct UserContext {
    active_root: OnceLock<ContextEntityRef>,
    pub(crate) metadata: Option<Box<dyn MetadataStore>>,
    pub(crate) entity_registry: Option<Box<dyn EntityRegistry>>,
    pub(crate) entity_graph_decoders: InMemoryEntityGraphDecoderRegistry,
    pub(crate) entity_data_service_behavior_registry:
        Option<Box<dyn EntityDataServiceBehaviorRegistry>>,
    pub(crate) request_policy: Option<Box<dyn RequestPolicy>>,
    pub(crate) checker_registry: Option<Box<dyn CheckerRegistry>>,
    pub(crate) event_sink: Option<Box<dyn RawAuditEventSink>>,
    pub(crate) custom_event_sink: Option<Box<dyn crate::SafeAuditEventSink>>,
    pub(crate) internal_id_generator: Option<Box<dyn InternalIdGenerator>>,
    schema_provider: Option<Box<dyn SchemaProvider>>,
    generated_schema_bootstraps: Vec<GeneratedSchemaBootstrap>,
    language: Language,
    i18n_catalog: Arc<crate::I18nCatalog>,
    typed_resources: HashMap<TypeId, Box<dyn Any + Send + Sync>>,
    named_resources: BTreeMap<String, Box<dyn Any + Send + Sync>>,
    locals: BTreeMap<String, Value>,
    pub(crate) initial_graphs: Vec<GraphNode>,
    pub(crate) root_graphs: Vec<GraphNode>,
    entity_runtime_state: EntityRuntimeState,
    sql_log_options: SqlLogOptions,
    sql_log_entries: Mutex<Vec<SqlLogEntry>>,
    user_identifier: Option<String>,
    timezone: Option<String>,
    trace_id: String,
    continuous_page_cursor_store: std::sync::Arc<dyn ContinuousPageCursorStore>,
    continuous_page_observation: Mutex<(String, Option<String>)>,
    id_set_store: Arc<dyn IdSetStore>,
    id_set_observation: Mutex<(String, Option<u64>)>,
    local_lock_owner: u64,
    remote_lock_owner: String,
    runtime_telemetry: Arc<dyn crate::RuntimeTelemetry>,
    last_fix_evidence: Mutex<Vec<FixEvidence>>,
}

impl Default for UserContext {
    fn default() -> Self {
        let pid = std::process::id();
        let thread_id_str = format!("{:?}", std::thread::current().id());
        let numeric_thread_id = thread_id_str
            .strip_prefix("ThreadId(")
            .and_then(|s| s.strip_suffix(")"))
            .unwrap_or(&thread_id_str);
        let os_user = std::env::var("USER")
            .or_else(|_| std::env::var("USERNAME"))
            .unwrap_or_else(|_| "main".to_owned());
        let user_id = format!("{os_user}@pid-{pid}.tid-{numeric_thread_id}");
        let owner_sequence = next_local_lock_owner();
        Self {
            active_root: OnceLock::new(),
            metadata: None,
            entity_registry: None,
            entity_graph_decoders: InMemoryEntityGraphDecoderRegistry::default(),
            entity_data_service_behavior_registry: None,
            request_policy: None,
            checker_registry: None,
            event_sink: None,
            custom_event_sink: None,
            internal_id_generator: None,
            schema_provider: None,
            language: Language::default(),
            i18n_catalog: crate::I18nCatalog::builtin().clone(),
            typed_resources: HashMap::new(),
            named_resources: BTreeMap::new(),
            locals: BTreeMap::new(),
            initial_graphs: Vec::new(),
            root_graphs: Vec::new(),
            generated_schema_bootstraps: Vec::new(),
            entity_runtime_state: EntityRuntimeState::default(),
            // Query and Mutation diagnostic SQL logging are on by default.
            // They remain independently controllable because performance and
            // retention policy may differ between the two operation families.
            sql_log_options: SqlLogOptions::default(),
            sql_log_entries: Mutex::new(Vec::new()),
            user_identifier: Some(user_id),
            timezone: Some("UTC".to_owned()),
            trace_id: format!(
                "req-{pid}-{numeric_thread_id}-{:x}",
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .unwrap_or_default()
                    .as_micros()
            ),
            continuous_page_cursor_store: std::sync::Arc::new(
                InMemoryContinuousPageCursorStore::default(),
            ),
            continuous_page_observation: Mutex::new(("DISABLED".to_owned(), None)),
            id_set_store: Arc::new(InMemoryIdSetStore::default()),
            id_set_observation: Mutex::new(("ID_SET_DISABLED".to_owned(), None)),
            local_lock_owner: owner_sequence,
            remote_lock_owner: format!(
                "teaql:{pid}:{owner_sequence}:{}",
                SystemTime::now()
                    .duration_since(SystemTime::UNIX_EPOCH)
                    .unwrap_or_default()
                    .as_nanos()
            ),
            runtime_telemetry: Arc::new(crate::NoopRuntimeTelemetry),
            last_fix_evidence: Mutex::new(Vec::new()),
        }
    }
}

#[async_trait::async_trait]
pub trait DataStore: Send + Sync + 'static {
    async fn get(&self, key: &str) -> Option<Value>;
    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>);
    async fn remove(&self, key: &str);
}

#[derive(Default)]
pub struct InMemoryDataStore {
    cache: std::sync::RwLock<HashMap<String, (Value, Option<std::time::Instant>)>>,
}

#[async_trait::async_trait]
impl DataStore for InMemoryDataStore {
    async fn get(&self, key: &str) -> Option<Value> {
        let lock = self.cache.read().unwrap();
        if let Some((val, expires_at)) = lock.get(key) {
            if let Some(exp) = expires_at
                && std::time::Instant::now() > *exp
            {
                return None;
            }
            return Some(val.clone());
        }
        None
    }

    async fn put(&self, key: &str, value: Value, timeout_seconds: Option<u64>) {
        let mut lock = self.cache.write().unwrap();
        let expires_at = timeout_seconds
            .map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
        lock.insert(key.to_string(), (value, expires_at));
    }

    async fn remove(&self, key: &str) {
        let mut lock = self.cache.write().unwrap();
        lock.remove(key);
    }
}

impl UserContext {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_active_root(self, entity_type: impl Into<String>, id: u64) -> Self {
        let entity_type = crate::canonical_id_space_entity(&entity_type.into());
        assert!(
            !entity_type.trim().is_empty(),
            "active root entity type is required"
        );
        assert!(id > 0, "active root id must be positive");
        self.active_root
            .set(ContextEntityRef { entity_type, id })
            .expect("active root may only be assigned once");
        self
    }

    #[doc(hidden)]
    pub fn set_generated_bootstrap_active_root(
        &self,
        entity_type: impl Into<String>,
        id: u64,
    ) -> Result<(), RuntimeError> {
        let entity_type = crate::canonical_id_space_entity(&entity_type.into());
        if entity_type.trim().is_empty() || id == 0 {
            return Err(RuntimeError::Schema(
                "invalid generated active root".to_owned(),
            ));
        }
        match self.active_root.get() {
            Some(existing) if existing.entity_type == entity_type && existing.id == id => Ok(()),
            Some(existing) => Err(RuntimeError::Schema(format!(
                "active root already set to {}:{}",
                existing.entity_type, existing.id
            ))),
            None => self
                .active_root
                .set(ContextEntityRef { entity_type, id })
                .map_err(|_| RuntimeError::Schema("active root initialization raced".to_owned())),
        }
    }

    pub fn require_active_root(
        &self,
        expected_entity_type: &str,
    ) -> Result<&ContextEntityRef, ContextRootError> {
        let canonical_expected = crate::canonical_id_space_entity(expected_entity_type);
        match self.active_root.get() {
            Some(root) if root.entity_type == canonical_expected => Ok(root),
            actual_root => Err(ContextRootError {
                expected_entity_type: canonical_expected,
                actual_root: actual_root.cloned(),
            }),
        }
    }

    pub(crate) fn active_root_ref(&self) -> Option<&ContextEntityRef> {
        self.active_root.get()
    }

    pub fn with_runtime_telemetry(mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) -> Self {
        self.runtime_telemetry = telemetry;
        self
    }

    pub fn set_runtime_telemetry(&mut self, telemetry: Arc<dyn crate::RuntimeTelemetry>) {
        self.runtime_telemetry = telemetry;
    }

    pub fn runtime_telemetry(&self) -> &Arc<dyn crate::RuntimeTelemetry> {
        &self.runtime_telemetry
    }

    pub(crate) fn runtime_telemetry_is_noop(&self) -> bool {
        self.runtime_telemetry.is_noop()
    }

    pub fn start_runtime_operation(
        &self,
        operation: crate::RuntimeOperation,
    ) -> crate::FailOpenRuntimeTelemetryScope {
        crate::start_runtime_operation(&self.runtime_telemetry, operation)
    }

    pub fn user_identifier(&self) -> Option<&str> {
        self.user_identifier.as_deref()
    }

    pub fn set_user_identifier(&mut self, user_identifier: impl Into<String>) {
        self.user_identifier = Some(user_identifier.into());
    }

    pub fn with_user_identifier(mut self, user_identifier: impl Into<String>) -> Self {
        self.user_identifier = Some(user_identifier.into());
        self
    }

    pub fn set_user_identifier_option(&mut self, user_identifier: Option<String>) {
        self.user_identifier = user_identifier;
    }

    pub fn with_user_identifier_option(mut self, user_identifier: Option<String>) -> Self {
        self.user_identifier = user_identifier;
        self
    }

    pub fn timezone(&self) -> Option<&str> {
        self.timezone.as_deref()
    }

    pub fn set_timezone(&mut self, timezone: impl Into<String>) {
        self.timezone = Some(timezone.into());
    }

    pub fn with_timezone(mut self, timezone: impl Into<String>) -> Self {
        self.timezone = Some(timezone.into());
        self
    }

    pub fn trace_id(&self) -> &str {
        &self.trace_id
    }

    pub fn set_trace_id(&mut self, trace_id: impl Into<String>) {
        self.trace_id = trace_id.into();
    }

    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
        self.trace_id = trace_id.into();
        self
    }

    pub fn with_module(mut self, module: crate::RuntimeModule) -> Self {
        module.apply_to(&mut self);
        self
    }

    pub fn entity_runtime_state(&self) -> EntityRuntimeState {
        // UserContext owns only the immutable identity-graph anchor. Every query/new-entity
        // operation receives fresh mutation state, even when the same context is reused.
        EntityRuntimeState::fresh_with_shared_graph(&self.entity_runtime_state)
    }

    pub fn initial_graphs(&self) -> &[GraphNode] {
        &self.initial_graphs
    }

    pub fn set_initial_graphs(&mut self, graphs: Vec<GraphNode>) {
        self.initial_graphs = graphs;
    }

    pub fn root_graphs(&self) -> &[GraphNode] {
        &self.root_graphs
    }

    pub fn set_root_graphs(&mut self, graphs: Vec<GraphNode>) {
        self.root_graphs = graphs;
    }

    pub fn with_metadata(mut self, metadata: impl MetadataStore + 'static) -> Self {
        self.metadata = Some(Box::new(metadata));
        self
    }

    pub fn set_metadata(&mut self, metadata: impl MetadataStore + 'static) {
        self.metadata = Some(Box::new(metadata));
    }

    pub fn with_entity_registry(mut self, registry: impl EntityRegistry + 'static) -> Self {
        self.entity_registry = Some(Box::new(registry));
        self
    }

    pub fn set_entity_registry(&mut self, registry: impl EntityRegistry + 'static) {
        self.entity_registry = Some(Box::new(registry));
    }

    pub fn set_entity_graph_decoder_registry(
        &mut self,
        registry: InMemoryEntityGraphDecoderRegistry,
    ) {
        self.entity_graph_decoders = registry;
    }

    pub(crate) fn has_entity_graph_decoder(&self, entity: &str) -> bool {
        self.entity_graph_decoders.contains(entity)
    }

    pub(crate) fn decode_compact_entity_into_graph(
        &self,
        entity: &str,
        row: teaql_core::CompactRow,
        root: &EntityRuntimeState,
        graph: &mut EntityGraphBuilder,
    ) -> Result<(), teaql_core::EntityError> {
        self.entity_graph_decoders
            .decode_compact(entity, row, root, graph)
    }

    #[allow(clippy::too_many_arguments)] // Mirrors the generated decoder contract.
    pub(crate) fn decode_compact_entity_list_into_graph(
        &self,
        entity: &str,
        rows: Vec<teaql_core::CompactRow>,
        root: &EntityRuntimeState,
        graph: &mut EntityGraphBuilder,
        owner_entity: &str,
        owner_id: u64,
        relation: &str,
    ) -> Result<(), teaql_core::EntityError> {
        self.entity_graph_decoders.decode_compact_list(
            entity,
            rows,
            root,
            graph,
            owner_entity,
            owner_id,
            relation,
        )
    }

    pub(crate) fn decode_compact_entity_batch_into_graph(
        &self,
        entity: &str,
        rows: Vec<teaql_core::CompactRow>,
        root: &EntityRuntimeState,
        graph: &mut EntityGraphBuilder,
    ) -> Result<(), teaql_core::EntityError> {
        self.entity_graph_decoders
            .decode_compact_batch(entity, rows, root, graph)
    }

    #[allow(clippy::too_many_arguments)] // Mirrors the generated decoder contract.
    pub(crate) fn decode_compact_entity_option_into_graph(
        &self,
        entity: &str,
        rows: Vec<teaql_core::CompactRow>,
        root: &EntityRuntimeState,
        graph: &mut EntityGraphBuilder,
        owner_entity: &str,
        owner_id: u64,
        relation: &str,
    ) -> Result<(), teaql_core::EntityError> {
        self.entity_graph_decoders.decode_compact_option(
            entity,
            rows,
            root,
            graph,
            owner_entity,
            owner_id,
            relation,
        )
    }

    pub fn with_entity_data_service_behavior_registry(
        mut self,
        registry: impl EntityDataServiceBehaviorRegistry + 'static,
    ) -> Self {
        self.entity_data_service_behavior_registry = Some(Box::new(registry));
        self
    }

    pub fn set_entity_data_service_behavior_registry(
        &mut self,
        registry: impl EntityDataServiceBehaviorRegistry + 'static,
    ) {
        self.entity_data_service_behavior_registry = Some(Box::new(registry));
    }

    pub fn with_request_policy(mut self, policy: impl RequestPolicy + 'static) -> Self {
        self.request_policy = Some(Box::new(policy));
        self
    }

    pub fn set_request_policy(&mut self, policy: impl RequestPolicy + 'static) {
        self.request_policy = Some(Box::new(policy));
    }

    pub fn clear_request_policy(&mut self) {
        self.request_policy = None;
    }

    pub fn with_checker_registry(mut self, registry: impl CheckerRegistry + 'static) -> Self {
        self.checker_registry = Some(Box::new(registry));
        self
    }

    pub fn set_checker_registry(&mut self, registry: impl CheckerRegistry + 'static) {
        self.checker_registry = Some(Box::new(registry));
    }

    #[cfg(test)]
    pub(crate) fn with_event_sink(mut self, sink: impl RawAuditEventSink + 'static) -> Self {
        self.event_sink = Some(Box::new(sink));
        self
    }

    pub(crate) fn set_event_sink(&mut self, sink: impl RawAuditEventSink + 'static) {
        self.event_sink = Some(Box::new(sink));
    }

    pub fn with_custom_event_sink(
        mut self,
        sink: impl crate::SafeAuditEventSink + 'static,
    ) -> Self {
        self.custom_event_sink = Some(Box::new(sink));
        self
    }

    pub fn set_custom_event_sink(&mut self, sink: impl crate::SafeAuditEventSink + 'static) {
        self.custom_event_sink = Some(Box::new(sink));
    }

    pub fn with_internal_id_generator(
        mut self,
        generator: impl InternalIdGenerator + 'static,
    ) -> Self {
        self.internal_id_generator = Some(Box::new(generator));
        self
    }

    pub fn set_internal_id_generator(&mut self, generator: impl InternalIdGenerator + 'static) {
        self.internal_id_generator = Some(Box::new(generator));
    }

    pub fn with_schema_provider(mut self, provider: impl SchemaProvider + 'static) -> Self {
        self.schema_provider = Some(Box::new(provider));
        self
    }

    pub fn set_schema_provider(&mut self, provider: impl SchemaProvider + 'static) {
        self.schema_provider = Some(Box::new(provider));
    }

    pub async fn ensure_schema(&self) -> Result<(), RuntimeError> {
        let provider = self
            .schema_provider
            .as_ref()
            .ok_or_else(|| RuntimeError::Schema("missing schema provider".to_owned()))?;
        let invocation = SchemaInvocation { _context_owned: () };
        provider.ensure_schema(self, &invocation).await?;
        if let Some(business_ids) = self.get_resource::<BusinessIdSchemaService>() {
            business_ids
                .ensure_schema(self)
                .map_err(|error| RuntimeError::Schema(error.to_string()))?;
        }
        GENERATED_SCHEMA_BOOTSTRAP_MODE
            .scope((), async {
                for bootstrap in &self.generated_schema_bootstraps {
                    bootstrap(self).await?;
                }
                Ok::<(), RuntimeError>(())
            })
            .await?;
        Ok(())
    }

    pub(crate) fn is_generated_schema_bootstrap(&self) -> bool {
        GENERATED_SCHEMA_BOOTSTRAP_MODE
            .try_with(|_| true)
            .unwrap_or(false)
    }

    pub(crate) fn set_generated_schema_bootstraps(
        &mut self,
        bootstraps: Vec<GeneratedSchemaBootstrap>,
    ) {
        self.generated_schema_bootstraps = bootstraps;
    }

    #[doc(hidden)]
    pub fn initialize_generated_bootstrap_entity<E: teaql_core::Entity>(
        &self,
        entity: &mut E,
        entity_name: &str,
        fixed_id: u64,
    ) -> Result<(), RuntimeError> {
        let generator = self.internal_id_generator.as_ref().ok_or_else(|| {
            RuntimeError::IdGeneration("missing internal ID generator".to_owned())
        })?;
        generator.ensure_floor(entity_name, fixed_id)?;
        entity.mark_as_new();
        Ok(())
    }

    pub fn with_language(mut self, language: Language) -> Self {
        self.language = language;
        self
    }

    pub fn set_language(&mut self, language: Language) {
        self.language = language;
    }

    pub fn with_i18n_catalog(mut self, catalog: Arc<crate::I18nCatalog>) -> Self {
        self.i18n_catalog = catalog;
        self
    }

    pub fn set_i18n_catalog(&mut self, catalog: Arc<crate::I18nCatalog>) {
        self.i18n_catalog = catalog;
    }

    pub fn language(&self) -> Language {
        self.language
    }

    pub fn set_language_code(&mut self, code: &str) -> Result<(), RuntimeError> {
        let Some(language) = Language::from_code(code) else {
            return Err(RuntimeError::UnsupportedLocale(code.to_owned()));
        };
        self.language = language;
        Ok(())
    }

    pub fn set_locale_code(&mut self, code: &str) -> Result<(), RuntimeError> {
        self.set_language_code(code)
    }

    pub fn generate_id(&self, entity: &str) -> Result<Option<u64>, RuntimeError> {
        self.internal_id_generator
            .as_ref()
            .map(|generator| generator.generate_id(entity))
            .transpose()
    }

    pub fn next_id(&self, entity: &str) -> Result<u64, RuntimeError> {
        match self.generate_id(entity)? {
            Some(id) => Ok(id),
            None => local_id_generator().generate_id(entity),
        }
    }

    pub fn entity(&self, name: &str) -> Option<&EntityDescriptor> {
        self.metadata
            .as_ref()
            .and_then(|metadata| metadata.entity(name))
    }

    pub fn all_entities(&self) -> Vec<&EntityDescriptor> {
        self.metadata
            .as_ref()
            .map(|metadata| metadata.all_entities())
            .unwrap_or_default()
    }

    pub fn require_entity(&self, name: &str) -> Result<&EntityDescriptor, RuntimeError> {
        self.entity(name)
            .ok_or_else(|| RuntimeError::MissingEntity(name.to_owned()))
    }

    pub fn insert_resource<T>(&mut self, resource: T)
    where
        T: Send + Sync + 'static,
    {
        self.typed_resources
            .insert(TypeId::of::<T>(), Box::new(resource));
    }

    pub fn with_business_id_service(mut self, service: BusinessIdService) -> Self {
        self.insert_resource(service);
        self
    }

    pub fn set_business_id_service(&mut self, service: BusinessIdService) {
        self.insert_resource(service);
    }

    /// Installs one provider-backed allocator as both the generated Fix service
    /// and an explicit `ensure_schema` infrastructure contributor.
    pub fn set_business_id_infrastructure<A>(&mut self, allocator: A)
    where
        A: BusinessIdAllocator + BusinessIdSchemaContributor + 'static,
    {
        let allocator = Arc::new(allocator);
        let allocation_service: Arc<dyn BusinessIdAllocator> = allocator.clone();
        let schema_contributor: Arc<dyn BusinessIdSchemaContributor> = allocator;
        self.insert_resource(BusinessIdService::from_shared(allocation_service));
        self.insert_resource(BusinessIdSchemaService::from_shared(schema_contributor));
    }

    pub fn business_ids(&self) -> Result<&BusinessIdService, ContextError> {
        self.require_resource::<BusinessIdService>()
    }

    pub fn get_resource<T>(&self) -> Option<&T>
    where
        T: Send + Sync + 'static,
    {
        self.typed_resources
            .get(&TypeId::of::<T>())
            .and_then(|value| value.downcast_ref::<T>())
    }

    pub fn require_resource<T>(&self) -> Result<&T, ContextError>
    where
        T: Send + Sync + 'static,
    {
        self.get_resource::<T>()
            .ok_or(ContextError::MissingTypedResource(
                std::any::type_name::<T>(),
            ))
    }

    pub fn insert_named_resource<T>(&mut self, name: impl Into<String>, resource: T)
    where
        T: Send + Sync + 'static,
    {
        self.named_resources.insert(name.into(), Box::new(resource));
    }

    pub fn get_named_resource<T>(&self, name: &str) -> Option<&T>
    where
        T: Send + Sync + 'static,
    {
        self.named_resources
            .get(name)
            .and_then(|value| value.downcast_ref::<T>())
    }

    pub fn require_named_resource<T>(&self, name: &str) -> Result<&T, ContextError>
    where
        T: Send + Sync + 'static,
    {
        self.get_named_resource::<T>(name)
            .ok_or_else(|| ContextError::MissingResource(name.to_owned()))
    }

    pub fn put_local(&mut self, key: impl Into<String>, value: impl Into<Value>) {
        self.locals.insert(key.into(), value.into());
    }

    pub fn local(&self, key: &str) -> Option<&Value> {
        self.locals.get(key)
    }

    pub fn remove_local(&mut self, key: &str) -> Option<Value> {
        self.locals.remove(key)
    }

    pub fn has_entity_data_service(&self, entity: &str) -> bool {
        let in_registry = self
            .entity_registry
            .as_ref()
            .map(|registry| registry.contains(entity))
            .unwrap_or(false);
        in_registry || self.entity(entity).is_some()
    }

    pub fn entity_data_service_behavior(
        &self,
        entity: &str,
    ) -> Option<std::sync::Arc<dyn EntityDataServiceBehavior>> {
        self.entity_data_service_behavior_registry
            .as_ref()
            .and_then(|registry| registry.behavior(entity))
    }

    pub fn has_checker(&self, entity: &str) -> bool {
        self.checker_registry
            .as_ref()
            .and_then(|registry| registry.checker(entity))
            .is_some()
    }

    /// One deterministic clock value for every Fix executed by the current
    /// graph save. The task-local scope keeps concurrent saves on one context
    /// isolated; standalone checker calls receive their own current value.
    pub fn fix_time(&self) -> teaql_core::time::Timestamp {
        crate::entity_save::current_graph_fix_time()
    }

    pub fn record_fix_evidence(&self, evidence: FixEvidence) {
        crate::entity_save::record_graph_fix_evidence(evidence);
    }

    pub(crate) fn replace_last_fix_evidence(&self, evidence: Vec<FixEvidence>) {
        *self.last_fix_evidence.lock().unwrap() = evidence;
    }

    pub fn last_fix_evidence(&self) -> Vec<FixEvidence> {
        self.last_fix_evidence.lock().unwrap().clone()
    }

    pub fn check_and_fix_values(
        &self,
        entity: &str,
        values: &mut crate::EntityValues,
    ) -> Result<(), RuntimeError> {
        self.check_and_fix_values_at(entity, values, &ObjectLocation::root())
    }

    pub fn check_and_fix_values_at(
        &self,
        entity: &str,
        values: &mut crate::EntityValues,
        location: &ObjectLocation,
    ) -> Result<(), RuntimeError> {
        let status = CheckObjectStatus::from_values(values);
        let checker = self
            .checker_registry
            .as_ref()
            .and_then(|registry| registry.checker(entity));
        let mut results = CheckResults::new();
        if let Some(checker) = checker {
            checker.check_and_fix(self, values, location, &mut results);
        }

        // Custom checkers get the first chance to supply or fix a value.
        self.collect_required_property_results(entity, values, status, location, &mut results);
        if results.is_empty() {
            return Ok(());
        }
        self.translate_check_results(&mut results);
        Err(RuntimeError::Check(results))
    }

    /// Validate the sparse values that a ledger CREATE will actually insert.
    /// This is intentionally metadata-only: the complete typed graph already
    /// ran Checker/Fix once, and running it again on sparse changes would
    /// misclassify unchanged loaded fields and repeat Fix side effects.
    pub(crate) fn validate_required_create_payload(
        &self,
        entity: &str,
        values: &crate::EntityValues,
        location: &ObjectLocation,
    ) -> Result<(), RuntimeError> {
        let mut results = CheckResults::new();
        self.collect_required_property_results(
            entity,
            values,
            CheckObjectStatus::Create,
            location,
            &mut results,
        );
        if results.is_empty() {
            return Ok(());
        }
        self.translate_check_results(&mut results);
        Err(RuntimeError::Check(results))
    }

    fn collect_required_property_results(
        &self,
        entity: &str,
        values: &crate::EntityValues,
        status: CheckObjectStatus,
        location: &ObjectLocation,
        results: &mut CheckResults,
    ) {
        // Keep runtime validation aligned with the schema generated from the
        // same metadata. Every NOT NULL property must be present on a create,
        // and an update must not explicitly clear one.
        if let Some(descriptor) = self
            .metadata
            .as_ref()
            .and_then(|metadata| metadata.entity(entity))
        {
            for property in descriptor
                .properties
                .iter()
                // The optimistic-lock version is runtime-managed. Insert
                // preparation assigns its initial value after check/fix, so a
                // create caller must never be required to provide it.
                .filter(|property| !property.nullable && !property.is_version)
            {
                let missing = !values.contains_key(&property.name);
                let null = matches!(values.get(&property.name), Some(Value::Null));
                let property_location = location.clone().member(&property.name);
                let already_reported = results.iter().any(|result| {
                    result.rule == crate::CheckRule::Required
                        && result.location == property_location
                });
                if ((status.is_create() && missing) || null) && !already_reported {
                    results.push(CheckResult::required(property_location));
                }
            }
        }
    }

    pub fn translate_check_results(&self, results: &mut CheckResults) {
        for result in results {
            if result.message.is_none() {
                result.message = Some(
                    self.i18n_catalog
                        .translate_check_result(self.language, result),
                );
            }
        }
    }

    pub fn send_event(&self, mut event: RawAuditEvent) -> Result<(), RuntimeError> {
        if self.is_generated_schema_bootstrap()
            && matches!(
                event.kind,
                crate::RawAuditEventKind::Created | crate::RawAuditEventKind::Updated
            )
        {
            let reason = event
                .trace_chain
                .last()
                .map(|node| node.comment.clone())
                .unwrap_or_else(|| "generated runtime bootstrap".to_owned());
            let resulting_version = event
                .new_values
                .as_ref()
                .and_then(|values| values.get("version"))
                .or_else(|| event.values.get("version"))
                .and_then(teaql_core::Value::try_i64);
            let occurred_at_millis = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis() as u64;
            event.bootstrap_audit = Some(crate::BootstrapAuditIdentity {
                actor: "teaql-generated-bootstrap".to_owned(),
                category: "runtime-bootstrap".to_owned(),
                reason,
                resulting_version,
                occurred_at_millis,
            });
        }
        let scope = self.start_runtime_operation(
            crate::RuntimeOperation::new("audit", format!("{}.event", event.entity))
                .attribute("teaql.entity.type", event.entity.clone()),
        );
        let result = self.send_event_inner(event);
        match &result {
            Ok(()) => scope.success(std::collections::BTreeMap::new()),
            Err(_) => scope.failure("audit_error"),
        }
        result
    }

    fn send_event_inner(&self, event: RawAuditEvent) -> Result<(), RuntimeError> {
        if let Some(sink) = self.event_sink.as_ref() {
            sink.on_event(self, &event)?;
        }
        if let Some(sink) = self.custom_event_sink.as_ref() {
            let (mask_fields, max_len) = self
                .metadata
                .as_ref()
                .and_then(|metadata| metadata.entity(&event.entity))
                .map(|desc| (desc.audit_mask_fields.clone(), desc.audit_value_max_len))
                .unwrap_or_else(|| (vec![], None));

            let safe_event = event.build_safe_event(&mask_fields, max_len);
            sink.on_safe_event(self, &safe_event)?;
        }

        crate::log_formatter::LogManager::write_audit_log(&event);

        Ok(())
    }

    pub async fn get_in_store(&self, key: &str) -> Option<Value> {
        let store = self.get_resource::<Box<dyn DataStore>>()?;
        store.get(key).await
    }

    pub async fn put_in_store(
        &self,
        key: &str,
        value: impl Into<Value>,
        timeout_seconds: Option<u64>,
    ) {
        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
            store.put(key, value.into(), timeout_seconds).await;
        }
    }

    pub async fn clear_in_store(&self, key: &str) {
        if let Some(store) = self.get_resource::<Box<dyn DataStore>>() {
            store.remove(key).await;
        }
    }
}

#[cfg(test)]
mod sql_log_option_tests {
    use super::*;

    #[test]
    fn diagnostic_sql_log_is_enabled_by_default_with_independent_switches() {
        let mut context = UserContext::default();
        assert_eq!(context.sql_log_options(), SqlLogOptions::all());
        assert!(context.sql_logs().is_empty());

        context.disable_select_sql_log();
        assert_eq!(context.sql_log_options(), SqlLogOptions::mutation_only());

        context.enable_select_sql_log();
        context.disable_mutation_sql_log();
        assert_eq!(context.sql_log_options(), SqlLogOptions::select_only());

        context.disable_sql_log();
        assert_eq!(context.sql_log_options(), SqlLogOptions::disabled());
    }

    #[test]
    fn disabled_sql_log_rejects_executor_metadata_before_recording() {
        let mut context = UserContext::default();
        context.disable_sql_log();
        let now = SystemTime::now();
        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
            backend: "sql".to_owned(),
            operation: teaql_data_service::DataServiceOperation::Query,
            started_at: now,
            ended_at: now,
            affected_rows: None,
            result_count: Some(1),
            trace_chain: Vec::new(),
            comment: Some("disabled log test".to_owned()),
            backend_request_id: None,
            parameterized_query: Some("SELECT id FROM sample WHERE id = $1".to_owned()),
            params: vec![Value::I64(1)],
            debug_query: Some("SELECT id FROM sample WHERE id = 1".to_owned()),
        });
        assert!(context.sql_logs().is_empty());
    }

    #[test]
    fn default_sql_telemetry_never_retains_bound_values_or_copy_paste_sql() {
        let mut context = UserContext::default();
        let buffer = UnifiedLogBuffer::default();
        context.insert_resource(buffer.clone());
        let now = SystemTime::now();
        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
            backend: "sqlite".to_owned(),
            operation: teaql_data_service::DataServiceOperation::Query,
            started_at: now,
            ended_at: now,
            affected_rows: None,
            result_count: Some(1),
            trace_chain: Vec::new(),
            comment: Some("what: load an account".to_owned()),
            backend_request_id: None,
            parameterized_query: Some("SELECT id FROM account WHERE secret = ?".to_owned()),
            params: vec![Value::from("private-value-0123")],
            debug_query: Some(
                "SELECT id FROM account WHERE secret = 'private-value-0123'".to_owned(),
            ),
        });
        let entry = context.sql_logs().pop().expect("safe SQL log");
        assert_eq!(entry.sql, "SELECT id FROM account WHERE secret = ?");
        assert_eq!(entry.result_count, Some(1));
        assert_eq!(entry.parameter_count(), 1);
        assert_eq!(entry.params, vec![Value::Null]);
        assert!(entry.debug_sql.is_empty());
        assert!(entry.pretty_sql.is_empty());
        let buffered = buffer.entries.lock().unwrap();
        let LogPayload::Sql(buffered_sql) = &buffered[0].payload else {
            panic!("expected SQL log");
        };
        assert_eq!(buffered_sql, &entry);
        let ordinary_text = crate::log_formatter::LogFormatter::format_sql_log(
            &crate::log_formatter::HumanReaderFormatter,
            &entry.trace_path,
            &entry,
        );
        assert!(!ordinary_text.contains("private-value-0123"));
        assert!(!ordinary_text.contains("Debug SQL:"));
        let mut diagnostic_entry = entry.clone();
        diagnostic_entry.params = vec![Value::from("private-value-0123")];
        diagnostic_entry.debug_sql =
            "SELECT id FROM account WHERE secret = 'private-value-0123'".to_owned();
        let diagnostic_text = crate::log_formatter::LogFormatter::format_sql_log(
            &crate::log_formatter::HumanReaderFormatter,
            &diagnostic_entry.trace_path,
            &diagnostic_entry,
        );
        assert!(diagnostic_text.contains("Debug SQL:"));
        assert!(diagnostic_text.contains("private-value-0123"));
    }

    #[test]
    fn metadata_log_retains_structured_intent_sql_forms_and_multilevel_trace() {
        let context = UserContext::default();
        let now = SystemTime::now();
        let query_trace = vec![
            teaql_core::TraceNode::typed(
                teaql_core::TraceKind::Comment,
                "School",
                None,
                "what: load school graph",
            ),
            teaql_core::TraceNode::typed(
                teaql_core::TraceKind::Purpose,
                "School",
                None,
                "why: render school details",
            ),
            teaql_core::TraceNode::typed(
                teaql_core::TraceKind::Relation,
                "platform",
                None,
                "School.platform",
            ),
            teaql_core::TraceNode::typed(
                teaql_core::TraceKind::Relation,
                "organization",
                None,
                "Platform.organization",
            ),
            teaql_core::TraceNode::typed(
                teaql_core::TraceKind::Relation,
                "region",
                None,
                "Organization.region",
            ),
        ];
        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
            backend: "sqlite".to_owned(),
            operation: teaql_data_service::DataServiceOperation::Query,
            started_at: now,
            ended_at: now,
            affected_rows: None,
            result_count: Some(2),
            trace_chain: query_trace.clone(),
            comment: None,
            backend_request_id: None,
            parameterized_query: Some("SELECT name FROM school_data WHERE id = ?".to_owned()),
            params: vec![Value::I64(7)],
            debug_query: Some("SELECT name FROM school_data WHERE id = 7".to_owned()),
        });

        let query = context.sql_logs().pop().expect("query log");
        assert_eq!(query.comment.as_deref(), Some("what: load school graph"));
        assert_eq!(query.purpose.as_deref(), Some("why: render school details"));
        assert_eq!(query.audit_reason, None);
        assert_eq!(
            query
                .trace_path
                .iter()
                .map(|node| node.kind)
                .collect::<Vec<_>>(),
            vec![
                teaql_core::TraceKind::Operation,
                teaql_core::TraceKind::Request,
                teaql_core::TraceKind::Relation,
                teaql_core::TraceKind::Relation,
                teaql_core::TraceKind::Relation,
                teaql_core::TraceKind::Provider,
                teaql_core::TraceKind::Sql,
            ]
        );
        assert_eq!(query.trace_path[0].entity_type, "School");
        assert_eq!(query.trace_path[5].entity_type, "sqlite");
        assert_eq!(query.trace_path[6].entity_type, "select");
        assert_eq!(query.sql, "SELECT name FROM school_data WHERE id = ?");
        assert_eq!(query.parameter_count(), 1);
        assert_eq!(query.params, vec![Value::Null]);
        assert!(query.debug_sql.is_empty());
        assert_eq!(query.result_count, Some(2));

        let mutation_trace = vec![teaql_core::TraceNode::typed(
            teaql_core::TraceKind::AuditReason,
            "School",
            Some(7),
            "correct school name",
        )];
        context.record_metadata_log(&teaql_data_service::ExecutionMetadata {
            backend: "sqlite".to_owned(),
            operation: teaql_data_service::DataServiceOperation::Update,
            started_at: now,
            ended_at: now,
            affected_rows: Some(1),
            result_count: None,
            trace_chain: mutation_trace.clone(),
            comment: None,
            backend_request_id: None,
            parameterized_query: Some("UPDATE school_data SET name = ? WHERE id = ?".to_owned()),
            params: vec![Value::from("Academy"), Value::I64(7)],
            debug_query: Some("UPDATE school_data SET name = 'Academy' WHERE id = 7".to_owned()),
        });
        let mutation = context.sql_logs().pop().expect("mutation log");
        assert_eq!(mutation.parameter_count(), 2);
        assert_eq!(mutation.params, vec![Value::Null, Value::Null]);
        assert_eq!(mutation.comment, None);
        assert_eq!(mutation.purpose, None);
        assert_eq!(
            mutation.audit_reason.as_deref(),
            Some("correct school name")
        );
        assert_eq!(
            mutation
                .trace_path
                .iter()
                .map(|node| node.kind)
                .collect::<Vec<_>>(),
            vec![
                teaql_core::TraceKind::Operation,
                teaql_core::TraceKind::Entity,
                teaql_core::TraceKind::Provider,
                teaql_core::TraceKind::Sql,
            ]
        );
        assert_eq!(mutation.affected_rows, Some(1));
        assert!(mutation.debug_sql.is_empty());
    }
}

#[cfg(test)]
mod entity_runtime_state_tests {
    use super::*;
    use crate::EntityKey;

    #[test]
    fn reused_user_context_returns_independent_mutation_ledgers() {
        let context = UserContext::default();
        let first = context.entity_runtime_state();
        let key = EntityKey::new("School", 1_u64);
        first.set(key.clone(), "name", "First");

        let second = context.entity_runtime_state();

        assert_eq!(first.changed_field_names(&key).len(), 1);
        assert!(second.changed_field_names(&key).is_empty());
    }
}