pharos_native/runtime/feature_runtime/

logical.rs

//! Declared feature execution through the configured logical database adapter.

use pharos_runtime_architect::database::ArchitectStore;
use pharos_runtime_database::DbBackendConfig;
use pharos_runtime_database::operations::{
    AppSchema, DatabaseError, DatabaseOperations, FieldDefault, FieldType, Mutation, Predicate,
    Query, ReservationExpiry, SortKey, Transaction, WriteValue,
};
use serde_json::{Map, Value};
use std::collections::{BTreeMap, BTreeSet};

#[cfg(not(target_os = "wasi"))]
use super::populate_generated_values;
use super::{
    FeatureRequestContext, FeatureRuntimeError, query_binding_value, resolve_value,
    transaction_operations,
};
use crate::config::AppRuntime;

/// Keeps one selected adapter and schema for the whole declared feature call.
enum LogicalStore {
    Architect(ArchitectStore),
    App(crate::runtime::database::LogicalAppStore),
}

impl LogicalStore {
    fn config(&self) -> &DbBackendConfig {
        match self {
            Self::Architect(store) => store.config(),
            Self::App(store) => &store.config,
        }
    }

    fn schema(&self) -> &AppSchema {
        match self {
            Self::Architect(store) => store.schema(),
            Self::App(store) => &store.schema,
        }
    }

    fn adapter(&self) -> &dyn DatabaseOperations {
        match self {
            Self::Architect(store) => store.adapter(),
            Self::App(store) => store.adapter.as_ref(),
        }
    }
}

/// Opens the app's logical entity catalog through its configured backend.
fn store_for_app(app: &AppRuntime) -> Result<LogicalStore, FeatureRuntimeError> {
    let app_id = app
        .app_id
        .as_deref()
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("app id".to_owned()))?;
    let store =
        crate::runtime::database::runtime_logical_app_store(app).map_err(map_database_error)?;
    if app_id == "architect" {
        return ArchitectStore::new(store.config, store.adapter, store.schema)
            .map(LogicalStore::Architect)
            .map_err(map_database_error);
    }
    if app.is_editor_preview && ArchitectStore::supports_preview_schema(&store.schema) {
        return ArchitectStore::new_isolated_preview(store.config, store.adapter, store.schema)
            .map(LogicalStore::Architect)
            .map_err(map_database_error);
    }
    Ok(LogicalStore::App(store))
}

/// Preserves the logical database error category at the declared feature boundary.
pub(crate) fn map_database_error(error: DatabaseError) -> FeatureRuntimeError {
    match error {
        DatabaseError::MissingRecord => {
            FeatureRuntimeError::MissingRecord("database record".to_owned())
        }
        DatabaseError::Conflict => FeatureRuntimeError::Conflict("database conflict".to_owned()),
        DatabaseError::InvalidInput => {
            FeatureRuntimeError::InvalidInput("database operation".to_owned())
        }
        _ => FeatureRuntimeError::Database(error),
    }
}

/// Converts HTML form booleans into the logical field type before any adapter sees them.
fn field_value(
    schema: &AppSchema,
    entity: &str,
    field: &str,
    value: Value,
) -> Result<Value, FeatureRuntimeError> {
    let Some(declaration) = schema
        .entities
        .iter()
        .find(|entry| entry.name == entity)
        .and_then(|entry| entry.fields.iter().find(|entry| entry.name == field))
    else {
        return Ok(value);
    };
    if declaration.kind == FieldType::Integer
        && let Value::String(text) = &value
    {
        if declaration.nullable && text.trim().is_empty() {
            return Ok(Value::Null);
        }
        return text
            .parse::<i64>()
            .map(Value::from)
            .map_err(|_| FeatureRuntimeError::InvalidInput(field.to_owned()));
    }
    if declaration.kind != FieldType::Boolean || value.is_null() || value.is_boolean() {
        return Ok(value);
    }
    match value {
        Value::Number(number) if number.as_i64() == Some(0) => Ok(Value::Bool(false)),
        Value::Number(number) if number.as_i64() == Some(1) => Ok(Value::Bool(true)),
        Value::String(text) if text == "0" => Ok(Value::Bool(false)),
        Value::String(text) if text == "1" => Ok(Value::Bool(true)),
        _ => Err(FeatureRuntimeError::InvalidInput(field.to_owned())),
    }
}

/// Resolves one declared field filter into a typed backend-neutral predicate.
pub(super) fn predicate_for_filters(
    filters: Option<&Map<String, Value>>,
    context: &FeatureRequestContext,
    schema: &AppSchema,
    entity: &str,
) -> Result<Option<Predicate>, FeatureRuntimeError> {
    let mut parts = Vec::new();
    for (field, declaration) in filters.into_iter().flat_map(|values| values.iter()) {
        let operator = declaration
            .get("operator")
            .and_then(Value::as_str)
            .unwrap_or("eq");
        if operator == "is_null" {
            let predicate = Predicate::IsNull(field.clone());
            let declaration = pharos_runtime_database::operations::entity(schema, entity)
                .map_err(map_database_error)?;
            pharos_runtime_database::operations::validate_predicate(declaration, &predicate)
                .map_err(map_database_error)?;
            parts.push(predicate);
            continue;
        }
        let source = declaration.get("value").unwrap_or(declaration);
        let optional = declaration
            .get("optional")
            .and_then(Value::as_bool)
            .unwrap_or(false);
        let Some(value) = resolve_value(source, context)? else {
            if optional {
                continue;
            }
            return Err(FeatureRuntimeError::InvalidInput(field.clone()));
        };
        if optional && value.as_str() == Some("") {
            continue;
        }
        if operator == "in" {
            let values = value
                .as_array()
                .filter(|values| values.len() <= 64)
                .ok_or_else(|| FeatureRuntimeError::InvalidInput(field.clone()))?;
            let values = values
                .iter()
                .map(|value| field_value(schema, entity, field, value.clone()))
                .collect::<Result<Vec<_>, _>>()?;
            let predicate = Predicate::In(field.clone(), values);
            let declaration = pharos_runtime_database::operations::entity(schema, entity)
                .map_err(map_database_error)?;
            pharos_runtime_database::operations::validate_predicate(declaration, &predicate)
                .map_err(map_database_error)?;
            parts.push(predicate);
            continue;
        }
        let value = field_value(schema, entity, field, value)?;
        let part = match operator {
            "eq" => Predicate::Equal(field.clone(), value),
            "ne" => Predicate::NotEqual(field.clone(), value),
            "gt" => Predicate::GreaterThan(field.clone(), value),
            "gte" => Predicate::AtLeast(field.clone(), value),
            "lt" => Predicate::LessThan(field.clone(), value),
            "lte" => Predicate::AtMost(field.clone(), value),
            "contains" => Predicate::Contains(
                field.clone(),
                value
                    .as_str()
                    .ok_or_else(|| FeatureRuntimeError::InvalidInput(field.clone()))?
                    .to_owned(),
            ),
            _ => {
                return Err(FeatureRuntimeError::InvalidContract(format!(
                    "{field}.{operator}"
                )));
            }
        };
        parts.push(part);
    }
    Ok(match parts.len() {
        0 => None,
        1 => parts.pop(),
        _ => Some(Predicate::All(parts)),
    })
}

/// Resolves declared reservation filters without losing an ensured row identity.
fn predicate_for_transaction_filters(
    filters: &Map<String, Value>,
    context: &FeatureRequestContext,
    bindings: &BTreeMap<String, usize>,
    schema: &AppSchema,
    entity: &str,
) -> Result<Predicate, FeatureRuntimeError> {
    if filters.is_empty() || filters.len() > 32 {
        return Err(FeatureRuntimeError::InvalidContract(
            "transaction filters".to_owned(),
        ));
    }
    let mut parts = Vec::with_capacity(filters.len());
    for (field, declaration) in filters {
        let source = declaration.get("value").unwrap_or(declaration);
        let generated = source
            .as_str()
            .and_then(|text| text.strip_prefix("query."))
            .and_then(|path| path.strip_suffix(".id"))
            .and_then(|name| bindings.get(name));
        if let Some(step) = generated {
            if declaration
                .get("operator")
                .and_then(Value::as_str)
                .is_some_and(|operator| operator != "eq")
            {
                return Err(FeatureRuntimeError::InvalidContract(
                    "generated identity filter operator".to_owned(),
                ));
            }
            parts.push(Predicate::GeneratedEqual(field.clone(), *step));
            continue;
        }
        let single = Map::from_iter([(field.clone(), declaration.clone())]);
        let part = predicate_for_filters(Some(&single), context, schema, entity)?
            .ok_or_else(|| FeatureRuntimeError::InvalidContract("transaction filter".to_owned()))?;
        parts.push(part);
    }
    Ok(if parts.len() == 1 {
        parts.remove(0)
    } else {
        Predicate::All(parts)
    })
}

/// Reuses a bounded, trusted request-local query binding without another storage read.
/// Persistence still checks its captured row values inside the mutation transaction.
fn working_query_rows(
    context: &FeatureRequestContext,
    query_id: &str,
) -> Result<Option<Vec<Map<String, Value>>>, FeatureRuntimeError> {
    let Some(binding) = context.bindings.get(query_id) else {
        return Ok(None);
    };
    let rows = match binding {
        Value::Object(row) => vec![row.clone()],
        Value::Array(rows) if rows.len() <= 100 => rows
            .iter()
            .map(|row| {
                row.as_object().cloned().ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("working query row".to_owned())
                })
            })
            .collect::<Result<Vec<_>, _>>()?,
        Value::Null => Vec::new(),
        _ => {
            return Err(FeatureRuntimeError::InvalidContract(
                "working query binding".to_owned(),
            ));
        }
    };
    let bytes = serde_json::to_vec(&rows)
        .map_err(|_| FeatureRuntimeError::InvalidContract("working query encoding".to_owned()))?;
    if bytes.len() > 8 * 1024 * 1024 {
        return Err(FeatureRuntimeError::Database(
            DatabaseError::CapacityExceeded,
        ));
    }
    Ok(Some(rows))
}

/// Executes declared feature queries without compiling SQL in the request path.
pub(super) fn execute_logical_queries(
    app: &AppRuntime,
    feature: &Value,
    query_ids: &[String],
    context: &FeatureRequestContext,
) -> Result<BTreeMap<String, Vec<Map<String, Value>>>, FeatureRuntimeError> {
    if query_ids.len() > 32 {
        return Err(FeatureRuntimeError::InvalidContract(
            "query count".to_owned(),
        ));
    }
    let store = store_for_app(app)?;
    let declarations = feature
        .get("queries")
        .and_then(Value::as_object)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("feature.queries".to_owned()))?;
    let mut output = BTreeMap::new();
    let mut scoped = context.clone();
    let mut owned_tasks = None;
    for query_id in query_ids {
        let declaration = declarations
            .get(query_id)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.clone()))?;
        if let Some(rows) = working_query_rows(&scoped, query_id)? {
            output.insert(query_id.clone(), rows);
            continue;
        }
        let entity = declaration
            .get("entity")
            .and_then(Value::as_str)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{query_id}.entity")))?;
        let fields = declaration
            .get("fields")
            .and_then(Value::as_array)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{query_id}.fields")))?
            .iter()
            .map(|value| {
                value.as_str().map(str::to_owned).ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract(format!("{query_id}.fields"))
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        if fields.is_empty() || fields.len() > 64 {
            return Err(FeatureRuntimeError::InvalidContract(query_id.clone()));
        }
        let limit = declaration.get("limit").and_then(Value::as_u64).unwrap_or(
            if declaration.get("result").and_then(Value::as_str) == Some("many") {
                24
            } else {
                1
            },
        );
        if limit == 0 || limit > 1000 {
            return Err(FeatureRuntimeError::InvalidContract(format!(
                "{query_id}.limit"
            )));
        }
        let order = declaration
            .get("order_by")
            .and_then(Value::as_array)
            .into_iter()
            .flatten()
            .map(|item| {
                Ok(SortKey {
                    field: item
                        .get("field")
                        .and_then(Value::as_str)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract(format!("{query_id}.order"))
                        })?
                        .to_owned(),
                    descending: item.get("direction").and_then(Value::as_str) == Some("desc"),
                })
            })
            .collect::<Result<Vec<_>, FeatureRuntimeError>>()?;
        let mut filter = predicate_for_filters(
            declaration.get("filters").and_then(Value::as_object),
            &scoped,
            store.schema(),
            entity,
        )?;
        if let Some(scope) = declaration.get("actor_task_field") {
            let field = scope.as_str().ok_or_else(|| {
                FeatureRuntimeError::InvalidContract(format!("{query_id}.actor_task_field"))
            })?;
            if !matches!(
                (entity, field),
                ("architect_task", "id")
                    | ("architect_task_event", "task_id")
                    | ("architect_context_item_view", "task_id")
            ) {
                return Err(FeatureRuntimeError::InvalidContract(format!(
                    "{query_id}.actor_task_field"
                )));
            }
            let LogicalStore::Architect(architect) = &store else {
                return Err(FeatureRuntimeError::InvalidContract(entity.to_owned()));
            };
            if owned_tasks.is_none() {
                owned_tasks = Some(
                    pharos_runtime_architect::task_ownership::owned_task_ids(
                        architect,
                        context.actor_id,
                    )
                    .map_err(map_database_error)?,
                );
            }
            let ids = owned_tasks
                .as_ref()
                .ok_or_else(|| FeatureRuntimeError::InvalidContract("task ownership".to_owned()))?;
            let current_ids = if declaration.get("latest_app_attempt") == Some(&Value::Bool(true)) {
                if entity != "architect_task" || field != "id" {
                    return Err(FeatureRuntimeError::InvalidContract(format!(
                        "{query_id}.latest_app_attempt"
                    )));
                }
                Some(
                    pharos_runtime_architect::task_ownership::latest_app_task_ids(architect, ids)
                        .map_err(map_database_error)?,
                )
            } else {
                None
            };
            let owner_filter = pharos_runtime_architect::task_ownership::id_filter(
                field,
                current_ids.as_ref().unwrap_or(ids),
            )
            .map_err(map_database_error)?;
            filter = Some(match filter {
                Some(filter) => Predicate::All(vec![filter, owner_filter]),
                None => owner_filter,
            });
        }
        let mut rows = if entity.ends_with("_view") {
            let LogicalStore::Architect(architect) = &store else {
                return Err(FeatureRuntimeError::InvalidContract(entity.to_owned()));
            };
            pharos_runtime_architect::projections::query_view(
                architect,
                entity,
                filter,
                &order,
                limit as u32,
            )
            .map_err(map_database_error)?
        } else {
            store
                .adapter()
                .query(
                    store.config(),
                    store.schema(),
                    &Query {
                        entity: entity.to_owned(),
                        fields: fields.clone(),
                        filter,
                        order,
                        limit: limit as u32,
                        offset: 0,
                    },
                )
                .map_err(map_database_error)?
        };
        if query_id == "recent_tasks"
            && entity == "architect_task"
            && let LogicalStore::Architect(architect) = &store
        {
            let ids = rows
                .iter()
                .filter_map(|row| row.get("id").and_then(Value::as_i64))
                .collect::<Vec<_>>();
            let metadata =
                pharos_runtime_architect::task_repository::task_action_metadata(architect, &ids)
                    .map_err(|_| {
                        FeatureRuntimeError::InvalidContract(
                            "recent_tasks.accepted_source".to_owned(),
                        )
                    })?;
            for row in &mut rows {
                if let Some(task) = row
                    .get("id")
                    .and_then(Value::as_i64)
                    .and_then(|id| metadata.get(&id))
                    && !task.validation_summary.is_empty()
                {
                    row.insert(
                        "validation_summary".to_owned(),
                        Value::String(task.validation_summary.clone()),
                    );
                }
                let accepted = row
                    .get("id")
                    .and_then(Value::as_i64)
                    .and_then(|id| metadata.get(&id))
                    .is_some_and(|task| task.validated);
                row.insert(
                    "accepted_source_status".to_owned(),
                    Value::String(
                        if accepted {
                            "validated"
                        } else {
                            "not validated"
                        }
                        .to_owned(),
                    ),
                );
                row.insert(
                    "validate_accepted_source".to_owned(),
                    Value::Bool(
                        accepted
                            && row
                                .get("id")
                                .and_then(Value::as_i64)
                                .and_then(|id| metadata.get(&id))
                                .is_some_and(|task| task.source_validation_due)
                            && row.get("task_status").and_then(Value::as_str) == Some("queued")
                            && row.get("app_model_digest").and_then(Value::as_str)
                                == Some("pending")
                            && row.get("brief").and_then(Value::as_str)
                                == Some(
                                    pharos_runtime_architect::task_repository::EMPTY_INTAKE_BRIEF,
                                ),
                    ),
                );
            }
            let candidates = rows
                .iter()
                .filter(|row| {
                    row.get("validate_accepted_source").and_then(Value::as_bool) == Some(true)
                })
                .filter_map(|row| row.get("id").and_then(Value::as_i64))
                .filter_map(|id| {
                    metadata
                        .get(&id)
                        .map(|task| (id, task.source_version_created_at.clone()))
                })
                .collect::<Vec<_>>();
            let eligible = pharos_runtime_architect::source_validation_inputs::snapshot_guards(
                architect,
                &candidates,
            )
            .map_err(|_| {
                FeatureRuntimeError::InvalidContract("recent_tasks.pending_context".to_owned())
            })?;
            for row in &mut rows {
                let allowed = row.get("validate_accepted_source").and_then(Value::as_bool)
                    == Some(true)
                    && row
                        .get("id")
                        .and_then(Value::as_i64)
                        .is_some_and(|id| eligible.contains_key(&id));
                row.insert(
                    "validate_accepted_source".to_owned(),
                    Value::String(if allowed { "true" } else { "false" }.to_owned()),
                );
            }
        }
        for row in &rows {
            if fields.iter().any(|field| !row.contains_key(field)) {
                return Err(FeatureRuntimeError::InvalidContract(format!(
                    "{query_id}.fields"
                )));
            }
        }
        let result_kind = declaration
            .get("result")
            .and_then(Value::as_str)
            .unwrap_or("many");
        scoped.bindings.insert(
            query_id.clone(),
            query_binding_value(query_id, result_kind, &rows)?,
        );
        output.insert(query_id.clone(), rows);
    }
    Ok(output)
}

/// Compiles declared write values while preserving a prior generated-id binding.
fn write_values(
    values: &Map<String, Value>,
    context: &FeatureRequestContext,
    bindings: &BTreeMap<String, usize>,
    row_generated: &BTreeMap<String, WriteValue>,
    schema: &AppSchema,
    entity: &str,
) -> Result<BTreeMap<String, WriteValue>, FeatureRuntimeError> {
    values
        .iter()
        .map(|(field, source)| {
            if let Some(path) = source.as_str().and_then(|text| text.strip_prefix("query."))
                && let Some(name) = path.strip_suffix(".id")
                && let Some(step) = bindings.get(name)
            {
                return Ok((field.clone(), WriteValue::GeneratedId(*step)));
            }
            if let Some(name) = source.as_str().and_then(|text| text.strip_prefix("row."))
                && let Some(step) = row_generated.get(name)
            {
                return Ok((field.clone(), step.clone()));
            }
            let value = resolve_value(source, context)?
                .ok_or_else(|| FeatureRuntimeError::InvalidInput(field.clone()))?;
            Ok((
                field.clone(),
                WriteValue::Literal(field_value(schema, entity, field, value)?),
            ))
        })
        .collect()
}

#[derive(Clone)]
struct InsertedRow {
    fields: Map<String, Value>,
    generated: BTreeMap<String, WriteValue>,
}

/// Distinguishes unrelated prospective writes from changes to a query's own entity.
fn planned_write_targets(steps: &[Mutation], entity: &str) -> bool {
    steps.iter().any(|step| match step {
        Mutation::Insert { entity: target, .. }
        | Mutation::InsertIfAbsent { entity: target, .. }
        | Mutation::Upsert { entity: target, .. }
        | Mutation::Update { entity: target, .. }
        | Mutation::Delete { entity: target, .. }
        | Mutation::Reserve { entity: target, .. } => target == entity,
        Mutation::RequireCount { .. } => false,
    })
}

/// Matches a declared equality filter against a transaction-local write.
fn local_row_matches(
    row: &InsertedRow,
    filters: &Map<String, Value>,
    context: &FeatureRequestContext,
    bindings: &BTreeMap<String, usize>,
) -> Result<bool, FeatureRuntimeError> {
    for (field, source) in filters {
        if let Some(path) = source.as_str().and_then(|text| text.strip_prefix("query."))
            && let Some(name) = path.strip_suffix(".id")
            && let Some(step) = bindings.get(name)
        {
            if row.generated.get(field) != Some(&WriteValue::GeneratedId(*step)) {
                return Ok(false);
            }
            continue;
        }
        let value = resolve_value(source, context)?.ok_or_else(|| {
            FeatureRuntimeError::InvalidContract("local select filter".to_owned())
        })?;
        if row.fields.get(field) != Some(&value) {
            return Ok(false);
        }
    }
    Ok(true)
}

fn inserted_row(
    values: &BTreeMap<String, WriteValue>,
    identity: Option<&str>,
    step: usize,
    ensured_unique_key: Option<&[String]>,
) -> InsertedRow {
    let mut fields = Map::new();
    let mut generated = BTreeMap::new();
    for (name, value) in values {
        match value {
            WriteValue::Literal(value) => {
                fields.insert(name.clone(), value.clone());
            }
            WriteValue::GeneratedId(_)
            | WriteValue::GeneratedText { .. }
            | WriteValue::GeneratedJson { .. } => {
                fields.insert(name.clone(), Value::Null);
                generated.insert(name.clone(), value.clone());
            }
        }
    }
    if let Some(identity) = identity {
        fields.insert(identity.to_owned(), Value::Null);
        generated.insert(identity.to_owned(), WriteValue::GeneratedId(step));
    }
    if let Some(unique_key) = ensured_unique_key {
        fields.retain(|name, _| Some(name.as_str()) == identity || unique_key.contains(name));
        generated.retain(|name, _| Some(name.as_str()) == identity || unique_key.contains(name));
    }
    InsertedRow { fields, generated }
}

/// Completes a planned insert using declared defaults so local reads match its write.
fn complete_insert_values(
    schema: &AppSchema,
    entity: &str,
    values: &mut BTreeMap<String, WriteValue>,
    context: &FeatureRequestContext,
) -> Result<(), FeatureRuntimeError> {
    let declaration = schema
        .entities
        .iter()
        .find(|entry| entry.name == entity)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(entity.to_owned()))?;
    for field in &declaration.fields {
        if values.contains_key(&field.name)
            || declaration.identity.generated_field() == Some(field.name.as_str())
        {
            continue;
        }
        let value = match &field.default {
            Some(FieldDefault::Literal(value)) => value.clone(),
            Some(FieldDefault::CurrentTimestamp) => {
                resolve_value(&Value::String("server.now".to_owned()), context)?
                    .ok_or_else(|| FeatureRuntimeError::InvalidContract(field.name.clone()))?
            }
            None if field.nullable => Value::Null,
            None => continue,
        };
        values.insert(
            field.name.clone(),
            WriteValue::Literal(field_value(schema, entity, &field.name, value)?),
        );
    }
    Ok(())
}

/// Captures an ensured row through the interface and guards its snapshot before writes.
fn ensured_existing_row(
    store: &LogicalStore,
    entity: &str,
    unique_key: &[String],
    values: &BTreeMap<String, WriteValue>,
    steps: &mut Vec<Mutation>,
) -> Result<Option<Map<String, Value>>, FeatureRuntimeError> {
    let predicates = unique_key
        .iter()
        .map(|field| {
            let Some(WriteValue::Literal(value)) = values.get(field) else {
                return Err(FeatureRuntimeError::InvalidContract(
                    "ensure requires a resolved unique key".to_owned(),
                ));
            };
            Ok(Predicate::Equal(field.clone(), value.clone()))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let filter = Predicate::All(predicates);
    let declaration = store
        .schema()
        .entities
        .iter()
        .find(|entry| entry.name == entity)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(entity.to_owned()))?;
    let mut rows = store
        .adapter()
        .query(
            store.config(),
            store.schema(),
            &Query {
                entity: entity.to_owned(),
                fields: declaration
                    .fields
                    .iter()
                    .map(|field| field.name.clone())
                    .collect(),
                filter: Some(filter.clone()),
                order: Vec::new(),
                limit: 2,
                offset: 0,
            },
        )
        .map_err(map_database_error)?;
    if rows.len() > 1 {
        return Err(FeatureRuntimeError::InvalidContract(
            "ensure unique key matched multiple records".to_owned(),
        ));
    }
    steps.push(Mutation::RequireCount {
        entity: entity.to_owned(),
        filter,
        minimum: rows.len() as u64,
        maximum: Some(rows.len() as u64),
    });
    if let Some(row) = rows.pop() {
        append_read_guards(store.schema(), entity, &[&row], steps)?;
        Ok(Some(row))
    } else {
        Ok(None)
    }
}

/// Builds a bounded reservation status transition from app-owned declarations.
fn reservation_transition(
    kind: &str,
    operation: &Value,
    context: &FeatureRequestContext,
    schema: &AppSchema,
) -> Result<(Mutation, Mutation), FeatureRuntimeError> {
    let entity = operation
        .get("entity")
        .and_then(Value::as_str)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.entity")))?;
    let status_field = operation
        .get("status_field")
        .and_then(Value::as_str)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.status_field")))?;
    let statuses = operation
        .get("active_statuses")
        .and_then(Value::as_array)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.active_statuses")))?;
    if statuses.is_empty()
        || statuses.len() > 16
        || statuses.iter().any(|status| {
            status
                .as_str()
                .is_none_or(|text| text.is_empty() || text.len() > 64)
        })
    {
        return Err(FeatureRuntimeError::InvalidContract(format!(
            "{kind}.active_statuses"
        )));
    }
    let owner = predicate_for_filters(
        operation.get("owner_filters").and_then(Value::as_object),
        context,
        schema,
        entity,
    )?
    .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.owner_filters")))?;
    let scope = predicate_for_filters(
        operation.get("scope_filters").and_then(Value::as_object),
        context,
        schema,
        entity,
    )?
    .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.scope_filters")))?;
    let capacity = operation
        .get("capacity")
        .and_then(Value::as_u64)
        .filter(|value| *value > 0 && *value <= 100_000)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.capacity")))?;
    let status = operation
        .get("to_status")
        .and_then(Value::as_str)
        .filter(|text| !text.is_empty() && text.len() <= 64)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract(format!("{kind}.to_status")))?;
    let filter = Predicate::All(vec![
        owner,
        scope,
        Predicate::In(status_field.to_owned(), statuses.clone()),
    ]);
    let mut values = operation
        .get("values")
        .and_then(Value::as_object)
        .map(|values| {
            write_values(
                values,
                context,
                &BTreeMap::new(),
                &BTreeMap::new(),
                schema,
                entity,
            )
        })
        .transpose()?
        .unwrap_or_default();
    values.insert(
        status_field.to_owned(),
        WriteValue::Literal(Value::String(status.to_owned())),
    );
    Ok((
        Mutation::RequireCount {
            entity: entity.to_owned(),
            filter: filter.clone(),
            minimum: 0,
            maximum: Some(capacity),
        },
        Mutation::Update {
            entity: entity.to_owned(),
            filter,
            values,
            expected_rows: None,
        },
    ))
}

/// Reuses a saved result only when every declared idempotency field still matches.
fn prior_idempotent_result(
    app: &AppRuntime,
    feature: &Value,
    idempotency: &Value,
    context: &FeatureRequestContext,
) -> Result<bool, FeatureRuntimeError> {
    let query_id = idempotency
        .get("query")
        .and_then(Value::as_str)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("idempotency.query".to_owned()))?;
    let matches = idempotency
        .get("matches")
        .and_then(Value::as_object)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("idempotency.matches".to_owned()))?;
    let rows = execute_logical_queries(app, feature, &[query_id.to_owned()], context)?;
    let Some(row) = rows.get(query_id).and_then(|rows| rows.first()) else {
        return Ok(false);
    };
    for (field, expected_source) in matches {
        let expected = resolve_value(expected_source, context)?
            .ok_or_else(|| FeatureRuntimeError::InvalidInput(field.to_owned()))?;
        if row.get(field) != Some(&expected) {
            return Err(FeatureRuntimeError::Conflict(
                "idempotency key reused with different input".to_owned(),
            ));
        }
    }
    Ok(true)
}

/// Binds a read set in bounded groups without consuming one transaction step per row.
/// Grouping requires complete identities so matching duplicates cannot hide a stale row.
fn append_read_guards(
    schema: &AppSchema,
    entity_name: &str,
    rows: &[&Map<String, Value>],
    steps: &mut Vec<Mutation>,
) -> Result<(), FeatureRuntimeError> {
    if rows.len() > 1000 {
        return Err(map_database_error(DatabaseError::CapacityExceeded));
    }
    let typed_rows = rows
        .iter()
        .map(|row| {
            row.iter()
                .map(|(field, value)| {
                    Ok((
                        field.clone(),
                        field_value(schema, entity_name, field, value.clone())?,
                    ))
                })
                .collect::<Result<Map<String, Value>, FeatureRuntimeError>>()
        })
        .collect::<Result<Vec<_>, _>>()?;
    let identity = pharos_runtime_database::operations::entity(schema, entity_name)
        .map_err(map_database_error)?
        .identity
        .fields();
    if typed_rows
        .iter()
        .all(|row| identity.iter().all(|field| row.contains_key(*field)))
    {
        let mut seen = BTreeSet::new();
        for row in &typed_rows {
            let key = identity
                .iter()
                .map(|field| &row[*field])
                .collect::<Vec<_>>();
            let digest = pharos_runtime_database::operations::operation_fingerprint(&key)
                .map_err(map_database_error)?;
            if !seen.insert(digest) {
                return Err(map_database_error(DatabaseError::InvalidInput));
            }
        }
    }
    let mut planned = Vec::new();
    for chunk in typed_rows.chunks(100) {
        let references = chunk.iter().collect::<Vec<_>>();
        planned.extend(
            pharos_runtime_database::operations::guard_read_set(schema, entity_name, &references)
                .map_err(map_database_error)?,
        );
    }
    if steps.len().saturating_add(planned.len())
        > pharos_runtime_database::operations::MAX_TRANSACTION_STEPS
    {
        return Err(map_database_error(DatabaseError::CapacityExceeded));
    }
    steps.extend(planned);
    Ok(())
}

/// Checks declared guards before writes and binds their read set into the transaction.
#[cfg(not(target_os = "wasi"))]
fn bind_guard_preconditions(
    app: &AppRuntime,
    feature: &Value,
    guard_contract: Option<&Value>,
    context: &mut FeatureRequestContext,
    schema: &AppSchema,
    steps: &mut Vec<Mutation>,
    action_id: &str,
) -> Result<(), FeatureRuntimeError> {
    let Some(contract) = guard_contract.filter(|value| !value.is_null()) else {
        return Ok(());
    };
    let guards = contract
        .get("guards")
        .and_then(Value::as_array)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("guard contract".to_owned()))?;
    if guards.len() > 16 {
        return Err(FeatureRuntimeError::InvalidContract(
            "guard count".to_owned(),
        ));
    }
    let queries = feature
        .get("queries")
        .and_then(Value::as_object)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("feature.queries".to_owned()))?;
    let mut guarded = BTreeSet::new();
    for guard in guards {
        let query_id = guard
            .get("query")
            .and_then(Value::as_str)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract("guard.query".to_owned()))?;
        if !guarded.insert(query_id.to_owned()) {
            continue;
        }
        let declaration = queries
            .get(query_id)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
        let entity = declaration
            .get("entity")
            .and_then(Value::as_str)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
        let generated = BTreeMap::new();
        for dependency in operation_query_references(guard)?
            .into_iter()
            .chain(std::iter::once(query_id.to_owned()))
        {
            QueryDependencyPlan {
                app,
                feature,
                context,
                generated: &generated,
                schema,
                steps,
                visiting: BTreeSet::new(),
            }
            .bind(&dependency)?;
        }
        let binding = context
            .bindings
            .get(query_id)
            .ok_or_else(|| FeatureRuntimeError::MissingRecord(query_id.to_owned()))?;
        let rows = super::guard_binding_rows(query_id, binding)?;
        let filter = predicate_for_filters(
            declaration.get("filters").and_then(Value::as_object),
            context,
            schema,
            entity,
        )?
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("guard.filters".to_owned()))?;
        steps.push(Mutation::RequireCount {
            entity: entity.to_owned(),
            filter,
            minimum: rows.len() as u64,
            maximum: Some(rows.len() as u64),
        });
        append_read_guards(schema, entity, &rows, steps)?;
    }
    super::evaluate_guard_contract(action_id, guard_contract, context)
}

/// Finds only declared query references used by one bounded operation.
fn operation_query_references(value: &Value) -> Result<BTreeSet<String>, FeatureRuntimeError> {
    let mut references = BTreeSet::new();
    let mut pending = vec![value];
    let mut nodes = 0;
    while let Some(value) = pending.pop() {
        nodes += 1;
        if nodes > 8192 {
            return Err(FeatureRuntimeError::InvalidContract(
                "operation query references".to_owned(),
            ));
        }
        match value {
            Value::String(value) => {
                if let Some(reference) = value.strip_prefix("query.") {
                    let name = reference.split('.').next().unwrap_or_default();
                    if !pharos_runtime_database::operations::valid_identifier(name)
                        || references.len() >= 32
                    {
                        return Err(FeatureRuntimeError::InvalidContract(
                            "query reference".to_owned(),
                        ));
                    }
                    references.insert(name.to_owned());
                }
            }
            Value::Array(values) => pending.extend(values),
            Value::Object(values) => pending.extend(values.values()),
            _ => {}
        }
    }
    Ok(references)
}

/// Owns bounded query-dependency resolution while a transaction is planned.
struct QueryDependencyPlan<'a> {
    app: &'a AppRuntime,
    feature: &'a Value,
    context: &'a mut FeatureRequestContext,
    generated: &'a BTreeMap<String, usize>,
    schema: &'a AppSchema,
    steps: &'a mut Vec<Mutation>,
    visiting: BTreeSet<String>,
}

impl QueryDependencyPlan<'_> {
    /// Resolves a declared read and guards it in the eventual mutation transaction.
    /// Generated bindings remain transaction-local and never trigger stale reads.
    fn bind(&mut self, query_id: &str) -> Result<(), FeatureRuntimeError> {
        if self.context.bindings.contains_key(query_id) || self.generated.contains_key(query_id) {
            return Ok(());
        }
        if self.visiting.len() >= 32 || !self.visiting.insert(query_id.to_owned()) {
            return Err(FeatureRuntimeError::InvalidContract(
                "cyclic query dependency".to_owned(),
            ));
        }
        let feature = self.feature;
        let declaration = feature
            .get("queries")
            .and_then(|queries| queries.get(query_id))
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
        for dependency in operation_query_references(declaration)? {
            self.bind(&dependency)?;
        }
        let entity = declaration
            .get("entity")
            .and_then(Value::as_str)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
        if planned_write_targets(self.steps, entity) {
            return Err(FeatureRuntimeError::InvalidContract(
                "query dependency after write requires transactional select".to_owned(),
            ));
        }
        let results =
            execute_logical_queries(self.app, feature, &[query_id.to_owned()], self.context)?;
        let rows = results
            .get(query_id)
            .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
        let filter = predicate_for_filters(
            declaration.get("filters").and_then(Value::as_object),
            self.context,
            self.schema,
            entity,
        )?;
        if let Some(filter) = filter {
            self.steps.push(Mutation::RequireCount {
                entity: entity.to_owned(),
                filter,
                minimum: rows.len() as u64,
                maximum: Some(rows.len() as u64),
            });
        }
        append_read_guards(
            self.schema,
            entity,
            &rows.iter().collect::<Vec<_>>(),
            self.steps,
        )?;
        self.context.bindings.insert(
            query_id.to_owned(),
            query_binding_value(
                query_id,
                declaration
                    .get("result")
                    .and_then(Value::as_str)
                    .unwrap_or("many"),
                rows,
            )?,
        );
        self.visiting.remove(query_id);
        Ok(())
    }
}

/// Executes declared writes and guards in one adapter transaction.
pub(super) fn execute_logical_transaction(
    app: &AppRuntime,
    feature: &Value,
    transaction: &Value,
    guard_contract: Option<&Value>,
    mut context: FeatureRequestContext,
    action_id: &str,
) -> Result<(), FeatureRuntimeError> {
    #[cfg(target_os = "wasi")]
    let _ = guard_contract;
    let idempotency = transaction.get("idempotency");
    if let Some(contract) = idempotency
        && prior_idempotent_result(app, feature, contract, &context)?
    {
        return Ok(());
    }
    #[cfg(not(target_os = "wasi"))]
    populate_generated_values(transaction, &mut context)?;
    let store = store_for_app(app)?;
    let queries = feature
        .get("queries")
        .and_then(Value::as_object)
        .ok_or_else(|| FeatureRuntimeError::InvalidContract("feature.queries".to_owned()))?;
    let declared_steps = transaction_operations(transaction)?;
    let reserve_error = declared_steps
        .iter()
        .find(|(kind, _)| kind == "reserve")
        .map(|(_, operation)| {
            let code = operation
                .get("on_conflict")
                .and_then(|value| value.get("code"))
                .and_then(Value::as_str)
                .filter(|text| !text.is_empty() && text.len() <= 128)
                .unwrap_or("reservation_unavailable");
            let message = operation
                .get("on_conflict")
                .and_then(|value| value.get("message"))
                .and_then(Value::as_str)
                .filter(|text| !text.is_empty() && text.len() <= 256)
                .unwrap_or("The requested reservation is not available.");
            (code.to_owned(), message.to_owned())
        });
    let mut bindings = BTreeMap::new();
    let mut steps = Vec::with_capacity(declared_steps.len());
    let mut inserted = BTreeMap::<String, Vec<InsertedRow>>::new();
    let mut inserted_queries = BTreeMap::<String, Vec<InsertedRow>>::new();
    let mut guard_bound = false;
    for (kind, operation) in declared_steps {
        for query_id in operation_query_references(&operation)? {
            QueryDependencyPlan {
                app,
                feature,
                context: &mut context,
                generated: &bindings,
                schema: store.schema(),
                steps: &mut steps,
                visiting: BTreeSet::new(),
            }
            .bind(&query_id)?;
        }
        if !guard_bound && !matches!(kind.as_str(), "require" | "select" | "assert") {
            #[cfg(not(target_os = "wasi"))]
            bind_guard_preconditions(
                app,
                feature,
                guard_contract,
                &mut context,
                store.schema(),
                &mut steps,
                action_id,
            )?;
            guard_bound = true;
        }
        if kind == "reserve" {
            let entity = operation
                .get("entity")
                .and_then(Value::as_str)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract("reserve.entity".to_owned()))?;
            let status_field = operation
                .get("status_field")
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("reserve.status_field".to_owned())
                })?;
            let statuses = operation
                .get("active_statuses")
                .and_then(Value::as_array)
                .filter(|values| !values.is_empty() && values.len() <= 16)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("reserve.active_statuses".to_owned())
                })?
                .clone();
            let owner_filters = operation
                .get("owner_filters")
                .and_then(Value::as_object)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("reserve.owner_filters".to_owned())
                })?;
            let owner_filter = predicate_for_transaction_filters(
                owner_filters,
                &context,
                &bindings,
                store.schema(),
                entity,
            )?;
            let scope_filters = operation
                .get("scope_filters")
                .and_then(Value::as_object)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("reserve.scope_filters".to_owned())
                })?;
            let scope_filter = predicate_for_transaction_filters(
                scope_filters,
                &context,
                &bindings,
                store.schema(),
                entity,
            )?;
            let lock_scope_filter = operation
                .get("lock_scope_filters")
                .map(|filters| {
                    predicate_for_transaction_filters(
                        filters.as_object().ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract(
                                "reserve.lock_scope_filters".to_owned(),
                            )
                        })?,
                        &context,
                        &bindings,
                        store.schema(),
                        entity,
                    )
                })
                .transpose()?;
            let capacity = operation
                .get("capacity")
                .and_then(|value| {
                    value.as_u64().or_else(|| {
                        resolve_value(value, &context)
                            .ok()
                            .flatten()
                            .and_then(|value| value.as_u64())
                    })
                })
                .filter(|value| *value <= 100_000)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("reserve.capacity".to_owned())
                })?;
            let expiry = operation
                .get("expires_after_seconds")
                .and_then(Value::as_u64)
                .map(|seconds| {
                    if seconds == 0 || seconds > 604_800 {
                        return Err(FeatureRuntimeError::InvalidContract(
                            "reserve.expires_after_seconds".to_owned(),
                        ));
                    }
                    let cutoff = resolve_value(
                        &serde_json::json!({"server_now_minus_seconds":seconds}),
                        &context,
                    )?
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("reserve.expiry".to_owned())
                    })?;
                    let now = resolve_value(&Value::String("server.now".to_owned()), &context)?
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("reserve.expiry".to_owned())
                        })?;
                    let declared = store
                        .schema()
                        .entities
                        .iter()
                        .find(|entry| entry.name == entity)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("reserve.entity".to_owned())
                        })?;
                    let released_at_name = operation
                        .get("expires_at_field")
                        .and_then(Value::as_str)
                        .unwrap_or("released_at");
                    let reason_name = operation
                        .get("expires_reason_field")
                        .and_then(Value::as_str)
                        .unwrap_or("release_reason");
                    Ok(ReservationExpiry {
                        filter: operation
                            .get("expires_filters")
                            .map(|filters| {
                                predicate_for_transaction_filters(
                                    filters.as_object().ok_or_else(|| {
                                        FeatureRuntimeError::InvalidContract(
                                            "reserve.expires_filters".to_owned(),
                                        )
                                    })?,
                                    &context,
                                    &bindings,
                                    store.schema(),
                                    entity,
                                )
                            })
                            .transpose()?,
                        timestamp_field: operation
                            .get("expires_from_field")
                            .and_then(Value::as_str)
                            .unwrap_or("reserved_at")
                            .to_owned(),
                        cutoff,
                        statuses: operation
                            .get("expire_statuses")
                            .and_then(Value::as_array)
                            .cloned()
                            .unwrap_or_else(|| vec![Value::String("reserved".to_owned())]),
                        to_status: Value::String(
                            operation
                                .get("expires_to_status")
                                .and_then(Value::as_str)
                                .unwrap_or("released")
                                .to_owned(),
                        ),
                        released_at: declared
                            .fields
                            .iter()
                            .any(|field| field.name == released_at_name)
                            .then(|| (released_at_name.to_owned(), now)),
                        reason: declared
                            .fields
                            .iter()
                            .any(|field| field.name == reason_name)
                            .then(|| {
                                (
                                    reason_name.to_owned(),
                                    Value::String(
                                        operation
                                            .get("expires_reason")
                                            .and_then(Value::as_str)
                                            .unwrap_or("reservation_expired")
                                            .to_owned(),
                                    ),
                                )
                            }),
                    })
                })
                .transpose()?;
            let values = operation
                .get("values")
                .and_then(Value::as_object)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract("reserve.values".to_owned()))?;
            steps.push(Mutation::Reserve {
                entity: entity.to_owned(),
                status_field: status_field.to_owned(),
                active_statuses: statuses,
                owner_filter: Box::new(owner_filter),
                scope_filter: Box::new(scope_filter),
                lock_scope_filter: lock_scope_filter.map(Box::new),
                capacity,
                expiry: expiry.map(Box::new),
                values: write_values(
                    values,
                    &context,
                    &bindings,
                    &BTreeMap::new(),
                    store.schema(),
                    entity,
                )?,
            });
            continue;
        }
        if matches!(kind.as_str(), "confirm" | "release") {
            let (guard, update) =
                reservation_transition(&kind, &operation, &context, store.schema())?;
            steps.push(guard);
            steps.push(update);
            continue;
        }
        if kind == "bind_inserted" {
            let query_id = operation
                .get("query")
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("bind_inserted.query".to_owned())
                })?;
            let declaration = queries
                .get(query_id)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
            let entity = declaration
                .get("entity")
                .and_then(Value::as_str)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
            let rows = inserted
                .get(entity)
                .cloned()
                .ok_or_else(|| FeatureRuntimeError::MissingRecord(query_id.to_owned()))?;
            if rows.len() > 100 {
                return Err(FeatureRuntimeError::InvalidContract(
                    "bind_inserted.limit".to_owned(),
                ));
            }
            context.bindings.insert(
                query_id.to_owned(),
                Value::Array(
                    rows.iter()
                        .map(|row| Value::Object(row.fields.clone()))
                        .collect(),
                ),
            );
            inserted_queries.insert(query_id.to_owned(), rows);
            continue;
        }
        if kind == "insert_from" {
            let query_id = operation
                .get("query")
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("insert_from.query".to_owned())
                })?;
            let entity = operation
                .get("entity")
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("insert_from.entity".to_owned())
                })?;
            let values = operation
                .get("values")
                .and_then(Value::as_object)
                .ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("insert_from.values".to_owned())
                })?;
            let max_rows = operation
                .get("max_rows")
                .and_then(Value::as_u64)
                .unwrap_or(20);
            if max_rows == 0 || max_rows > 100 {
                return Err(FeatureRuntimeError::InvalidContract(
                    "insert_from.max_rows".to_owned(),
                ));
            }
            let source = context
                .bindings
                .get(query_id)
                .and_then(Value::as_array)
                .ok_or_else(|| FeatureRuntimeError::MissingRecord(query_id.to_owned()))?
                .clone();
            if source.len() > max_rows as usize || steps.len().saturating_add(source.len()) > 64 {
                return Err(FeatureRuntimeError::InvalidContract(
                    "insert_from.limit".to_owned(),
                ));
            }
            let source_generated = inserted_queries.get(query_id);
            let identity = store
                .schema()
                .entities
                .iter()
                .find(|entry| entry.name == entity)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(entity.to_owned()))?
                .identity
                .generated_field()
                .map(str::to_owned);
            for (index, row) in source.into_iter().enumerate() {
                context.row = row.as_object().cloned().ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract("insert_from.row".to_owned())
                })?;
                let generated = source_generated
                    .and_then(|rows| rows.get(index))
                    .map(|row| &row.generated);
                let mut values = write_values(
                    values,
                    &context,
                    &bindings,
                    generated.unwrap_or(&BTreeMap::new()),
                    store.schema(),
                    entity,
                )?;
                complete_insert_values(store.schema(), entity, &mut values, &context)?;
                let step = steps.len();
                inserted
                    .entry(entity.to_owned())
                    .or_default()
                    .push(inserted_row(&values, identity.as_deref(), step, None));
                steps.push(Mutation::Insert {
                    entity: entity.to_owned(),
                    values,
                });
            }
            context.row.clear();
            continue;
        }
        if kind == "select" {
            let query_id = operation
                .get("query")
                .and_then(Value::as_str)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract("select.query".to_owned()))?;
            let query = queries
                .get(query_id)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
            let entity = query
                .get("entity")
                .and_then(Value::as_str)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
            if planned_write_targets(&steps, entity) {
                let candidate_rows = inserted.get(entity).ok_or_else(|| {
                    FeatureRuntimeError::InvalidContract(
                        "select after write has no transaction-local record".to_owned(),
                    )
                })?;
                let filters = query
                    .get("filters")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("local select filters".to_owned())
                    })?;
                let rows = candidate_rows
                    .iter()
                    .filter_map(
                        |row| match local_row_matches(row, filters, &context, &bindings) {
                            Ok(true) => Some(Ok(row.clone())),
                            Ok(false) => None,
                            Err(error) => Some(Err(error)),
                        },
                    )
                    .collect::<Result<Vec<_>, _>>()?;
                let many = query.get("result").and_then(Value::as_str) == Some("many");
                if !many && rows.is_empty() {
                    return Err(FeatureRuntimeError::MissingRecord(query_id.to_owned()));
                }
                if rows.len() > 100 || (!many && rows.len() != 1) {
                    return Err(FeatureRuntimeError::InvalidContract(
                        "select after write has an invalid transaction-local record count"
                            .to_owned(),
                    ));
                }
                let requested_fields = query
                    .get("fields")
                    .and_then(Value::as_array)
                    .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
                if rows.iter().any(|row| {
                    requested_fields.iter().any(|field| {
                        field
                            .as_str()
                            .is_none_or(|field| !row.fields.contains_key(field))
                    })
                }) {
                    return Err(FeatureRuntimeError::InvalidContract(
                        "transaction-local record lacks a selected field".to_owned(),
                    ));
                }
                if many {
                    context.bindings.insert(
                        query_id.to_owned(),
                        Value::Array(
                            rows.iter()
                                .map(|row| Value::Object(row.fields.clone()))
                                .collect(),
                        ),
                    );
                    inserted_queries.insert(query_id.to_owned(), rows.clone());
                } else {
                    if let Some(WriteValue::GeneratedId(step)) = rows[0].generated.get("id") {
                        bindings.insert(query_id.to_owned(), *step);
                    }
                    context
                        .bindings
                        .insert(query_id.to_owned(), Value::Object(rows[0].fields.clone()));
                }
                continue;
            }
            if query.get("result").and_then(Value::as_str) == Some("many") {
                return Err(FeatureRuntimeError::InvalidContract(
                    "select.many".to_owned(),
                ));
            }
            let rows = execute_logical_queries(app, feature, &[query_id.to_owned()], &context)?;
            let row = rows
                .get(query_id)
                .and_then(|rows| rows.first())
                .ok_or_else(|| FeatureRuntimeError::MissingRecord(query_id.to_owned()))?;
            let entity = query
                .get("entity")
                .and_then(Value::as_str)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
            let identity = store
                .schema()
                .entities
                .iter()
                .find(|entry| entry.name == entity)
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?
                .identity
                .fields();
            if identity.iter().any(|field| !row.contains_key(*field)) {
                return Err(FeatureRuntimeError::InvalidContract(format!(
                    "{query_id}.identity"
                )));
            }
            let predicates = row
                .iter()
                .map(|(field, value)| {
                    Ok(Predicate::Equal(
                        field.clone(),
                        field_value(store.schema(), entity, field, value.clone())?,
                    ))
                })
                .collect::<Result<Vec<_>, FeatureRuntimeError>>()?;
            steps.push(Mutation::RequireCount {
                entity: entity.to_owned(),
                filter: Predicate::All(predicates),
                minimum: 1,
                maximum: None,
            });
            context
                .bindings
                .insert(query_id.to_owned(), Value::Object(row.clone()));
            continue;
        }
        let entity = operation
            .get("entity")
            .and_then(Value::as_str)
            .unwrap_or_default();
        let mut step = match kind.as_str() {
            "allocate_sequence" => {
                let counter = operation
                    .get("counter")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("allocate_sequence.counter".to_owned())
                    })?;
                let counter_entity =
                    counter
                        .get("entity")
                        .and_then(Value::as_str)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("counter.entity".to_owned())
                        })?;
                let counter_field =
                    counter
                        .get("field")
                        .and_then(Value::as_str)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("counter.field".to_owned())
                        })?;
                let scope = counter
                    .get("scope")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("counter.scope".to_owned())
                    })?
                    .iter()
                    .map(|(field, source)| {
                        Ok((
                            field.clone(),
                            resolve_value(source, &context)?
                                .ok_or_else(|| FeatureRuntimeError::InvalidInput(field.clone()))?,
                        ))
                    })
                    .collect::<Result<BTreeMap<_, _>, FeatureRuntimeError>>()?;
                let initial = if let Some(seed) = counter.get("seed").and_then(Value::as_object) {
                    let seed_entity =
                        seed.get("entity").and_then(Value::as_str).ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("counter.seed.entity".to_owned())
                        })?;
                    let seed_field =
                        seed.get("field").and_then(Value::as_str).ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("counter.seed.field".to_owned())
                        })?;
                    let filter = predicate_for_filters(
                        seed.get("filters").and_then(Value::as_object),
                        &context,
                        store.schema(),
                        seed_entity,
                    )?;
                    let maximum = store
                        .adapter()
                        .aggregate(
                            store.config(),
                            store.schema(),
                            &pharos_runtime_database::operations::AggregateQuery {
                                entity: seed_entity.to_owned(),
                                filter,
                                aggregate: pharos_runtime_database::operations::Aggregate::Maximum(
                                    seed_field.to_owned(),
                                ),
                            },
                        )
                        .map_err(map_database_error)?;
                    if maximum.is_null() {
                        0
                    } else {
                        maximum.as_i64().ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("counter.seed.value".to_owned())
                        })?
                    }
                } else {
                    0
                };
                let maximum = counter
                    .get("maximum")
                    .and_then(Value::as_i64)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("counter.maximum".to_owned())
                    })?;
                let (value, counter_steps) =
                    pharos_runtime_database::sequence::plan_sequence_allocation(
                        store.adapter(),
                        store.config(),
                        store.schema(),
                        pharos_runtime_database::sequence::SequenceAllocation {
                            entity: counter_entity,
                            field: counter_field,
                            scope,
                            initial,
                            maximum,
                        },
                    )
                    .map_err(map_database_error)?;
                steps.extend(counter_steps);
                context
                    .generated
                    .insert("sequence_value".to_owned(), Value::from(value));
                if let Some(binding) = operation.get("bind").and_then(Value::as_str)
                    && bindings.insert(binding.to_owned(), steps.len()).is_some()
                {
                    return Err(FeatureRuntimeError::InvalidContract(
                        "allocate_sequence.bind".to_owned(),
                    ));
                }
                Mutation::Insert {
                    entity: entity.to_owned(),
                    values: write_values(
                        operation
                            .get("values")
                            .and_then(Value::as_object)
                            .ok_or_else(|| {
                                FeatureRuntimeError::InvalidContract(
                                    "allocate_sequence.values".to_owned(),
                                )
                            })?,
                        &context,
                        &bindings,
                        &BTreeMap::new(),
                        store.schema(),
                        entity,
                    )?,
                }
            }
            "require" => {
                let query_id = operation
                    .get("query")
                    .and_then(Value::as_str)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("require.query".to_owned())
                    })?;
                if bindings.contains_key(query_id) {
                    // The earlier insert or ensure step already owns the record in this
                    // transaction. Re-reading it on another connection would be stale.
                    continue;
                }
                let query = queries
                    .get(query_id)
                    .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
                let entity = query
                    .get("entity")
                    .and_then(Value::as_str)
                    .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
                if planned_write_targets(&steps, entity) {
                    return Err(FeatureRuntimeError::InvalidContract(
                        "require after write requires transactional read".to_owned(),
                    ));
                }
                let filter = predicate_for_filters(
                    query.get("filters").and_then(Value::as_object),
                    &context,
                    store.schema(),
                    entity,
                )?
                .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
                let rows = execute_logical_queries(app, feature, &[query_id.to_owned()], &context)?;
                let rows = rows
                    .get(query_id)
                    .ok_or_else(|| FeatureRuntimeError::InvalidContract(query_id.to_owned()))?;
                if rows.is_empty() {
                    return Err(FeatureRuntimeError::MissingRecord(query_id.to_owned()));
                }
                let result_kind = query
                    .get("result")
                    .and_then(Value::as_str)
                    .unwrap_or("many");
                context.bindings.insert(
                    query_id.to_owned(),
                    query_binding_value(query_id, result_kind, rows)?,
                );
                steps.push(Mutation::RequireCount {
                    entity: entity.to_owned(),
                    filter,
                    minimum: rows.len() as u64,
                    maximum: Some(rows.len() as u64),
                });
                let guarded_rows = rows.iter().collect::<Vec<_>>();
                append_read_guards(store.schema(), entity, &guarded_rows, &mut steps)?;
                continue;
            }
            "insert" => {
                let values = operation
                    .get("values")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("insert.values".to_owned())
                    })?;
                if let Some(binding) = operation.get("bind").and_then(Value::as_str) {
                    bindings.insert(binding.to_owned(), steps.len());
                }
                Mutation::Insert {
                    entity: entity.to_owned(),
                    values: write_values(
                        values,
                        &context,
                        &bindings,
                        &BTreeMap::new(),
                        store.schema(),
                        entity,
                    )?,
                }
            }
            "upsert" => {
                let update_on_conflict = match operation.get("update_on_conflict") {
                    None => false,
                    Some(Value::Bool(value)) => *value,
                    Some(_) => {
                        return Err(FeatureRuntimeError::InvalidContract(
                            "upsert.update_on_conflict".to_owned(),
                        ));
                    }
                };
                if let Some(binding) = operation.get("bind").and_then(Value::as_str)
                    && (binding.is_empty()
                        || update_on_conflict
                        || bindings.insert(binding.to_owned(), steps.len()).is_some())
                {
                    return Err(FeatureRuntimeError::InvalidContract(
                        "upsert.bind".to_owned(),
                    ));
                }
                let unique_key = operation
                    .get("unique_by")
                    .and_then(Value::as_array)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("upsert.unique_by".to_owned())
                    })?
                    .iter()
                    .map(|field| {
                        field.as_str().map(str::to_owned).ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("upsert.unique_by".to_owned())
                        })
                    })
                    .collect::<Result<Vec<_>, _>>()?;
                let values = operation
                    .get("values")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("upsert.values".to_owned())
                    })?;
                let values = write_values(
                    values,
                    &context,
                    &bindings,
                    &BTreeMap::new(),
                    store.schema(),
                    entity,
                )?;
                if update_on_conflict {
                    Mutation::Upsert {
                        entity: entity.to_owned(),
                        unique_key,
                        values,
                    }
                } else {
                    Mutation::InsertIfAbsent {
                        entity: entity.to_owned(),
                        unique_key,
                        values,
                    }
                }
            }
            "update" => {
                let filter = predicate_for_transaction_filters(
                    operation
                        .get("filters")
                        .and_then(Value::as_object)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("update.filters".to_owned())
                        })?,
                    &context,
                    &bindings,
                    store.schema(),
                    entity,
                )?;
                let values = operation
                    .get("values")
                    .and_then(Value::as_object)
                    .ok_or_else(|| {
                        FeatureRuntimeError::InvalidContract("update.values".to_owned())
                    })?;
                let expected_rows = steps
                    .iter()
                    .rev()
                    .find_map(|step| match step {
                        Mutation::RequireCount {
                            entity: guarded_entity,
                            filter: guarded_filter,
                            minimum,
                            maximum: Some(maximum),
                        } if guarded_entity == entity
                            && guarded_filter == &filter
                            && minimum == maximum =>
                        {
                            Some(*minimum)
                        }
                        _ => None,
                    })
                    .unwrap_or(1);
                Mutation::Update {
                    entity: entity.to_owned(),
                    filter,
                    values: write_values(
                        values,
                        &context,
                        &bindings,
                        &BTreeMap::new(),
                        store.schema(),
                        entity,
                    )?,
                    expected_rows: Some(expected_rows),
                }
            }
            "delete" => {
                let filter = predicate_for_transaction_filters(
                    operation
                        .get("filters")
                        .and_then(Value::as_object)
                        .ok_or_else(|| {
                            FeatureRuntimeError::InvalidContract("delete.filters".to_owned())
                        })?,
                    &context,
                    &bindings,
                    store.schema(),
                    entity,
                )?;
                Mutation::Delete {
                    entity: entity.to_owned(),
                    filter,
                    expected_rows: None,
                }
            }
            _ => return Err(FeatureRuntimeError::InvalidContract(kind)),
        };
        if let Mutation::Insert { entity, values }
        | Mutation::InsertIfAbsent { entity, values, .. } = &mut step
        {
            complete_insert_values(store.schema(), entity, values, &context)?;
        }
        let ensured_row = if let Mutation::InsertIfAbsent {
            entity,
            unique_key,
            values,
        } = &step
        {
            Some(ensured_existing_row(
                &store, entity, unique_key, values, &mut steps,
            )?)
        } else {
            None
        };
        if ensured_row.is_some()
            && let Some(binding) = operation.get("bind").and_then(Value::as_str)
        {
            bindings.insert(binding.to_owned(), steps.len());
        }
        if let Mutation::Insert { entity, values }
        | Mutation::InsertIfAbsent { entity, values, .. } = &step
        {
            let identity = store
                .schema()
                .entities
                .iter()
                .find(|entry| entry.name == *entity)
                .and_then(|entry| entry.identity.generated_field());
            let mut row = inserted_row(values, identity, steps.len(), None);
            if let Some(Some(fields)) = ensured_row {
                row.fields = fields;
            }
            inserted.entry(entity.clone()).or_default().push(row);
        }
        steps.push(step);
    }
    if !guard_bound {
        #[cfg(not(target_os = "wasi"))]
        bind_guard_preconditions(
            app,
            feature,
            guard_contract,
            &mut context,
            store.schema(),
            &mut steps,
            action_id,
        )?;
    }
    let effect_input = context
        .input
        .iter()
        .map(|(key, value)| {
            (
                key.clone(),
                if value.is_null() {
                    None
                } else {
                    Some(
                        value
                            .as_str()
                            .map(ToOwned::to_owned)
                            .unwrap_or_else(|| value.to_string()),
                    )
                },
            )
        })
        .collect::<BTreeMap<_, _>>();
    let effect_request = crate::runtime::http::HttpRequest {
        method: "POST".to_owned(),
        target: context.request_target.clone(),
        cookie: None,
        content_type: None,
        headers: String::new(),
        body: Vec::new(),
        body_file_path: None,
        body_len: 0,
        is_multipart: false,
    };
    let effects = crate::runtime::dynamic::declared_effect_steps(
        app,
        store.schema(),
        &effect_request,
        action_id,
        &effect_input,
        0,
        context.actor_id,
    )
    .map_err(|_| FeatureRuntimeError::InvalidContract("declared action effects".to_owned()))?;
    let resource = crate::runtime::dynamic::action_resource_binding(app, store.schema(), action_id);
    let outcome = pharos_runtime_database::effects::transact_with_resource_effects(
        store.adapter(),
        store.config(),
        store.schema(),
        &Transaction {
            steps,
            idempotency_key: None,
        },
        &effects,
        resource.as_ref(),
    );
    if matches!(outcome, Err(DatabaseError::Conflict))
        && let Some(contract) = idempotency
    {
        for attempt in 0..3 {
            if prior_idempotent_result(app, feature, contract, &context)? {
                return Ok(());
            }
            if attempt < 2 {
                std::thread::sleep(std::time::Duration::from_millis(25));
            }
        }
    }
    if matches!(outcome, Err(DatabaseError::CapacityExceeded))
        && let Some((code, message)) = reserve_error
    {
        return Err(FeatureRuntimeError::ValidationReport(
            super::validation_report(
                action_id,
                vec![code.clone()],
                None,
                Some((code.clone(), code, message)),
            ),
        ));
    }
    outcome.map_err(map_database_error)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    /// Large declared source sets retain all read assertions in one transaction without widening the per-batch database bound.
    #[test]
    fn read_guards_batch_full_source_sets_without_widening_scope() {
        let schema=serde_json::from_value(serde_json::json!({"app_id":"guard_test","entities":[{"name":"record","identity":{"auto_integer":"id"},"fields":[{"name":"id","kind":"integer","nullable":false,"constraints":[]},{"name":"name","kind":"text","nullable":false,"constraints":[]}],"unique":[],"indexes":[],"conditional_indexes":[]}]})).unwrap();
        let mut rows = (0..125)
            .map(|id| {
                serde_json::json!({"id":id,"name":"pending"})
                    .as_object()
                    .unwrap()
                    .clone()
            })
            .collect::<Vec<_>>();
        let mut steps = Vec::new();
        super::append_read_guards(
            &schema,
            "record",
            &rows.iter().collect::<Vec<_>>(),
            &mut steps,
        )
        .unwrap();
        assert!(steps.len() > 1 && steps.len() < 125);
        rows[124]["id"] = serde_json::json!(0);
        let mut rejected = Vec::new();
        assert!(
            super::append_read_guards(
                &schema,
                "record",
                &rows.iter().collect::<Vec<_>>(),
                &mut rejected
            )
            .is_err()
        );
        assert!(rejected.is_empty());
        let excess = vec![&rows[0]; 1001];
        assert!(super::append_read_guards(&schema, "record", &excess, &mut rejected).is_err());
        assert!(rejected.is_empty());
    }
    /// Blank optional references become NULL while malformed integer identities remain invalid.
    #[test]
    fn optional_integer_references_accept_blank_without_relaxing_required_values() {
        use pharos_runtime_database::operations::{AppSchema, Entity, Field, FieldType, Identity};
        let field = Field {
            name: "account_id".into(),
            kind: FieldType::Integer,
            nullable: true,
            max_length: None,
            default: None,
            references: None,
            on_delete: None,
            constraints: vec![],
        };
        let mut schema = AppSchema {
            app_id: "sample".into(),
            entities: vec![Entity {
                name: "contact".into(),
                identity: Identity::AutoInteger("account_id".into()),
                fields: vec![field],
                unique: vec![],
                indexes: vec![],
                conditional_indexes: vec![],
            }],
        };
        assert_eq!(
            super::field_value(&schema, "contact", "account_id", json!("")).unwrap(),
            Value::Null
        );
        assert_eq!(
            super::field_value(&schema, "contact", "account_id", json!("42")).unwrap(),
            json!(42)
        );
        assert!(super::field_value(&schema, "contact", "account_id", json!("42 OR 1=1")).is_err());
        schema.entities[0].fields[0].nullable = false;
        assert!(super::field_value(&schema, "contact", "account_id", json!("")).is_err());
    }
    use std::path::PathBuf;
    use std::time::{SystemTime, UNIX_EPOCH};

    use pharos_runtime_compiler::database::load_source as load_contract;
    use pharos_runtime_database::DbBackendConfig;
    #[cfg(not(target_os = "wasi"))]
    use pharos_runtime_database::operations::Mutation;
    use pharos_runtime_database::operations::{Predicate, Query};
    use serde_json::{Value, json};

    #[cfg(not(target_os = "wasi"))]
    use super::bind_guard_preconditions;
    use super::{FeatureRequestContext, execute_logical_transaction};
    use crate::config::AppRuntime;

    /// Operational and configuration failures retain their exact public error category.
    #[test]
    fn database_error_mapping_preserves_operational_categories() {
        for error in [
            pharos_runtime_database::operations::DatabaseError::CapacityExceeded,
            pharos_runtime_database::operations::DatabaseError::InvalidContract,
            pharos_runtime_database::operations::DatabaseError::AccessDenied,
            pharos_runtime_database::operations::DatabaseError::Unavailable,
            pharos_runtime_database::operations::DatabaseError::StorageFailure,
            pharos_runtime_database::operations::DatabaseError::Unsupported,
        ] {
            assert!(matches!(super::map_database_error(error),
                super::FeatureRuntimeError::Database(actual) if actual == error));
        }
    }

    /// Cached working rows bypass an unavailable database until a guarded commit needs it.
    #[test]
    fn working_queries_reuse_rows_without_opening_storage() {
        let repo = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let temporary = repo
            .join("tmp")
            .join(format!("pharos-working-query-{}", std::process::id()));
        std::fs::create_dir_all(temporary.join("data/contracts"))
            .expect("working contract directory");
        pharos_runtime_compiler::database::materialize_json(
            repo.join("apps/ucal/data/contracts/database.yml"),
            temporary.join("data/contracts/database.json"),
        )
        .expect("logical contract");
        std::fs::write(
            temporary.join("pharos.app.yml"),
            "app_id: ucal\ndatabase:\n  migration_contract: data/contracts/database.json\n",
        )
        .expect("manifest");
        let config = temporary.join("runtime.conf");
        let missing_database = temporary.join("missing/database.sqlite3");
        std::fs::write(
            &config,
            format!(
                "db_backend=sqlite\ndb_path={}\n",
                missing_database.display()
            ),
        )
        .expect("configuration");
        let app = AppRuntime {
            app_id: Some("ucal".to_owned()),
            app_root: Some(temporary.display().to_string()),
            runtime_conf_path: Some(config.display().to_string()),
            ..AppRuntime::default()
        };
        let feature = json!({"queries":{"loaded":{"entity":"business","fields":["id","name"],"result":"many"}}});
        let mut context = FeatureRequestContext::default();
        context.bindings.insert(
            "loaded".to_owned(),
            json!([{"id":1,"name":"kept in process"}]),
        );
        for _ in 0..3 {
            let rows =
                super::execute_logical_queries(&app, &feature, &["loaded".to_owned()], &context)
                    .expect("reuse existing query");
            assert_eq!(rows["loaded"][0]["name"], "kept in process");
        }
        assert!(
            !missing_database.exists(),
            "cached reads must not open or create storage"
        );
        context.bindings.clear();
        assert!(
            super::execute_logical_queries(&app, &feature, &["loaded".to_owned()], &context)
                .is_err(),
            "missing records require authoritative storage"
        );
        context
            .bindings
            .insert("loaded".to_owned(), Value::Array(vec![json!({}); 101]));
        assert!(
            super::working_query_rows(&context, "loaded").is_err(),
            "working views remain bounded"
        );
        std::fs::remove_dir_all(temporary).expect("clean working test");
    }

    /// Copying a transaction-local row preserves its typed generated expression.
    #[test]
    fn inserted_rows_preserve_generated_text_references() {
        use pharos_runtime_database::operations::{AppSchema, WriteValue};
        use std::collections::BTreeMap;
        let expression = WriteValue::GeneratedText {
            index: 3,
            prefix: "metric:".to_owned(),
        };
        let row = super::inserted_row(
            &BTreeMap::from([("resource_id".to_owned(), expression.clone())]),
            Some("id"),
            4,
            None,
        );
        let values = super::write_values(
            &serde_json::Map::from_iter([("resource_id".to_owned(), json!("row.resource_id"))]),
            &FeatureRequestContext::default(),
            &BTreeMap::new(),
            &row.generated,
            &AppSchema {
                app_id: "expression_test".to_owned(),
                entities: Vec::new(),
            },
            "outbox",
        )
        .expect("transaction-local expression");
        assert_eq!(values["resource_id"], expression);
        assert_eq!(row.generated["id"], WriteValue::GeneratedId(4));
    }

    /// Keeps a declared cart cap and its checked row set in one adapter transaction.
    #[cfg(not(target_os = "wasi"))]
    #[test]
    fn logical_cart_guard_adds_transactional_read_checks() {
        let source = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("apps/commerce_spa/data/contracts/database.yml");
        let schema = load_contract(&source, "commerce_spa")
            .expect("Commerce logical contract")
            .schema;
        let feature = json!({"queries": {"current_cart_items": {
            "entity": "shopping_cart_item",
            "fields": ["id", "user_id", "price_minor"],
            "filters": {"user_id": "subject.user_id"},
            "result": "many"
        }}});
        let guard = json!({"guards": [{
            "guard_id": "cart_total_cap", "kind": "sum", "query": "current_cart_items",
            "field": "price_minor", "add": "query.published_product.price_minor",
            "maximum": 6000
        }]});
        let mut context = FeatureRequestContext {
            actor_id: 42,
            ..FeatureRequestContext::default()
        };
        context.bindings.insert(
            "current_cart_items".to_owned(),
            json!([{"id": 1, "user_id": 42, "price_minor": 3000}]),
        );
        context
            .bindings
            .insert("published_product".to_owned(), json!({"price_minor": 2900}));
        let mut steps = Vec::new();
        bind_guard_preconditions(
            &AppRuntime::default(),
            &feature,
            Some(&guard),
            &mut context,
            &schema,
            &mut steps,
            "add_pack",
        )
        .expect("guard accepts a cart below the cap");
        assert_eq!(steps.len(), 2);
        assert!(matches!(
            steps[0],
            Mutation::RequireCount {
                minimum: 1,
                maximum: Some(1),
                ..
            }
        ));
        context
            .bindings
            .insert("published_product".to_owned(), json!({"price_minor": 3100}));
        let rejected = bind_guard_preconditions(
            &AppRuntime::default(),
            &feature,
            Some(&guard),
            &mut context,
            &schema,
            &mut Vec::new(),
            "add_pack",
        );
        assert!(matches!(
            rejected,
            Err(super::FeatureRuntimeError::ValidationReport(_))
        ));
    }

    /// Runs every UCAL generated mutation against one configured logical adapter.
    fn exercise_ucal_generated_mutations(backend: &str) {
        let repository = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock")
            .as_nanos();
        let temporary = repository
            .join("tmp/database-operation-tests")
            .join(format!(
                "ucal_feature_{backend}_{}_{nonce}",
                std::process::id()
            ));
        let contract_dir = temporary.join("data/contracts");
        std::fs::create_dir_all(&contract_dir).expect("isolated app contract directory");
        std::fs::write(
            temporary.join("pharos.app.yml"),
            "app_id: ucal\ndatabase:\n  migration_contract: data/contracts/database.json\n",
        )
        .expect("isolated logical app manifest");
        let source = repository.join("apps/ucal/data/contracts/database.yml");
        pharos_runtime_compiler::database::materialize_json(
            &source,
            contract_dir.join("database.json"),
        )
        .expect("isolated logical app contract");
        let path = temporary.join("ucal.sqlite3");
        let database = format!("ucal_feature_{}_{nonce}", std::process::id());
        let config = match backend {
            "sqlite" => DbBackendConfig {
                backend: "sqlite".to_owned(),
                sqlite_path: Some(path),
                ..DbBackendConfig::default()
            },
            "postgresql" | "mysql" | "mongodb" => DbBackendConfig {
                backend: backend.to_owned(),
                host: "127.0.0.1".to_owned(),
                port: match backend {
                    "postgresql" => "25432",
                    "mysql" => "23306",
                    _ => "27018",
                }
                .to_owned(),
                database,
                user: match backend {
                    "postgresql" => "pharos_parity",
                    "mysql" => "root",
                    _ => "",
                }
                .to_owned(),
                password_env_name: match backend {
                    "postgresql" => "PHAROS_DB_PARITY_PG_PASSWORD",
                    "mysql" => "PHAROS_TEST_MYSQL_ROOT_PASSWORD",
                    _ => "",
                }
                .to_owned(),
                tls_mode: "disable".to_owned(),
                options: (backend == "mongodb").then(|| "directConnection=true".to_owned()),
                ..DbBackendConfig::default()
            },
            _ => panic!("unsupported test backend"),
        };
        let adapter = crate::runtime::database::operations::resolve(&config)
            .expect("configured database adapter");
        let contract = load_contract(&source, "ucal").expect("UCAL contract");
        adapter
            .ensure_database(&config, &contract.schema)
            .expect("isolated UCAL database");
        adapter
            .apply_contract(&config, &contract)
            .expect("UCAL logical schema and seeds");
        let runtime_conf = temporary.join("runtime.conf");
        std::fs::write(
            &runtime_conf,
            format!(
                "db_backend={}\ndb_path={}\ndb_host={}\ndb_port={}\ndb_name={}\ndb_user={}\ndb_password_env={}\ndb_tls_mode={}\ndb_options={}\n",
                config.backend,
                config.sqlite_path.as_ref().map(|path| path.display().to_string()).unwrap_or_default(),
                config.host,
                config.port,
                config.database,
                config.user,
                config.password_env_name,
                config.tls_mode,
                config.options.as_deref().unwrap_or_default(),
            ),
        )
        .expect("isolated runtime config");
        let app = AppRuntime {
            app_id: Some("ucal".to_owned()),
            app_root: Some(temporary.to_string_lossy().into_owned()),
            runtime_conf_path: Some(runtime_conf.to_string_lossy().into_owned()),
            ..AppRuntime::default()
        };
        assert!(
            crate::runtime::database::selected_logical_contract(&temporary, "ucal")
                .expect("logical app selection")
                .is_some()
        );
        let feature: Value = serde_json::from_slice(
            &std::fs::read(repository.join("apps/ucal/data/contracts/features.json"))
                .expect("UCAL feature contract"),
        )
        .expect("UCAL feature JSON");
        let transaction = &feature["actions"]["api_create_service"]["transaction"];
        let context = FeatureRequestContext {
            path_params: serde_json::from_value(json!({"business_slug":"lotus-studio"}))
                .expect("path parameters"),
            input: serde_json::from_value(json!({
                "name":"Logical service",
                "service_type":"appointment",
                "location_type":"business",
                "duration_minutes":30,
                "buffer_minutes":0,
                "cost":"15.50",
                "description":"Adapter-backed action",
                "sort_order":1,
                "image_url":""
            }))
            .expect("declared input"),
            ..FeatureRequestContext::default()
        };
        execute_logical_transaction(
            &app,
            &feature,
            transaction,
            None,
            context,
            "api_create_service",
        )
        .expect("logical select and insert");
        let rows = adapter
            .query(
                &config,
                &contract.schema,
                &Query {
                    entity: "service".to_owned(),
                    fields: vec!["id".to_owned(), "business_id".to_owned(), "cost".to_owned()],
                    filter: Some(Predicate::Equal(
                        "name".to_owned(),
                        json!("Logical service"),
                    )),
                    order: Vec::new(),
                    limit: 1,
                    offset: 0,
                },
            )
            .expect("read inserted service");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["business_id"], json!(1));
        assert_eq!(rows[0]["cost"], json!("15.50"));
        let upsert_feature = json!({"queries":{}});
        let upsert = json!({"steps":[{"upsert":{
            "entity":"business", "unique_by":["slug"], "update_on_conflict":true,
            "values":{"slug":"lotus-studio", "name":"Logical studio", "owner_id":{"literal":1}}
        }}]});
        execute_logical_transaction(
            &app,
            &upsert_feature,
            &upsert,
            None,
            FeatureRequestContext::default(),
            "update_business",
        )
        .expect("logical upsert updates an existing business");
        let insert_if_absent = json!({"steps":[{"upsert":{
            "entity":"business", "unique_by":["slug"],
            "values":{"slug":"lotus-studio", "name":"Ignored change", "owner_id":{"literal":1}}
        }}]});
        execute_logical_transaction(
            &app,
            &upsert_feature,
            &insert_if_absent,
            None,
            FeatureRequestContext::default(),
            "insert_business_if_absent",
        )
        .expect("logical insert-if-absent preserves the existing business");
        let businesses = adapter
            .query(
                &config,
                &contract.schema,
                &Query {
                    entity: "business".to_owned(),
                    fields: vec!["id".to_owned(), "name".to_owned()],
                    filter: Some(Predicate::Equal("slug".to_owned(), json!("lotus-studio"))),
                    order: Vec::new(),
                    limit: 2,
                    offset: 0,
                },
            )
            .expect("read upserted business");
        assert_eq!(businesses.len(), 1);
        assert_eq!(businesses[0]["name"], json!("Logical studio"));
        let insert_new_business = json!({"steps":[{"upsert":{
            "entity":"business", "unique_by":["slug"],
            "values":{"slug":"logical-new-studio", "name":"New studio", "owner_id":{"literal":1}}
        }}]});
        execute_logical_transaction(
            &app,
            &upsert_feature,
            &insert_new_business,
            None,
            FeatureRequestContext::default(),
            "insert_business_if_absent",
        )
        .expect("logical insert-if-absent creates a missing business");
        let inserted = adapter
            .query(
                &config,
                &contract.schema,
                &Query {
                    entity: "business".to_owned(),
                    fields: vec!["id".to_owned(), "name".to_owned()],
                    filter: Some(Predicate::Equal(
                        "slug".to_owned(),
                        json!("logical-new-studio"),
                    )),
                    order: Vec::new(),
                    limit: 1,
                    offset: 0,
                },
            )
            .expect("read insert-if-absent business");
        assert_eq!(inserted.len(), 1);
        assert_eq!(inserted[0]["name"], json!("New studio"));
        let required_feature = json!({"queries":{"selected_business":{
            "entity":"business", "fields":["id", "slug", "name"],
            "filters":{"slug":"path.business_slug"}, "result":"one"
        }}});
        let guarded_update = json!({"steps":[
            {"require":{"query":"selected_business"}},
            {"update":{"entity":"business", "filters":{"id":"query.selected_business.id"},
                "values":{"name":"Guarded studio"}}}
        ]});
        execute_logical_transaction(
            &app,
            &required_feature,
            &guarded_update,
            None,
            FeatureRequestContext {
                path_params: serde_json::from_value(json!({"business_slug":"lotus-studio"}))
                    .expect("business path"),
                ..FeatureRequestContext::default()
            },
            "update_business",
        )
        .expect("logical require binds and guards the selected business");
        let service_id = rows[0]["id"].as_i64().expect("created service id");
        let cross_business = FeatureRequestContext {
            path_params: serde_json::from_value(json!({
                "business_slug":"north-wellness", "service_id":service_id
            }))
            .expect("cross-business path"),
            input: serde_json::from_value(json!({
                "name":"Unauthorized change", "service_type":"appointment",
                "location_type":"business", "duration_minutes":30, "buffer_minutes":0,
                "cost":"0.00", "description":"", "sort_order":0, "image_url":""
            }))
            .expect("cross-business input"),
            ..FeatureRequestContext::default()
        };
        assert!(matches!(
            execute_logical_transaction(
                &app,
                &feature,
                &feature["actions"]["api_update_service"]["transaction"],
                None,
                cross_business,
                "api_update_service",
            ),
            Err(super::FeatureRuntimeError::MissingRecord(_))
        ));

        /// Executes one UCAL action with the selected record and declared input.
        fn apply(app: &AppRuntime, feature: &Value, action: &str, input: Value, selected: Value) {
            let mut path: serde_json::Map<String, Value> =
                serde_json::from_value(json!({"business_slug":"lotus-studio"}))
                    .expect("business path");
            path.extend(
                serde_json::from_value::<serde_json::Map<String, Value>>(selected)
                    .expect("selected path"),
            );
            let context = FeatureRequestContext {
                path_params: path,
                input: serde_json::from_value(input).expect("declared input"),
                ..FeatureRequestContext::default()
            };
            execute_logical_transaction(
                app,
                feature,
                &feature["actions"][action]["transaction"],
                None,
                context,
                action,
            )
            .unwrap_or_else(|error| panic!("{action}: {error:?}"));
        }

        let service_input = json!({
            "name":"Updated service", "service_type":"appointment", "location_type":"business",
            "duration_minutes":45, "buffer_minutes":5, "cost":"22.00",
            "description":"Updated through adapter", "sort_order":2, "image_url":""
        });
        let service_path = json!({"service_id": service_id});
        apply(
            &app,
            &feature,
            "api_update_service",
            service_input,
            service_path.clone(),
        );
        apply(
            &app,
            &feature,
            "api_delete_service",
            json!({}),
            service_path.clone(),
        );
        apply(
            &app,
            &feature,
            "api_restore_service",
            json!({}),
            service_path,
        );

        let availability_input =
            json!({"day_of_week":"Weekends","start_time":"19:00","end_time":"20:00"});
        apply(
            &app,
            &feature,
            "api_create_availability",
            availability_input,
            json!({}),
        );
        let availability = adapter
            .query(
                &config,
                &contract.schema,
                &Query {
                    entity: "business_availability".to_owned(),
                    fields: vec!["id".to_owned()],
                    filter: Some(Predicate::All(vec![
                        Predicate::Equal("business_id".to_owned(), json!(1)),
                        Predicate::Equal("day_of_week".to_owned(), json!("Weekends")),
                        Predicate::Equal("start_time".to_owned(), json!("19:00")),
                    ])),
                    order: Vec::new(),
                    limit: 1,
                    offset: 0,
                },
            )
            .expect("created availability");
        let availability_id = availability[0]["id"].as_i64().expect("availability id");
        let availability_path = json!({"availability_id":availability_id});
        apply(
            &app,
            &feature,
            "api_update_availability",
            json!({
                "day_of_week":"Weekends","start_time":"20:00","end_time":"21:00"
            }),
            availability_path.clone(),
        );
        apply(
            &app,
            &feature,
            "api_delete_availability",
            json!({}),
            availability_path,
        );

        apply(
            &app,
            &feature,
            "api_create_technician",
            json!({
                "name":"Logical technician","email":"logical@example.invalid"
            }),
            json!({}),
        );
        let technicians = adapter
            .query(
                &config,
                &contract.schema,
                &Query {
                    entity: "technician".to_owned(),
                    fields: vec!["id".to_owned()],
                    filter: Some(Predicate::Equal(
                        "email".to_owned(),
                        json!("logical@example.invalid"),
                    )),
                    order: Vec::new(),
                    limit: 1,
                    offset: 0,
                },
            )
            .expect("created technician");
        let technician_id = technicians[0]["id"].as_i64().expect("technician id");
        let technician_path = json!({"technician_id":technician_id});
        apply(
            &app,
            &feature,
            "api_update_technician",
            json!({
                "name":"Updated technician","email":"updated@example.invalid","is_active":true
            }),
            technician_path.clone(),
        );
        apply(
            &app,
            &feature,
            "api_delete_technician",
            json!({}),
            technician_path.clone(),
        );
        apply(
            &app,
            &feature,
            "api_restore_technician",
            json!({}),
            technician_path,
        );
    }

    /// Keeps the local SQLite action suite in the default unit-test gate.
    #[test]
    fn ucal_generated_mutations_use_logical_select_bindings() {
        exercise_ucal_generated_mutations("sqlite");
    }

    /// Certifies the same UCAL actions against the three local service adapters.
    #[test]
    #[ignore = "requires local PostgreSQL, MySQL, and MongoDB test containers"]
    fn ucal_generated_mutations_cross_backend() {
        for backend in ["postgresql", "mysql", "mongodb"] {
            exercise_ucal_generated_mutations(backend);
        }
    }
}