freenet 0.2.134

//! Executes one conformance property against one contract.
//!
//! The logic here is deterministic given the oracle, which is what lets a peer ship
//! a case to another peer and have it reach the same conclusion independently, and
//! what lets `fdev` reproduce a network finding offline.
//!
//! The *contract* is not guaranteed to be, and that is the interesting part. A
//! contract can reach for the host clock, so the same case run twice can genuinely
//! produce different outputs. [`verify_case`] therefore re-runs any check that was
//! about to report a violation and requires the same finding, with the same outputs,
//! before it will accuse anyone — see the note there. If the module were as
//! deterministic as the first sentence sounds, that machinery would be dead code.

use std::collections::HashSet;
use std::sync::Arc;

use freenet_stdlib::prelude::{RelatedContracts, State, StateDelta, UpdateData, ValidateResult};

use super::oracle::{ConformanceOracle, OracleError, OracleErrorKind};
use super::property::{
    ConformanceProperty, IdempotenceSettling, Inconclusive, OutputDigest, PropertyOutcome,
    Violation,
};

/// Shared, cheaply-clonable input bytes.
pub type Bytes = Arc<[u8]>;

/// How many reconciliation rounds to simulate before giving up.
///
/// The RFC allows legitimate multi-round convergence, so this cannot be small.
/// Running out of rounds is [`Inconclusive::RoundLimit`], never a violation — the
/// only way to fail this check is to revisit an exact state pair, which is proof of
/// a cycle rather than of slowness.
pub const MAX_RECONCILIATION_ROUNDS: usize = 16;

/// How many times to repeat a call when checking determinism.
pub const DETERMINISM_REPEATS: usize = 3;

/// How many times `merge(A, A)` may change the state before the state is judged
/// never to settle.
///
/// Matches `IDENTITY_PROBE_MAX_APPLIES` in the executor's own probe, which arrived
/// at the same number for the same reason: a canonicalizing contract legitimately
/// rewrites a non-canonical stored state once and then stabilizes.
pub const MAX_CANONICALIZATION_APPLIES: usize = 3;

/// One property plus the exact inputs to check it with.
///
/// Cases are the unit of work everywhere: the generator produces them, the verifier
/// consumes them, and [`ConformanceEvidence`](super::ConformanceEvidence) is a
/// serializable case plus enough provenance to interpret it.
#[derive(Debug, Clone)]
pub struct ConformanceCase {
    pub property: ConformanceProperty,
    /// Input states, in the order the property's definition names them (`A`, `B`, `C`).
    pub states: Vec<Bytes>,
    /// Input deltas, where the property takes them.
    pub deltas: Vec<Bytes>,
    /// The summary a delta was generated against, where that matters for replay.
    pub summary: Option<Bytes>,
    /// Related-contract state the contract needs in order to validate the inputs.
    pub related: RelatedContracts<'static>,
}

impl ConformanceCase {
    pub fn new(property: ConformanceProperty, states: Vec<Bytes>) -> Self {
        Self {
            property,
            states,
            deltas: Vec::new(),
            summary: None,
            related: RelatedContracts::default(),
        }
    }

    pub fn with_deltas(mut self, deltas: Vec<Bytes>) -> Self {
        self.deltas = deltas;
        self
    }

    pub fn with_summary(mut self, summary: Bytes) -> Self {
        self.summary = Some(summary);
        self
    }

    pub fn with_related(mut self, related: RelatedContracts<'static>) -> Self {
        self.related = related;
        self
    }

    /// Total input bytes. Used to keep evidence and corpora inside their budgets.
    pub fn input_bytes(&self) -> usize {
        self.states.iter().map(|s| s.len()).sum::<usize>()
            + self.deltas.iter().map(|d| d.len()).sum::<usize>()
            + self.summary.as_ref().map_or(0, |s| s.len())
    }

    fn arity_ok(&self) -> Result<(), Inconclusive> {
        let want_states = self.property.state_arity();
        if self.states.len() < want_states {
            return Err(Inconclusive::MalformedCase(format!(
                "{} needs {want_states} states, got {}",
                self.property,
                self.states.len()
            )));
        }
        let want_deltas = self.property.delta_arity();
        if self.deltas.len() < want_deltas {
            return Err(Inconclusive::MalformedCase(format!(
                "{} needs {want_deltas} deltas, got {}",
                self.property,
                self.deltas.len()
            )));
        }
        Ok(())
    }
}

/// Check one property. This is the shared entry point: `fdev` and the node both
/// call exactly this function, so a finding means the same thing in both places.
pub fn verify_case<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    case: &ConformanceCase,
) -> PropertyOutcome {
    let first = match run(oracle, case) {
        Ok(outcome) => outcome,
        Err(reason) => return PropertyOutcome::Inconclusive(reason),
    };

    // A violation must reproduce before it is reported.
    //
    // Nothing here is free of the clock. A contract that prunes expired entries
    // against `time::now()` — the shape of the in-tree ping contract, with a
    // five-second TTL — can have two merges milliseconds apart straddle an expiry
    // boundary, and the two sides then genuinely differ for a reason that has
    // nothing to do with commutativity. Reported once, that is an accusation of
    // breaking a merge law, aimed at the wrong law and at a contract that is fine.
    //
    // Re-running costs one extra execution and only on the path that was about to
    // accuse someone. A defect that is actually algebraic reproduces every time; a
    // straddled deadline does not.
    if first.is_violation() {
        let second = match run(oracle, case) {
            Ok(outcome) => outcome,
            Err(reason) => return PropertyOutcome::Inconclusive(reason),
        };
        return match (&first, &second) {
            (PropertyOutcome::Violated(a), PropertyOutcome::Violated(b))
                if a.property == b.property && reproduced_identically(a, b) =>
            {
                second
            }
            // The violation reproduced; only its CLASSIFICATION did not.
            //
            // Same inputs, same outputs, both runs — `merge(A, A) != A` was
            // established twice. What flapped is the settling class, which this
            // change insists is DATA about the finding and not the finding itself.
            // Discarding the violation because the data was unstable would throw
            // away a reproduced defect, which is the exact failure this whole
            // change exists to remove.
            //
            // Marked `Indeterminate` rather than reported as either class, because
            // asserting a classification observed once is the other half of the
            // same mistake. That is the identical treatment an ERRORING
            // classification already gets a few lines up in `run`: "we could not
            // determine the class" is one situation and it must not be handled two
            // ways in one file.
            //
            // A clock-reading contract straddling a TTL boundary between the two
            // runs is the ordinary way to reach here, not a contrived one.
            (PropertyOutcome::Violated(a), PropertyOutcome::Violated(b))
                if a.property == b.property
                    && a.left == b.left
                    && a.right == b.right
                    && a.settling != b.settling =>
            {
                let PropertyOutcome::Violated(mut v) = second else {
                    unreachable!("matched on Violated above")
                };
                v.settling = Some(IdempotenceSettling::Indeterminate);
                PropertyOutcome::Violated(v)
            }
            // Disagreed with itself on the OUTPUTS. The contract is
            // nondeterministic, which is its own defect with its own property, and
            // is emphatically not proof of the law this case was checking. Say so
            // under the right name if we can: reporting nothing at all would turn a
            // real defect into a silent miss.
            //
            // Note for `StateIdempotence` specifically: this escalation cannot
            // name it. `escalate_to_determinism` returns early unless the case has
            // enough states for `UpdateDeterminism` (arity 2), and an idempotence
            // case carries one. So the outcome here is always `NotReproducible` for
            // that property, which is why the settling-flap arm above must keep the
            // violation rather than fall through to this.
            _ => escalate_to_determinism(oracle, case),
        };
    }

    first
}

/// A check that failed once and not again means something outside the inputs moved.
/// Try to name that directly instead of just declining to answer.
///
/// Without this, a contract that varies its merge output between identical calls
/// escapes entirely: the law it appeared to break does not reproduce, and the
/// determinism check that would name it correctly repeats back-to-back and can miss
/// a coarse-grained clock. Reporting `UpdateDeterminism` when it does fire turns a
/// silent miss into a finding under the right property.
fn escalate_to_determinism<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    case: &ConformanceCase,
) -> PropertyOutcome {
    if case.states.len() < ConformanceProperty::UpdateDeterminism.state_arity() {
        return PropertyOutcome::Inconclusive(Inconclusive::NotReproducible);
    }
    let determinism = ConformanceCase {
        property: ConformanceProperty::UpdateDeterminism,
        states: case.states.clone(),
        deltas: Vec::new(),
        summary: None,
        related: case.related.clone(),
    };
    match run(oracle, &determinism) {
        Ok(outcome @ PropertyOutcome::Violated(_)) => outcome,
        _ => PropertyOutcome::Inconclusive(Inconclusive::NotReproducible),
    }
}

/// Did the second run reproduce the *same* failure, or merely another failure?
///
/// Matching on the property alone is not enough, and getting this wrong makes the
/// whole re-run pointless for the case it exists to catch: a clock-dependent merge
/// reports `StateCommutativity` on every run, with different bytes each time, so a
/// property-only comparison waves it through as reproducible. The outputs have to
/// match too.
///
/// The determinism properties are the deliberate exception. For those, outputs
/// differing between runs *is* the defect being reported, so demanding that the
/// digests agree would suppress exactly the finding the check exists to make.
fn reproduced_identically(first: &Violation, second: &Violation) -> bool {
    match first.property {
        ConformanceProperty::UpdateDeterminism
        | ConformanceProperty::SummaryDeterminism
        | ConformanceProperty::DeltaDeterminism => true,
        // `settling` too, not just the digests. It is decision-relevant data — the
        // whole point of carrying it is that an enforcement policy branches on it —
        // so a classification that differs between runs must not be reported as
        // though it had reproduced. A contract whose settling varies IS
        // non-deterministic, and falling through to `escalate_to_determinism` names
        // that under the property which describes it.
        ConformanceProperty::StateIdempotence => {
            first.left == second.left
                && first.right == second.right
                && first.settling == second.settling
        }
        ConformanceProperty::StateCommutativity
        | ConformanceProperty::StateAssociativity
        | ConformanceProperty::EmittedStateValidity
        | ConformanceProperty::DeltaIdempotence
        | ConformanceProperty::DeltaPermutationInvariance
        | ConformanceProperty::SelfDeltaEmpty
        | ConformanceProperty::WholeStateSelfDelta
        | ConformanceProperty::ReconciliationCycle
        | ConformanceProperty::PathAgreement
        | ConformanceProperty::TransitionPathAgreement => {
            first.left == second.left && first.right == second.right
        }
    }
}

fn run<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    case: &ConformanceCase,
) -> Result<PropertyOutcome, Inconclusive> {
    case.arity_ok()?;

    // Every property below is only meaningful over states the contract itself
    // considers valid. Checking laws against states a contract would never have
    // accepted is how you manufacture false positives.
    for state in &case.states {
        require_valid(oracle, state, &case.related)?;
    }

    match case.property {
        ConformanceProperty::StateIdempotence => {
            // `merge(A, A) != A` is reported. The fixpoint iteration below
            // CLASSIFIES the finding; it does not decide whether there is one.
            //
            // It used to decide. The loop returned `Holds` if the state settled
            // within `MAX_CANONICALIZATION_APPLIES`, which silently merged two very
            // different outcomes into a pass. The reasoning was that a correct
            // CANONICALIZING contract fails a single-apply check for a reason that
            // is not its fault: the PUT install path stores the client's raw bytes
            // without ever running `update_state`, so a peer can hold a
            // non-canonical state and the first merge legitimately rewrites it.
            //
            // That reasoning is true and it does not license a pass. Two peers
            // handed the same updates then hold different bytes — permanently, not
            // until some deadline, but until unrelated traffic happens to arrive —
            // and their summaries differ, so anti-entropy fires over a difference
            // carrying no information. That is idempotence broken on any honest
            // reading, and the loop was suppressing a true finding.
            //
            // The choice looked binary — flag every canonicalizing contract as
            // removal-eligible, or suppress — and it is not. The distinction is
            // kept as DATA on the finding ([`IdempotenceSettling`]) rather than as
            // a severity fork, so an eventual removal policy can branch on it
            // without the verifier having pre-decided the answer. Bending the
            // classification to soften an enforcement consequence is the specific
            // mistake this replaced (#5462).
            //
            // DELIBERATE DIVERGENCE from the executor's in-tree probe — do NOT
            // "reconcile" them without reading both. `probe_identical_input_idempotency`
            // (contract/executor/runtime/executor_impl.rs) still returns WITHOUT
            // flagging when the state settles after a rewrite, and its rustdoc says
            // flagging on the first change alone "would false-flag every such
            // contract". That is correct FOR THAT MECHANISM and must stay: it has
            // teeth today, suppressing commit, broadcast and full-state egress, so a
            // false flag there degrades a live node. This is a REPORT, acted on by
            // nobody, so it can afford to be honest and carry the distinction as
            // data. The two answer the same question differently on purpose, and the
            // direction a future reader picks decides whether a live egress
            // suppression starts firing on canonicalizing contracts (#5462).
            //
            // Sequencing note: canonicalising at install removes the benign class
            // entirely, and is what makes this honest report safe to enforce on
            // later. Until then, expect `SettledAfter(1)` findings against correct
            // contracts and read them as "the install path has not been fixed yet",
            // not as "this contract is broken".
            let a: &[u8] = &case.states[0];
            let once = merge(oracle, a, a)?;
            if once == a {
                return Ok(PropertyOutcome::Holds);
            }

            // Idempotence is already broken. Everything below only decides WHICH
            // kind, and must never turn the finding back into a pass.
            let mut current = once.clone();
            let mut rewrites: u32 = 1;
            let mut settling = IdempotenceSettling::NeverSettled;
            let mut classification_failure: Option<Inconclusive> = None;
            // One merge is already spent above, so the remaining budget is
            // `MAX_CANONICALIZATION_APPLIES - 1` and the total is unchanged.
            for _ in 1..MAX_CANONICALIZATION_APPLIES {
                // Deliberately NOT `?`. Propagating here would return
                // `Inconclusive` from the whole check and DISCARD the violation
                // established above — reporting "we could not tell" about a
                // contract already caught. The never-settling class grows its
                // state on every apply, so it is the likeliest to hit a fuel or
                // size ceiling on exactly these follow-up merges: the class
                // enforcement can act on would be the one that silently
                // disappeared, and a contract could arrange that in one line by
                // trapping on the second identical merge.
                let merged = match merge(oracle, &current, &current) {
                    Ok(merged) => merged,
                    Err(reason) => {
                        // BIND the reason; do not discard it. A `let`-else here threw
                        // the `Inconclusive` away unread, collapsing a RUNTIME trap
                        // and a CONTRACT rejection into one indistinguishable
                        // outcome — destroying the only signal that would tell us the
                        // harness has a bug rather than the contract. That
                        // distinction is the whole point of #5509/#5517, whose own
                        // rustdoc says a runtime failure must never wear a contract
                        // error's label.
                        settling = IdempotenceSettling::Indeterminate;
                        classification_failure = Some(reason);
                        break;
                    }
                };
                if merged == current {
                    settling = IdempotenceSettling::SettledAfter(rewrites);
                    break;
                }
                current = merged;
                rewrites += 1;
            }

            let mut detail = match settling {
                IdempotenceSettling::SettledAfter(n) => format!(
                    "merge(A, A) != A: the state was rewritten {n} time(s) and then \
                     stopped changing. Idempotence is broken either way; a \
                     canonicalizing contract lands here because the install path \
                     stores raw client bytes without running update_state"
                ),
                IdempotenceSettling::NeverSettled => format!(
                    "merge(A, A) != A and no fixpoint within \
                     {MAX_CANONICALIZATION_APPLIES} applies: the state changes on \
                     every re-apply, so redelivery of the same state keeps mutating it"
                ),
                IdempotenceSettling::Indeterminate => {
                    // Name WHICH side failed. Saying "the contract errored" when the
                    // cause was a runtime trap accuses the wrong party, and this
                    // report is the one that would otherwise carry the harness-bug
                    // signal (#5509).
                    let cause = classification_failure
                        .as_ref()
                        .map_or_else(|| "cause not recorded".to_string(), |r| format!("{r}"));
                    format!(
                        "merge(A, A) != A; classification could not complete \
                         ({cause}), so whether it settles is unknown. The violation \
                         stands on the merge that succeeded. If that cause is a \
                         runtime or harness failure rather than the contract's own \
                         rejection, the unknown classification is OURS, not the \
                         contract's"
                    )
                }
            };
            // Name a reordering as an ENCODING problem, the way every
            // `compare`-based property does. Without this the detail above sends the
            // author to the PUT install path when the real fix is a deterministic
            // encoding — and a merge that only permutes bytes (a HashMap serialized
            // in iteration order, the #4295 shape) settles after one apply, so it
            // lands squarely in the `SettledAfter` arm. The commutativity path
            // already carries a test against exactly this misdiagnosis.
            if is_reordering(&once, a) {
                detail.push_str(
                    ". NOTE: the rewritten state holds exactly the same bytes in a \
                     different order, so the merge agreed on content and the ENCODING \
                     is not canonical (e.g. a HashMap serialized in iteration order). \
                     Peers compare state by hash, so this still prevents convergence, \
                     but the fix is a deterministic encoding rather than a change to \
                     the merge",
                );
            }
            Ok(idempotence_violation(
                case.property,
                &once,
                a,
                &detail,
                settling,
            ))
        }

        ConformanceProperty::StateCommutativity => {
            let (a, b) = (&case.states[0], &case.states[1]);
            let ab = merge(oracle, a, b)?;
            let ba = merge(oracle, b, a)?;
            Ok(compare(
                case.property,
                &ab,
                &ba,
                "merge(A, B) must equal merge(B, A)",
            ))
        }

        ConformanceProperty::StateAssociativity => {
            let (a, b, c) = (&case.states[0], &case.states[1], &case.states[2]);
            let ab = merge(oracle, a, b)?;
            let bc = merge(oracle, b, c)?;
            // The intermediates get the same validity precondition as the inputs.
            // This function's own rule is that checking laws against states a
            // contract would never accept manufactures false positives, and an
            // intermediate the contract rejects is exactly such a state: the real
            // network drops it, so no peer ever merges on top of it.
            require_valid(oracle, &ab, &case.related)?;
            require_valid(oracle, &bc, &case.related)?;
            let ab_c = merge(oracle, &ab, c)?;
            let a_bc = merge(oracle, a, &bc)?;
            Ok(compare(
                case.property,
                &ab_c,
                &a_bc,
                "merge(merge(A, B), C) must equal merge(A, merge(B, C))",
            ))
        }

        ConformanceProperty::EmittedStateValidity => {
            // Merging two valid states must emit a state the contract still accepts.
            // A contract that produces states it would reject cannot converge: the
            // next peer to receive that state drops it.
            //
            // Two distinct states rather than a self-merge, because a self-merge is
            // already covered by StateIdempotence (if merge(A,A) == A and A is valid,
            // the result is trivially valid). The interesting failures happen when
            // combining different states produces something that breaks the
            // contract's own invariant, e.g. a union that exceeds a size cap.
            let (a, b) = (&case.states[0], &case.states[1]);
            let merged = merge(oracle, a, b)?;
            match oracle
                .validate_state(&merged, &case.related)
                .map_err(inconclusive_from)?
            {
                ValidateResult::Valid => Ok(PropertyOutcome::Holds),
                ValidateResult::RequestRelated(_) => Err(Inconclusive::RelatedRequired),
                ValidateResult::Invalid => Ok(PropertyOutcome::Violated(Violation {
                    property: case.property,
                    severity: case.property.severity(),
                    left: OutputDigest::of(&merged),
                    right: OutputDigest::of(a),
                    detail: "update_state emitted a state the contract rejects as invalid"
                        .to_string(),
                    settling: None,
                })),
            }
        }

        ConformanceProperty::UpdateDeterminism => {
            let (a, b) = (&case.states[0], &case.states[1]);
            let first = merge(oracle, a, b)?;
            for _ in 1..DETERMINISM_REPEATS {
                oracle.reset_instance();
                let again = merge(oracle, a, b)?;
                if again != first {
                    return Ok(PropertyOutcome::Violated(Violation {
                        property: case.property,
                        severity: case.property.severity(),
                        left: OutputDigest::of(&first),
                        right: OutputDigest::of(&again),
                        detail: "merge(A, B) returned different bytes on repeated identical calls"
                            .to_string(),
                        settling: None,
                    }));
                }
            }
            Ok(PropertyOutcome::Holds)
        }

        ConformanceProperty::SummaryDeterminism => {
            let a = &case.states[0];
            let first = oracle.summarize_state(a).map_err(inconclusive_from)?;
            for _ in 1..DETERMINISM_REPEATS {
                oracle.reset_instance();
                let again = oracle.summarize_state(a).map_err(inconclusive_from)?;
                if again != first {
                    return Ok(PropertyOutcome::Violated(Violation {
                        property: case.property,
                        severity: case.property.severity(),
                        left: OutputDigest::of(&first),
                        right: OutputDigest::of(&again),
                        detail: "summarize_state returned different bytes for the same state"
                            .to_string(),
                        settling: None,
                    }));
                }
            }
            Ok(PropertyOutcome::Holds)
        }

        ConformanceProperty::DeltaDeterminism => {
            let a = &case.states[0];
            let summary = match &case.summary {
                Some(s) => s.to_vec(),
                None => oracle.summarize_state(a).map_err(inconclusive_from)?,
            };
            let first = oracle
                .get_state_delta(a, &summary)
                .map_err(inconclusive_from)?;
            for _ in 1..DETERMINISM_REPEATS {
                oracle.reset_instance();
                let again = oracle
                    .get_state_delta(a, &summary)
                    .map_err(inconclusive_from)?;
                if again != first {
                    return Ok(PropertyOutcome::Violated(Violation {
                        property: case.property,
                        severity: case.property.severity(),
                        left: OutputDigest::of(&first),
                        right: OutputDigest::of(&again),
                        detail: "get_state_delta returned different bytes for the same inputs"
                            .to_string(),
                        settling: None,
                    }));
                }
            }
            Ok(PropertyOutcome::Holds)
        }

        ConformanceProperty::DeltaIdempotence => {
            let a = &case.states[0];
            let d = &case.deltas[0];
            let once = apply_delta(oracle, a, d)?;
            let twice = apply_delta(oracle, &once, d)?;
            Ok(compare(
                case.property,
                &once,
                &twice,
                "applying the same delta twice must equal applying it once",
            ))
        }

        ConformanceProperty::DeltaPermutationInvariance => {
            let a = &case.states[0];
            let (d1, d2) = (&case.deltas[0], &case.deltas[1]);
            // Ordering only. Re-delivery is deliberately NOT checked here.
            //
            // This property is `Severity::Violation`; delta idempotence is contested
            // and therefore `Severity::Diagnostic`. Re-applying D1 a third time here
            // would be a delta-idempotence check wearing an ordering check's name and
            // severity, so downgrading `DeltaIdempotence` would not actually spare a
            // counter-style contract — it would still be accused, under a property
            // whose description says nothing about re-delivery. One law per property.
            let a_then_d1 = apply_delta(oracle, a, d1)?;
            let forward = apply_delta(oracle, &a_then_d1, d2)?;
            let a_then_d2 = apply_delta(oracle, a, d2)?;
            let reverse = apply_delta(oracle, &a_then_d2, d1)?;
            Ok(compare(
                case.property,
                &forward,
                &reverse,
                "applying independent deltas in a different order reached a different state",
            ))
        }

        ConformanceProperty::SelfDeltaEmpty => {
            let a = &case.states[0];
            let summary = oracle.summarize_state(a).map_err(inconclusive_from)?;
            let delta = oracle
                .get_state_delta(a, &summary)
                .map_err(inconclusive_from)?;
            if delta.is_empty() {
                Ok(PropertyOutcome::Holds)
            } else {
                Ok(PropertyOutcome::Violated(Violation {
                    property: case.property,
                    severity: case.property.severity(),
                    left: OutputDigest::of(&delta),
                    right: OutputDigest::of(&[]),
                    detail: format!(
                        "delta against an exact summary of the same state is {} bytes, not empty",
                        delta.len()
                    ),
                    settling: None,
                }))
            }
        }

        ConformanceProperty::WholeStateSelfDelta => {
            // An empty state has nothing to save, so `delta.len() < a.len()` would be
            // `0 < 0` and report a "self-delta is 0 bytes against a 0 byte state"
            // diagnostic on every contract with an empty initial state. Noise, and
            // noise in a diagnostic is what makes people stop reading diagnostics.
            if case.states[0].is_empty() {
                return Ok(PropertyOutcome::Holds);
            }
            let a = &case.states[0];
            let summary = oracle.summarize_state(a).map_err(inconclusive_from)?;
            let delta = oracle
                .get_state_delta(a, &summary)
                .map_err(inconclusive_from)?;
            if delta.len() < a.len() {
                Ok(PropertyOutcome::Holds)
            } else {
                Ok(PropertyOutcome::Violated(Violation {
                    property: case.property,
                    severity: case.property.severity(),
                    left: OutputDigest::of(&delta),
                    right: OutputDigest::of(a),
                    detail: format!(
                        "self-delta is {} bytes against a {} byte state: synchronization saves nothing",
                        delta.len(),
                        a.len()
                    ),
                    settling: None,
                }))
            }
        }

        ConformanceProperty::ReconciliationCycle => {
            reconciliation_cycle(oracle, &case.states[0], &case.states[1], &case.related)
        }

        ConformanceProperty::PathAgreement => {
            // Both directions, because which state is "base" and which is "other" is
            // an artifact of the order the corpus happened to be listed in, not of
            // anything the contract does. The generator emits each unordered pair
            // once (`i < j`), so checking one direction would make catching a defect
            // depend on file order — the shape of a test that passes for the wrong
            // reason.
            let (a, b) = (&case.states[0], &case.states[1]);
            let forward = path_agreement(oracle, a, b, &case.related);
            if matches!(forward, Ok(PropertyOutcome::Violated(_))) {
                return forward;
            }
            let reverse = path_agreement(oracle, b, a, &case.related);
            if matches!(reverse, Ok(PropertyOutcome::Violated(_))) {
                return reverse;
            }
            // Neither direction found anything. One usable verdict is a verdict;
            // only refuse if neither direction could be evaluated at all.
            match (forward, reverse) {
                (Ok(outcome), _) => Ok(outcome),
                (Err(_), Ok(outcome)) => Ok(outcome),
                (Err(reason), Err(_)) => Err(reason),
            }
        }

        ConformanceProperty::TransitionPathAgreement => {
            // `states[0]` is the base a peer held, `states[1]` the result it
            // actually reached. The generator only ever builds this case from a
            // recorded transition, which is what makes the order meaningful: for an
            // arbitrary pair, "merging B into A yields B" is last-write-wins.
            let (base, result) = (&case.states[0], &case.states[1]);

            // Drive `result` to a fixpoint of its own merge before comparing.
            //
            // A canonicalizing contract legitimately rewrites a stored state the
            // first time it is merged — the PUT install path stores the client's raw
            // bytes without ever running `update_state` — so the state a peer holds
            // may not be canonical yet. Comparing against the raw bytes would report
            // that rewrite as a merge-law break. NOTE: `StateIdempotence` no longer
            // iterates for this reason — since #5462 it reports the break and uses the
            // iteration only to CLASSIFY it. This arm still normalizes, which leaves
            // issue #5462's item 4 ("apply canonicalization uniformly or not at all")
            // open; the decision recorded on that issue covers `StateIdempotence`
            // only. Kept as-is here deliberately rather than changed in passing.
            // Historically both iterated for the
            // same reason and to the same budget.
            let mut settled = result.to_vec();
            let mut reached_fixpoint = false;
            for _ in 0..MAX_CANONICALIZATION_APPLIES {
                let again = merge(oracle, &settled, &settled)?;
                if again == settled {
                    reached_fixpoint = true;
                    break;
                }
                settled = again;
            }
            if !reached_fixpoint {
                // A state that rewrites itself on every re-apply cannot be asked
                // whether some OTHER state absorbs into it, and the defect already
                // has a name. Accusing it here would name the wrong law.
                return Err(Inconclusive::StateNotSettled);
            }
            require_valid(oracle, &settled, &case.related)?;

            let merged = merge(oracle, base, &settled)?;
            // Both sides of the comparison get the same validity precondition, for
            // the same reason `path_agreement` validates both of its outputs: a
            // state the contract itself rejects never reaches another peer, so
            // reasoning about it is reasoning about a history that cannot happen.
            //
            // Without this, a merge that EMITS an invalid state is reported under
            // this property rather than under `EmittedStateValidity`, which is the
            // law that actually names that defect. One law per property is the rule
            // this module follows everywhere else.
            require_valid(oracle, &merged, &case.related)?;
            Ok(compare(
                case.property,
                &merged,
                &settled,
                "merging the settled form of a state a peer actually REACHED back \
                 into the state it came from did not reproduce that settled form, so \
                 the merge path cannot reach what the update path reached: every \
                 peer that receives this state merges it into something else, and \
                 the two can never agree",
            ))
        }
    }
}

/// One direction of [`ConformanceProperty::PathAgreement`]: does `base` reach the
/// same state whether it receives `other`'s information as a delta or whole?
///
/// The two calls mirror the two things production actually does with the same
/// update. `get_state_delta(other, summarize(base))` is the delta the SENDER
/// computes against the RECIPIENT's summary — the same orientation
/// [`reconciliation_cycle`] uses, and getting it backwards would be checking a call
/// the network never makes.
fn path_agreement<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    base: &Bytes,
    other: &Bytes,
    related: &RelatedContracts<'static>,
) -> Result<PropertyOutcome, Inconclusive> {
    let base_summary = oracle.summarize_state(base).map_err(inconclusive_from)?;
    let delta = oracle
        .get_state_delta(other, &base_summary)
        .map_err(inconclusive_from)?;

    // No delta path exists to compare against, in either of the two ways production
    // can decline to take one.
    //
    // An empty delta is the protocol's "nothing to send", and a summary too coarse
    // to express the divergence produces one legitimately — the same contract shape
    // `reconciliation_cycle` models the full-state fallback for. An oversized delta
    // is refused by production's own gate and replaced with a whole-state send, so
    // the delta path is not the path this update would take. Checking either one
    // would be checking a call that never happens.
    if delta.is_empty() || delta_would_be_refused(&delta, other) {
        return Err(Inconclusive::NoDeltaPath);
    }

    let delta_path = apply_delta_bytes(oracle, base, &delta)?;
    let merge_path = merge(oracle, base, other)?;
    if delta_path == merge_path {
        return Ok(PropertyOutcome::Holds);
    }

    // Same validity precondition the inputs get, for the same reason: a state the
    // contract rejects never reaches another peer, so reasoning about it is
    // reasoning about a history that cannot happen.
    require_valid(oracle, &delta_path, related)?;
    require_valid(oracle, &merge_path, related)?;

    // Distinguish "the delta carried less than the whole state" from "the two paths
    // computed different things". This is the guard that keeps this property off a
    // large and legitimate class of contract, and without it the check is a
    // false-positive generator.
    //
    // A contract whose summary cannot express a particular divergence ships a delta
    // carrying only PART of what `other` holds. Its delta path lands short of the
    // merged state on this round and catches up on the next one — a weak delta
    // encoding, not a broken merge, and this module already declines to accuse it
    // under `ReconciliationCycle` for exactly this reason.
    //
    // Merging the whole state on top separates the two. For any merge that is a
    // genuine join, a delta derived from `other` can only move `base` somewhere
    // between `base` and `base ⊔ other`, so joining `other` back on top lands on
    // `base ⊔ other` either way and a partial delta heals. A delta path that
    // computed something the merge path cannot reach does not heal.
    //
    // The guard errs toward silence in every direction, which is the right direction
    // for a removal-eligible property. That includes when the merge itself is
    // unsound: a last-write-wins merge heals trivially, so this declines to pile a
    // second accusation onto a defect `StateCommutativity` already names.
    let healed = merge(oracle, &delta_path, other)?;
    if healed == merge_path {
        return Ok(PropertyOutcome::Holds);
    }

    Ok(compare(
        ConformanceProperty::PathAgreement,
        &delta_path,
        &merge_path,
        "the delta path and the merge path disagree: applying the delta the network \
         would have sent reached a different state from merging the sender's whole \
         state, and re-merging the whole state does not repair the difference, so a \
         peer that received this update as a delta can never agree with one that \
         received it as a state",
    ))
}

/// Simulate two peers exchanging summaries and deltas until they agree.
///
/// Both deltas in a round are computed against the pre-round states, because that is
/// what actually happens on the wire: the peers do not take turns. Convergence ends
/// the check; an exactly-repeated state pair is a proven cycle; running out of
/// rounds while still moving is inconclusive.
///
/// # Why a repeated pair is proof, and the two assumptions it rests on
///
/// The simulation is deterministic and carries no state between rounds, so the next
/// round is a pure function of the current `(left, right)` pair. Revisiting a pair
/// therefore means the process has entered a loop it can never leave, which is why
/// this is reported as a violation rather than as "slow". Legitimate multi-round
/// convergence never trips it: each round moves to a pair not seen before, and
/// running out of rounds is [`Inconclusive::RoundLimit`].
///
/// That argument depends on two things, and both are worth stating because a finding
/// from this check is the most model-dependent one this module produces:
///
/// 1. **The contract is deterministic.** If `summarize_state` or `get_state_delta`
///    varies between identical calls — a contract stamping the host clock into its
///    summary, say — then a repeated pair no longer implies a loop, and the finding
///    would be real but mislabelled: the defect is the nondeterminism, which
///    [`ConformanceProperty::SummaryDeterminism`] names directly. Such a contract
///    genuinely cannot converge either way, so this never accuses a *correct*
///    contract, but the property named may be the wrong one.
/// 2. **This exchange resembles the protocol's.** Peers are modelled as exchanging
///    simultaneously, each computing its delta against the other's pre-round summary.
///    A protocol that reconciles by some other schedule could in principle converge
///    where this model loops. Shadow mode is where that assumption gets tested
///    against real traffic, and it is a reason to treat early findings from this
///    check with more suspicion than the direct algebraic ones.
fn reconciliation_cycle<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    a: &Bytes,
    b: &Bytes,
    related: &RelatedContracts<'static>,
) -> Result<PropertyOutcome, Inconclusive> {
    let mut left = a.to_vec();
    let mut right = b.to_vec();
    let mut seen: HashSet<([u8; 32], [u8; 32])> = HashSet::new();
    seen.insert((digest(&left), digest(&right)));

    for _ in 0..MAX_RECONCILIATION_ROUNDS {
        if left == right {
            return Ok(PropertyOutcome::Holds);
        }

        let left_summary = oracle.summarize_state(&left).map_err(inconclusive_from)?;
        let right_summary = oracle.summarize_state(&right).map_err(inconclusive_from)?;
        let to_left = oracle
            .get_state_delta(&right, &left_summary)
            .map_err(inconclusive_from)?;
        let to_right = oracle
            .get_state_delta(&left, &right_summary)
            .map_err(inconclusive_from)?;

        // Mirror the production size gate, per direction and before applying.
        //
        // `ring::interest::gate_delta_size` refuses a delta that is not meaningfully
        // smaller than the state and sends the whole state instead, so a contract
        // whose delta encoding is oversized never has that delta applied on the
        // network at all. Simulating the application anyway can walk the pair into a
        // repeat — for instance a delta that swaps the two peers' states — and
        // produce an enforceable cycle finding against a contract whose merge is
        // sound and which converges in production.
        // Gate against the SENDER's state in each direction: `to_left` was computed
        // from `right`, so `right` is what production would compare it against.
        let mut next_left = if delta_would_be_refused(&to_left, &right) {
            merge(oracle, &left, &right)?
        } else {
            apply_delta_bytes(oracle, &left, &to_left)?
        };
        let mut next_right = if delta_would_be_refused(&to_right, &left) {
            merge(oracle, &right, &left)?
        } else {
            apply_delta_bytes(oracle, &right, &to_right)?
        };

        // Model the protocol's full-state fallback.
        //
        // Without this the check accuses correct contracts. A summary that cannot
        // express a particular divergence — a coarse version clock under concurrent
        // writes, a compact or probabilistic digest — yields an empty delta in both
        // directions, the pair repeats unchanged, and a delta-only simulation calls
        // that a proven cycle on the very first round. The real protocol does not
        // give up there: when the delta path cannot carry the difference it sends the
        // whole state instead (`ring::interest::gate_delta_size` refuses a
        // state-sized delta and hands back a full-state send). A full-state send is
        // just `update_state(local, [State(remote)])`.
        //
        // Modelling it is what separates the two cases that otherwise look identical
        // here: a contract with a weak delta path converges once the states are
        // exchanged whole, while genuine mutual rejection (#5153) still refuses to
        // move, because refusing is what it does with any input.
        if next_left == left && next_right == right {
            next_left = merge(oracle, &left, &right)?;
            next_right = merge(oracle, &right, &left)?;
        }

        // A state the contract would reject never reaches another peer: the real
        // network drops it. Continuing to iterate on one would be reasoning about a
        // history that cannot happen.
        require_valid(oracle, &next_left, related)?;
        require_valid(oracle, &next_right, related)?;

        left = next_left;
        right = next_right;

        if left == right {
            return Ok(PropertyOutcome::Holds);
        }
        if !seen.insert((digest(&left), digest(&right))) {
            return Ok(PropertyOutcome::Violated(Violation {
                property: ConformanceProperty::ReconciliationCycle,
                severity: ConformanceProperty::ReconciliationCycle.severity(),
                left: OutputDigest::of(&left),
                right: OutputDigest::of(&right),
                detail: "two valid states revisited an exact state pair while still divergent: \
                     reconciliation cannot converge"
                    .to_string(),
                settling: None,
            }));
        }
    }

    Err(Inconclusive::RoundLimit)
}

fn digest(bytes: &[u8]) -> [u8; 32] {
    *blake3::hash(bytes).as_bytes()
}

/// Would the network refuse to send this delta and ship the whole state instead?
///
/// Deliberately conservative relative to `ring::interest::gate_delta_size`: a delta
/// at least as large as the state it would replace carries no saving, and the
/// production path treats that case as a full-state send. Erring toward the fallback
/// only ever makes the simulation converge more readily, which is the safe direction
/// for a check whose failure mode is accusing a correct contract.
/// Whether production would refuse this delta and send the whole state instead.
///
/// `sender_state` is the state of the peer COMPUTING the delta, which is what
/// production compares against (`gate_delta_size`'s `our_state_size`) — not the
/// recipient's. Getting that backwards is not a rounding error: with asymmetric
/// state sizes it applies a delta the network would have replaced with a full state,
/// which can walk the pair into a repeat and manufacture an enforceable
/// reconciliation-cycle finding against a contract that converges in production. It
/// can also do the reverse and hide a real one.
///
/// The margin is production's, imported rather than approximated, for the same
/// reason.
fn delta_would_be_refused(delta: &[u8], sender_state: &[u8]) -> bool {
    !delta.is_empty()
        && delta.len()
            >= sender_state
                .len()
                .saturating_add(crate::ring::interest::MIN_FULL_STATE_SAVING_BYTES)
}

fn require_valid<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    state: &[u8],
    related: &RelatedContracts<'static>,
) -> Result<(), Inconclusive> {
    match oracle
        .validate_state(state, related)
        .map_err(inconclusive_from)?
    {
        ValidateResult::Valid => Ok(()),
        ValidateResult::Invalid => Err(Inconclusive::InputNotValid),
        ValidateResult::RequestRelated(_) => Err(Inconclusive::RelatedRequired),
    }
}

/// `merge(base, other)` — apply `other` to `base` as a full-state update.
///
/// There is no separate `merge` entry point in the contract interface; a full-state
/// update *is* the merge, which is why a contract that "rejects" an incoming state by
/// returning its own state unchanged shows up as an ordinary commutativity failure.
fn merge<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    base: &[u8],
    other: &[u8],
) -> Result<Vec<u8>, Inconclusive> {
    let update = UpdateData::State(State::from(other.to_vec()));
    apply_updates(oracle, base, &[update])
}

fn apply_delta<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    base: &[u8],
    delta: &[u8],
) -> Result<Vec<u8>, Inconclusive> {
    apply_delta_bytes(oracle, base, delta)
}

fn apply_delta_bytes<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    base: &[u8],
    delta: &[u8],
) -> Result<Vec<u8>, Inconclusive> {
    if delta.is_empty() {
        // An empty delta is the protocol's "nothing to send". Handing it to the
        // contract would test the contract's tolerance of a no-op, not a merge law.
        return Ok(base.to_vec());
    }
    let update = UpdateData::Delta(StateDelta::from(delta.to_vec()));
    apply_updates(oracle, base, &[update])
}

fn apply_updates<O: ConformanceOracle + ?Sized>(
    oracle: &mut O,
    base: &[u8],
    updates: &[UpdateData<'_>],
) -> Result<Vec<u8>, Inconclusive> {
    let modification = oracle
        .update_state(base, updates)
        .map_err(inconclusive_from)?;
    match modification.new_state {
        Some(state) => Ok(state.into_bytes()),
        None if modification.requires_dependencies() => Err(Inconclusive::RelatedRequired),
        // No state and no related-contract request. `UpdateModification` is
        // `#[non_exhaustive]` with only `valid()` and `requires()` as constructors,
        // so safe Rust cannot build this shape and no unit test here constructs it.
        // It is still reachable in production: the runtime deserializes whatever the
        // guest emitted, and a malformed or hostile contract can serialize exactly
        // this. Inconclusive rather than a violation, because it says the contract
        // answered nothing, not that a law was broken.
        None => Err(Inconclusive::NoOutputState),
    }
}

fn inconclusive_from(err: OracleError) -> Inconclusive {
    match err.kind {
        OracleErrorKind::Resource => Inconclusive::ResourceLimit(err.message),
        // A contract rejection and a host/WASM failure are both non-enforceable —
        // neither is a merge-law proof, and treating a trap as a violation would
        // mean removing a contract for a bug in the runtime executing it. That
        // severity call is shared and stays shared. The LABEL is not: a trap in the
        // host or the WASM module is not the contract's fault, and folding it into
        // `ContractError` names the wrong culprit and hides the one signal that
        // would tell us the harness itself has a bug (#5509).
        OracleErrorKind::Contract => Inconclusive::ContractError(err.message),
        OracleErrorKind::Runtime => Inconclusive::RuntimeError(err.message),
    }
}

fn compare(
    property: ConformanceProperty,
    left: &[u8],
    right: &[u8],
    detail: &str,
) -> PropertyOutcome {
    if left == right {
        return PropertyOutcome::Holds;
    }
    // Distinguish a reordering from a genuinely different result.
    //
    // Two outputs holding the same bytes in a different order are the signature of
    // non-canonical serialization — a `HashMap` iterated in insertion order, most
    // often — rather than of a merge that computed something different. The
    // distinction matters because the two call for opposite fixes: make the encoding
    // canonical, versus fix the merge. It is also the shape of the #4295
    // false-positive class, which is why the executor's in-tree probe compares byte
    // MULTISETS (`byte_multiset_eq`) rather than exact bytes.
    //
    // This still reports a violation. A reordering is a real problem for this
    // network: peers compare state by hash, so two peers holding the same logical
    // state in different byte order never recognise each other as converged. But
    // saying WHICH of the two it is costs one sort of an already-materialised buffer
    // on a path that is about to report a finding anyway, and an author told
    // "your merge is not commutative" when the truth is "your encoding is not
    // canonical" will look in the wrong place.
    let detail = if is_reordering(left, right) {
        &format!(
            "{detail}. NOTE: both results hold exactly the same bytes in a different \
             order, so the merge agreed on content and the ENCODING is not canonical \
             (e.g. a HashMap serialized in iteration order). Peers compare state by \
             hash, so this still prevents convergence, but the fix is a deterministic \
             encoding rather than a change to the merge"
        )
    } else {
        detail
    };
    violation(property, left, right, detail)
}

/// Whether two differing outputs are permutations of each other.
///
/// Sorting copies is affordable here because this runs only on the path that has
/// already decided to report a finding, never on the common passing path.
fn is_reordering(left: &[u8], right: &[u8]) -> bool {
    if left.len() != right.len() {
        return false;
    }
    let mut a = left.to_vec();
    let mut b = right.to_vec();
    a.sort_unstable();
    b.sort_unstable();
    a == b
}

fn violation(
    property: ConformanceProperty,
    left: &[u8],
    right: &[u8],
    detail: &str,
) -> PropertyOutcome {
    PropertyOutcome::Violated(Violation {
        property,
        severity: property.severity(),
        left: OutputDigest::of(left),
        right: OutputDigest::of(right),
        detail: detail.to_string(),
        settling: None,
    })
}

/// [`violation`] plus the settling classification, for
/// [`ConformanceProperty::StateIdempotence`].
///
/// Separate constructor rather than an extra parameter on `violation`, so the
/// other six properties cannot accidentally acquire a field that means nothing
/// for them, and so the severity stays `property.severity()` for both — the
/// settling data must not reach into the severity.
fn idempotence_violation(
    property: ConformanceProperty,
    left: &[u8],
    right: &[u8],
    detail: &str,
    settling: IdempotenceSettling,
) -> PropertyOutcome {
    PropertyOutcome::Violated(Violation {
        property,
        severity: property.severity(),
        left: OutputDigest::of(left),
        right: OutputDigest::of(right),
        detail: detail.to_string(),
        settling: Some(settling),
    })
}