fraiseql-core 2.15.0

//! Mutation execution runner.
//!
//! [`MutationRunner`] executes GraphQL mutations with compile-time capability enforcement.
//!
//! The [`execute_mutation_impl`] free function contains the core logic and is bounded only on
//! `A: DatabaseAdapter`, allowing it to be called from both the compile-time-checked
//! [`MutationRunner`] path and the runtime-guarded `execute_mutation_query` path on
//! [`Executor`](super::super::core::Executor).

mod invalidation;
mod payload_gates;

use std::sync::Arc;

use fraiseql_db::{ChangeLogWrite, ViewName};
pub use payload_gates::{StampContract, mutation_contract_errors};

use self::payload_gates::{PayloadGates, PayloadPosition};
use super::{
    super::{context::ExecutorContext, mutation::WriteSelections, resolve_inject_value},
    query_projection::selections_contain_field,
};
use crate::{
    error::{FraiseQLError, Result},
    graphql::{DirectiveEvaluator, FieldSelection},
    runtime::{
        ResultProjector,
        mutation_result::{MutationOutcome, parse_mutation_row},
        projection::effective_selections,
        suggest_similar,
    },
    schema::{CompiledSchema, InputStyle, MutationOperation, NamingConvention},
    security::SecurityContext,
};

/// Serve one entity of a payload: projected through `selections` as a `type_name` at
/// `position` under the payload's read gates (`payload_gates`), then put to the #423 field
/// authorizer.
///
/// # Errors
///
/// A type `position` cannot hold (a contract error, ruling AA 1), a type whose role the
/// caller lacks, and what the read plan refuses over the entity.
// Reason: the entity, where it sits, and the request's principal, gates and variables are
// each a separate input to one decision; a struct would only relocate them.
#[allow(clippy::too_many_arguments)]
fn serve_entity(
    ctx: &ExecutorContext,
    security_ctx: Option<&SecurityContext>,
    gates: &PayloadGates,
    position: PayloadPosition,
    type_name: &str,
    selections: &[FieldSelection],
    entity: &serde_json::Value,
    variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    gates.serve(ctx, security_ctx, position, type_name, selections, entity, variables)
}

// ── Typed cascade payload projection ─────────────────────────────────────────
//
// A `cascade = true` mutation returns a synthesized `<Name>Payload { entity,
// cascade, updatedFields }` (see `cli::converter::cascade_types`). The DB function
// emits the spec-nested cascade shape — `{ updated: [{ __typename, id, operation,
// entity }], deleted: [{ __typename, id, deletedAt }] }` — and the runtime builds
// the payload filtered to the client's selection set, projecting (camelCase) and
// field-authorizing the primary entity AND every cascade `entity` exactly like a
// queried entity. This is the load-bearing security fix: cascade entities no
// longer bypass the field authorizer and projection.

/// The synthesized cascade type names (mirror `cli::converter::cascade_types`).
const CASCADE_UPDATES_TYPE: &str = "CascadeUpdates";
const CASCADE_METADATA_TYPE: &str = "CascadeMetadata";
const QUERY_INVALIDATION_TYPE: &str = "QueryInvalidation";
const UPDATED_ENTITY_TYPE: &str = "UpdatedEntity";
const DELETED_ENTITY_TYPE: &str = "DeletedEntity";

/// One mebibyte, for the `max_response_size_mb` cascade ceiling.
const MIB: usize = 1024 * 1024;

/// Resolve a cascade mutation's concrete payload type: the first non-error member
/// of the result union, or the return type itself when it is not a union.
fn resolve_payload_type(return_type: &str, schema: &CompiledSchema) -> String {
    schema
        .find_union(return_type)
        .and_then(|u| {
            u.member_types
                .iter()
                .find(|t| schema.find_type(t).is_none_or(|td| !td.is_error))
                .cloned()
        })
        .unwrap_or_else(|| return_type.to_string())
}

/// Resolve the distinct views backing every entity type in a cascade, via the
/// compiled schema (`type name → sql_source`) — the same lookup the primary-entity
/// invalidation uses, so `tv_*` / multi-word / custom view names resolve correctly
/// instead of being guessed as `v_<lowercase>`. Types with no `sql_source`, and
/// unknown types, are skipped.
fn resolve_cascade_views(cascade: &serde_json::Value, schema: &CompiledSchema) -> Vec<ViewName> {
    let mut seen = std::collections::HashSet::new();
    let mut views = Vec::new();
    for arm in ["updated", "deleted"] {
        let Some(entries) = cascade.get(arm).and_then(serde_json::Value::as_array) else {
            continue;
        };
        for entry in entries {
            let Some(typename) = entry.get("__typename").and_then(serde_json::Value::as_str) else {
                continue;
            };
            if let Some(td) = schema.types.iter().find(|t| t.name.as_str() == typename) {
                let src = td.sql_source.as_str();
                if !src.is_empty() && seen.insert(src.to_string()) {
                    views.push(ViewName::from(src));
                }
            }
        }
    }
    views
}

/// The one type an unstamped success of a mutation returning `return_type` can be, when there
/// is exactly one: for a cascade mutation, its payload's declared entity type; otherwise the
/// only non-error member of a returned union, the only implementor of a returned interface,
/// or the returned type. `None` when a stamp is needed to tell (ruling AA 1).
fn unstamped_type(schema: &CompiledSchema, return_type: &str, is_cascade: bool) -> Option<String> {
    if is_cascade {
        return payload_entity_type(&resolve_payload_type(return_type, schema), schema);
    }
    match payload_gates::success_types(schema, return_type).as_slice() {
        [only] => Some(only.clone()),
        _ => None,
    }
}

/// The contract error for an unstamped success that could be several types.
fn unstamped_ambiguity(return_type: &str, candidates: &[String]) -> FraiseQLError {
    payload_gates::contract_error(format!(
        "the mutation function did not stamp entity_type, and a '{return_type}' success can be \
         any of: {} — stamp the one it produced; the write was rolled back",
        candidates.join(", ")
    ))
}

/// The contract error for an unstamped failure that could be several error types.
fn unstamped_error_ambiguity(return_type: &str, candidates: &[String]) -> FraiseQLError {
    payload_gates::contract_error(format!(
        "the mutation function did not stamp entity_type on a failure, and a '{return_type}' \
         failure can be any of: {} — stamp the one it produced \
         (`fraiseql.mutation_err(…, p_entity_type => '<ErrorType>')`); the write was rolled \
         back",
        candidates.join(", ")
    ))
}

/// The concrete entity type a payload wraps, read from its `entity` field type.
fn payload_entity_type(payload_type: &str, schema: &CompiledSchema) -> Option<String> {
    schema
        .find_type(payload_type)?
        .fields
        .iter()
        .find(|f| f.name.as_str() == "entity")?
        .field_type
        .type_name()
        .map(std::string::ToString::to_string)
}

/// Build the typed cascade payload `{ entity, cascade, updatedFields }`, filtered
/// to the client's selection set. Projects + field-authorizes the primary entity
/// and delegates the cascade envelope to [`build_cascade_updates`].
// Reason: cohesive per-call inputs for cascade projection (context, the two
// resolved type names, and the three outcome payloads); grouping into a struct
// would only relocate them.
#[allow(clippy::too_many_arguments)]
fn build_cascade_payload(
    ctx: &ExecutorContext,
    security_ctx: Option<&SecurityContext>,
    gates: &PayloadGates,
    payload_type: &str,
    entity_type: &str,
    entity: &serde_json::Value,
    cascade: Option<&serde_json::Value>,
    updated_fields: &[String],
    selections: &[FieldSelection],
    variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    let mut out = serde_json::Map::new();
    for sel in effective_selections(selections, payload_type, &ctx.schema) {
        match sel.name.as_str() {
            "__typename" => {
                out.insert(
                    sel.response_key().to_string(),
                    serde_json::Value::String(payload_type.to_string()),
                );
            },
            "entity" => {
                let projected = serve_entity(
                    ctx,
                    security_ctx,
                    gates,
                    PayloadPosition::CascadeEntity,
                    entity_type,
                    &sel.nested_fields,
                    entity,
                    variables,
                )?;
                out.insert(sel.response_key().to_string(), projected);
            },
            "cascade" => {
                let built = build_cascade_updates(
                    ctx,
                    security_ctx,
                    gates,
                    cascade,
                    &sel.nested_fields,
                    variables,
                )?;
                out.insert(sel.response_key().to_string(), built);
            },
            "updatedFields" => {
                out.insert(
                    sel.response_key().to_string(),
                    serde_json::Value::Array(
                        updated_fields.iter().cloned().map(serde_json::Value::String).collect(),
                    ),
                );
            },
            _ => {},
        }
    }
    Ok(serde_json::Value::Object(out))
}

/// Build the `CascadeUpdates` envelope (`updated` / `deleted` / `metadata`),
/// filtered to the client's selection set. Enforces the affected-entity ceiling
/// (truncating + flagging `metadata.truncated`) and the response-size ceiling
/// (rejecting an over-large cascade), per graphql-cascade `16_security`.
fn build_cascade_updates(
    ctx: &ExecutorContext,
    security_ctx: Option<&SecurityContext>,
    gates: &PayloadGates,
    cascade: Option<&serde_json::Value>,
    selections: &[FieldSelection],
    variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    let cascade_obj = cascade.and_then(serde_json::Value::as_object);
    let empty: Vec<serde_json::Value> = Vec::new();
    let updated: &[serde_json::Value] = cascade_obj
        .and_then(|c| c.get("updated"))
        .and_then(serde_json::Value::as_array)
        .map_or(empty.as_slice(), Vec::as_slice);
    let deleted: &[serde_json::Value] = cascade_obj
        .and_then(|c| c.get("deleted"))
        .and_then(serde_json::Value::as_array)
        .map_or(empty.as_slice(), Vec::as_slice);

    // Affected-entity ceiling: greedily fill up to the limit — `updated` first
    // (they carry entity data), remainder to `deleted` — and record the truncation
    // in metadata (spec `16_security` size-limiting). Greedy fill honors the full
    // limit for a lopsided cascade, where a half-each split would waste headroom.
    let limits = ctx.config.cascade_limits;
    let total = updated.len() + deleted.len();
    let truncated = total > limits.max_updated_entities;
    let (updated, deleted) = if truncated {
        let updated_keep = updated.len().min(limits.max_updated_entities);
        let deleted_keep = deleted.len().min(limits.max_updated_entities - updated_keep);
        (&updated[..updated_keep], &deleted[..deleted_keep])
    } else {
        (updated, deleted)
    };
    let affected_count = updated.len() + deleted.len();

    let mut out = serde_json::Map::new();
    for sel in effective_selections(selections, CASCADE_UPDATES_TYPE, &ctx.schema) {
        match sel.name.as_str() {
            "__typename" => {
                out.insert(
                    sel.response_key().to_string(),
                    serde_json::Value::String(CASCADE_UPDATES_TYPE.to_string()),
                );
            },
            "updated" => {
                let arr = build_updated_entities(
                    ctx,
                    security_ctx,
                    gates,
                    updated,
                    &sel.nested_fields,
                    variables,
                )?;
                out.insert(sel.response_key().to_string(), arr);
            },
            "deleted" => {
                let arr = build_deleted_entities(deleted, &sel.nested_fields, &ctx.schema)?;
                out.insert(sel.response_key().to_string(), arr);
            },
            "metadata" => {
                let meta = build_cascade_metadata(
                    &ctx.schema,
                    cascade_obj.and_then(|c| c.get("metadata")),
                    affected_count,
                    truncated,
                    truncated.then_some(total),
                    &sel.nested_fields,
                );
                out.insert(sel.response_key().to_string(), meta);
            },
            "invalidations" => {
                let arr = build_invalidations(
                    &ctx.schema,
                    cascade_obj.and_then(|c| c.get("invalidations")),
                    &sel.nested_fields,
                );
                out.insert(sel.response_key().to_string(), arr);
            },
            _ => {},
        }
    }
    let cascade_value = serde_json::Value::Object(out);

    // Response-size ceiling: reject an over-large cascade (spec `16_security`).
    let max_bytes = limits.max_response_size_mb.saturating_mul(MIB);
    if max_bytes > 0 {
        let size = serde_json::to_vec(&cascade_value).map_or(0, |v| v.len());
        if size > max_bytes {
            return Err(FraiseQLError::Validation {
                message: format!(
                    "cascade response too large: {size} bytes exceeds the \
                     max_response_size_mb={} ceiling",
                    limits.max_response_size_mb
                ),
                path:    Some("cascade".to_string()),
            });
        }
    }
    Ok(cascade_value)
}

/// Build the `CascadeMetadata` object, filtered to the client's selection set.
/// The runtime owns `affectedCount` / `truncated` / `originalCount` (computed from
/// what it actually returns) and the non-null `timestamp` (the function's if given,
/// else the server clock); `transactionId` / `depth` pass through from the
/// function's `cascade.metadata` (`depth` defaults to 0 if absent).
fn build_cascade_metadata(
    schema: &CompiledSchema,
    db_metadata: Option<&serde_json::Value>,
    affected_count: usize,
    truncated: bool,
    original_count: Option<usize>,
    selections: &[FieldSelection],
) -> serde_json::Value {
    let db = db_metadata.and_then(serde_json::Value::as_object);
    let mut out = serde_json::Map::new();
    for sel in effective_selections(selections, CASCADE_METADATA_TYPE, schema) {
        let key = sel.response_key().to_string();
        match sel.name.as_str() {
            "__typename" => {
                out.insert(key, serde_json::Value::String(CASCADE_METADATA_TYPE.to_string()));
            },
            "affectedCount" => {
                out.insert(key, serde_json::Value::from(affected_count as u64));
            },
            "truncated" => {
                out.insert(key, serde_json::Value::Bool(truncated));
            },
            "originalCount" => {
                if let Some(oc) = original_count {
                    out.insert(key, serde_json::Value::from(oc as u64));
                }
            },
            "depth" => {
                let depth = db
                    .and_then(|m| m.get("depth"))
                    .cloned()
                    .unwrap_or_else(|| serde_json::Value::from(0u64));
                out.insert(key, depth);
            },
            "timestamp" => {
                // The runtime owns the non-null `DateTime!` guarantee (like
                // `affectedCount`): prefer the function's transaction timestamp,
                // else stamp the current server time so a function that omits it
                // still yields an SDL-valid response rather than a null violation.
                let ts = db
                    .and_then(|m| m.get("timestamp"))
                    .cloned()
                    .unwrap_or_else(|| serde_json::Value::String(chrono::Utc::now().to_rfc3339()));
                out.insert(key, ts);
            },
            "transactionId" => {
                if let Some(v) =
                    db.and_then(|m| m.get("transactionId").or_else(|| m.get("transaction_id")))
                {
                    out.insert(key, v.clone());
                }
            },
            _ => {},
        }
    }
    serde_json::Value::Object(out)
}

/// Build the `invalidations: [QueryInvalidation!]!` array — the client-side cache
/// hints the function emitted, filtered to the client's selection set. Plain
/// passthrough: these are advisory hints, not entity data, so there is nothing to
/// project or field-authorize.
fn build_invalidations(
    schema: &CompiledSchema,
    entries: Option<&serde_json::Value>,
    selections: &[FieldSelection],
) -> serde_json::Value {
    let Some(arr) = entries.and_then(serde_json::Value::as_array) else {
        return serde_json::Value::Array(Vec::new());
    };
    let effective = effective_selections(selections, QUERY_INVALIDATION_TYPE, schema);
    let mut result = Vec::with_capacity(arr.len());
    for entry in arr {
        let Some(entry_obj) = entry.as_object() else {
            continue;
        };
        let mut item = serde_json::Map::new();
        for sel in &effective {
            if sel.name == "__typename" {
                item.insert(
                    sel.response_key().to_string(),
                    serde_json::Value::String(QUERY_INVALIDATION_TYPE.to_string()),
                );
            } else if let Some(v) = entry_obj.get(sel.name.as_str()) {
                item.insert(sel.response_key().to_string(), v.clone());
            }
        }
        result.push(serde_json::Value::Object(item));
    }
    serde_json::Value::Array(result)
}

/// Build the `updated: [UpdatedEntity!]!` array, projecting + field-authorizing
/// each entry's `entity` under its concrete `__typename`.
///
/// Fail-closed on any malformed entry: a missing or
/// unknown `__typename`, a missing `id`, or a missing/invalid `operation` aborts
/// the response rather than shipping an unprojectable entity or an SDL-invalid
/// non-null violation — regardless of what the client selected.
fn build_updated_entities(
    ctx: &ExecutorContext,
    security_ctx: Option<&SecurityContext>,
    gates: &PayloadGates,
    entries: &[serde_json::Value],
    selections: &[FieldSelection],
    variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    let effective = effective_selections(selections, UPDATED_ENTITY_TYPE, &ctx.schema);
    let mut result = Vec::with_capacity(entries.len());
    for entry in entries {
        let Some(entry_obj) = entry.as_object() else {
            return Err(FraiseQLError::Validation {
                message: "cascade.updated entry is not an object".to_string(),
                path:    Some("cascade.updated".to_string()),
            });
        };
        let Some(typename) = entry_obj.get("__typename").and_then(serde_json::Value::as_str) else {
            return Err(FraiseQLError::Validation {
                message: "cascade.updated entry is missing __typename".to_string(),
                path:    Some("cascade.updated.__typename".to_string()),
            });
        };
        // Fail-closed on a type we cannot deliver. Two rejections:
        //  - unknown type — we cannot project or authorize it;
        //  - framework-`internal` projection (#665, gate 5) — a change-log / checkpoint bookkeeping
        //    view is the change-capture *mechanism*, never a cascade-deliverable entity. The
        //    compiler excludes `internal` types from cascade classification, so they do not
        //    implement `CascadeNode`; an entry naming one (only hand-crafted JSONB can produce this
        //    — the framework never does) would otherwise yield a payload whose `entity` violates
        //    its declared interface. This reads `internal` off the re-loaded compiled schema — the
        //    flag's first runtime consumer, hence Phase 01 serializes it rather than `serde(skip)`.
        match ctx.schema.find_type(typename) {
            None => {
                return Err(FraiseQLError::Validation {
                    message: format!("cascade.updated entry has unknown __typename '{typename}'"),
                    path:    Some("cascade.updated.__typename".to_string()),
                });
            },
            Some(ty) if ty.internal => {
                return Err(FraiseQLError::Validation {
                    message: format!(
                        "cascade.updated entry names framework-internal type '{typename}', a \
                         bookkeeping projection that is not a cascade-deliverable entity"
                    ),
                    path:    Some("cascade.updated.__typename".to_string()),
                });
            },
            Some(_) => {},
        }
        // Non-null `id: ID!` and a valid `operation: CascadeOperation!` are required.
        if !entry_obj.contains_key("id") {
            return Err(FraiseQLError::Validation {
                message: "cascade.updated entry is missing id".to_string(),
                path:    Some("cascade.updated.id".to_string()),
            });
        }
        match entry_obj.get("operation").and_then(serde_json::Value::as_str) {
            Some("CREATED" | "UPDATED" | "DELETED") => {},
            other => {
                return Err(FraiseQLError::Validation {
                    message: format!(
                        "cascade.updated entry has a missing or invalid operation: {other:?} \
                         (expected CREATED, UPDATED, or DELETED)"
                    ),
                    path:    Some("cascade.updated.operation".to_string()),
                });
            },
        }
        let mut item = serde_json::Map::new();
        for sel in &effective {
            match sel.name.as_str() {
                "__typename" => {
                    item.insert(
                        sel.response_key().to_string(),
                        serde_json::Value::String(UPDATED_ENTITY_TYPE.to_string()),
                    );
                },
                "id" => {
                    if let Some(id) = entry_obj.get("id") {
                        item.insert(sel.response_key().to_string(), id.clone());
                    }
                },
                "operation" => {
                    if let Some(op) = entry_obj.get("operation") {
                        item.insert(sel.response_key().to_string(), op.clone());
                    }
                },
                "entity" => {
                    let entity_blob =
                        entry_obj.get("entity").cloned().unwrap_or(serde_json::Value::Null);
                    let projected = serve_entity(
                        ctx,
                        security_ctx,
                        gates,
                        PayloadPosition::UpdatedEntity,
                        typename,
                        &sel.nested_fields,
                        &entity_blob,
                        variables,
                    )?;
                    item.insert(sel.response_key().to_string(), projected);
                },
                _ => {},
            }
        }
        result.push(serde_json::Value::Object(item));
    }
    Ok(serde_json::Value::Array(result))
}

/// Build the `deleted: [DeletedEntity!]!` array. Deleted entries carry no entity
/// body (the row is gone) — only `id` + `deletedAt` — so there is nothing to
/// project or field-authorize.
///
/// Fail-closed: an entry missing the non-null `id` or
/// `deletedAt` aborts the response rather than emitting an SDL-invalid violation.
fn build_deleted_entities(
    entries: &[serde_json::Value],
    selections: &[FieldSelection],
    schema: &CompiledSchema,
) -> Result<serde_json::Value> {
    let effective = effective_selections(selections, DELETED_ENTITY_TYPE, schema);
    let mut result = Vec::with_capacity(entries.len());
    for entry in entries {
        let Some(entry_obj) = entry.as_object() else {
            return Err(FraiseQLError::Validation {
                message: "cascade.deleted entry is not an object".to_string(),
                path:    Some("cascade.deleted".to_string()),
            });
        };
        if !entry_obj.contains_key("id") {
            return Err(FraiseQLError::Validation {
                message: "cascade.deleted entry is missing id".to_string(),
                path:    Some("cascade.deleted.id".to_string()),
            });
        }
        let deleted_at = entry_obj.get("deletedAt").or_else(|| entry_obj.get("deleted_at"));
        let Some(deleted_at) = deleted_at else {
            return Err(FraiseQLError::Validation {
                message: "cascade.deleted entry is missing deletedAt".to_string(),
                path:    Some("cascade.deleted.deletedAt".to_string()),
            });
        };
        let mut item = serde_json::Map::new();
        for sel in &effective {
            match sel.name.as_str() {
                "__typename" => {
                    item.insert(
                        sel.response_key().to_string(),
                        serde_json::Value::String(DELETED_ENTITY_TYPE.to_string()),
                    );
                },
                "id" => {
                    if let Some(id) = entry_obj.get("id") {
                        item.insert(sel.response_key().to_string(), id.clone());
                    }
                },
                "deletedAt" => {
                    item.insert(sel.response_key().to_string(), deleted_at.clone());
                },
                _ => {},
            }
        }
        result.push(serde_json::Value::Object(item));
    }
    Ok(serde_json::Value::Array(result))
}

/// Executes GraphQL mutations.
///
/// Carries no capability bound. It used to require `A: SupportsMutations`, which was
/// described as making a write through `FraiseWireAdapter` a compiler error — true, and
/// the reason `execute_mutation` skipped the runtime check, which was the defect: the
/// bound speaks for the marker and not for `supports_mutations()`. Capability now lives
/// in the executor's write slot, resolved from both gates at construction and consulted
/// at step 0 of [`execute_mutation_impl`].
pub(in super::super) struct MutationRunner {
    ctx: Arc<ExecutorContext>,
}

impl MutationRunner {
    /// Create a new `MutationRunner` from a shared executor context.
    ///
    /// Zero-cost: `Arc` is already shared — this is just a newtype wrapper.
    pub(in super::super) const fn new(ctx: Arc<ExecutorContext>) -> Self {
        Self { ctx }
    }

    /// Execute a GraphQL mutation.
    ///
    /// # Errors
    ///
    /// Same as [`execute_mutation_impl`].
    pub(in super::super) async fn execute_mutation(
        &self,
        mutation_name: &str,
        variables: Option<&serde_json::Value>,
        selections: WriteSelections<'_>,
    ) -> Result<serde_json::Value> {
        // The typed API supplies the input via `variables`; it has no inline-literal
        // root arguments to resolve.
        execute_mutation_impl(
            &self.ctx,
            mutation_name,
            mutation_name,
            variables,
            None,
            selections,
            &[],
        )
        .await
        .map(|execution| execution.data)
    }
}

/// Re-case a mutation input payload's keys from the GraphQL surface naming
/// convention to canonical `snake_case`, recursing into nested input objects and
/// arrays of input objects.
///
/// The compiled schema stores field names in their GraphQL *surface* form: SDKs
/// pre-case them (the Python SDK camelCases at `registry.py:233`; the
/// `TypeScript` SDK stores whatever the author wrote, idiomatically
/// `camelCase`), and the
/// introspection layer renders them verbatim. With
/// [`NamingConvention::CamelCase`] a client therefore sends `camelCase` keys. The
/// Insert path maps those to positional SQL args by name (casing handled
/// implicitly), but the Update / `input_style = jsonb` path forwards the whole
/// object as one JSONB arg — so without this the SQL function receives
/// surface-cased keys it cannot read (#400, #456).
///
/// The input type's per-field map is used to *match* the incoming surface key and
/// to drive recursion into nested input types; every emitted canonical key is
/// then normalised with the engine's acronym-aware
/// [`to_snake_case`](crate::utils::to_snake_case) so writes round-trip exactly as
/// reads do (`s3Key → s3_key`, `dns1Id → dns_1_id`) — the same transform the
/// fallback below uses, regardless of which SDK authored the schema (#456).
///
/// When no per-field map is available — `input_type_name` is `None` (a raw `JSON`
/// input arg) or names a type absent from the compiled schema — it falls back to
/// the same key transform ([`recase_jsonb_keys_to_snake`]) so a single-JSONB
/// payload still reaches the function as `snake_case` (#400). A
/// [`NamingConvention::Preserve`] schema is always left untouched.
fn recase_input_payload(
    value: serde_json::Value,
    input_type_name: Option<&str>,
    schema: &CompiledSchema,
) -> serde_json::Value {
    // Preserve convention: the GraphQL surface already uses the canonical names.
    if schema.naming_convention != NamingConvention::CamelCase {
        return value;
    }
    // No per-field name map (raw `JSON` arg, or an Input type missing from the
    // compiled schema): recase the keys directly with the canonical reverse so the
    // single-JSONB path is not left forwarding verbatim camelCase keys (#400).
    let Some(input_type) = input_type_name.and_then(|n| schema.find_input_type(n)) else {
        return recase_jsonb_keys_to_snake(value);
    };
    match value {
        serde_json::Value::Object(map) => {
            let mut out = serde_json::Map::with_capacity(map.len());
            for (key, val) in map {
                // Match the incoming surface key against each field's surface name,
                // then emit the canonical key in `snake_case`. The stored field
                // name is the GraphQL *surface* name (SDKs pre-case it — e.g. the
                // Python SDK camelCases at `registry.py:233`), so it cannot be
                // forwarded verbatim; normalising with the engine's acronym-aware
                // `to_snake_case` — the same transform the raw-`JSON` fallback and
                // the read path use — is what reaches the function as `snake_case`
                // regardless of which SDK authored the schema (#456 / #400).
                let field = input_type.fields.iter().find(|f| schema.display_name(&f.name) == key);
                let canonical = field.map_or_else(
                    || crate::utils::to_snake_case(&key),
                    |f| crate::utils::to_snake_case(&f.name),
                );
                let recased = match field {
                    Some(f) => recase_input_field_value(val, &f.field_type, schema),
                    None => val,
                };
                out.insert(canonical, recased);
            }
            serde_json::Value::Object(out)
        },
        other => other,
    }
}

/// Recase one input field's *value*: when `field_type` names a nested input
/// object (or a list of them), recurse into the object's keys; scalars, enums,
/// free-form JSON, and lists of scalars are returned untouched.
///
/// Shared by the Update path (the whole payload is one JSONB arg) and the Insert
/// path (each composite field is one positional JSONB arg), so both recase nested
/// composite keys identically — without it, a `jsonb_populate_record(NULL::config,
/// $arg)` on the Insert path sees the surface-cased keys it cannot read (#400).
fn recase_input_field_value(
    value: serde_json::Value,
    field_type: &str,
    schema: &CompiledSchema,
) -> serde_json::Value {
    let Some(nested) = nested_input_type_name(field_type, schema) else {
        return value;
    };
    match value {
        serde_json::Value::Object(_) => recase_input_payload(value, Some(nested.as_str()), schema),
        serde_json::Value::Array(items) => serde_json::Value::Array(
            items
                .into_iter()
                .map(|it| match it {
                    serde_json::Value::Object(_) => {
                        recase_input_payload(it, Some(nested.as_str()), schema)
                    },
                    other => other,
                })
                .collect(),
        ),
        other => other,
    }
}

/// Recase every object key of a single-JSONB input payload to canonical
/// `snake_case` with the engine's one canonical acronym-aware reverse
/// ([`to_snake_case`](crate::utils::to_snake_case)), recursing into nested objects
/// and arrays; scalar values are untouched.
///
/// The fallback for [`recase_input_payload`] when no registered Input type
/// supplies a per-field name map — a custom `mutation(input: JSON)`, or an Update
/// whose Input type is absent from the compiled schema. Sharing `to_snake_case`
/// with the read path makes write keys round-trip exactly as reads do: `dns1Id` →
/// `dns_1_id` (digit split), `s3Key` → `s3_key` (registered acronym kept whole).
fn recase_jsonb_keys_to_snake(value: serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::Object(map) => {
            let mut out = serde_json::Map::with_capacity(map.len());
            for (key, val) in map {
                out.insert(crate::utils::to_snake_case(&key), recase_jsonb_keys_to_snake(val));
            }
            serde_json::Value::Object(out)
        },
        serde_json::Value::Array(items) => {
            serde_json::Value::Array(items.into_iter().map(recase_jsonb_keys_to_snake).collect())
        },
        other => other,
    }
}

/// If `field_type` (e.g. `"BillingAddressInput"`, `"[TagInput!]"`) names a known
/// input object type, return that type's bare name (`[`, `]`, `!` stripped) so
/// [`recase_input_payload`] can recurse; otherwise `None` (scalars, enums, and
/// unknown types are left untouched).
fn nested_input_type_name(field_type: &str, schema: &CompiledSchema) -> Option<String> {
    let base = field_type.trim_matches(|c| c == '[' || c == ']' || c == '!');
    schema.find_input_type(base).map(|_| base.to_string())
}

/// Core mutation execution logic.
///
/// Called from:
/// - `MutationRunner::execute_mutation` — the typed entries a non-GraphQL transport uses
/// - `Executor::execute_mutation_query` — raw GraphQL dispatch
/// - `execute_with_security_internal` — authenticated GraphQL dispatch
///
/// The caller is **not** responsible for establishing write capability. It used to be,
/// by one of two means — a `SupportsMutations` bound or a `supports_mutations()` guard —
/// and the first caller above met it with the bound, which speaks only for the marker.
/// Step 0 below resolves the executor's write handle — present exactly when a constructor
/// bounded on `Writer` built it — and that handle is the only way to reach the database
/// from here.
///
/// # Errors
///
/// * [`FraiseQLError::Validation`] — mutation not found, no `sql_source`, missing security context
///   for `inject` params, or database function returned no rows.
/// * [`FraiseQLError::Database`] — the adapter's `execute_write` failed.
///
/// `response_key` is the key the result appears under in `data` — the document's
/// alias when it has one, otherwise `mutation_name`. Two roots calling the same
/// mutation are told apart only by it.
/// What a mutation produced: the GraphQL-shaped response **and** the parsed
/// `mutation_response` envelope (#1330).
///
/// A GraphQL client reads the projection; a transport whose own wire format *is*
/// the envelope — gRPC's `MutationResponse{success, id, error}` — needs
/// `entity_id` and the failure message, and the projection cannot carry them: on
/// success it holds the *entity*, whose `id` is the row's, not the envelope's, and
/// on failure it holds only the error class. Consuming the envelope and discarding
/// it is what forced gRPC to call the database directly in the first place.
#[derive(Debug)]
#[non_exhaustive]
pub struct MutationExecution {
    /// The `{"data": {...}}` envelope, projected by the selection set.
    pub data:    serde_json::Value,
    /// The parsed `app.mutation_response` row this write produced.
    pub outcome: MutationOutcome,
}

/// Build a mutation's GraphQL response value from the `mutation_response` row the
/// SQL function returned.
///
/// Pure: it reads `ctx.schema`, the request's principal and selection set, and the
/// parsed outcome — and touches the database not at all. That is what lets the
/// caller run it *inside* the write's transaction, so the field authorizer (#423)
/// can refuse the write rather than only its result (#1353).
///
/// # Errors
///
/// Returns [`FraiseQLError::Authorization`] when the field authorizer refuses a
/// selected policy-gated field (or cannot be consulted), and
/// [`FraiseQLError::Internal`] when a gated field's arguments cannot be read. The caller
/// rolls the write back on any error, whatever its variant.
// Reason: the outcome, the mutation's shape, and the request's principal, gates, selection
// and variables are each a separate input; a struct would only relocate them.
#[allow(clippy::too_many_arguments)]
fn build_mutation_result(
    ctx: &ExecutorContext,
    security_ctx: Option<&SecurityContext>,
    gates: &PayloadGates,
    outcome: MutationOutcome,
    mutation_return_type: &str,
    is_cascade: bool,
    selections: &[FieldSelection],
    authz_variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    match outcome {
        MutationOutcome::Success {
            entity,
            entity_type,
            cascade,
            updated_fields,
            ..
        } if is_cascade => {
            // Cascade mutation: build the typed payload `{ entity, cascade,
            // updatedFields }`, projecting + field-authorizing the primary entity
            // and every cascade entity (findings 1, 5). The payload type is the
            // success member of the (possibly error-union) return type; the
            // concrete entity type is the DB-stamped `entity_type`, else the
            // payload's `entity` field type.
            let payload_type = resolve_payload_type(mutation_return_type, &ctx.schema);
            // The stamp, else the payload's declared entity type (ruling AA 1). A stamp the
            // cascade entity cannot hold is refused where it is served (`PayloadGates`).
            let entity_type_name = match entity_type {
                Some(stamp) => stamp,
                None => {
                    unstamped_type(&ctx.schema, mutation_return_type, true).ok_or_else(|| {
                        unstamped_ambiguity(
                            mutation_return_type,
                            &payload_gates::cascade_node_types(&ctx.schema),
                        )
                    })?
                },
            };
            build_cascade_payload(
                ctx,
                security_ctx,
                gates,
                &payload_type,
                &entity_type_name,
                &entity,
                cascade.as_ref(),
                &updated_fields,
                selections,
                authz_variables,
            )
        },
        MutationOutcome::Success {
            entity,
            entity_type,
            cascade,
            updated_fields,
            ..
        } => {
            // Resolve the concrete GraphQL type of the success entity (ruling AA 1): the
            // stamp, which must name a type a success can be — a non-error member of the
            // returned union, an implementor of the returned interface, or the returned type —
            // else, unstamped, the one type a success can be. Where there are several, the
            // function's silence must not pick one: a contract error, and the write rolls back.
            let successes = &gates.contract().success;
            let typename = match entity_type {
                Some(stamp) if successes.contains(&stamp) => stamp,
                Some(stamp) => {
                    return Err(payload_gates::off_contract(
                        PayloadPosition::Root,
                        &stamp,
                        successes,
                    ));
                },
                None => unstamped_type(&ctx.schema, mutation_return_type, false)
                    .ok_or_else(|| unstamped_ambiguity(mutation_return_type, successes))?,
            };

            // Project the entity through the single canonical projector — the same
            // snake_case source keys, surface output keys, depth-aware recursion and
            // selection-gated __typename as the query path — so a mutation's success
            // payload and a query over the same entity return an identical shape. Served
            // under the read gates a query of the same entity meets (`payload_gates`), then
            // the dynamic field authorizer (#423), per the resolved concrete type.
            // Fail-closed.
            let mut projected = serve_entity(
                ctx,
                security_ctx,
                gates,
                PayloadPosition::Root,
                &typename,
                selections,
                &entity,
                authz_variables,
            )?;

            // Cascade is opt-in (`cascade = true`, handled by the guarded arm
            // above): a non-cascade mutation never surfaces cascade, even if its
            // function returns a `cascade` JSONB. `cascade` is a typed,
            // selection-gated payload field on cascade mutations, or nothing.
            let _ = cascade;

            // Surface `updated_fields` (the GraphQL field names this mutation
            // changed) as `updatedFields`, selection-gated — present only when the
            // client selects it, so a mutation that does not ask for it keeps an
            // exact projected shape (#433). An empty list (noop) still surfaces as
            // `[]` when selected.
            if selections_contain_field(selections, "updatedFields") {
                if let serde_json::Value::Object(ref mut map) = projected {
                    map.insert(
                        "updatedFields".to_string(),
                        serde_json::Value::Array(
                            updated_fields.into_iter().map(serde_json::Value::String).collect(),
                        ),
                    );
                }
            }

            Ok(projected)
        },
        MutationOutcome::Error {
            error_class,
            message,
            http_status,
            entity_type,
            metadata,
        } => {
            let status = error_class.as_str();

            // Build the error projection source from the error_detail JSONB, enriched
            // with the composite's first-class fields under snake_case keys so a
            // declared error type can surface them as ordinary projected fields
            // (project_entity omits any field whose source key is absent). The
            // always-injected `status` is attached after projection, below.
            let mut source_map = match metadata {
                serde_json::Value::Object(map) => map,
                _ => serde_json::Map::new(),
            };
            if !message.is_empty() {
                source_map
                    .entry("message".to_string())
                    .or_insert_with(|| serde_json::Value::String(message));
            }
            if let Some(code) = http_status {
                source_map
                    .entry("http_status".to_string())
                    .or_insert_with(|| serde_json::json!(code));
            }
            source_map
                .entry("error_class".to_string())
                .or_insert_with(|| serde_json::Value::String(status.to_string()));
            let source = serde_json::Value::Object(source_map);

            // Resolve the concrete error type to project (ruling AG 3), symmetric with the
            // success arm: the stamp — `fraiseql.mutation_err(…, p_entity_type => …)` — must be
            // one of the error types this mutation can return (a union's or an interface's
            // error members; for an object return, the #465 pattern, any declared error type),
            // routing onto the *specific* member (`DuplicateEmailError` vs `ValidationError`).
            // Unstamped, a returned union or interface with one error member is that member;
            // with several, the function's silence must not pick one: a contract error, and
            // the write rolls back. An unstamped failure of an object return, or of a union
            // with no error member, is the untyped failure: there is no member to pick.
            let errors = &gates.contract().error;
            let returns_members = ctx.schema.find_union(mutation_return_type).is_some()
                || ctx.schema.find_interface(mutation_return_type).is_some();
            let error_type = match entity_type {
                Some(stamp) if errors.contains(&stamp) => ctx.schema.find_type(&stamp),
                Some(stamp) => {
                    return Err(payload_gates::off_contract(PayloadPosition::Root, &stamp, errors));
                },
                None if !returns_members => None,
                None => match errors.as_slice() {
                    [] => None,
                    [only] => ctx.schema.find_type(only),
                    _ => return Err(unstamped_error_ambiguity(mutation_return_type, errors)),
                },
            };

            // Project the error source through the same canonical projector when the
            // schema declares a matching error type. Otherwise emit just __typename
            // (only when selected, matching the query contract); status is attached
            // below in both cases.
            // Served under the read gates of the error type (`payload_gates`) and the
            // dynamic field authorizer (#423), so a scoped or gated field on an error type
            // cannot leak through the error arm.
            let mut result = if let Some(td) = error_type {
                serve_entity(
                    ctx,
                    security_ctx,
                    gates,
                    PayloadPosition::Root,
                    td.name.as_str(),
                    selections,
                    &source,
                    authz_variables,
                )?
            } else {
                let mut map = serde_json::Map::new();
                // Scan recursively: `__typename` may be nested inside an inline
                // fragment (`... on T { __typename }`), not just at the top level.
                if selections_contain_field(selections, "__typename") {
                    map.insert(
                        "__typename".to_string(),
                        serde_json::Value::String(mutation_return_type.to_string()),
                    );
                }
                serde_json::Value::Object(map)
            };

            // Inject the synthetic `status` field — not part of the type definition,
            // but required by clients to discriminate error outcomes.
            if let serde_json::Value::Object(ref mut map) = result {
                map.insert("status".to_string(), serde_json::Value::String(status.to_string()));
            }

            Ok(result)
        },
    }
}

pub(in super::super) async fn execute_mutation_impl(
    ctx: &Arc<ExecutorContext>,
    mutation_name: &str,
    response_key: &str,
    variables: Option<&serde_json::Value>,
    security_ctx: Option<&SecurityContext>,
    selections: WriteSelections<'_>,
    inline_arguments: &[crate::graphql::GraphQLArgument],
) -> Result<MutationExecution> {
    let selections = selections.as_slice();

    // 0. May this executor write at all? First, before every other gate.
    //
    // `78f91c9e2` fixed this for the document path and named the reason: a capability
    // refusal that arrives at the dispatch has already let the operation authorizer,
    // `requires_role`, `requires_actor`, argument validation and the `before:mutation`
    // chain run — and `before:mutation` runs app-authored rule code, placed after every
    // static gate precisely so an unauthorized caller never reaches it. The five typed
    // write entries still had no check of any kind: they trusted the `SupportsMutations`
    // bound on their impl block, which settles the *marker* and cannot settle the
    // *override* that `execute_function_call` is actually keyed on.
    //
    // Both paths now resolve one slot, here. The returned handle is also the only way to
    // reach the database further down, so this cannot decay back into a check that some
    // later entry point forgets to make.
    let writer = ctx.writer(mutation_name)?;

    // #1336 backstop: the same question the read path asks, at the write chokepoint
    // every transport converges on (#1327, #1330).
    crate::runtime::executor::support::security::enforce_enrichment_resolved(
        &ctx.schema,
        security_ctx,
    )?;

    // 1. Locate the mutation definition
    let mutation_def = ctx.schema.find_mutation(mutation_name).ok_or_else(|| {
        let display_names: Vec<String> =
            ctx.schema.mutations.iter().map(|m| ctx.schema.display_name(&m.name)).collect();
        let candidate_refs: Vec<&str> = display_names.iter().map(String::as_str).collect();
        let suggestion = suggest_similar(mutation_name, &candidate_refs);
        let message = match suggestion.as_slice() {
            [s] => {
                format!("Mutation '{mutation_name}' not found in schema. Did you mean '{s}'?")
            },
            [a, b] => format!(
                "Mutation '{mutation_name}' not found in schema. Did you mean '{a}' or '{b}'?"
            ),
            [a, b, c, ..] => format!(
                "Mutation '{mutation_name}' not found in schema. Did you mean '{a}', '{b}', or \
                 '{c}'?"
            ),
            _ => format!("Mutation '{mutation_name}' not found in schema"),
        };
        FraiseQLError::Validation {
            message,
            path: None,
        }
    })?;

    // 1a. Operation-level authorization (#422): the universal mutation chokepoint.
    //     EVERY mutation entry path converges here — the two `*_internal` GraphQL
    //     branches, `execute_mutation_query`, and the direct `SupportsMutations` API
    //     used by the anonymous-REST write path (which bypasses both chokepoints).
    //     Runs after `find_mutation` so an unknown name keeps its "not found" message
    //     (no enumeration leak), and before `requires_role` (AND-composition).
    //     Fail-closed: a `Deny` or any policy error returns 403.
    if let Some(authorizer) = ctx.config.authorizer.as_ref() {
        let ops = [crate::security::AuthzOperation::root(
            crate::security::OperationKind::Mutation,
            mutation_name,
            Some(mutation_def.return_type.as_str()),
        )];
        crate::security::authorizer::enforce_authz(
            authorizer.as_ref(),
            security_ctx,
            &ops,
            variables,
        )?;
    }

    // 1b. Enforce requires_role — "not found" (not "forbidden") to prevent enumeration.
    crate::security::role_gate::enforce_requires_role(
        "Mutation",
        mutation_name,
        mutation_def.requires_role.as_deref(),
        security_ctx,
    )?;

    // 1c. Enforce requires_actor (#966). This chokepoint is why "every transport"
    //     is a fact rather than a claim: every mutation entry path — both GraphQL
    //     branches, `execute_mutation_query`, `execute_mutation_as` (gRPC, #1330),
    //     and the direct `SupportsMutations` API the REST write path uses —
    //     converges here.
    //
    //     ⚠ That sentence was FALSE between #966 and #1330, and the enumeration
    //     above is why it read as true: gRPC was simply not in the list, because it
    //     reached `execute_function_call` on the adapter and never arrived here at
    //     all. A claim of universality is only worth the audit that produced its
    //     list — `tools/check-mutation-dispatch-sites.sh` is now that audit, run on
    //     every push, and its exemption list is empty.
    crate::security::actor_type::enforce_requires_actor(
        "Mutation",
        mutation_name,
        &mutation_def.requires_actor,
        security_ctx,
    )?;

    // 1e. #1005: the fields selected on the payload must be defined on it
    //      (GraphQL § 5.3.1). #939 wired this into `QueryMatcher::match_query` and
    //      `execute_node_query`; mutations classify and dispatch on their own path
    //      and reached neither, so the silent-null behaviour #939 removed from the
    //      read path survived on the write path — where it costs more, because the
    //      row is committed before the client reads the null.
    //
    //      Validated *here*, at the chokepoint, and therefore before the write.
    //      A payload type that is a union is scoped per inline fragment by
    //      `validate_selection_set`; a bare key on one passes unadjudicated,
    //      because scoring it against the wrong variant would reject a working
    //      mutation.
    //
    //      Named fragment spreads are already expanded by classification, and
    //      § 5.3.1 is a static rule, so this runs on the unfiltered selection set
    //      rather than after `@skip`/`@include` — exactly as the query path does.
    crate::graphql::validate_selection_set(&ctx.schema, &mutation_def.return_type, selections)?;

    // 1f. #1357: and the payload's composite fields must each *have* a selection set
    //      (GraphQL § 5.3.3). On a write an absent one is not under-specification but
    //      permission: `project_entity` returns the stored entity unchanged for an
    //      empty slice, and `selection_set_selects_gated_field` reports nothing gated
    //      is selected, so `mutation { createUser }` answered with every
    //      `authorize`-gated field of the row while the #423 field authorizer took
    //      zero calls. `{ id profile }` did the same one level down, returning the
    //      whole `profile` sub-blob.
    //
    //      Like § 5.3.1 above this is a static rule, so it runs on `selections` —
    //      the written set, before `@skip`/`@include`. That ordering is load-bearing
    //      here in a way it is not above: an all-skipped selection set is a valid
    //      document whose answer is `{}`, and adjudicating the post-directive set
    //      would refuse it.
    crate::graphql::validate_leaf_field_selections(
        &ctx.schema,
        &mutation_def.return_type,
        selections,
    )?;

    // 1d. #1154: the arguments written on the mutation field must be defined on
    //      it (GraphQL § 5.4.1). Step 3 below binds positionally from
    //      `mutation_def.arguments`, so an undeclared argument was dropped and
    //      the write ran without it — the silent failure the query path had, on
    //      the surface where it costs the most. `inject_params` are deliberately
    //      *not* accepted: they are sourced server-side from the caller's claims
    //      and the schema documents them as not exposed as GraphQL arguments, so
    //      a client naming one is writing an argument the field does not have.
    crate::runtime::validate_argument_names(
        &format!("Mutation.{}", ctx.schema.display_name(&mutation_def.name)),
        &mutation_def.arguments.iter().map(|arg| arg.name.clone()).collect::<Vec<_>>(),
        inline_arguments,
    )?;

    // 2. Require a sql_source (PostgreSQL function name).
    //
    // Fall back to the operation's table field when sql_source is absent.
    // The CLI compiler stores the SQL function name in both places
    // (sql_source and operation.{Insert|Update|Delete}.table), but older or
    // alternate compilation paths (e.g. fraiseql-core's own codegen) may only
    // populate operation.table and leave sql_source as None.
    let sql_source_owned: String;
    let sql_source: &str = if let Some(src) = mutation_def.sql_source.as_deref() {
        src
    } else {
        match &mutation_def.operation {
            MutationOperation::Insert { table }
            | MutationOperation::Update { table }
            | MutationOperation::Delete { table }
                if !table.is_empty() =>
            {
                sql_source_owned = table.clone();
                &sql_source_owned
            },
            _ => {
                return Err(FraiseQLError::Validation {
                    message: format!("Mutation '{mutation_name}' has no sql_source configured"),
                    path:    None,
                });
            },
        }
    };

    // 3. Build positional args Vec from variables in ArgumentDefinition order. Validate that every
    //    required (non-nullable, no default) argument is present.
    //
    //    Input object unwrapping: when the mutation has a single argument named "input"
    //    whose type is an Input type, AND the client sends a JSON object for that argument,
    //    unwrap the object's fields and pass them positionally in the order defined by the
    //    input type's field list.  This keeps the SQL function signature flat while letting
    //    the GraphQL API use the standard input object pattern.
    //
    //    Merge the root field's inline arguments (e.g. `createMachine(input: { ... })`)
    //    into the variables view first, so an inline-literal input — including one whose
    //    fields reference nested `$var`s — is visible below. This mirrors the query path
    //    (QueryMatcher merges inline args; whole-arg request variables take precedence).
    //    `merged_variables` owns the value the shadowed `variables` borrows for the rest
    //    of the function; it is `Some` only when an inline arg added a new key.
    let merged_variables: Option<serde_json::Value> = if inline_arguments.is_empty() {
        None
    } else {
        let var_map = crate::runtime::matcher::QueryMatcher::extract_arguments(variables);
        let mut obj = match variables {
            Some(serde_json::Value::Object(m)) => m.clone(),
            _ => serde_json::Map::new(),
        };
        let mut added = false;
        for arg in inline_arguments {
            if !obj.contains_key(&arg.name) {
                // A malformed stored argument is refused, not skipped: skipping
                // it would run the mutation with a silently missing input (#719).
                if let Some(val) =
                    crate::runtime::matcher::QueryMatcher::resolve_inline_arg(arg, &var_map)?
                {
                    obj.insert(arg.name.clone(), val);
                    added = true;
                }
            }
        }
        added.then_some(serde_json::Value::Object(obj))
    };
    let variables = merged_variables.as_ref().or(variables);

    // 3a. `before:mutation` enforcement (#1327). Here, and not in the HTTP handler,
    //     because `before:mutation` is enforcement and the handler could only run it
    //     once per request: it keyed the chain on `parse_query(…).root_field` — the
    //     *first* root — against `request.variables`, and never ran on the REST write
    //     route at all. Those were the three bypasses. At this chokepoint the chain
    //     runs once per executed root, in document order (the caller iterates
    //     `Mutation { roots }`), on every transport, immediately before the write.
    //
    //     Placed after every static gate so an unauthorized caller never reaches
    //     app-authored rule code, and after the inline-argument merge above so the
    //     gate sees the arguments the write will bind from — literals included —
    //     rather than only the request's `variables` map. Keyed on `mutation_name`,
    //     never `response_key`: two roots calling the same mutation differ only by
    //     alias, so keying on the alias would run one chain twice and skip the other.
    //
    //     A rewrite replaces the whole argument view, so it reaches the bound args and
    //     the field-authorizer view below, not just the request variables.
    //     `gated_variables` owns what the shadowed `variables` borrows from here on.
    //     The chain reads through a caller-scoped, read-only bridge (#1328) built
    //     from *this* context and *this* principal — see
    //     `support::hook_reader::CallerScopedReader`. It is passed as a closure so a
    //     build with no gate installed still pays only the `Option` check.
    let gated_variables = crate::security::mutation_gate::enforce_before_mutation(
        ctx.config.before_mutation_gate.as_deref(),
        security_ctx,
        mutation_name,
        response_key,
        variables,
        || {
            std::sync::Arc::new(
                crate::runtime::executor::support::hook_reader::CallerScopedReader::new(
                    std::sync::Arc::clone(ctx),
                    security_ctx,
                ),
            )
        },
    )
    .await?;
    let variables = gated_variables.as_ref().or(variables);

    // The same map, in the shape the field authorizer resolves against: a policy
    // that matches on a gated field's argument must see the value the client sent,
    // not the `{"$var": …}` reference marker (#903).
    let authz_variables = crate::runtime::matcher::QueryMatcher::extract_arguments(variables);

    let vars_obj = variables.and_then(|v| v.as_object());

    // #1362: an enum written at an argument — or at a field of the input object
    // under one — must name a member the schema declares. Here rather than in the
    // GraphQL matcher because this is the seam every transport that writes passes
    // through (#1327): REST, gRPC and MCP arrive with a JSON payload and no
    // document, so a check that reads `[GraphQLArgument]` would cover one caller in
    // four. Before the branch below, so the single-JSONB path and the flatten path
    // are adjudicated by one call — the two differ in how the value reaches SQL,
    // never in whether it should.
    crate::runtime::validate_enum_argument_values(
        &ctx.schema,
        &format!("Mutation.{}", ctx.schema.display_name(mutation_name)),
        &mutation_def.arguments,
        variables,
    )?;

    let mut missing_required: Vec<&str> = Vec::new();
    let total_args = mutation_def.arguments.len() + mutation_def.inject_params.len();
    let mut args: Vec<serde_json::Value> = Vec::with_capacity(total_args);

    // Detect the single-`input`-object pattern: exactly one argument named "input".
    // `input_type_name` is its declared Input type when it has one (vs a raw `JSON`
    // scalar — a custom `mutation(input: JSON)` whose SQL function takes
    // `(input jsonb, …)`); it drives the field-level flatten/recase path below.
    let single_input_arg =
        mutation_def.arguments.len() == 1 && mutation_def.arguments[0].name == "input";
    // The compiler emits an input-type reference as `FieldType::Object(name)`, never
    // `FieldType::Input` (the converter's `parse_field_type` has no `Input` variant),
    // so an `Object` whose name resolves to a registered input type IS a structured
    // input arg. Without recognising it here the arg matches neither the
    // single-JSONB nor the flatten branch and falls through to the verbatim
    // standard-arg path — forwarding camelCase keys to the SQL function with no
    // recasing, regardless of `naming_convention` (#456). The `find_input_type`
    // guard is exact: only names registered as input types match, so output-object
    // args are never misclassified.
    let input_type_name = if single_input_arg {
        match &mutation_def.arguments[0].arg_type {
            crate::schema::FieldType::Input(name) => Some(name.as_str()),
            crate::schema::FieldType::Object(name)
                if ctx.schema.find_input_type(name).is_some() =>
            {
                Some(name.as_str())
            },
            _ => None,
        }
    } else {
        None
    };
    // A single `input` arg shaped as a JSON payload — a structured Input type
    // (`FieldType::Input`, or the `Object` the compiler actually emits for one) or a
    // raw `JSON` scalar — as opposed to a plain scalar (`input: String`), which is a
    // positional arg, not a JSONB blob.
    let input_arg_is_structured = single_input_arg
        && match &mutation_def.arguments[0].arg_type {
            crate::schema::FieldType::Input(_) | crate::schema::FieldType::Json => true,
            crate::schema::FieldType::Object(name) => ctx.schema.find_input_type(name).is_some(),
            _ => false,
        };

    // Update mutations pass the entire input object as a single JSONB arg, which
    // preserves all three field states that typed positional args cannot express:
    //   - key absent            → leave the database value unchanged
    //   - key present, null     → SET field = NULL
    //   - key present, value    → SET field = <value>
    // SQL update functions use `input_payload ? 'field'` to test key presence.
    //
    // Insert / Delete / Custom flatten the Input type fields to positional args as
    // before (no three-state problem: absent ≡ NULL for creates; deletes need only
    // the PK).
    let is_update = matches!(&mutation_def.operation, MutationOperation::Update { .. });

    // An explicit `input_style = jsonb` opt-in (orthogonal to the DML verb):
    // forward the whole input as one JSONB arg regardless of the operation, so a
    // backend using the single-`jsonb`-wrapper convention can register the real
    // verb (Insert/Delete/Custom) instead of being forced to Update purely to opt
    // into single-JSONB passing — letting the Change Spine record the true
    // `modification_type` (#400 / `input_style`).
    let jsonb_input_style = matches!(mutation_def.input_style, InputStyle::Jsonb);

    // The whole `input` object is forwarded as ONE JSONB arg — never flattened to
    // positional columns — when the operation is Update (three-state semantics
    // above), the mutation opts in via `input_style = jsonb`, OR the structured
    // `input` arg is not a known Input type (a custom `mutation(input: JSON)`, or
    // an Update whose Input type is absent from the compiled schema). On that path
    // the keys must be recased from the camelCase GraphQL surface to canonical
    // snake_case so the function can read them — field-driven when the Input type
    // is known (preserves intentional names), acronym-aware key-driven
    // `to_snake_case` otherwise (#400). The flatten path below gets recasing for
    // free via positional args + recase_input_field_value.
    let known_input_type = input_type_name.and_then(|n| ctx.schema.find_input_type(n)).is_some();
    let pass_input_as_single_jsonb =
        input_arg_is_structured && (is_update || jsonb_input_style || !known_input_type);

    if pass_input_as_single_jsonb {
        // Forward the whole `input` value as ONE JSONB arg, re-cased from the
        // GraphQL surface naming convention to the schema's canonical (stored)
        // field names so the SQL function — which reads `payload->>'snake_field'`
        // / jsonb_populate_record — sees the values (#400). The Insert path below
        // gets this for free via positional args; this path must do it explicitly.
        //
        // Usually an object (an Input type or `input: JSON`); a raw `input: JSON`
        // arg may also be an array or scalar, so recase recurses into objects and
        // lists and leaves scalars untouched. An absent or explicit-null value on a
        // non-null no-default arg is a missing required argument; otherwise the
        // arg's default (or SQL NULL) is forwarded so the function keeps its arity.
        let arg = &mutation_def.arguments[0];
        match vars_obj.and_then(|obj| obj.get("input")) {
            Some(value) if !value.is_null() => {
                args.push(recase_input_payload(value.clone(), input_type_name, &ctx.schema));
            },
            _ if !arg.nullable && arg.default_value.is_none() => missing_required.push("input"),
            _ => args
                .push(arg.default_value.as_ref().map_or(serde_json::Value::Null, |v| v.to_json())),
        }
    } else if let Some(input_type) = input_type_name.and_then(|n| ctx.schema.find_input_type(n)) {
        // Insert / Delete / Custom: flatten Input type fields to positional typed args.
        let input_obj = vars_obj.and_then(|obj| obj.get("input")).and_then(|v| v.as_object());
        if let Some(input_obj) = input_obj {
            // #414: enforce required (non-null, no-default) input fields before the
            // database call, rejecting an omitted-or-explicit-null required field
            // with a GraphQL validation error instead of passing SQL NULL through.
            //
            // Look up each field by its GraphQL surface name (`display_name`):
            // under `NamingConvention::CamelCase` the client sends camelCase keys
            // and `field.name` is itself the surface name (`display_name` is then a
            // no-op), so matching on it makes the required check correct and finds
            // the value to forward.
            let mut missing_input_fields: Vec<&str> = Vec::new();
            for field in &input_type.fields {
                let key = ctx.schema.display_name(&field.name);
                let value = input_obj.get(&key);
                if field.is_required() && value.is_none_or(serde_json::Value::is_null) {
                    missing_input_fields.push(field.name.as_str());
                }
                // Top-level keys map to columns positionally (casing implicit), but a
                // field whose type is a nested input object is passed as one JSONB arg
                // — recase its keys so the SQL function can read them (#400).
                let raw = value.cloned().unwrap_or(serde_json::Value::Null);
                args.push(recase_input_field_value(raw, &field.field_type, &ctx.schema));
            }
            if !missing_input_fields.is_empty() {
                return Err(FraiseQLError::Validation {
                    message: format!(
                        "Mutation '{mutation_name}': required input field(s) not provided or \
                         null: {}",
                        missing_input_fields.join(", ")
                    ),
                    path:    None,
                });
            }
        } else if !mutation_def.arguments[0].nullable {
            missing_required.push("input");
        }
    } else {
        // Standard argument handling (flat arguments, no input object)
        args.extend(mutation_def.arguments.iter().map(|arg| {
            let value = vars_obj.and_then(|obj| obj.get(&arg.name)).cloned();
            if let Some(v) = value {
                v
            } else {
                if !arg.nullable && arg.default_value.is_none() {
                    missing_required.push(&arg.name);
                }
                arg.default_value.as_ref().map_or(serde_json::Value::Null, |v| v.to_json())
            }
        }));
    }

    if !missing_required.is_empty() {
        return Err(FraiseQLError::Validation {
            message: format!(
                "Mutation '{mutation_name}' is missing required argument(s): {}",
                missing_required.join(", ")
            ),
            path:    None,
        });
    }

    // 3a. Append server-injected parameters (after client args, in injection order).
    //
    // CONTRACT: inject params are always the *last* positional parameters of the SQL
    // function, in the order they appear in `inject_params` (insertion-ordered IndexMap).
    // The SQL function signature in the database MUST declare injected parameters after
    // all client-supplied parameters. Violating this order silently passes inject values
    // to the wrong SQL parameters. The CLI compiler (`fraiseql-cli compile`) validates
    // inject key names and source syntax when producing `schema.compiled.json`, but
    // cannot verify SQL function arity — that remains a developer responsibility.
    if !mutation_def.inject_params.is_empty() {
        let sec_ctx = security_ctx.ok_or_else(|| FraiseQLError::Validation {
            message: format!(
                "Mutation '{}' requires inject params but no security context is available \
                 (unauthenticated request)",
                mutation_name
            ),
            path:    None,
        })?;
        for (param_name, source) in &mutation_def.inject_params {
            args.push(resolve_inject_value(param_name, source, sec_ctx)?);
        }
    }

    // 4. Execute the mutation's stored function.
    //
    //    There used to be a second arm here, selected by `adapter.mutation_strategy()`:
    //    `DirectSql`, which built INSERT/DELETE straight from the mutation contract for
    //    adapters without stored functions (SQLite). No adapter has been able to return
    //    that strategy since the non-PostgreSQL backends were removed (#374) — the trait
    //    default is `FunctionCall` and the caching adapter forwards — so the arm was
    //    unreachable, and no test ever covered it. Unreachable code inside the write
    //    chokepoint is worse than unreachable code elsewhere: it is where a reader goes to
    //    learn which gates a write faces.
    // Resolve everything the response shape depends on BEFORE the write. None of it
    // reads the database — it is `mutation_def` plus the request's own variables — and
    // computing it here means a `@skip`/`@include` expression the evaluator refuses is
    // refused before the function runs rather than after it committed, matching every
    // other gate in this function. It also lets the projection run inside the write's
    // transaction below, which is what lets the field authorizer refuse the write
    // itself (#1353) instead of only its result.
    // Clone name and return_type to avoid borrow issues after schema lookups
    let mutation_return_type = mutation_def.return_type.clone();
    let response_key_owned = response_key.to_string();
    // Whether this mutation exposes the typed cascade payload surface.
    let is_cascade = mutation_def.cascade;

    // Evaluate @skip / @include against the request variables before projecting, so
    // conditional fields are honoured exactly as on the query path. (Named fragment
    // spreads were already resolved at classification time, where the document's
    // fragment definitions are available.)
    let variables_map: std::collections::HashMap<String, serde_json::Value> = match variables {
        Some(serde_json::Value::Object(map)) => {
            map.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
        },
        _ => std::collections::HashMap::new(),
    };
    let filtered_selections = DirectiveEvaluator::filter_selections(selections, &variables_map)
        .map_err(|e| FraiseQLError::Validation {
            message: e.to_string(),
            path:    Some("directives".to_string()),
        })?;
    let selections: &[FieldSelection] = &filtered_selections;

    // 4a. The payload selection meets the read gates a query of the same entities meets —
    //      `requires_scope` at every level, a nested level's read role, actor type and #422
    //      authorizer, and its row security over the returned document (`payload_gates`).
    //      Everything that can refuse is decided here, before the write, so a refused
    //      selection never runs the function; masking and the row filter follow it. On the
    //      post-directive selection, as the query path classifies.
    let payload_gates = PayloadGates::classify(
        ctx,
        security_ctx,
        variables,
        &mutation_def.return_type,
        mutation_def.cascade,
        selections,
    )?;

    let (envelope, result_json) = {
        // 3b. Resolve session variables once and pass them to the adapter call so
        //     they are applied on the same connection / transaction as the function
        //     (fixes #329 — set_config(..., true) is transaction-local, so applying
        //     it on a separate pooled connection left it invisible to the function).
        //
        // Only resolved when there are variables to inject or inject_started_at is
        // enabled, and only on the authenticated path (security context present).
        // The no-op default on non-PostgreSQL adapters means an empty slice here is
        // effectively free there.
        let resolved_session_vars = {
            let sv = &ctx.schema.session_variables;
            match security_ctx {
                Some(sec_ctx) if !sv.variables.is_empty() || sv.inject_started_at => {
                    crate::runtime::executor::security::resolve_session_variables(sv, sec_ctx)?
                },
                _ => Vec::new(),
            }
        };
        let session_pairs: Vec<(&str, &str)> =
            resolved_session_vars.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();

        // 4. Call the database function (session variables pinned to its connection) AND write the
        //    change-log outbox row in the same transaction — the Change Spine transactional outbox.
        //    The framework owns this write by default; apps drop their hand-rolled
        //    per-mutation-function inserts on upgrade (a documented breaking change). The adapter
        //    reads the changed-entity columns (object_id / object_data / updated_fields / cascade)
        //    from the function's own mutation_response row; only the DML verb and a NOT-NULL
        //    object_type fallback (the GraphQL return type) are threaded down here. Non-PostgreSQL
        //    adapters ignore the change-log descriptor (multi-DB parity lands in phase-03).
        //
        //    Opt-out (default-on): a row is written only when the global switch
        //    (`RuntimeConfig.changelog_enabled`) is on AND this mutation is not
        //    individually opted out (`MutationDefinition.changelog`). Passing `None`
        //    makes the adapter behave exactly like the session-affine path.
        let modification_type = mutation_def.operation.kind_str().to_uppercase();
        let write_changelog = ctx.config.changelog_enabled && mutation_def.changelog;
        // Envelope stamp (phase-03): stamp the tenant partition id EXPLICITLY from the
        // SecurityContext — never reconstructed from connection / RLS state, because
        // out-of-session spine consumers (poller, NATS bridge) bypass RLS and must
        // re-authz fan-out from the row itself. `tenant_id` is the Trinity
        // public-facing UUID; a request with no tenant, or a tenant identifier that
        // is not UUID-shaped, leaves it NULL (we never abort a user's mutation over a
        // log-row stamp). `trace_id` is the originating request's W3C trace id (#375),
        // stamped onto the SecurityContext from the inbound `traceparent` header — NULL
        // for a request with no trace context. `schema_version` is the compiled
        // schema's content hash (#377), a per-deployment constant precomputed once on
        // the ExecutorContext — NOT request-scoped — so a row records which deployment
        // produced it (the #378 replay-correctness handle). `trace_context` is the
        // request's full W3C trace context (#375), serialized JSON from the
        // SecurityContext — NULL for a request with no trace context. `actor_type`
        // is the request's actor classification and `acting_for` the delegated human
        // an agent acts for (#390), both derived onto the SecurityContext at auth time
        // — NULL for an unauthenticated mutation (no SecurityContext to stamp).
        let tenant_uuid = security_ctx
            .and_then(|c| c.tenant_id.as_ref())
            .and_then(|t| uuid::Uuid::parse_str(t.as_str()).ok());
        let trace_id = security_ctx.and_then(SecurityContext::trace_id);
        // Serialize the trace context once, into a binding that outlives the write call
        // (ChangeLogWrite borrows it as JSON text).
        let trace_context_json = security_ctx
            .and_then(SecurityContext::trace_context)
            .map(serde_json::Value::to_string);
        let actor_type = security_ctx.map(|c| c.actor_type().as_str());
        let acting_for = security_ctx.and_then(SecurityContext::acting_for);
        // `transport` is the ingress door the request declared (#376) — set by a
        // transport that stamps itself onto the SecurityContext (today: MCP);
        // NULL for the HTTP GraphQL path and unauthenticated mutations.
        let transport = security_ctx.and_then(SecurityContext::transport);
        // The change-log records the type the write produced: its stamp (the CTE prefers it),
        // else the one type an unstamped success can be — never a union's or a cascade
        // payload's name (ruling AA 1). An off-contract or ambiguous stamp rolls the whole
        // write back, change-log row included.
        let changelog_object_type =
            unstamped_type(&ctx.schema, &mutation_def.return_type, mutation_def.cascade)
                .unwrap_or_else(|| mutation_def.return_type.clone());
        let changelog = write_changelog.then(|| {
            ChangeLogWrite::new(&changelog_object_type, &modification_type)
                .with_tenant_id(tenant_uuid)
                .with_trace_id(trace_id)
                .with_schema_version(Some(&ctx.schema_version))
                .with_trace_context(trace_context_json.as_deref())
                .with_actor_type(actor_type)
                .with_acting_for(acting_for)
                .with_transport(transport)
                // Opt-in pre-image: when this mutation sets `changelog_pre_image`, the
                // outbox CTE also records the entity's before-state (from the
                // function's `entity_before`) into `object_data_before`. Off by
                // default → no extra column, byte-for-byte today's behavior.
                .with_pre_image(mutation_def.changelog_pre_image)
        });
        // 5/6. Row → outcome → response, as one step, because on the gated path
        //      below all of it has to happen before the transaction commits.
        let adjudicate = |rows: &[std::collections::HashMap<String, serde_json::Value>]| {
            let row = rows.first().ok_or_else(|| FraiseQLError::Validation {
                message: format!("Mutation '{mutation_name}': function returned no rows"),
                path:    None,
            })?;
            let outcome = parse_mutation_row(row)?;
            // The caller needs the envelope; `build_mutation_result` consumes the
            // outcome to build the projection.
            let envelope = outcome.clone();
            let result_json = build_mutation_result(
                ctx,
                security_ctx,
                &payload_gates,
                outcome,
                &mutation_return_type,
                is_cascade,
                selections,
                &authz_variables,
            )
            .inspect_err(|e| {
                // A contract error is a bug in the mutation function: say which one, where
                // operators read (ruling AJ 3); it is also counted.
                if payload_gates::is_contract_error(e) {
                    tracing::warn!(
                        mutation = %mutation_name,
                        function = %sql_source,
                        error = %e,
                        "mutation contract error: the write was rolled back"
                    );
                }
            })?;
            Ok::<_, FraiseQLError>((envelope, result_json))
        };

        // Every write commits only once it has been adjudicated (ruling Z 1): the
        // response is built from the row inside the write's transaction, and its
        // verdict decides the commit.
        //
        // #1353 is why a refusal must be able to take the write with it: the field
        // authorizer's contract takes the resolved entity as `parent`, so it cannot be
        // asked before the row exists, and a caller it refused used to lose the field
        // and keep the side effect. Everything that *can* be asked earlier already ran
        // before the write (the operation `Authorizer`, `requires_role`,
        // `requires_actor`, `before:mutation`, and the payload's static gates above).
        //
        // But building the response can fail on any schema, gated or not — no rows, a
        // row that is not a `mutation_response`, a malformed cascade entry, the cascade
        // size ceiling. A write that committed and then failed told its caller it failed
        // while it had landed, and stored no idempotency record, so a retry ran it again.
        // So it is not only a refusal that decides: every write takes this path. With the
        // change-log outbox on (the default) the write is in this transaction anyway; the
        // gate only moves the adjudication before `COMMIT`.
        //
        // The write commits only if adjudication completed without error: `adjudicate`
        // returned `Ok`. That is not the same as every field being allowed — a `Mask`
        // decision nulls one field on one row and returns `Ok`, because it is a statement
        // about the value, not the operation. Any `Err` rolls back and is returned
        // unchanged, whatever its variant.
        let built = std::sync::OnceLock::new();
        let gate = |rows: &[std::collections::HashMap<String, serde_json::Value>]| {
            drop(built.set(adjudicate(rows)?));
            Ok(())
        };
        // A dry run (#501, `dry_run_mutations`) is the same write in
        // `WriteMode::DryRun` (ruling AA 6): the same transaction, session variables,
        // outbox row and gate, ending in a rollback — so it answers exactly as the write
        // would, and nothing persists.
        //
        // The adapter returns the gate's error verbatim once the rollback has
        // landed — so a refusal, a failed rollback and a failed commit all surface
        // here, and a failed rollback surfaces as itself rather than as the error
        // it was trying to honour.
        let mode = if ctx.config.dry_run_mutations {
            crate::backend::WriteMode::DryRun
        } else {
            crate::backend::WriteMode::Commit
        };
        let request = crate::backend::WriteRequest::new(sql_source, &args)
            .with_session_vars(&session_pairs)
            .with_changelog(changelog.as_ref())
            .with_mode(mode);
        writer.execute_write(&request, &gate).await?;
        built.into_inner().ok_or_else(|| FraiseQLError::Internal {
            message: format!(
                "Mutation '{mutation_name}': the commit gate never ran, so the write \
                     committed unadjudicated"
            ),
            source:  None,
        })?
    };

    // What follows a commit follows only a commit (ruling AD 3): a dry run rolled back,
    // so no fact table moved, no cached read is stale, and nothing was executed to audit.
    let committed = !ctx.config.dry_run_mutations;

    // 6a. Bump fact table versions after a successful mutation.
    //
    // This invalidates cached aggregation results for any fact tables listed
    // in `MutationDefinition.invalidates_fact_tables`.  We bump versions on
    // Success only — an Error outcome means no data was written, so caches
    // remain valid.  Non-cached adapters return Ok(()) from the default trait
    // implementation (no-op); only `CachedDatabaseAdapter` performs actual work.
    if committed
        && matches!(envelope, MutationOutcome::Success { .. })
        && !mutation_def.invalidates_fact_tables.is_empty()
    {
        // Through the write handle, like the dispatch above: this bumps a version row
        // in the database, so it is write-shaped even though it is cache bookkeeping.
        // It is unreachable without the handle anyway — it runs only after a mutation
        // succeeded — but taking it from the same place keeps "the handle is the
        // permission" true of every write in this function rather than nearly every one.
        writer.bump_fact_table_versions(&mutation_def.invalidates_fact_tables).await?;
    }

    // Invalidate query result cache for views/entities touched by this mutation.
    //
    // The decision is one pure function over (mutation_def, outcome, schema) —
    // see `invalidation::plan_invalidation` for why it is a view sweep rather
    // than a list-vs-point classification (#741, #742, #763). Declared views,
    // the return type, the stamped entity type and cascade side-effects all
    // resolve in the same place, so no caller can reach a different answer.
    let plan = if committed {
        invalidation::plan_invalidation(mutation_def, &envelope, &ctx.schema)
    } else {
        invalidation::InvalidationPlan::default()
    };
    if !plan.views.is_empty() {
        ctx.adapter.invalidate_views(&plan.views).await?;
        if let Some(ref rc) = ctx.response_cache {
            let _ = rc.invalidate_views(&plan.views);
        }
    }
    // Supplementary, not a substitute: covers entries indexed under this entity
    // when the entity's own view could not be resolved from the schema.
    if let Some((etype, eid)) = &plan.entity {
        ctx.adapter.invalidate_by_entity(etype, eid).await?;
    }

    // 7. Emit structured mutation audit event when audit_mutations is enabled.
    //
    // This is the single chokepoint for all mutation paths (GraphQL handler,
    // REST handler, typed execute_mutation, bulk filter). Zero-cost when disabled:
    // the branch is not taken and no string formatting or allocation occurs.
    if committed && ctx.config.audit_mutations {
        tracing::info!(
            target: "fraiseql::mutation_audit",
            mutation_name = mutation_name,
            entity_type = %mutation_def.return_type,
            operation = %mutation_def.operation.kind_str(),
            tenant_id = %security_ctx
                .and_then(|c| c.tenant_id.as_ref().map(|t| t.as_str()))
                .unwrap_or(""),
            "mutation.executed"
        );
    }

    let response = ResultProjector::wrap_in_data_envelope(result_json, &response_key_owned);
    Ok(MutationExecution {
        data:    response,
        outcome: envelope,
    })
}

#[cfg(test)]
mod tests;