fraiseql-core 2.17.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 derived_cascade;
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_mutation_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.
///
/// An `invalidations` hint names a query whose cached reads are stale (#1575): it resolves
/// to the views that query's entries are cached under, its `sql_source` and its
/// `additional_views`. Unknown query names 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();
    let mut push = |src: &str| {
        if !src.is_empty() && seen.insert(src.to_string()) {
            views.push(ViewName::from(src));
        }
    };
    let entries = |arm: &str| {
        cascade
            .get(arm)
            .and_then(serde_json::Value::as_array)
            .map_or(&[][..], Vec::as_slice)
    };
    for arm in ["updated", "deleted"] {
        for entry in entries(arm) {
            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) {
                push(td.sql_source.as_str());
            }
        }
    }
    for hint in entries("invalidations") {
        let Some(name) = hint.get("queryName").and_then(serde_json::Value::as_str) else {
            continue;
        };
        if let Some(query) = schema.queries.iter().find(|q| q.name == name) {
            for view in query.sql_source.iter().chain(&query.additional_views) {
                push(view);
            }
        }
    }
    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 schema.success_types(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],
    result: Option<&serde_json::Value>,
    selections: &[FieldSelection],
    variables: &std::collections::HashMap<String, serde_json::Value>,
) -> Result<serde_json::Value> {
    let declared = ctx.schema.find_type(payload_type);
    // A success field the function returned wrong refuses the write whatever the client
    // selected (#1397): what commits must not depend on the selection set.
    for field in declared.iter().flat_map(|t| &t.fields) {
        let name = field.name.as_str();
        if PAYLOAD_OWN_FIELDS.contains(&name) {
            continue;
        }
        // A row with no `result` column cannot carry the field at all: refused as the
        // function's contract error, never served as a run of nulls.
        if result.is_none() {
            return Err(payload_gates::contract_error_in(
                "result",
                format!(
                    "the mutation declares success field `{name}`, but its function's response \
                     row has no `result` column; the write was rolled back"
                ),
            ));
        }
        success_field_value_ok(&ctx.schema, field, &success_value(result, name)).map_err(
            |held| {
                payload_gates::contract_error_in(
                    "result",
                    format!(
                        "the mutation function returned {held} for success field `{name}` ({}); \
                         the write was rolled back",
                        field.field_type
                    ),
                )
            },
        )?;
    }
    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(),
                    ),
                );
            },
            // A declared success field (#1397), checked above, read from `result` under its
            // stored name; a key `result` carries that no field declares is never read.
            name => {
                if declared.is_some_and(|t| t.fields.iter().any(|f| f.name.as_str() == name)) {
                    out.insert(sel.response_key().to_string(), success_value(result, name));
                }
            },
        }
    }
    Ok(serde_json::Value::Object(out))
}

/// The fields a cascade payload has of its own (`cli::converter::cascade_types`); every
/// other field of it is a declared success field.
const PAYLOAD_OWN_FIELDS: [&str; 3] = ["entity", "cascade", "updatedFields"];

/// The value `result` holds for success field `name`, under its stored (`snake_case`) key;
/// `null` when it holds none.
fn success_value(result: Option<&serde_json::Value>, name: &str) -> serde_json::Value {
    result
        .and_then(|r| r.get(crate::utils::casing::to_snake_case(name)))
        .cloned()
        .unwrap_or(serde_json::Value::Null)
}

/// Whether `value` is one `field` can hold, for a success field (#1397): `Err` names what
/// it holds instead. A non-null field the function returned no value for is as much the
/// function's error as a value of the wrong type: refused alike, on every transport, before
/// the write commits (REST has no partial answer to carry a field error in).
///
/// Each type is held to the JSON shape it is served as. A rich scalar is a leaf (stored as
/// text); `Json` and a custom scalar are served as the function returned them. A list or a
/// composite type is not a success field type (the compiler refuses it), so it holds nothing.
/// The refusal names the kind of value, never the value: it reaches the client and the logs.
fn success_field_value_ok(
    schema: &CompiledSchema,
    field: &crate::schema::FieldDefinition,
    value: &serde_json::Value,
) -> std::result::Result<(), String> {
    use serde_json::Value;

    use crate::schema::FieldType;
    let ok = match (&field.field_type, value) {
        (_, Value::Null) => field.nullable,
        (FieldType::Int, Value::Number(n)) => n.as_i64().is_some_and(|i| i32::try_from(i).is_ok()),
        (FieldType::Float | FieldType::Decimal, Value::Number(_))
        | (
            FieldType::String
            | FieldType::Id
            | FieldType::DateTime
            | FieldType::Date
            | FieldType::Time
            | FieldType::Uuid
            | FieldType::Decimal
            | FieldType::BitVector
            | FieldType::SparseVector,
            Value::String(_),
        )
        | (FieldType::Boolean, Value::Bool(_))
        | (FieldType::Json, _) => true,
        (FieldType::Id, Value::Number(n)) => n.is_i64() || n.is_u64(),
        (FieldType::Vector | FieldType::HalfVector, Value::Array(items)) => {
            items.iter().all(Value::is_number)
        },
        (FieldType::Enum(name), Value::String(s)) => {
            schema.find_enum(name).is_some_and(|e| e.values.iter().any(|v| v.name == *s))
        },
        (FieldType::Scalar(name), _) => {
            !crate::schema::RICH_SCALARS.contains(&name.as_str())
                || !matches!(value, Value::Array(_) | Value::Object(_))
        },
        (
            FieldType::Int
            | FieldType::Float
            | FieldType::String
            | FieldType::Boolean
            | FieldType::Id
            | FieldType::DateTime
            | FieldType::Date
            | FieldType::Time
            | FieldType::Uuid
            | FieldType::Decimal
            | FieldType::Vector
            | FieldType::BitVector
            | FieldType::HalfVector
            | FieldType::SparseVector
            | FieldType::Enum(_)
            | FieldType::List { item: _, .. }
            | FieldType::Object(_)
            | FieldType::Input(_)
            | FieldType::Interface(_)
            | FieldType::Union(_),
            _,
        ) => false,
    };
    if ok {
        return Ok(());
    }
    Err(match value {
        Value::String(_) => "a string",
        Value::Number(_) => "a number",
        Value::Bool(_) => "a boolean",
        Value::Array(_) => "a list",
        Value::Object(_) => "an object",
        Value::Null => "no value",
    }
    .to_string())
}

/// 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,
            result,
            ..
        } 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,
                result.as_ref(),
                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. A
            // selected field the source has no value for completes to `null` (#1618).
            // `status` is a source key too, so a selected one takes its place in the
            // selection; an unselected one is injected 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()));
            source_map.insert("status".to_string(), 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 when the selection did not place it: the
            // generated clients declare it on every error type, whether or not the type
            // does, to discriminate error outcomes.
            if let serde_json::Value::Object(ref mut map) = result {
                map.entry("status".to_string())
                    .or_insert_with(|| serde_json::Value::String(status.to_string()));
            }

            Ok(result)
        },
    }
}

/// The typed failure an integrity-constraint violation (SQLSTATE class 23) raised by
/// a mutation's function becomes (#1424), or `None` to keep it a request error.
///
/// Only a mutation whose result union or interface has exactly one error member has a
/// typed failure to give: that member answers, as it does for an unstamped failure
/// the function returns. The message is the same generic text error sanitization uses,
/// whatever the configuration: the database's text never reaches the response.
///
/// What it says about the constraint is `entry` (#1531, reversing #1424's "the
/// constraint's name is not exposed"): one `errors[]` entry, the shape a function's
/// own `mutation_err_entries` gives, built by [`constraint_entry`] under the server's
/// `mutation_constraint_metadata`. Its `code` is the status decided here.
fn constraint_violation_outcome(
    error: &FraiseQLError,
    schema: &crate::schema::CompiledSchema,
    return_type: &str,
    contract: &payload_gates::StampContract,
    entry: Option<serde_json::Map<String, serde_json::Value>>,
) -> Option<MutationOutcome> {
    let FraiseQLError::Database {
        sql_state: Some(sql_state),
        ..
    } = error
    else {
        return None;
    };
    let returns_members =
        schema.find_union(return_type).is_some() || schema.find_interface(return_type).is_some();
    if !returns_members || contract.error.len() != 1 {
        return None;
    }
    let (error_class, http_status, message) = match sql_state.as_str() {
        // not_null_violation, check_violation: the input was invalid.
        "23502" | "23514" => (
            crate::runtime::cascade::MutationErrorClass::Validation,
            422,
            "The request contains an invalid value",
        ),
        // unique, exclusion and foreign-key violations, and the rest of the class.
        state if state.starts_with("23") => (
            crate::runtime::cascade::MutationErrorClass::Conflict,
            409,
            "The request conflicts with the current state of the data",
        ),
        _ => return None,
    };
    let metadata = entry.map_or(serde_json::Value::Null, |mut entry| {
        entry.insert("code".to_string(), serde_json::json!(http_status));
        entry.insert("message".to_string(), serde_json::json!(message));
        serde_json::json!({ "errors": [entry] })
    });
    Some(MutationOutcome::Error {
        error_class,
        message: message.to_string(),
        http_status: Some(http_status),
        entity_type: None,
        metadata,
    })
}

/// The `errors[]` entry a constraint violation's typed error carries (#1531), without its
/// `code` and `message` (the classification's): `identifier` is the constraint's name and
/// `details` its SQLSTATE, plus its table and columns under `full`. `None` under `none`,
/// for an error that is not a class-23 violation, and when nothing names the constraint.
///
/// Schema identifiers only: the database's `DETAIL`, which carries the row's values, is
/// never read. What the error does not say (a not-null constraint's name, the columns) is
/// asked of the catalog, on the failure path only; a failed lookup is logged and the entry
/// keeps what the error itself gave.
async fn constraint_entry(
    ctx: &ExecutorContext,
    error: &FraiseQLError,
) -> Option<serde_json::Map<String, serde_json::Value>> {
    use crate::runtime::ConstraintMetadata;
    let full = match ctx.config.constraint_metadata {
        ConstraintMetadata::Identifier => false,
        ConstraintMetadata::Full => true,
        // `ConstraintMetadata` is `#[non_exhaustive]`: an unknown setting says nothing.
        _ => return None,
    };
    let FraiseQLError::Database {
        sql_state: Some(sql_state),
        constraint: Some(violation),
        ..
    } = error
    else {
        return None;
    };
    if !sql_state.starts_with("23") {
        return None;
    }
    let described = if full || violation.name.is_none() {
        match ctx.adapter.describe_constraint(violation).await {
            Ok(described) => described,
            Err(lookup) => {
                tracing::warn!(error = %lookup, "could not describe a violated constraint");
                crate::backend::ConstraintDescription {
                    name:    violation.name.clone(),
                    columns: None,
                }
            },
        }
    } else {
        crate::backend::ConstraintDescription {
            name:    violation.name.clone(),
            columns: None,
        }
    };
    let identifier = described.name?;
    let mut details = serde_json::Map::new();
    details.insert("sqlstate".to_string(), serde_json::json!(sql_state));
    if full {
        if let Some(table) = &violation.table {
            details.insert("table".to_string(), serde_json::json!(table));
        }
        if let Some(columns) = described.columns {
            details.insert("columns".to_string(), serde_json::json!(columns));
        }
    }
    let mut entry = serde_json::Map::new();
    entry.insert("identifier".to_string(), serde_json::json!(identifier));
    entry.insert("details".to_string(), serde_json::Value::Object(details));
    Some(entry)
}

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> {
    // 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)?;
    let prepared = prepare_write(
        ctx,
        mutation_name,
        response_key,
        variables,
        security_ctx,
        selections,
        inline_arguments,
    )
    .await?;
    let parts = prepared.parts();
    let request = parts.request(&prepared);

    // 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, all in
    // `prepare_write`).
    //
    // 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(prepared.adjudicate(rows)?));
        Ok(())
    };
    // 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 adjudicated = match writer.execute_write(&request, &gate).await {
        Ok(_) => 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,
        })?,
        Err(error) => prepared.answer_violation(error).await?,
    };

    after_commit(ctx, writer, prepared.mutation_def, mutation_name, security_ctx, &[&adjudicated])
        .await?;
    Ok(prepared.execution(adjudicated))
}

/// Steps 1–4 of a write, everything asked before its transaction: the gates that do not
/// need the written row, the arguments bound, the selection settled, the session variables
/// and the change-log stamp resolved.
async fn prepare_write<'a>(
    ctx: &'a Arc<ExecutorContext>,
    mutation_name: &'a str,
    response_key: &str,
    variables: Option<&serde_json::Value>,
    security_ctx: Option<&'a SecurityContext>,
    selections: WriteSelections<'_>,
    inline_arguments: &[crate::graphql::GraphQLArgument],
) -> Result<PreparedWrite<'a>> {
    let selections = selections.as_slice();

    // #633: a tenant over its storage quota writes nothing, whatever the transport. Asked
    // before anything of the write runs, so a refused write executes no hook and no
    // statement.
    if let Some(gate) = ctx.config.write_gate.as_ref() {
        gate.admit(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);

    // #1513: a localized value is coerced to the locale map its function receives, and
    // refused if it is not one, before any statement. Every transport's arguments meet here.
    let localized = crate::runtime::coerce_localized_arguments(
        &ctx.schema,
        &format!("Mutation.{}", ctx.schema.display_name(mutation_name)),
        &mutation_def.arguments,
        variables,
    )?;
    let variables = localized.as_ref().or(variables);

    // #1362, #1528: every value written at an argument — or at a field of the input
    // object under one, at any depth — must have its declared type: an enum must name a
    // member, a built-in scalar must hold one, and an input object may carry only the
    // fields it 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. After the localized coercion above (a localized
    // value is a locale map by now, and is skipped), and 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_mutation_argument_values(
        &ctx.schema,
        &format!("Mutation.{}", ctx.schema.display_name(mutation_name)),
        &mutation_def.arguments,
        variables,
    )?;
    // #1580: and every enum member named there becomes the value it is stored as, the value
    // the function receives. After the membership check, so every name here is a member.
    let enum_coerced = crate::runtime::coerce_enum_argument_object(
        &ctx.schema,
        &mutation_def.arguments,
        variables,
    );
    let variables = enum_coerced.as_ref().or(variables);
    let vars_obj = variables.and_then(|v| v.as_object());

    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_mutation_inject_value(
                param_name,
                source,
                sec_ctx,
                ctx.schema.tenant_claim(),
            )?);
        }
    }

    // 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.

    // 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,
    )?;

    // 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. An anonymous write gets its `literal` and `header` sources and the
    // timestamp (#1520); claim and enrichment sources need a principal.
    let session_vars = {
        let sv = &ctx.schema.session_variables;
        if sv.variables.is_empty() && !sv.inject_started_at {
            Vec::new()
        } else {
            crate::runtime::executor::support::security::resolve_session_variables(
                sv,
                security_ctx,
                ctx.schema.tenant_claim(),
                crate::runtime::executor::support::security::SessionPurpose::Write,
            )?
        }
    };

    // 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.
    //
    //    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 write_changelog = ctx.config.changelog_enabled && mutation_def.changelog;
    // 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);
    // 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());

    // #1391: a mutation whose cascade also comes from pg_tviews asks the adapter for the
    // rows its transaction changed, read on its connection, before the gate. A type whose
    // reads the engine filters is not read (#1573): the merge lists none of its rows.
    let derived_views = (mutation_def.cascade_source == crate::schema::CascadeSource::PgTviews)
        .then(|| {
            derived_cascade::views(&ctx.schema)
                .into_iter()
                .filter(|(name, _)| {
                    !derived_cascade::engine_filtered(
                        &ctx.schema,
                        ctx.config.rls_policy.is_some(),
                        name,
                    )
                })
                .collect()
        });

    Ok(PreparedWrite {
        ctx,
        mutation_def,
        mutation_name,
        security_ctx,
        sql_source: sql_source.to_string(),
        args,
        response_key: response_key.to_string(),
        selections: filtered_selections,
        authz_variables,
        payload_gates,
        session_vars,
        modification_type: mutation_def.operation.kind_str().to_uppercase(),
        changelog_object_type: write_changelog.then_some(changelog_object_type),
        trace_context_json,
        derived_views,
    })
}

/// A mutation call, decided: every gate that can be asked before the write has been asked,
/// and what the write sends and how its rows are judged are fixed.
struct PreparedWrite<'a> {
    ctx:                   &'a Arc<ExecutorContext>,
    mutation_def:          &'a crate::schema::MutationDefinition,
    mutation_name:         &'a str,
    security_ctx:          Option<&'a SecurityContext>,
    sql_source:            String,
    args:                  Vec<serde_json::Value>,
    response_key:          String,
    /// The payload selection, after `@skip`/`@include`.
    selections:            Vec<FieldSelection>,
    authz_variables:       std::collections::HashMap<String, serde_json::Value>,
    payload_gates:         PayloadGates,
    session_vars:          Vec<(String, String)>,
    modification_type:     String,
    /// The type the change-log row records, when this write writes one.
    changelog_object_type: Option<String>,
    trace_context_json:    Option<String>,
    /// The types a derived cascade reads, with their views (#1391), when it asks for one.
    derived_views:         Option<Vec<(&'a str, &'a str)>>,
}

/// What a write's rows were judged to be: the parsed outcome, the response built from it, and
/// the types a derived cascade's report named (#1575).
/// A call's outcome, its response, every type `pg_tviews` reported it changed (#1575), and the
/// rows it changed when a report says so (#1576; `None`: unknown).
type Adjudicated = (
    MutationOutcome,
    serde_json::Value,
    std::collections::BTreeSet<String>,
    Option<crate::backend::ChangedRows>,
);

/// The borrowed pieces a [`crate::backend::WriteRequest`] points into.
struct WriteParts<'p> {
    session_pairs:  Vec<(&'p str, &'p str)>,
    changelog:      Option<ChangeLogWrite<'p>>,
    derived:        Option<crate::backend::DerivedCascade<'p>>,
    /// Report the rows the call changes, at most this many (#1576).
    report_changes: Option<i32>,
    mode:           crate::backend::WriteMode,
}

impl WriteParts<'_> {
    fn request<'r>(&'r self, prepared: &'r PreparedWrite<'_>) -> crate::backend::WriteRequest<'r> {
        crate::backend::WriteRequest::new(&prepared.sql_source, &prepared.args)
            .with_session_vars(&self.session_pairs)
            .with_changelog(self.changelog.as_ref())
            .with_mode(self.mode)
            .with_derived_cascade(self.derived.as_ref())
            .with_changes_reported(self.report_changes)
    }
}

impl PreparedWrite<'_> {
    fn parts(&self) -> WriteParts<'_> {
        let ctx = self.ctx;
        let security_ctx = self.security_ctx;
        // Envelope stamp: 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).
        // `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 changelog = self.changelog_object_type.as_deref().map(|object_type| {
            ChangeLogWrite::new(object_type, &self.modification_type)
                .with_tenant_id(
                    security_ctx
                        .and_then(|c| c.tenant_id.as_ref())
                        .and_then(|t| uuid::Uuid::parse_str(t.as_str()).ok()),
                )
                .with_trace_id(security_ctx.and_then(SecurityContext::trace_id))
                .with_schema_version(Some(&ctx.schema_version))
                .with_trace_context(self.trace_context_json.as_deref())
                .with_actor_type(security_ctx.map(|c| c.actor_type().as_str()))
                .with_acting_for(security_ctx.and_then(SecurityContext::acting_for))
                .with_transport(security_ctx.and_then(SecurityContext::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(self.mutation_def.changelog_pre_image)
        });
        // 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.
        let mode = if ctx.config.dry_run_mutations {
            crate::backend::WriteMode::DryRun
        } else {
            crate::backend::WriteMode::Commit
        };
        WriteParts {
            session_pairs: self
                .session_vars
                .iter()
                .map(|(k, v)| (k.as_str(), v.as_str()))
                .collect(),
            changelog,
            derived: self
                .derived_views
                .as_deref()
                .map(|views| crate::backend::DerivedCascade::new(max_entities(ctx), views)),
            // The cache evicts its reads by id by the rows the write changed (#1576).
            report_changes: ctx.adapter.reports_changed_rows().then(|| max_entities(ctx)),
            mode,
        }
    }

    /// 5/6. Row → outcome → response, as one step, because all of it has to happen before
    /// the transaction commits (see [`execute_mutation_impl`]).
    fn adjudicate(
        &self,
        rows: &[std::collections::HashMap<String, serde_json::Value>],
    ) -> Result<Adjudicated> {
        let ctx = self.ctx;
        let (mutation_name, sql_source) = (self.mutation_name, self.sql_source.as_str());
        let row = rows.first().ok_or_else(|| FraiseQLError::Validation {
            message: format!("Mutation '{mutation_name}': function returned no rows"),
            path:    None,
        })?;
        // #1575: what the report says changed, before the merge drops what it may not
        // serve; every such type's cached reads are stale.
        let mut reported = derived_cascade::reported_types(row);
        // #1576: the rows it changed, as either report says; their types' reads are stale too.
        let changed = derived_cascade::changed_rows(row);
        if let Some(changed) = &changed {
            reported.extend(changed.type_names().map(str::to_string));
        }
        // #1391: pg_tviews' report, merged into the cascade it is served from.
        let row = derived_cascade::merge(&ctx.schema, row, |type_name| {
            derived_cascade::engine_filtered(
                &ctx.schema,
                ctx.config.rls_policy.is_some(),
                type_name,
            )
        });
        let outcome = parse_mutation_row(&row)?;
        crate::runtime::mutation_result::check_error_shape(
            ctx.config.mutation_error_shape_check,
            &outcome,
            mutation_name,
            sql_source,
        );
        // The caller needs the envelope; `build_mutation_result` consumes the
        // outcome to build the projection.
        let envelope = outcome.clone();
        let result_json = self.build(outcome).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((envelope, result_json, reported, changed))
    }

    fn build(&self, outcome: MutationOutcome) -> Result<serde_json::Value> {
        build_mutation_result(
            self.ctx,
            self.security_ctx,
            &self.payload_gates,
            outcome,
            &self.mutation_def.return_type,
            self.mutation_def.cascade,
            &self.selections,
            &self.authz_variables,
        )
    }

    /// #1424: an integrity constraint the function violated is the data's answer to this
    /// write, and the mutation's result union has a member for it. The transaction rolled
    /// back with the error; the client gets the typed failure the function would have
    /// returned had it pre-checked the constraint. Any other error is returned.
    async fn answer_violation(&self, error: FraiseQLError) -> Result<Adjudicated> {
        let ctx = self.ctx;
        let entry = constraint_entry(ctx, &error).await;
        let Some(outcome) = constraint_violation_outcome(
            &error,
            &ctx.schema,
            &self.mutation_def.return_type,
            self.payload_gates.contract(),
            entry,
        ) else {
            return Err(error);
        };
        tracing::info!(
            mutation = %self.mutation_name,
            function = %self.sql_source,
            error = %error,
            "constraint violation answered as the mutation's typed error"
        );
        let envelope = outcome.clone();
        let result_json = self.build(outcome)?;
        // Rolled back: nothing changed.
        let unchanged = Some(crate::backend::ChangedRows::default());
        Ok((envelope, result_json, std::collections::BTreeSet::new(), unchanged))
    }

    fn execution(&self, (outcome, result_json, ..): Adjudicated) -> MutationExecution {
        MutationExecution {
            data: ResultProjector::wrap_in_data_envelope(result_json, &self.response_key),
            outcome,
        }
    }
}

/// The error a batch returns for an item its function refused (#1574): the batch commits all
/// of its items or none, so a refused item refuses it. The variant follows the function's
/// error class, so the transport's status does too.
fn refused_item(
    index: usize,
    count: usize,
    mutation_name: &str,
    outcome: &MutationOutcome,
) -> Option<FraiseQLError> {
    use crate::runtime::cascade::MutationErrorClass as Class;
    let MutationOutcome::Error {
        error_class,
        message,
        ..
    } = outcome
    else {
        return None;
    };
    let message = format!(
        "Mutation '{mutation_name}' refused it ({}): {message}; nothing was written",
        error_class.as_str()
    );
    let error = match error_class {
        Class::Validation => FraiseQLError::Validation {
            message,
            path: None,
        },
        Class::NotFound => FraiseQLError::NotFound {
            resource_type: format!("Bulk item {} of {count}", index + 1),
            identifier:    message,
        },
        Class::Unauthorized => FraiseQLError::Authentication { message },
        Class::Forbidden => FraiseQLError::Authorization {
            message,
            action: Some("mutation".to_string()),
            resource: Some(mutation_name.to_string()),
        },
        Class::Internal => FraiseQLError::Internal {
            message,
            source: None,
        },
        Class::RateLimited | Class::ServiceUnavailable => FraiseQLError::ServiceUnavailable {
            message,
            retry_after: None,
        },
        Class::Conflict | Class::TransactionFailed | Class::Timeout => {
            FraiseQLError::Conflict { message }
        },
    };
    Some(error.in_batch_item(index, count))
}

/// The most rows `pg_tviews` reports for one call: the cascade ceiling.
fn max_entities(ctx: &ExecutorContext) -> i32 {
    i32::try_from(ctx.config.cascade_limits.max_updated_entities).unwrap_or(i32::MAX)
}

/// What follows a commit follows only a commit (ruling AD 3), once for every call a
/// transaction committed together: a dry run rolled back, so no fact table moved, no cached
/// read is stale, and nothing was executed to audit.
async fn after_commit(
    ctx: &ExecutorContext,
    writer: &dyn crate::backend::Writer,
    mutation_def: &crate::schema::MutationDefinition,
    mutation_name: &str,
    security_ctx: Option<&SecurityContext>,
    written: &[&Adjudicated],
) -> Result<()> {
    if ctx.config.dry_run_mutations {
        return Ok(());
    }
    let succeeded = || {
        written
            .iter()
            .filter(|(envelope, ..)| matches!(envelope, MutationOutcome::Success { .. }))
    };

    // 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 succeeded().next().is_some() && !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.
    // A batch's calls committed together and are swept together (#1574).
    //
    // #1576: when every call reported the rows it changed, the cache evicts its reads of one
    // row by id by those rows instead of by the sweep. A call that wrote and failed counts
    // too: evicting more is never wrong.
    let mut plan = invalidation::InvalidationPlan::default();
    let mut entities = Vec::new();
    let mut changed = Some(crate::backend::ChangedRows::default());
    for (envelope, _, reported, call_changed) in written.iter().copied() {
        let call = invalidation::plan_invalidation(mutation_def, envelope, &ctx.schema)
            .with_reported_types(reported, &ctx.schema);
        entities.extend(plan.absorb(call));
        match (changed.as_mut(), call_changed) {
            (Some(all), Some(call_changed)) => all.absorb(call_changed.clone()),
            _ => changed = None,
        }
    }
    match changed.filter(|_| ctx.adapter.reports_changed_rows()) {
        Some(changed) => {
            ctx.adapter.invalidate_written(&plan.views, &changed).await?;
        },
        None if !plan.views.is_empty() => {
            ctx.adapter.invalidate_views(&plan.views).await?;
        },
        None => {},
    }
    // Supplementary, not a substitute: covers entries indexed under this entity
    // when the entity's own view could not be resolved from the schema.
    for (etype, eid) in &entities {
        ctx.adapter.invalidate_by_entity(etype, eid).await?;
    }

    for (envelope, ..) in written.iter().copied() {
        // 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 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"
            );
        }

        // 8. Tell the after-mutation observer (#1340, #1440). Here, rather than in a transport, so
        //    every transport fires the same `after:mutation` work: this is the one place that knows
        //    the write committed and what it produced. A failure is `MutationOutcome::Error` even
        //    when it is served as union data, and a dry run committed nothing, so neither reaches
        //    the observer. The row is the function's whole entity, not the client's selection.
        if let (
            Some(observer),
            MutationOutcome::Success {
                entity,
                entity_type,
                entity_id,
                ..
            },
        ) = (ctx.config.after_mutation_observer.as_deref(), envelope)
        {
            // The produced type: the stamp the gates validated, else the one type an
            // unstamped success can be. An ambiguous unstamped success was refused, and
            // rolled back, before the commit, so one of the two is always present here.
            let produced = entity_type.clone().or_else(|| {
                unstamped_type(&ctx.schema, &mutation_def.return_type, mutation_def.cascade)
            });
            if let Some(produced) = produced {
                observer.on_committed(&crate::runtime::CommittedMutation {
                    mutation_name,
                    entity_type: &produced,
                    operation: &mutation_def.operation,
                    entity,
                    entity_id: entity_id.as_deref(),
                    security_ctx,
                    dispatch_depth: crate::runtime::current_dispatch_depth(),
                });
            }
        }
    }
    Ok(())
}

/// Run `mutation_name` once per item of `items`, in **one** transaction (#1574): every item
/// commits, or none does.
///
/// Every item is prepared first — the operation authorizer, `requires_role`,
/// `requires_actor`, argument validation and `before:mutation` all run for every item before
/// any statement — so an item refused there writes nothing. The calls then run in one
/// transaction ([`Writer::execute_writes`](crate::backend::Writer::execute_writes)), each row
/// adjudicated before the commit; an item its function refuses (`succeeded = false`) refuses
/// the batch. The error names the item. One invalidation follows the commit, for every item.
///
/// # Errors
///
/// The first item's refusal, from any gate or the database, named by its position; nothing
/// is written.
pub(in super::super) async fn execute_mutation_batch_impl(
    ctx: &Arc<ExecutorContext>,
    mutation_name: &str,
    items: &[serde_json::Value],
    security_ctx: Option<&SecurityContext>,
    selections: WriteSelections<'_>,
) -> Result<Vec<MutationExecution>> {
    let writer = ctx.writer(mutation_name)?;
    let count = items.len();
    let mut prepared = Vec::with_capacity(count);
    for (index, item) in items.iter().enumerate() {
        let write = prepare_write(
            ctx,
            mutation_name,
            mutation_name,
            Some(item),
            security_ctx,
            selections,
            &[],
        )
        .await
        .map_err(|e| e.in_batch_item(index, count))?;
        prepared.push(write);
    }
    let parts: Vec<WriteParts<'_>> = prepared.iter().map(PreparedWrite::parts).collect();
    let requests: Vec<crate::backend::WriteRequest<'_>> =
        parts.iter().zip(&prepared).map(|(parts, write)| parts.request(write)).collect();

    // Every call commits only once it has been adjudicated, as a single write does; here a
    // failed outcome also refuses, because the batch cannot commit in part.
    let built: Vec<std::sync::OnceLock<Adjudicated>> =
        std::iter::repeat_with(std::sync::OnceLock::new).take(count).collect();
    let gate = |index: usize, rows: &[std::collections::HashMap<String, serde_json::Value>]| {
        let adjudicated =
            prepared[index].adjudicate(rows).map_err(|e| e.in_batch_item(index, count))?;
        if let Some(refused) = refused_item(index, count, mutation_name, &adjudicated.0) {
            return Err(refused);
        }
        drop(built[index].set(adjudicated));
        Ok(())
    };
    writer.execute_writes(&requests, &gate).await?;
    let built: Vec<Adjudicated> = built
        .into_iter()
        .map(std::sync::OnceLock::into_inner)
        .collect::<Option<_>>()
        .ok_or_else(|| FraiseQLError::Internal {
            message: format!(
                "Mutation '{mutation_name}': the batch committed with an item the commit gate \
                 never adjudicated"
            ),
            source:  None,
        })?;

    if let Some(first) = prepared.first() {
        let written: Vec<&Adjudicated> = built.iter().collect();
        after_commit(ctx, writer, first.mutation_def, mutation_name, security_ctx, &written)
            .await?;
    }
    Ok(prepared
        .iter()
        .zip(built)
        .map(|(write, built)| write.execution(built))
        .collect())
}

#[cfg(test)]
mod tests;