zirium 0.5.0

//! Shared diff execution with native comparison handles and host-owned operations.

use super::{
    EvaluationError, EvaluationLimits, EvaluationOptions, OperationTraversal, core_error,
    evaluation_budget, program,
};
use crate::{
    diff::{ChangeField, ChangeId, ChangeKind, Diff, DiffOperation, DiffSide},
    semantic::{AttributeValue, OperationId},
};
use std::sync::Arc;
use std::{borrow::Cow, collections::BTreeMap, ops::Range};
use zirql::ProjectionHost;
use zirql::{
    Bindings, Budget, ExpressionKind, ExpressionStep, Host, Located, NativeInput, ProgramHost,
    Stage, Truth,
};

/// Native results from a diff text program. Selections retain comparison and
/// side ownership until an explicit JSON or MLIR rendering boundary. JSON,
/// Markdown and print stages produce report values rather than emission events.
#[derive(Clone, Debug, PartialEq)]
pub enum DiffProgramOutput {
    Changes(Vec<ChangeId>),
    Operations(DiffSide, Vec<DiffOperation>),
    Strings(Vec<String>),
    Count(usize),
    Constructed(super::comprehension::ConstructedReport),
    Map(Vec<(String, DiffProgramOutput)>),
    Json(serde_json::Value, Option<DiffSide>),
    Text(String),
}

/// A parsed diff program. The diff frontend retains its existing syntax and
/// statement-end output policy while Zirql executes statements and expressions.
#[derive(Debug)]
pub struct DiffProgram {
    pub(super) program: zirql::Program<DiffStage, SetOperator, ()>,
    pub(super) parse_work: usize,
    pub(super) parse_results: usize,
}

impl Drop for DiffProgram {
    fn drop(&mut self) {
        let program = std::mem::replace(&mut self.program, zirql::Program::new(Vec::new(), 0..0));
        dispose_statements(program.into_statements());
    }
}

impl DiffProgram {
    /// Check the diff edit policy before loading either input, including a silent
    /// CLI request whose query is otherwise unevaluated.
    ///
    /// # Errors
    /// Refuses edit stage names outside quoted strings.
    pub fn check_read_only(source: &str) -> Result<(), EvaluationError> {
        super::diff_text::reject_diff_mutations(source).map_err(EvaluationError::new)
    }

    /// Parse and bound a read-only diff program without accessing documents.
    ///
    /// # Errors
    /// Refuses unsupported syntax, edits and excessive expression nesting.
    pub fn parse(source: &str) -> Result<Self, EvaluationError> {
        super::diff_text::parse(source)
    }

    /// Evaluate with legacy per-stream and work limits.
    ///
    /// # Errors
    /// Returns located language, ownership, traversal or callback failures.
    pub fn evaluate(
        &self,
        diff: &Diff<'_>,
        document_names: (&str, &str),
        limits: EvaluationLimits,
        options: EvaluationOptions,
        emit: impl FnMut(&Diff<'_>, DiffProgramOutput, &mut Budget) -> Result<(), EvaluationError>,
    ) -> Result<(), EvaluationError> {
        self.evaluate_with_budget(
            diff,
            document_names,
            options,
            &mut evaluation_budget(limits),
            emit,
        )
    }

    /// Run with one caller-owned allowance through nested bodies and callbacks.
    /// Native operation/change values retain comparison ownership, including
    /// the side of an empty operation selection. Earlier callbacks remain
    /// observable after a later failure; callers choose output buffering.
    ///
    /// # Errors
    /// Returns located language, resource, ownership or callback failures.
    pub fn evaluate_with_budget(
        &self,
        diff: &Diff<'_>,
        document_names: (&str, &str),
        options: EvaluationOptions,
        budget: &mut Budget,
        mut emit: impl FnMut(&Diff<'_>, DiffProgramOutput, &mut Budget) -> Result<(), EvaluationError>,
    ) -> Result<(), EvaluationError> {
        budget
            .charge(
                self.parse_work,
                zirql::Location::Query(self.program.range()),
            )
            .map_err(core_error)?;
        budget
            .charge_results(
                self.parse_results,
                zirql::Location::Query(self.program.range()),
            )
            .map_err(core_error)?;
        let names = BTreeMap::from([
            ("input".to_owned(), Kind::Changes),
            ("before_document".to_owned(), Kind::Strings),
            ("after_document".to_owned(), Kind::Strings),
        ]);
        validate_program(&self.program, names, budget)?;
        let input = initial_changes(diff, budget, &self.program.range())?;
        let mut initial = vec![(
            Located::new("input".to_owned(), 0..0),
            copy(&input, budget, &self.program.range())?,
        )];
        for (name, value) in [
            ("before_document", document_names.0),
            ("after_document", document_names.1),
        ] {
            admit(budget, 1, &(0..0))?;
            program::charge_bytes(budget, value.len(), &(0..0))?;
            initial.push((
                Located::new(name.to_owned(), 0..0),
                DiffProgramOutput::Strings(vec![value.to_owned()]),
            ));
        }
        let bindings = Bindings::new(initial).map_err(core_error)?;
        let mut context = Context {
            diff,
            options,
            emit: &mut emit,
        };
        zirql::evaluate_program(
            &self.program,
            &DiffHost,
            input,
            &mut context,
            bindings,
            budget,
        )
        .map_err(program::evaluation_error)?;
        Ok(())
    }
}

pub(super) type Expression = zirql::Expression<DiffStage, SetOperator, ()>;

#[derive(Debug)]
pub(super) struct OwnedExpression(Option<Expression>);

impl OwnedExpression {
    pub(super) fn new(expression: Expression) -> Self {
        Self(Some(expression))
    }
    pub(super) fn into_inner(mut self) -> Expression {
        self.0.take().expect("retained expression")
    }
}
impl std::ops::Deref for OwnedExpression {
    type Target = Expression;
    fn deref(&self) -> &Expression {
        self.0
            .as_ref()
            .expect("a retained expression has not been transferred")
    }
}
impl Drop for OwnedExpression {
    fn drop(&mut self) {
        if let Some(expression) = self.0.take() {
            dispose_expression(expression);
        }
    }
}

pub(super) fn dispose_statements(
    statements: Vec<Located<zirql::Statement<DiffStage, SetOperator, ()>>>,
) {
    for statement in statements {
        let expression = match statement.value {
            zirql::Statement::Bind { expression, .. }
            | zirql::Statement::BindFields { expression, .. }
            | zirql::Statement::Evaluate { expression, .. } => expression,
        };
        dispose_expression(expression);
    }
}

fn dispose_expression(expression: Expression) {
    // A flat set spine is deliberately unbounded by logical nesting. Release
    // its boxes through a worklist instead of recursive Rust drop glue.
    let mut pending = vec![expression];
    while let Some(expression) = pending.pop() {
        match expression.value {
            ExpressionKind::Input | ExpressionKind::Binding(_) => {}
            ExpressionKind::Group(expression) => pending.push(*expression),
            ExpressionKind::Field { value, .. } => pending.push(*value),
            ExpressionKind::Set { left, right, .. } => {
                pending.push(*left);
                pending.push(*right);
            }
            ExpressionKind::Sequence { input, steps } => {
                pending.push(*input);
                // Nested host payloads have their own owned-expression guard.
                drop(steps);
            }
            ExpressionKind::Record(fields) => {
                pending.extend(fields.into_iter().map(|(_, expression)| expression))
            }
        }
    }
}

pub(super) use zirql::NativeSetOperator as SetOperator;

#[derive(Debug)]
pub(super) struct SharedValue(pub(super) Arc<zirql::ValueExpression<Expression>>);
impl Clone for SharedValue {
    fn clone(&self) -> Self {
        Self(Arc::clone(&self.0))
    }
}
impl Drop for SharedValue {
    fn drop(&mut self) {
        if let Some(value) = Arc::get_mut(&mut self.0) {
            let value = std::mem::replace(
                value,
                Located::new(
                    zirql::ValueExpressionKind::Literal(zirql::ConstructedValue::Absent),
                    0..0,
                ),
            );
            zirql::dispose_native_value(value);
        }
    }
}
#[derive(Clone, Debug)]
pub(super) enum Leaf {
    Expression(SharedValue),
    Dynamic {
        name: String,
        arguments: Vec<SharedValue>,
    },
    Bool(bool),
    Change(ChangeKind),
    Changed(ChangeField),
    Op(String),
    Dialect(String),
    ResultType(String),
    HasAttr(String),
    Attr(String, String),
}

#[derive(Debug)]
pub(super) enum DiffStage {
    UnsupportedProjection,
    Value(SharedValue),
    Nested(Box<OwnedExpression>),
    Stream(zirql::Pipeline<Leaf>),
    SelectionUnique(zirql::Pipeline<Leaf>),
    Input,
    Project(DiffSide),
    Users(Option<usize>),
    Defs(Option<usize>),
    Parent,
    Children,
    Root(zirql::Pipeline<Leaf>),
    Subtree,
    Graph(OperationTraversal),
    Fixpoint(Box<OwnedExpression>),
    WhereExists(Box<OwnedExpression>),
    Reverse,
    Head(usize),
    Tail(usize),
    Count,
    SortMapValue,
    Names,
    Attr(String),
    ResultTypes,
    OperandTypes,
    Sort,
    Extreme {
        maximum: bool,
        all: bool,
    },
    Order {
        kind: Order,
        selector: Box<OwnedExpression>,
    },
    MapBy {
        key: Box<OwnedExpression>,
        value: Box<OwnedExpression>,
    },
    Check(Option<usize>),
    Json,
    Markdown,
    Print(String),
    Noop,
}

#[derive(Clone, Copy, Debug)]
pub(super) enum Order {
    Sort,
    Min,
    Max,
    MinAll,
    MaxAll,
}

pub(super) fn sequence(stages: Vec<Located<DiffStage>>, range: Range<usize>) -> Expression {
    Located::new(
        ExpressionKind::Sequence {
            input: Box::new(Located::new(ExpressionKind::Input, range.clone())),
            steps: stages
                .into_iter()
                .map(|stage| Located::new(ExpressionStep::Stage(stage.value), stage.range))
                .collect(),
        },
        range,
    )
}

pub(super) fn nested(expression: Expression) -> Box<OwnedExpression> {
    let range = expression.range.clone();
    Box::new(OwnedExpression::new(Located::new(
        ExpressionKind::Group(Box::new(expression)),
        range,
    )))
}

pub(super) fn stream(
    stage: Stage<Leaf>,
    range: Range<usize>,
) -> Result<DiffStage, EvaluationError> {
    Ok(DiffStage::Stream(
        zirql::Pipeline::with_group_not_nesting(
            vec![Located::new(stage, range.clone())],
            range,
            zirql::MAX_NESTING,
        )
        .map_err(core_error)?,
    ))
}

pub(super) fn initial_changes(
    diff: &Diff<'_>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let mut changes = Vec::new();
    for change in diff.change_ids() {
        charge(budget, 1, range)?;
        push(&mut changes, change, budget, range)?;
    }
    Ok(DiffProgramOutput::Changes(changes))
}

pub(super) fn evaluate_typed(
    expression: &Expression,
    diff: &Diff<'_>,
    options: EvaluationOptions,
    budget: &mut Budget,
) -> Result<DiffProgramOutput, EvaluationError> {
    validate_expression(expression, Kind::Changes, &BTreeMap::new())?;
    let input = initial_changes(diff, budget, &expression.range)?;
    let mut discard = |_: &Diff<'_>, _: DiffProgramOutput, _: &mut Budget| Ok(());
    let mut context = Context {
        diff,
        options,
        emit: &mut discard,
    };
    evaluate_expression(
        expression,
        input,
        &mut context,
        &Bindings::default(),
        budget,
    )
}

struct Context<'a, 'document> {
    diff: &'a Diff<'document>,
    options: EvaluationOptions,
    emit:
        &'a mut dyn FnMut(&Diff<'_>, DiffProgramOutput, &mut Budget) -> Result<(), EvaluationError>,
}
struct DiffHost;

fn charge(budget: &mut Budget, units: usize, range: &Range<usize>) -> Result<(), EvaluationError> {
    budget
        .charge(units, zirql::Location::Query(range.clone()))
        .map_err(core_error)
}
fn admit(budget: &mut Budget, units: usize, range: &Range<usize>) -> Result<(), EvaluationError> {
    budget
        .charge_results(units, zirql::Location::Query(range.clone()))
        .map_err(core_error)
}
fn check(budget: &Budget, units: usize, range: &Range<usize>) -> Result<(), EvaluationError> {
    budget
        .check_items(units, zirql::Location::Query(range.clone()))
        .map_err(core_error)
}
fn push<T>(
    values: &mut Vec<T>,
    value: T,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<(), EvaluationError> {
    check(budget, values.len().saturating_add(1), range)?;
    admit(budget, 1, range)?;
    values.push(value);
    Ok(())
}
fn failure(message: impl Into<String>, range: &Range<usize>) -> EvaluationError {
    EvaluationError::new(message).at(range.clone())
}
fn diff_error(error: crate::diff::DiffError) -> EvaluationError {
    EvaluationError::new(error.to_string())
}

impl DiffProgramOutput {
    pub(super) fn len(&self) -> usize {
        match self {
            Self::Changes(v) => v.len(),
            Self::Operations(_, v) => v.len(),
            Self::Strings(v) => v.len(),
            Self::Map(v) => v.len(),
            Self::Count(_) | Self::Json(_, _) | Self::Text(_) | Self::Constructed(_) => 1,
        }
    }
}

fn copy(
    value: &DiffProgramOutput,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    check(budget, value.len(), range)?;
    admit(budget, value.len().max(1), range)?;
    match value {
        DiffProgramOutput::Constructed(value) => value
            .copy(budget, range)
            .map(DiffProgramOutput::Constructed),
        DiffProgramOutput::Map(entries) => {
            let mut result = Vec::with_capacity(entries.len());
            for (key, value) in entries {
                program::charge_bytes(budget, key.len(), range)?;
                result.push((key.clone(), copy(value, budget, range)?));
            }
            Ok(DiffProgramOutput::Map(result))
        }
        DiffProgramOutput::Strings(values) => {
            program::charge_bytes(budget, values.iter().map(String::len).sum(), range)?;
            Ok(DiffProgramOutput::Strings(values.clone()))
        }
        DiffProgramOutput::Text(text) => {
            program::charge_bytes(budget, text.len(), range)?;
            Ok(DiffProgramOutput::Text(text.clone()))
        }
        DiffProgramOutput::Json(value, side) => {
            admit(budget, super::json_size(value), range)?;
            program::charge_bytes(budget, json_bytes(value), range)?;
            Ok(DiffProgramOutput::Json(value.clone(), *side))
        }
        DiffProgramOutput::Changes(_)
        | DiffProgramOutput::Operations(_, _)
        | DiffProgramOutput::Count(_) => Ok(value.clone()),
    }
}

impl ProgramHost<Context<'_, '_>> for DiffHost {
    type Value = DiffProgramOutput;
    type Stage = DiffStage;
    type SetOperator = SetOperator;
    type Emitter = ();
    type Error = EvaluationError;

    fn copy_value(
        &self,
        value: &DiffProgramOutput,
        _: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<DiffProgramOutput, zirql::EvaluationError<EvaluationError>> {
        copy(value, budget, range).map_err(zirql::EvaluationError::Host)
    }
    fn evaluate_stage(
        &self,
        stage: &DiffStage,
        value: DiffProgramOutput,
        context: &mut Context<'_, '_>,
        bindings: &Bindings<DiffProgramOutput>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<DiffProgramOutput, zirql::EvaluationError<EvaluationError>> {
        let result =
            if let DiffStage::Stream(pipeline) | DiffStage::SelectionUnique(pipeline) = stage {
                if matches!(stage, DiffStage::SelectionUnique(_))
                    && !matches!(
                        value,
                        DiffProgramOutput::Changes(_) | DiffProgramOutput::Operations(_, _)
                    )
                {
                    Err(failure(
                        "unique requires a change or operation stream",
                        range,
                    ))
                } else {
                    evaluate_stream(pipeline, value, context, bindings, budget)
                }
            } else {
                charge(budget, value.len().max(1), range)
                    .and_then(|()| evaluate_stage(stage, value, context, bindings, budget, range))
            };
        result
            .and_then(|value| {
                check(budget, value.len(), range)?;
                Ok(value)
            })
            .map_err(|mut error| {
                if error.range.is_none() || error.range == Some(0..0) {
                    error.range = Some(range.clone());
                }
                zirql::EvaluationError::Host(error)
            })
    }
    fn combine(
        &self,
        operator: &SetOperator,
        left: DiffProgramOutput,
        right: DiffProgramOutput,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<DiffProgramOutput, zirql::EvaluationError<EvaluationError>> {
        combine(context.diff, left, right, *operator, budget, range)
            .map_err(zirql::EvaluationError::Host)
    }
    fn construct_record(
        &self,
        _: Vec<(String, DiffProgramOutput)>,
        _: &mut Context<'_, '_>,
        _: &mut Budget,
        range: &Range<usize>,
    ) -> Result<DiffProgramOutput, zirql::EvaluationError<EvaluationError>> {
        Err(zirql::EvaluationError::Host(failure(
            "records are unsupported in diff mode",
            range,
        )))
    }
    fn emit(
        &self,
        _: &(),
        value: &DiffProgramOutput,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<(), zirql::EvaluationError<EvaluationError>> {
        let snapshot = copy(value, budget, range).map_err(zirql::EvaluationError::Host)?;
        (context.emit)(context.diff, snapshot, budget).map_err(zirql::EvaluationError::Host)
    }
}

fn evaluate_expression(
    expression: &Expression,
    input: DiffProgramOutput,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
) -> Result<DiffProgramOutput, EvaluationError> {
    zirql::evaluate_expression(expression, &DiffHost, input, context, bindings, budget)
        .map_err(program::evaluation_error)
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
enum Row {
    Change(ChangeId),
    Operation(DiffOperation),
    String(String),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Kind {
    Changes,
    Operations(DiffSide),
    Strings,
    Count,
    Map,
    Json,
    Text,
    Constructed,
}

fn rows(value: DiffProgramOutput) -> Result<(Kind, Vec<Row>), EvaluationError> {
    Ok(match value {
        DiffProgramOutput::Changes(v) => (Kind::Changes, v.into_iter().map(Row::Change).collect()),
        DiffProgramOutput::Operations(s, v) => (
            Kind::Operations(s),
            v.into_iter().map(Row::Operation).collect(),
        ),
        DiffProgramOutput::Strings(v) => (Kind::Strings, v.into_iter().map(Row::String).collect()),
        _ => return Err(EvaluationError::new("stage requires a stream")),
    })
}
fn from_rows(kind: Kind, rows: Vec<Row>) -> DiffProgramOutput {
    match kind {
        Kind::Changes => DiffProgramOutput::Changes(
            rows.into_iter()
                .map(|row| {
                    let Row::Change(id) = row else {
                        unreachable!("stream kind matches rows")
                    };
                    id
                })
                .collect(),
        ),
        Kind::Operations(side) => DiffProgramOutput::Operations(
            side,
            rows.into_iter()
                .map(|row| {
                    let Row::Operation(id) = row else {
                        unreachable!("stream kind matches rows")
                    };
                    id
                })
                .collect(),
        ),
        Kind::Strings => DiffProgramOutput::Strings(
            rows.into_iter()
                .map(|row| {
                    let Row::String(text) = row else {
                        unreachable!("stream kind matches rows")
                    };
                    text
                })
                .collect(),
        ),
        _ => unreachable!("only stream kinds contain rows"),
    }
}
struct StreamContext<'a, 'c, 'document> {
    context: &'a mut Context<'c, 'document>,
    bindings: &'a Bindings<DiffProgramOutput>,
    kind: Kind,
}

fn evaluate_stream(
    pipeline: &zirql::Pipeline<Leaf>,
    input: DiffProgramOutput,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
) -> Result<DiffProgramOutput, EvaluationError> {
    let (kind, rows) = rows(input)?;
    let input = NativeInput::from_admitted(
        StreamContext {
            context,
            bindings,
            kind,
        },
        rows,
        budget,
        zirql::Location::Query(pipeline.range()),
    )
    .map_err(core_error)?;
    let result =
        zirql::evaluate(pipeline, &DiffHost, input, budget).map_err(program::evaluation_error)?;
    Ok(match result.value {
        zirql::NativeValue::Stream(rows) => from_rows(result.context.kind, rows),
        zirql::NativeValue::Count(count) => DiffProgramOutput::Count(count),
        zirql::NativeValue::Tally(entries) => DiffProgramOutput::Map(
            entries
                .into_iter()
                .map(|(key, count)| (key, DiffProgramOutput::Count(count)))
                .collect(),
        ),
    })
}

impl Host<Row, StreamContext<'_, '_, '_>, Leaf> for DiffHost {
    type Error = EvaluationError;
    type Key = Row;
    fn stage_work(&self, rows: &[Row], _: &StreamContext<'_, '_, '_>) -> usize {
        rows.len().max(1)
    }
    fn collection_item_work(&self) -> usize {
        0
    }
    fn prepare_stage(
        &self,
        stage: &Located<Stage<Leaf>>,
        context: &mut StreamContext<'_, '_, '_>,
        _: &mut Budget,
    ) -> Result<(), zirql::EvaluationError<EvaluationError>> {
        if matches!(stage.value, Stage::Filter(_))
            && !matches!(context.kind, Kind::Changes | Kind::Operations(_))
        {
            return Err(zirql::EvaluationError::Host(failure(
                "filter requires a change or operation stream",
                &stage.range,
            )));
        }
        if matches!(stage.value, Stage::Tally) && context.kind != Kind::Strings {
            return Err(zirql::EvaluationError::Host(failure(
                "tally requires a value stream",
                &stage.range,
            )));
        }
        if let Stage::Filter(predicate) = &stage.value {
            let mut pending = vec![predicate];
            while let Some(predicate) = pending.pop() {
                match &predicate.value {
                    zirql::PredicateKind::Leaf(Leaf::Change(_) | Leaf::Changed(_))
                        if matches!(context.kind, Kind::Operations(_)) =>
                    {
                        return Err(zirql::EvaluationError::Host(failure(
                            "change predicates require a change stream",
                            &stage.range,
                        )));
                    }
                    zirql::PredicateKind::Not(p) | zirql::PredicateKind::Group(p) => {
                        pending.push(p)
                    }
                    zirql::PredicateKind::And(ps) | zirql::PredicateKind::Or(ps) => {
                        pending.extend(ps)
                    }
                    _ => {}
                }
            }
        }
        Ok(())
    }
    fn prepare_filter_item(
        &self,
        row: &Row,
        context: &mut StreamContext<'_, '_, '_>,
        _: &mut Budget,
        _: &Range<usize>,
    ) -> Result<(), zirql::EvaluationError<EvaluationError>> {
        validate_row(context.context.diff, row).map_err(zirql::EvaluationError::Host)
    }
    fn evaluate_leaf(
        &self,
        leaf: &Leaf,
        row: &Row,
        context: &mut StreamContext<'_, '_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Truth, zirql::EvaluationError<EvaluationError>> {
        let input =
            row_value(row, context.kind, budget, range).map_err(zirql::EvaluationError::Host)?;
        let matched = match leaf {
            Leaf::Expression(value) => {
                let zirql::ValueExpressionKind::PredicateScope { predicate, .. } = &value.0.value
                else {
                    unreachable!("bound predicate")
                };
                zirql::evaluate_native_predicate(
                    predicate,
                    &DiffProjectionHost,
                    input,
                    context.context,
                    context.bindings,
                    budget,
                )?
            }
            Leaf::Dynamic { name, arguments } => {
                let leaf = resolve_leaf(
                    name,
                    arguments,
                    input,
                    context.context,
                    context.bindings,
                    budget,
                )
                .map_err(zirql::EvaluationError::Host)?;
                matches_leaf(context.context.diff, row, &leaf)
                    .map_err(zirql::EvaluationError::Host)?
            }
            _ => matches_leaf(context.context.diff, row, leaf)
                .map_err(zirql::EvaluationError::Host)?,
        };
        Ok(if matched { Truth::True } else { Truth::False })
    }
    fn identity_key(
        &self,
        row: &Row,
        context: &mut StreamContext<'_, '_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Row, zirql::EvaluationError<EvaluationError>> {
        validate_row(context.context.diff, row).map_err(zirql::EvaluationError::Host)?;
        if let Row::String(text) = row {
            admit(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            program::charge_bytes(budget, text.len(), range)
                .map_err(zirql::EvaluationError::Host)?;
        }
        Ok(row.clone())
    }
    fn tally_key(
        &self,
        row: Row,
        _: &mut StreamContext<'_, '_, '_>,
        _: &mut Budget,
        _: &Range<usize>,
    ) -> Result<String, zirql::EvaluationError<EvaluationError>> {
        let Row::String(text) = row else {
            unreachable!("tally preparation requires strings")
        };
        Ok(text)
    }
}

fn validate_row(diff: &Diff<'_>, row: &Row) -> Result<(), EvaluationError> {
    match row {
        Row::Change(id) => {
            diff.change(*id).map_err(diff_error)?;
        }
        Row::Operation(op) => {
            diff.operation_id(*op).map_err(diff_error)?;
        }
        Row::String(_) => {}
    }
    Ok(())
}
fn matches_leaf(diff: &Diff<'_>, row: &Row, leaf: &Leaf) -> Result<bool, EvaluationError> {
    match leaf {
        Leaf::Bool(value) => return Ok(*value),
        Leaf::Change(kind) => {
            let Row::Change(id) = row else {
                return Err(EvaluationError::new("change predicates require changes"));
            };
            let change = diff.change(*id).map_err(diff_error)?;
            return Ok(if *kind == ChangeKind::Moved {
                change.moved()
            } else {
                change.kind() == *kind
            });
        }
        Leaf::Changed(field) => {
            let Row::Change(id) = row else {
                return Err(EvaluationError::new("change predicates require changes"));
            };
            return Ok(diff
                .change(*id)
                .map_err(diff_error)?
                .fields()
                .contains(field));
        }
        _ => {}
    }
    let (side, operation) = match row {
        Row::Change(id) => diff.representative_id(*id).map_err(diff_error)?,
        Row::Operation(op) => (op.side(), diff.operation_id(*op).map_err(diff_error)?),
        Row::String(_) => {
            return Err(EvaluationError::new(
                "operation predicate requires operations",
            ));
        }
    };
    let document = diff.document(side);
    Ok(match leaf {
        Leaf::Op(name) => document.operation_name(operation) == Some(name),
        Leaf::Dialect(name) => document
            .operation_name(operation)
            .and_then(|name| name.split_once('.'))
            .is_some_and(|(dialect, _)| dialect == name),
        Leaf::ResultType(spelling) => document
            .result_types(operation)
            .unwrap_or(&[])
            .iter()
            .any(|&ty| document.type_spelling(ty) == Some(spelling)),
        Leaf::HasAttr(name) => document.attribute_id(operation, name).is_some(),
        Leaf::Attr(name, value) => {
            document
                .attribute_id(operation, name)
                .and_then(|id| document.attribute_value(id))
                .and_then(AttributeValue::decoded_string)
                .as_deref()
                == Some(value)
        }
        Leaf::Expression(_)
        | Leaf::Dynamic { .. }
        | Leaf::Bool(_)
        | Leaf::Change(_)
        | Leaf::Changed(_) => {
            unreachable!("handled change and Boolean leaves")
        }
    })
}

fn evaluate_stage(
    stage: &DiffStage,
    value: DiffProgramOutput,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let diff = context.diff;
    match stage {
        DiffStage::UnsupportedProjection => Err(failure(
            "projection records are unsupported in diff mode",
            range,
        )),
        DiffStage::Value(value_expression) => {
            let result = zirql::evaluate_native_value_with_types(
                &value_expression.0,
                &DiffProjectionHost,
                value,
                context,
                bindings,
                budget,
            )
            .map_err(program::evaluation_error)?;
            super::comprehension::ConstructedReport::with_types(
                result.value,
                result.types,
                range.clone(),
                budget,
            )
            .map(DiffProgramOutput::Constructed)
        }
        DiffStage::Stream(_) | DiffStage::SelectionUnique(_) => {
            unreachable!("stream stages use the shared runner")
        }
        DiffStage::Nested(value_expression) => {
            evaluate_expression(value_expression, value, context, bindings, budget)
        }
        DiffStage::Input => initial_changes(diff, budget, range),
        DiffStage::Project(side) => {
            let DiffProgramOutput::Changes(changes) = value else {
                return Err(failure("side projection requires changes", range));
            };
            let mut operations = Vec::new();
            for change in changes {
                let operation = match side {
                    DiffSide::Before => diff.before_operation(change),
                    DiffSide::After => diff.after_operation(change),
                }
                .map_err(diff_error)?;
                if let Some(operation) = operation {
                    push(&mut operations, operation, budget, range)?;
                }
            }
            Ok(DiffProgramOutput::Operations(*side, operations))
        }
        DiffStage::Users(_)
        | DiffStage::Defs(_)
        | DiffStage::Parent
        | DiffStage::Children
        | DiffStage::Root(_)
        | DiffStage::Subtree
        | DiffStage::Graph(_) => navigate(value, stage, context, bindings, budget, range),
        DiffStage::WhereExists(body) => where_exists(value, body, context, bindings, budget, range),
        DiffStage::Fixpoint(body) => fixed_point(value, body, context, bindings, budget, range),
        DiffStage::Names
        | DiffStage::Attr(_)
        | DiffStage::ResultTypes
        | DiffStage::OperandTypes => project(diff, value, stage, budget, range),
        DiffStage::Count => Ok(DiffProgramOutput::Count(value.len())),
        DiffStage::SortMapValue => {
            let DiffProgramOutput::Map(entries) = value else {
                return Err(failure("sort_by(value) requires a map", range));
            };
            zirql::sort_map_by_value(
                entries,
                budget,
                zirql::Location::Query(range.clone()),
                |value, budget| match value {
                    DiffProgramOutput::Count(count) => Ok(zirql::ScalarMapKey::Unsigned(
                        u64::try_from(*count).map_err(|_| {
                            zirql::EvaluationError::Host(failure(
                                "count exceeds unsigned map key range",
                                range,
                            ))
                        })?,
                    )),
                    DiffProgramOutput::Json(value, _) => {
                        super::scalar_map_key(value, budget, range)
                    }
                    _ => Err(zirql::EvaluationError::Host(failure(
                        "sort_by(value) requires scalar map values",
                        range,
                    ))),
                },
            )
            .map(DiffProgramOutput::Map)
            .map_err(program::evaluation_error)
        }
        DiffStage::Reverse | DiffStage::Head(_) | DiffStage::Tail(_) => {
            if let DiffProgramOutput::Map(mut entries) = value {
                let selection = match stage {
                    DiffStage::Reverse => zirql::MapSelection::Reverse,
                    DiffStage::Head(count) => zirql::MapSelection::Head(*count),
                    DiffStage::Tail(count) => zirql::MapSelection::Tail(*count),
                    _ => unreachable!(),
                };
                zirql::select_map(
                    &mut entries,
                    selection,
                    budget,
                    zirql::Location::Query(range.clone()),
                )
                .map_err(core_error)?;
                return Ok(DiffProgramOutput::Map(entries));
            }
            let (kind, mut items) = rows(value)?;
            match stage {
                DiffStage::Reverse => zirql::reverse(&mut items),
                DiffStage::Head(count) => items.truncate(*count),
                DiffStage::Tail(count) => zirql::tail(&mut items, *count),
                _ => {}
            }
            Ok(from_rows(kind, items))
        }
        DiffStage::Sort | DiffStage::Extreme { .. } => {
            if matches!(stage, DiffStage::Sort)
                && let DiffProgramOutput::Map(entries) = value
            {
                return super::order_map_keys(entries, budget, range).map(DiffProgramOutput::Map);
            }
            let DiffProgramOutput::Strings(strings) = value else {
                return Err(failure("sort and extrema require strings", range));
            };
            let order = match stage {
                DiffStage::Extreme {
                    maximum: false,
                    all: false,
                } => zirql::NativeOrder::Min,
                DiffStage::Extreme {
                    maximum: true,
                    all: false,
                } => zirql::NativeOrder::Max,
                DiffStage::Extreme {
                    maximum: false,
                    all: true,
                } => zirql::NativeOrder::MinAll,
                DiffStage::Extreme {
                    maximum: true,
                    all: true,
                } => zirql::NativeOrder::MaxAll,
                _ => zirql::NativeOrder::Sort,
            };
            super::order_strings(strings, order, budget, range).map(DiffProgramOutput::Strings)
        }
        DiffStage::Order { kind, selector } => {
            order(value, *kind, selector, context, bindings, budget, range)
        }
        DiffStage::MapBy {
            key,
            value: expression,
        } => map(value, key, expression, context, bindings, budget, range),
        DiffStage::Check(expected) => {
            let actual = value.len();
            zirql::check_cardinality(actual, *expected, None, range.clone())
                .map_err(super::core_error)?;
            Ok(value)
        }
        DiffStage::Json => {
            if matches!(value, DiffProgramOutput::Json(_, _)) {
                return Err(failure("json cannot be applied twice", range));
            }
            let side = if let DiffProgramOutput::Operations(side, _) = &value {
                Some(*side)
            } else {
                None
            };
            Ok(DiffProgramOutput::Json(
                to_json(diff, &value, budget, range)?,
                side,
            ))
        }
        DiffStage::Markdown => markdown(diff, value, budget, range).map(DiffProgramOutput::Text),
        DiffStage::Print(template) => {
            interpolate(template, bindings, budget, range).map(DiffProgramOutput::Text)
        }
        DiffStage::Noop => Ok(value),
    }
}

fn operations(
    diff: &Diff<'_>,
    value: DiffProgramOutput,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<(DiffSide, Vec<OperationId>), EvaluationError> {
    let DiffProgramOutput::Operations(side, items) = value else {
        return Err(failure("navigation requires projected operations", range));
    };
    admit(budget, items.len(), range)?;
    let ids = items
        .into_iter()
        .map(|item| {
            if item.side() != side {
                return Err(failure(
                    "operation selection has an incompatible diff side",
                    range,
                ));
            }
            diff.operation_id(item).map_err(diff_error)
        })
        .collect::<Result<_, _>>()?;
    Ok((side, ids))
}
fn navigate(
    value: DiffProgramOutput,
    stage: &DiffStage,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let diff = context.diff;
    let options = context.options;
    let (side, selected) = operations(diff, value, budget, range)?;
    let document = diff.document(side);
    let ids = match stage {
        DiffStage::Users(index) => super::evaluate_users(document, &selected, *index, budget)?,
        DiffStage::Defs(index) => operation_defs(document, &selected, *index, budget)?,
        DiffStage::Parent => super::evaluate_parent(document, &selected, budget)?,
        DiffStage::Children => super::evaluate_children(document, &selected, budget)?,
        DiffStage::Subtree => breadth_first_subtrees(document, &selected, budget, range)?,
        DiffStage::Root(filter) => {
            let mut roots = Vec::new();
            for operation in selected {
                let mut candidate = Some(operation);
                while let Some(operation) = candidate {
                    charge(budget, 1, range)?;
                    let handle = diff.scoped_operation(side, operation).map_err(diff_error)?;
                    admit(budget, 1, range)?;
                    let matches = evaluate_stream(
                        filter,
                        DiffProgramOutput::Operations(side, vec![handle]),
                        context,
                        bindings,
                        budget,
                    )?;
                    if matches.len() != 0 {
                        push(&mut roots, operation, budget, range)?;
                        break;
                    }
                    candidate = super::operation_parent(document, operation);
                }
            }
            roots
        }
        DiffStage::Graph(traversal) => match traversal {
            OperationTraversal::Reachable => {
                super::evaluate_reachable(document, selected, diff.registry(), options, budget)?
            }
            OperationTraversal::Closure => {
                super::evaluate_closure(document, selected, diff.registry(), false, budget)?
            }
            OperationTraversal::Slice => super::evaluate_slice(document, selected, budget)?,
            OperationTraversal::ForwardSlice => {
                super::evaluate_forward_slice(document, selected, budget)?
            }
        },
        _ => unreachable!("only navigation stages enter navigate"),
    };
    // The domain traversal has already admitted the returned stream.
    Ok(DiffProgramOutput::Operations(
        side,
        ids.into_iter()
            .map(|id| diff.scoped_operation(side, id).map_err(diff_error))
            .collect::<Result<_, _>>()?,
    ))
}

// Diff navigation retains its established operation-result-only definition edge.
// Ordinary defs additionally maps block arguments to their owning operation.
fn operation_defs(
    document: &crate::semantic::Document,
    selected: &[OperationId],
    index: Option<usize>,
    budget: &mut Budget,
) -> Result<Vec<OperationId>, EvaluationError> {
    super::collect(
        budget,
        selected
            .iter()
            .flat_map(|&operation| {
                let operands = document.operands(operation).unwrap_or(&[]);
                match index {
                    Some(index) => operands.get(index).map(std::slice::from_ref).unwrap_or(&[]),
                    None => operands,
                }
            })
            .filter_map(|operand| match operand {
                crate::semantic::ValueReference::Resolved(
                    crate::semantic::ValueId::OperationResult { operation, .. },
                ) => Some(*operation),
                crate::semantic::ValueReference::Resolved(
                    crate::semantic::ValueId::BlockArgument { .. },
                )
                | crate::semantic::ValueReference::Invalid(_) => None,
            }),
    )
}

fn breadth_first_subtrees(
    document: &crate::semantic::Document,
    selected: &[OperationId],
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<Vec<OperationId>, EvaluationError> {
    let mut expanded = Vec::new();
    for &root in selected {
        let mut cursor = expanded.len();
        push(&mut expanded, root, budget, range)?;
        while cursor < expanded.len() {
            charge(budget, 1, range)?;
            let children = super::evaluate_children(document, &[expanded[cursor]], budget)?;
            for child in children {
                check(budget, expanded.len().saturating_add(1), range)?;
                // Moving the already admitted child does not produce another item.
                expanded.push(child);
            }
            cursor += 1;
        }
    }
    Ok(expanded)
}

fn where_exists(
    value: DiffProgramOutput,
    body: &Expression,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let (kind, items) = rows(value)?;
    zirql::filter_native_exists(
        items,
        budget,
        zirql::Location::Query(range.clone()),
        |item, budget| {
            validate_row(context.diff, item).map_err(zirql::EvaluationError::Host)?;
            charge(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            admit(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            let input = from_rows(kind, vec![item.clone()]);
            let result = evaluate_expression(body, input, context, bindings, budget)
                .map_err(zirql::EvaluationError::Host)?;
            let (result_kind, result) = rows(result).map_err(zirql::EvaluationError::Host)?;
            if result_kind != kind {
                return Err(zirql::EvaluationError::Host(failure(
                    "where_exists relation must preserve its native kind and projected diff side",
                    range,
                )));
            }
            for item in &result {
                validate_row(context.diff, item).map_err(zirql::EvaluationError::Host)?;
            }
            Ok(result)
        },
    )
    .map(|items| from_rows(kind, items))
    .map_err(program::evaluation_error)
}

fn validate_relation_purity(expression: &Expression) -> Result<(), EvaluationError> {
    let mut pending = vec![expression];
    while let Some(expression) = pending.pop() {
        match &expression.value {
            ExpressionKind::Input | ExpressionKind::Binding(_) => {}
            ExpressionKind::Group(value) | ExpressionKind::Field { value, .. } => {
                pending.push(value)
            }
            ExpressionKind::Set { left, right, .. } => {
                pending.push(left);
                pending.push(right);
            }
            ExpressionKind::Record(fields) => pending.extend(fields.iter().map(|(_, value)| value)),
            ExpressionKind::Sequence { input, steps } => {
                pending.push(input);
                for step in steps {
                    let ExpressionStep::Stage(stage) = &step.value else {
                        return Err(failure("where_exists relation cannot emit", &step.range));
                    };
                    match stage {
                        DiffStage::Noop
                        | DiffStage::Json
                        | DiffStage::Markdown
                        | DiffStage::Print(_) => {
                            return Err(failure(
                                "where_exists relation cannot render or emit",
                                &step.range,
                            ));
                        }
                        DiffStage::Nested(body)
                        | DiffStage::Fixpoint(body)
                        | DiffStage::WhereExists(body) => pending.push(body),
                        DiffStage::Order { selector, .. } => pending.push(selector),
                        DiffStage::MapBy { key, value } => {
                            pending.push(key);
                            pending.push(value);
                        }
                        _ => {}
                    }
                }
            }
        }
    }
    Ok(())
}

fn fixed_point(
    value: DiffProgramOutput,
    body: &Expression,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let DiffProgramOutput::Operations(side, current) = value else {
        return Err(failure("fixpoint requires projected operations", range));
    };
    zirql::evaluate_native_fixed_point(
        current,
        context,
        budget,
        zirql::Location::Query(range.clone()),
        |current, context, budget| {
            let input = copy_operations(side, current, budget, range)
                .map_err(zirql::EvaluationError::Host)?;
            let next = evaluate_expression(body, input, context, bindings, budget)
                .map_err(zirql::EvaluationError::Host)?;
            let DiffProgramOutput::Operations(next_side, next) = next else {
                return Err(zirql::EvaluationError::Host(failure(
                    "fixpoint body must return operations",
                    range,
                )));
            };
            if next_side != side {
                return Err(zirql::EvaluationError::Host(failure(
                    "fixpoint body must preserve the projected diff side",
                    range,
                )));
            }
            Ok(next)
        },
        |current, _, budget| {
            let DiffProgramOutput::Operations(_, copied) =
                copy_operations(side, current, budget, range)
                    .map_err(zirql::EvaluationError::Host)?
            else {
                unreachable!("copied operations");
            };
            Ok(copied)
        },
        |left, right, _, budget| {
            zirql::ordered_native_equal(
                left,
                right,
                budget,
                zirql::Location::Query(range.clone()),
                |left, right, _| Ok(left == right),
            )
        },
    )
    .map(|current| DiffProgramOutput::Operations(side, current))
    .map_err(program::evaluation_error)
}

fn copy_operations(
    side: DiffSide,
    operations: &[DiffOperation],
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    check(budget, operations.len(), range)?;
    let work = operations
        .len()
        .checked_add(1)
        .filter(|_| {
            operations
                .len()
                .checked_mul(std::mem::size_of::<DiffOperation>())
                .is_some_and(|bytes| bytes <= isize::MAX as usize)
        })
        .ok_or_else(|| failure("fixpoint allocation size overflow", range))?;
    charge(budget, work, range)?;
    admit(budget, operations.len(), range)?;
    Ok(DiffProgramOutput::Operations(side, operations.to_vec()))
}

fn project(
    diff: &Diff<'_>,
    value: DiffProgramOutput,
    stage: &DiffStage,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let mut strings = Vec::new();
    let (kind, rows) = rows(value)?;
    if !matches!(kind, Kind::Changes | Kind::Operations(_)) {
        return Err(failure("projection requires changes or operations", range));
    }
    for row in rows {
        let (side, operation) = match row {
            Row::Change(id) => diff.representative_id(id).map_err(diff_error)?,
            Row::Operation(op) => (op.side(), diff.operation_id(op).map_err(diff_error)?),
            Row::String(_) => unreachable!("projection validated stream"),
        };
        let document = diff.document(side);
        match stage {
            DiffStage::Names => {
                if let Some(name) = document.operation_name(operation) {
                    add_string(&mut strings, name, budget, range)?;
                }
            }
            DiffStage::Attr(name) => {
                if let Some(id) = document.attribute_id(operation, name) {
                    match document.attribute_value(id) {
                        Some(value @ AttributeValue::String(spelling)) => {
                            check(budget, strings.len().saturating_add(1), range)?;
                            admit(budget, 1, range)?;
                            program::charge_bytes(budget, spelling.len(), range)?;
                            if let Some(text) = value.decoded_string() {
                                strings.push(text);
                            }
                        }
                        Some(AttributeValue::Symbol(path)) => {
                            check(budget, strings.len().saturating_add(1), range)?;
                            admit(budget, 1, range)?;
                            let bytes = path
                                .iter()
                                .map(String::len)
                                .sum::<usize>()
                                .saturating_add(path.len().saturating_sub(1).saturating_mul(2));
                            program::charge_bytes(budget, bytes, range)?;
                            strings.push(path.join("::"));
                        }
                        _ => {
                            if let Some(text) = document.attribute_spelling_value(id) {
                                add_string(&mut strings, text, budget, range)?;
                            }
                        }
                    }
                }
            }
            DiffStage::ResultTypes => {
                for &ty in document.result_types(operation).unwrap_or(&[]) {
                    if let Some(text) = document.type_spelling(ty) {
                        add_string(&mut strings, text, budget, range)?;
                    }
                }
            }
            DiffStage::OperandTypes => {
                for &operand in document.operands(operation).unwrap_or(&[]) {
                    if let Some(text) = document.value_type(operand) {
                        add_string(&mut strings, text, budget, range)?;
                    }
                }
            }
            _ => unreachable!("only projection stages enter project"),
        }
    }
    Ok(DiffProgramOutput::Strings(strings))
}

fn add_string(
    strings: &mut Vec<String>,
    text: &str,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<(), EvaluationError> {
    check(budget, strings.len().saturating_add(1), range)?;
    admit(budget, 1, range)?;
    program::charge_bytes(budget, text.len(), range)?;
    strings.push(text.to_owned());
    Ok(())
}

fn combine(
    diff: &Diff<'_>,
    left: DiffProgramOutput,
    right: DiffProgramOutput,
    operator: SetOperator,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let location = zirql::Location::Query(range.clone());
    match (left, right) {
        (DiffProgramOutput::Changes(left), DiffProgramOutput::Changes(right)) => {
            for id in left.iter().chain(&right) {
                charge(budget, 1, range)?;
                diff.change(*id).map_err(diff_error)?;
            }
            zirql::combine_native_set(
                operator,
                left.iter().copied(),
                right.iter().copied(),
                left.iter().chain(&right).copied().map(Ok),
                budget,
                location,
                |_, _| Ok(()),
            )
            .map(DiffProgramOutput::Changes)
            .map_err(program::evaluation_error)
        }
        (
            DiffProgramOutput::Operations(side, left),
            DiffProgramOutput::Operations(right_side, right),
        ) => {
            if side != right_side {
                return Err(failure(
                    "set operations require operation selections from the same diff side",
                    range,
                ));
            }
            for operation in left.iter().chain(&right) {
                charge(budget, 1, range)?;
                if operation.side() != side {
                    return Err(failure(
                        "set operation handle has a different diff side",
                        range,
                    ));
                }
                let id = diff.operation_id(*operation).map_err(diff_error)?;
                diff.document(side)
                    .check_operation(id)
                    .map_err(|error| failure(error.to_string(), range))?;
            }
            let selected = zirql::combine_native_set(
                operator,
                left.iter().copied(),
                right.iter().copied(),
                left.iter().chain(&right).copied().map(Ok),
                budget,
                location,
                |_, _| Ok(()),
            )
            .map_err(program::evaluation_error)?;
            Ok(DiffProgramOutput::Operations(side, selected))
        }
        (
            DiffProgramOutput::Changes(_)
            | DiffProgramOutput::Operations(_, _)
            | DiffProgramOutput::Strings(_),
            DiffProgramOutput::Changes(_)
            | DiffProgramOutput::Operations(_, _)
            | DiffProgramOutput::Strings(_),
        ) => Err(failure(
            "set operations require selections of the same kind",
            range,
        )),
        _ => Err(failure("stage requires a stream", range)),
    }
}

fn order(
    value: DiffProgramOutput,
    kind: Order,
    selector: &Expression,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let (stream, items) = rows(value)?;
    if !matches!(stream, Kind::Changes | Kind::Operations(_)) {
        return Err(failure(
            "keyed ordering requires changes or operations",
            range,
        ));
    }
    let order = match kind {
        Order::Sort => zirql::NativeOrder::Sort,
        Order::Min => zirql::NativeOrder::Min,
        Order::Max => zirql::NativeOrder::Max,
        Order::MinAll => zirql::NativeOrder::MinAll,
        Order::MaxAll => zirql::NativeOrder::MaxAll,
    };
    let items = zirql::order_native_by(
        items,
        order,
        budget,
        zirql::Location::Query(range.clone()),
        |item, budget| {
            admit(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            let input = from_rows(stream, vec![item.clone()]);
            let key = evaluate_expression(selector, input, context, bindings, budget)
                .map_err(zirql::EvaluationError::Host)?;
            Ok(match key {
                DiffProgramOutput::Count(count) => zirql::NativeOrderKey::Count(count),
                DiffProgramOutput::Strings(mut strings) if strings.len() == 1 => {
                    zirql::NativeOrderKey::Text(strings.pop().unwrap())
                }
                _ => {
                    return Err(zirql::EvaluationError::Host(failure(
                        "ordering key must produce exactly one scalar",
                        range,
                    )));
                }
            })
        },
    )
    .map_err(program::evaluation_error)?;
    Ok(from_rows(stream, items))
}

fn map(
    value: DiffProgramOutput,
    key: &Expression,
    expression: &Expression,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    let (kind, items) = rows(value)?;
    if !matches!(kind, Kind::Changes | Kind::Operations(_)) {
        return Err(failure(
            "map_by requires a change or operation stream",
            range,
        ));
    }
    let mapping = zirql::NativeMap {
        key,
        value: expression,
    };
    let entries = zirql::map_native_by(
        items,
        context,
        budget,
        zirql::Location::Query(range.clone()),
        |item, context, budget| {
            admit(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            let input = from_rows(kind, vec![item.clone()]);
            let key_input = copy(&input, budget, range).map_err(zirql::EvaluationError::Host)?;
            let key = evaluate_expression(mapping.key, key_input, context, bindings, budget)
                .map_err(zirql::EvaluationError::Host)?;
            Ok(if let DiffProgramOutput::Strings(keys) = key {
                Some(keys)
            } else {
                None
            })
        },
        |item, context, budget| {
            admit(budget, 1, range).map_err(zirql::EvaluationError::Host)?;
            evaluate_expression(
                mapping.value,
                from_rows(kind, vec![item]),
                context,
                bindings,
                budget,
            )
            .map_err(zirql::EvaluationError::Host)
        },
    )
    .map_err(program::evaluation_error)?;
    Ok(DiffProgramOutput::Map(entries))
}

fn json_bytes(value: &serde_json::Value) -> usize {
    match value {
        serde_json::Value::String(text) => text.len(),
        serde_json::Value::Array(values) => values.iter().map(json_bytes).sum(),
        serde_json::Value::Object(entries) => entries
            .iter()
            .map(|(key, value)| key.len() + json_bytes(value))
            .sum(),
        _ => 0,
    }
}

/// Render native diff data only at an explicit JSON boundary.
///
/// # Errors
/// Refuses foreign handles or exhaustion of the caller's result allowance.
pub fn diff_value_to_json(
    diff: &Diff<'_>,
    value: &DiffProgramOutput,
    budget: &mut Budget,
) -> Result<serde_json::Value, EvaluationError> {
    to_json(diff, value, budget, &(0..0))
}
fn to_json(
    diff: &Diff<'_>,
    value: &DiffProgramOutput,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<serde_json::Value, EvaluationError> {
    Ok(match value {
        DiffProgramOutput::Constructed(value) => value.to_json_with_budget(budget)?,
        DiffProgramOutput::Changes(items) => {
            let (units, bytes) = diff.selection_report_size(items).map_err(diff_error)?;
            // First admit copied report data, then charge each serialized chunk
            // before the writer extends its buffer. Parsing the JSON tree is a
            // second materialization of the same bounded report.
            admit(budget, units.saturating_mul(2), range)?;
            program::charge_bytes(budget, bytes.saturating_mul(2), range)?;
            let mut bytes = Vec::new();
            diff.write_selection_json(
                program::BudgetWriter {
                    bytes: &mut bytes,
                    budget,
                    range,
                },
                items,
            )
            .map_err(|error| failure(error.to_string(), range))?;
            serde_json::from_slice(&bytes).map_err(|error| failure(error.to_string(), range))?
        }
        DiffProgramOutput::Operations(side, items) => {
            admit(budget, 1 + items.len().saturating_mul(7), range)?;
            let mut values = Vec::with_capacity(items.len());
            for item in items {
                if item.side() != *side {
                    return Err(failure(
                        "operation selection has an incompatible diff side",
                        range,
                    ));
                }
                let operation = diff.operation_id(*item).map_err(diff_error)?;
                let name = diff.document(*side).operation_name(operation);
                program::charge_bytes(budget, 9 + name.map_or(0, str::len), range)?;
                values.push(serde_json::json!({ "name": name, "range": diff.document(*side).operation_source_range(operation).map(|range| [range.start(),range.end()]) }));
            }
            serde_json::Value::Array(values)
        }
        DiffProgramOutput::Strings(items) => {
            admit(budget, 1 + items.len(), range)?;
            program::charge_bytes(budget, items.iter().map(String::len).sum(), range)?;
            serde_json::json!(items)
        }
        DiffProgramOutput::Count(count) => {
            admit(budget, 1, range)?;
            serde_json::json!(count)
        }
        DiffProgramOutput::Map(entries) => {
            admit(budget, 1 + entries.len(), range)?;
            let mut json = serde_json::Map::new();
            for (key, value) in entries {
                program::charge_bytes(budget, key.len(), range)?;
                json.insert(key.clone(), to_json(diff, value, budget, range)?);
            }
            serde_json::Value::Object(json)
        }
        DiffProgramOutput::Json(value, _) => {
            admit(budget, super::json_size(value), range)?;
            program::charge_bytes(budget, json_bytes(value), range)?;
            value.clone()
        }
        DiffProgramOutput::Text(text) => {
            admit(budget, 1, range)?;
            program::charge_bytes(budget, text.len(), range)?;
            serde_json::Value::String(text.clone())
        }
    })
}
fn markdown(
    diff: &Diff<'_>,
    value: DiffProgramOutput,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<String, EvaluationError> {
    fn cell(text: &str) -> String {
        text.replace('\\', "\\\\")
            .replace('|', "\\|")
            .replace('\n', "<br>")
    }
    admit(budget, 1, range)?;
    let text = match value {
        DiffProgramOutput::Changes(items) => {
            let (units, bytes) = diff.selection_report_size(&items).map_err(diff_error)?;
            program::charge_bytes(
                budget,
                bytes
                    .saturating_add(units.saturating_mul(20))
                    .saturating_mul(6),
                range,
            )?;
            diff.selection_to_markdown(&items).map_err(diff_error)?
        }
        DiffProgramOutput::Strings(items) => {
            program::charge_bytes(budget, 18, range)?;
            let mut output = String::from("| Value |\n| --- |\n");
            for item in items {
                program::charge_bytes(budget, item.len().saturating_mul(4) + 6, range)?;
                output.push_str(&format!("| {} |\n", cell(&item)));
            }
            output
        }
        DiffProgramOutput::Map(entries) => {
            program::charge_bytes(budget, 30, range)?;
            let mut output = String::from("| Key | Value |\n| --- | --- |\n");
            for (key, value) in entries {
                let json = to_json(diff, &value, budget, range)?;
                let mut value = program::serialize_json(&json, budget, range, false)?;
                value.pop();
                program::charge_bytes(
                    budget,
                    (key.len() + value.len()).saturating_mul(4) + 9,
                    range,
                )?;
                output.push_str(&format!("| {} | {} |\n", cell(&key), cell(&value)));
            }
            output
        }
        _ => {
            return Err(failure(
                "markdown requires changes, values, or a map",
                range,
            ));
        }
    };
    Ok(text)
}
enum PrintFragment<'a> {
    Literal(&'a str),
    Binding(&'a str),
}

// Diff print escapes only doubled opening braces; closing braces are literal.
struct PrintFragments<'a> {
    rest: &'a str,
}

impl<'a> PrintFragments<'a> {
    fn new(template: &'a str) -> Self {
        Self { rest: template }
    }
}

impl<'a> Iterator for PrintFragments<'a> {
    type Item = Result<PrintFragment<'a>, &'static str>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.rest.is_empty() {
            return None;
        }
        let Some(open) = self.rest.find('{') else {
            return Some(Ok(PrintFragment::Literal(std::mem::take(&mut self.rest))));
        };
        if open != 0 {
            let text = &self.rest[..open];
            self.rest = &self.rest[open..];
            return Some(Ok(PrintFragment::Literal(text)));
        }
        self.rest = &self.rest[1..];
        if self.rest.starts_with('{') {
            self.rest = &self.rest[1..];
            return Some(Ok(PrintFragment::Literal("{")));
        }
        let Some(close) = self.rest.find('}') else {
            self.rest = "";
            return Some(Err("unterminated interpolation in print"));
        };
        let name = &self.rest[..close];
        self.rest = &self.rest[close + 1..];
        Some(Ok(PrintFragment::Binding(name)))
    }
}

fn interpolate(
    template: &str,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<String, EvaluationError> {
    // Admit the host's token scan; the core admits output copies separately.
    charge(budget, template.len(), range)?;
    let mut output = zirql::expand_fragments(
        PrintFragments::new(template),
        &mut (),
        budget,
        zirql::Location::Query(range.clone()),
        |part, _, budget| {
            let part = part.map_err(|message| zirql::EvaluationError::Host(failure(message, range)))?;
            Ok(match part {
                PrintFragment::Literal(text) => Cow::Borrowed(text),
                PrintFragment::Binding(name) => {
                    let value = bindings.get(name, range.clone())?;
                    match value {
                        DiffProgramOutput::Count(count) => {
                            let bytes = usize::try_from(count.checked_ilog10().unwrap_or(0))
                                .expect("usize represents the number of decimal digits") + 1;
                            budget.charge(bytes, zirql::Location::Query(range.clone()))?;
                            budget.charge_results(bytes, zirql::Location::Query(range.clone()))?;
                            Cow::Owned(count.to_string())
                        }
                        DiffProgramOutput::Strings(values) if values.len() == 1 => {
                            Cow::Borrowed(values[0].as_str())
                        }
                        _ => {
                            return Err(zirql::EvaluationError::Host(failure(
                                format!("string interpolation `{{{name}}}` requires a count or exactly one string"),
                                range,
                            )));
                        }
                    }
                }
            })
        },
    ).map_err(program::evaluation_error)?;
    output
        .len()
        .checked_add(1)
        .filter(|bytes| *bytes <= isize::MAX as usize)
        .ok_or_else(|| failure("print storage limit exceeded", range))?;
    charge(budget, 1, range)?;
    admit(budget, 1, range)?;
    output.push('\n');
    Ok(output)
}

fn validate_program(
    program: &zirql::Program<DiffStage, SetOperator, ()>,
    mut names: BTreeMap<String, Kind>,
    budget: &mut Budget,
) -> Result<(), EvaluationError> {
    let mut shared_names = value_scope(&names);
    for statement in program.statements() {
        let (expression, name) = match &statement.value {
            zirql::Statement::BindFields { .. } => {
                return Err(EvaluationError::new(
                    "pattern bindings are unsupported in diff queries",
                ));
            }
            zirql::Statement::Bind { name, expression } => (expression, Some(name)),
            zirql::Statement::Evaluate { expression, .. } => (expression, None),
        };
        let kind = validate_expression(expression, Kind::Changes, &names)?;
        let shared_kind = validate_shared(
            expression,
            ValueKind::Native(Kind::Changes),
            &shared_names,
            budget,
        )?;
        if let Some(name) = name {
            if matches!(
                name.value.as_str(),
                "input" | "before" | "after" | "before_document" | "after_document"
            ) {
                return Err(failure(
                    format!("binding name `{}` is reserved in diff mode", name.value),
                    &name.range,
                ));
            }
            shared_names.insert(name.value.clone(), shared_kind);
            if names.insert(name.value.clone(), kind).is_some() {
                return Err(failure(
                    format!("binding `{}` is already defined", name.value),
                    &name.range,
                ));
            }
        }
    }
    Ok(())
}
fn validate_expression(
    expression: &Expression,
    input: Kind,
    names: &BTreeMap<String, Kind>,
) -> Result<Kind, EvaluationError> {
    // The left-associated set spine is flat language structure, so inspect it
    // iteratively rather than treating each set term as another nesting level.
    let mut rights = Vec::new();
    let mut first = expression;
    while let ExpressionKind::Set { left, right, .. } = &first.value {
        rights.push(right.as_ref());
        first = left;
    }
    let kind = match &first.value {
        ExpressionKind::Input => input,
        ExpressionKind::Binding(name) => *names
            .get(name)
            .ok_or_else(|| failure(format!("undefined binding: {name}"), &first.range))?,
        ExpressionKind::Group(inner) => validate_expression(inner, input, names)?,
        ExpressionKind::Sequence {
            input: expression,
            steps,
        } => {
            let mut kind = validate_expression(expression, input, names)?;
            for step in steps {
                let ExpressionStep::Stage(stage) = &step.value else {
                    unreachable!("diff emitters are value stages")
                };
                kind = validate_stage(stage, kind, names, &step.range)?;
            }
            kind
        }
        ExpressionKind::Set { .. } => unreachable!("the flat set spine was removed"),
        ExpressionKind::Record(_) | ExpressionKind::Field { .. } => {
            return Err(failure(
                "records are unsupported in diff mode",
                &first.range,
            ));
        }
    };
    for right in rights.into_iter().rev() {
        let other = validate_expression(right, input, names)?;
        if kind != other || !matches!(kind, Kind::Changes | Kind::Operations(_)) {
            return Err(failure(
                if matches!((kind, other), (Kind::Operations(_), Kind::Operations(_))) {
                    "set operations require operation selections from the same diff side"
                } else {
                    "set operations require selections of the same kind"
                },
                &expression.range,
            ));
        }
    }
    Ok(kind)
}
fn validate_stage(
    stage: &DiffStage,
    input: Kind,
    names: &BTreeMap<String, Kind>,
    range: &Range<usize>,
) -> Result<Kind, EvaluationError> {
    let require = |valid: bool, message: &str| {
        if valid {
            Ok(())
        } else {
            Err(failure(message, range))
        }
    };
    let selection = matches!(input, Kind::Changes | Kind::Operations(_));
    let stream = selection || input == Kind::Strings;
    Ok(match stage {
        DiffStage::Stream(pipeline) | DiffStage::SelectionUnique(pipeline) => {
            if matches!(stage, DiffStage::SelectionUnique(_)) {
                require(selection, "unique requires a change or operation stream")?;
            }
            let mut kind = input;
            for stage in pipeline.stages() {
                match &stage.value {
                    Stage::Filter(predicate) => {
                        require(selection, "filter requires a change or operation stream")?;
                        let mut pending = vec![predicate];
                        while let Some(predicate) = pending.pop() {
                            match &predicate.value {
                                zirql::PredicateKind::Leaf(Leaf::Change(_) | Leaf::Changed(_)) => {
                                    require(
                                        input == Kind::Changes,
                                        "change predicates require a change stream",
                                    )?
                                }
                                zirql::PredicateKind::Not(p) | zirql::PredicateKind::Group(p) => {
                                    pending.push(p)
                                }
                                zirql::PredicateKind::And(ps) | zirql::PredicateKind::Or(ps) => {
                                    pending.extend(ps)
                                }
                                _ => {}
                            }
                        }
                    }
                    Stage::Head(_)
                    | Stage::Tail(_)
                    | Stage::Reverse
                    | Stage::Check { .. }
                    | Stage::Unique => require(stream, "stage requires a stream")?,
                    Stage::Count => {
                        require(stream, "count requires a stream")?;
                        kind = Kind::Count;
                    }
                    Stage::Tally => {
                        require(input == Kind::Strings, "tally requires a value stream")?;
                        kind = Kind::Map;
                    }
                    _ => unreachable!("diff native pipelines contain only shared stages"),
                }
            }
            kind
        }
        DiffStage::UnsupportedProjection => {
            return Err(failure(
                "projection records are unsupported in diff mode",
                range,
            ));
        }
        DiffStage::Value(_) => Kind::Constructed,
        DiffStage::Nested(expression) => validate_expression(expression, input, names)?,
        DiffStage::Input => Kind::Changes,
        DiffStage::Project(side) => {
            require(input == Kind::Changes, "side projection requires changes")?;
            Kind::Operations(*side)
        }
        DiffStage::Users(_)
        | DiffStage::Defs(_)
        | DiffStage::Parent
        | DiffStage::Children
        | DiffStage::Root(_)
        | DiffStage::Subtree
        | DiffStage::Graph(_) => {
            require(
                matches!(input, Kind::Operations(_)),
                "navigation requires projected operations",
            )?;
            input
        }
        DiffStage::WhereExists(body) => {
            require(
                selection,
                "where_exists requires changes or projected operations",
            )?;
            validate_relation_purity(body)?;
            let result = validate_expression(body, input, names)?;
            require(
                result == input,
                "where_exists relation must preserve its native kind and projected diff side",
            )?;
            input
        }
        DiffStage::Fixpoint(body) => {
            require(
                matches!(input, Kind::Operations(_)),
                "fixpoint requires projected operations",
            )?;
            let result = validate_expression(body, input, names)?;
            require(
                matches!(result, Kind::Operations(_)),
                "fixpoint body must return operations",
            )?;
            require(
                result == input,
                "fixpoint body must preserve the projected diff side",
            )?;
            input
        }
        DiffStage::Names
        | DiffStage::Attr(_)
        | DiffStage::ResultTypes
        | DiffStage::OperandTypes => {
            require(selection, "projection requires changes or operations")?;
            Kind::Strings
        }
        DiffStage::Count => {
            require(
                stream || input == Kind::Map,
                "count requires a stream or map",
            )?;
            Kind::Count
        }
        DiffStage::SortMapValue => {
            require(input == Kind::Map, "sort_by(value) requires a map")?;
            input
        }
        DiffStage::Reverse | DiffStage::Head(_) | DiffStage::Tail(_) => {
            require(
                stream || input == Kind::Map,
                "stage requires a stream or map",
            )?;
            input
        }
        DiffStage::Sort if input == Kind::Map => input,
        DiffStage::Sort | DiffStage::Extreme { .. } => {
            require(input == Kind::Strings, "sort and extrema require strings")?;
            input
        }
        DiffStage::Order { selector, .. } => {
            require(selection, "keyed ordering requires changes or operations")?;
            let kind = validate_expression(selector, input, names)?;
            require(
                matches!(kind, Kind::Strings | Kind::Count),
                "ordering key must produce exactly one scalar",
            )?;
            input
        }
        DiffStage::MapBy { key, value } => {
            require(selection, "map_by requires a change or operation stream")?;
            require(
                validate_expression(key, input, names)? == Kind::Strings,
                "map_by key must produce exactly one string",
            )?;
            validate_expression(value, input, names)?;
            Kind::Map
        }
        DiffStage::Check(_) | DiffStage::Noop => input,
        DiffStage::Json => {
            require(input != Kind::Json, "json cannot be applied twice")?;
            Kind::Json
        }
        DiffStage::Markdown => {
            require(
                matches!(input, Kind::Changes | Kind::Strings | Kind::Map),
                "markdown requires changes, values, or a map",
            )?;
            Kind::Text
        }
        DiffStage::Print(template) => {
            for part in PrintFragments::new(template) {
                if let PrintFragment::Binding(name) =
                    part.map_err(|message| failure(message, range))?
                {
                    require(
                        names
                            .get(name)
                            .is_some_and(|kind| matches!(kind, Kind::Count | Kind::Strings)),
                        &format!("binding `{name}` is unavailable or incompatible in diff mode"),
                    )?;
                }
            }
            Kind::Text
        }
    })
}

#[derive(Clone, Debug, PartialEq, Eq)]
enum ValueKind {
    Native(Kind),
    Constructed(Arc<Vec<zirql::ValueType>>),
}
struct DiffProjectionHost;

fn value_scope(names: &BTreeMap<String, Kind>) -> BTreeMap<String, ValueKind> {
    names
        .iter()
        .map(|(name, kind)| (name.clone(), ValueKind::Native(*kind)))
        .collect()
}
fn projection_error(error: EvaluationError) -> zirql::QueryError {
    zirql::QueryError::query(error.range.unwrap_or(0..0), error.message)
}
impl ProgramHost<Context<'_, '_>> for DiffProjectionHost {
    type Value = DiffProgramOutput;
    type Stage = DiffStage;
    type SetOperator = SetOperator;
    type Emitter = ();
    type Error = EvaluationError;
    fn copy_value(
        &self,
        value: &Self::Value,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Self::Value, zirql::EvaluationError<EvaluationError>> {
        DiffHost.copy_value(value, context, budget, range)
    }
    fn evaluate_stage(
        &self,
        stage: &DiffStage,
        value: Self::Value,
        context: &mut Context<'_, '_>,
        bindings: &Bindings<Self::Value>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Self::Value, zirql::EvaluationError<EvaluationError>> {
        ProgramHost::evaluate_stage(&DiffHost, stage, value, context, bindings, budget, range)
    }
    fn combine(
        &self,
        operator: &SetOperator,
        left: Self::Value,
        right: Self::Value,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Self::Value, zirql::EvaluationError<EvaluationError>> {
        DiffHost.combine(operator, left, right, context, budget, range)
    }
    fn construct_record(
        &self,
        fields: Vec<(String, Self::Value)>,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Self::Value, zirql::EvaluationError<EvaluationError>> {
        DiffHost.construct_record(fields, context, budget, range)
    }
    fn emit(
        &self,
        emitter: &(),
        value: &Self::Value,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<(), zirql::EvaluationError<EvaluationError>> {
        DiffHost.emit(emitter, value, context, budget, range)
    }
}
impl ProjectionHost<Context<'_, '_>> for DiffProjectionHost {
    type Kind = ValueKind;
    fn kind(
        &self,
        value: &DiffProgramOutput,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<ValueKind, zirql::QueryError> {
        budget.charge(1, zirql::Location::Query(range.clone()))?;
        Ok(match value {
            DiffProgramOutput::Changes(_) => ValueKind::Native(Kind::Changes),
            DiffProgramOutput::Operations(side, _) => ValueKind::Native(Kind::Operations(*side)),
            DiffProgramOutput::Strings(_) => ValueKind::Native(Kind::Strings),
            DiffProgramOutput::Count(_) => ValueKind::Native(Kind::Count),
            DiffProgramOutput::Map(_) => ValueKind::Native(Kind::Map),
            DiffProgramOutput::Json(_, _) => ValueKind::Native(Kind::Json),
            DiffProgramOutput::Text(_) => ValueKind::Native(Kind::Text),
            DiffProgramOutput::Constructed(value) => {
                ValueKind::Constructed(Arc::clone(&value.types))
            }
        })
    }
    fn scalar_types(
        &self,
        kind: &ValueKind,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Option<Vec<zirql::ValueType>>, zirql::QueryError> {
        use zirql::{ArrayType, ValueType as T};
        Ok(match kind {
            ValueKind::Constructed(types) => Some(zirql::copy_value_types(
                types,
                budget,
                zirql::Location::Query(range.clone()),
            )?),
            ValueKind::Native(Kind::Strings) => {
                Some(vec![T::Array(ArrayType::Elements(vec![T::String]))])
            }
            ValueKind::Native(Kind::Count) => Some(vec![T::Integer]),
            _ => None,
        })
    }
    fn constructed_kind(&self, types: Vec<zirql::ValueType>) -> ValueKind {
        ValueKind::Constructed(Arc::new(types))
    }
    fn stage_kind(
        &self,
        stage: &DiffStage,
        input: &ValueKind,
        names: &BTreeMap<String, ValueKind>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<ValueKind, zirql::QueryError> {
        if let DiffStage::Value(value) = stage {
            let types =
                zirql::native_value_types(&value.0, self, input.clone(), names.clone(), budget)?;
            return Ok(self.constructed_kind(types));
        }
        match stage {
            DiffStage::Nested(expression) => {
                return validate_shared(expression, input.clone(), names, budget)
                    .map_err(projection_error);
            }
            DiffStage::WhereExists(value) | DiffStage::Fixpoint(value) => {
                let result = zirql::native_expression_kind(value, self, input, names, budget)?;
                if result != *input {
                    return Err(zirql::QueryError::query(
                        range.clone(),
                        "native body must preserve its diff kind and side",
                    ));
                }
            }
            DiffStage::Order { selector, .. } => {
                zirql::native_expression_kind(selector, self, input, names, budget)?;
            }
            DiffStage::MapBy { key, value } => {
                zirql::native_expression_kind(key, self, input, names, budget)?;
                zirql::native_expression_kind(value, self, input, names, budget)?;
            }
            _ => {}
        }
        let ValueKind::Native(input_kind) = input else {
            return Err(zirql::QueryError::query(
                range.clone(),
                "diff stage requires its native stream kind",
            ));
        };
        let native_names = names
            .iter()
            .filter_map(|(name, kind)| match kind {
                ValueKind::Native(kind) => Some((name.clone(), *kind)),
                _ => None,
            })
            .collect();
        let result =
            validate_stage(stage, *input_kind, &native_names, range).map_err(projection_error)?;
        match stage {
            DiffStage::Stream(pipeline) | DiffStage::Root(pipeline) => {
                for stage in pipeline.stages() {
                    if let Stage::Filter(predicate) = &stage.value {
                        validate_filter(predicate, input.clone(), names, budget)?;
                    }
                }
            }
            _ => {}
        }
        Ok(ValueKind::Native(result))
    }
    fn is_pure(&self, stage: &DiffStage) -> bool {
        if let DiffStage::Nested(expression) = stage {
            return validate_relation_purity(expression).is_ok();
        }
        !matches!(
            stage,
            DiffStage::UnsupportedProjection
                | DiffStage::Json
                | DiffStage::Markdown
                | DiffStage::Print(_)
                | DiffStage::Noop
        )
    }
    fn nested_queries(
        &self,
        stage: &DiffStage,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<zirql::NativeOperands<Context<'_, '_>, Self>, zirql::QueryError> {
        let _ = (stage, budget, range);
        Ok(Vec::new())
    }
    fn subjects(
        &self,
        input: DiffProgramOutput,
        context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<Vec<DiffProgramOutput>, zirql::EvaluationError<EvaluationError>> {
        let (kind, rows) = rows(input).map_err(zirql::EvaluationError::Host)?;
        let mut values = Vec::new();
        for row in rows {
            validate_row(context.diff, &row).map_err(zirql::EvaluationError::Host)?;
            values
                .push(row_value(&row, kind, budget, range).map_err(zirql::EvaluationError::Host)?);
        }
        Ok(values)
    }
    fn project_value(
        &self,
        input: DiffProgramOutput,
        _context: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<zirql::ConstructedValue, zirql::EvaluationError<EvaluationError>> {
        use zirql::ConstructedValue as V;
        match input {
            DiffProgramOutput::Constructed(value) => value
                .into_value(budget, range)
                .map_err(zirql::EvaluationError::Host),
            DiffProgramOutput::Count(count) => {
                zirql::count_value(count, budget, zirql::Location::Query(range.clone()))
                    .map_err(Into::into)
            }
            DiffProgramOutput::Strings(values) => {
                admit(budget, values.len() + 1, range).map_err(zirql::EvaluationError::Host)?;
                Ok(V::Array(values.into_iter().map(V::String).collect()))
            }
            _ => Err(zirql::QueryError::query(
                range.clone(),
                "native diff subjects require an explicit scalar projection",
            )
            .into()),
        }
    }
    fn lift_constructed(
        &self,
        value: zirql::ConstructedValue,
        types: Vec<zirql::ValueType>,
        _: &mut Context<'_, '_>,
        budget: &mut Budget,
        range: &Range<usize>,
    ) -> Result<DiffProgramOutput, zirql::EvaluationError<EvaluationError>> {
        super::comprehension::ConstructedReport::with_types(value, types, range.clone(), budget)
            .map(DiffProgramOutput::Constructed)
            .map_err(zirql::EvaluationError::Host)
    }
    fn pattern_constructed<'v>(
        &self,
        value: &'v DiffProgramOutput,
    ) -> Option<&'v zirql::ConstructedValue> {
        match value {
            DiffProgramOutput::Constructed(value) => Some(value.value()),
            _ => None,
        }
    }
}
fn row_value(
    row: &Row,
    kind: Kind,
    budget: &mut Budget,
    range: &Range<usize>,
) -> Result<DiffProgramOutput, EvaluationError> {
    admit(budget, 1, range)?;
    Ok(match row {
        Row::Change(id) => DiffProgramOutput::Changes(vec![*id]),
        Row::Operation(operation) => {
            let Kind::Operations(side) = kind else {
                unreachable!("operation kind")
            };
            DiffProgramOutput::Operations(side, vec![*operation])
        }
        Row::String(value) => {
            program::charge_bytes(budget, value.len(), range)?;
            DiffProgramOutput::Strings(vec![value.clone()])
        }
    })
}
fn validate_filter(
    predicate: &zirql::Predicate<Leaf>,
    input: ValueKind,
    names: &BTreeMap<String, ValueKind>,
    budget: &mut Budget,
) -> Result<(), zirql::QueryError> {
    let mut pending = vec![predicate];
    while let Some(predicate) = pending.pop() {
        budget.charge(1, zirql::Location::Query(predicate.range.clone()))?;
        match &predicate.value {
            zirql::PredicateKind::Leaf(Leaf::Expression(value)) => {
                let zirql::ValueExpressionKind::PredicateScope { predicate, .. } = &value.0.value
                else {
                    unreachable!("bound predicate")
                };
                zirql::native_predicate_types(
                    predicate,
                    &DiffProjectionHost,
                    input.clone(),
                    names.clone(),
                    budget,
                )?;
            }
            zirql::PredicateKind::Leaf(Leaf::Dynamic { name, arguments }) => {
                if matches!(name.as_str(), "change" | "changed")
                    && input != ValueKind::Native(Kind::Changes)
                {
                    return Err(zirql::QueryError::query(
                        predicate.range.clone(),
                        "change predicates require a change stream",
                    ));
                }
                for argument in arguments {
                    let zirql::ValueExpressionKind::SubjectScope {
                        argument: predicate,
                        ..
                    } = &argument.0.value
                    else {
                        unreachable!("bound native argument")
                    };
                    let types = zirql::native_predicate_value_types(
                        predicate,
                        &DiffProjectionHost,
                        input.clone(),
                        names.clone(),
                        budget,
                    )?;
                    if types.iter().any(|ty| *ty != zirql::ValueType::String) {
                        return Err(zirql::QueryError::query(
                            argument.0.range.clone(),
                            "diff predicate argument needs a string",
                        ));
                    }
                }
            }
            zirql::PredicateKind::Not(value) | zirql::PredicateKind::Group(value) => {
                pending.push(value)
            }
            zirql::PredicateKind::And(values) | zirql::PredicateKind::Or(values) => {
                pending.extend(values)
            }
            _ => {}
        }
    }
    Ok(())
}
fn resolve_leaf(
    name: &str,
    arguments: &[SharedValue],
    input: DiffProgramOutput,
    context: &mut Context<'_, '_>,
    bindings: &Bindings<DiffProgramOutput>,
    budget: &mut Budget,
) -> Result<Leaf, EvaluationError> {
    let mut values = Vec::new();
    for argument in arguments {
        let input = copy(&input, budget, &argument.0.range)?;
        let zirql::ValueExpressionKind::SubjectScope {
            argument: predicate,
            ..
        } = &argument.0.value
        else {
            unreachable!("bound native argument")
        };
        let result = zirql::evaluate_native_predicate_value(
            predicate,
            &DiffProjectionHost,
            input,
            context,
            bindings,
            budget,
        )
        .map_err(program::evaluation_error)?;
        let zirql::ConstructedValue::String(value) = result.value.into_data() else {
            return Err(failure(
                "diff predicate argument needs a string",
                &argument.0.range,
            ));
        };
        values.push(value);
    }
    let mut values = values.into_iter();
    let value = values.next().expect("domain signature has one argument");
    Ok(match name {
        "change" => Leaf::Change(value.parse().map_err(diff_error)?),
        "changed" => Leaf::Changed(value.parse().map_err(diff_error)?),
        "op" => Leaf::Op(value),
        "dialect" => Leaf::Dialect(value),
        "result_type" => Leaf::ResultType(value),
        "has_attr" => Leaf::HasAttr(value),
        "attr" => Leaf::Attr(
            value,
            values
                .next()
                .expect("attribute signature has two arguments"),
        ),
        _ => unreachable!("registered diff predicate leaf"),
    })
}
fn validate_shared(
    expression: &Expression,
    input: ValueKind,
    names: &BTreeMap<String, ValueKind>,
    budget: &mut Budget,
) -> Result<ValueKind, EvaluationError> {
    let mut rights = Vec::new();
    let mut first = expression;
    while let ExpressionKind::Set { left, right, .. } = &first.value {
        rights.push(right.as_ref());
        first = left;
    }
    budget
        .charge(1, zirql::Location::Query(first.range.clone()))
        .map_err(core_error)?;
    let result = match &first.value {
        ExpressionKind::Input => input.clone(),
        ExpressionKind::Binding(name) => names
            .get(name)
            .cloned()
            .ok_or_else(|| failure(format!("undefined binding: {name}"), &first.range))?,
        ExpressionKind::Group(value) => validate_shared(value, input.clone(), names, budget)?,
        ExpressionKind::Sequence {
            input: source,
            steps,
        } => {
            let mut kind = validate_shared(source, input.clone(), names, budget)?;
            for step in steps {
                let ExpressionStep::Stage(stage) = &step.value else {
                    return Err(failure("unsupported diff emitter", &step.range));
                };
                kind = DiffProjectionHost
                    .stage_kind(stage, &kind, names, budget, &step.range)
                    .map_err(core_error)?;
            }
            kind
        }
        _ => return Err(failure("unsupported diff expression", &first.range)),
    };
    for right in rights.into_iter().rev() {
        let other = validate_shared(right, input.clone(), names, budget)?;
        DiffProjectionHost
            .set_kind(&result, &other, budget, &right.range)
            .map_err(core_error)?;
    }
    Ok(result)
}