rlmesh_proto/

lib.rs

//! Generated RLMesh protobuf bindings and protocol-level constants.
#![deny(rustdoc::broken_intra_doc_links)]

use std::collections::HashMap;
use std::sync::RwLock;

/// Identity of the frozen wire substrate: the `core` handshake plus the `spaces`
/// value transport the runtime relays between env and model. NOT the package
/// version, NOT the product semver, and NOT a per-service version — it names the
/// one shared byte contract every component is built against, decoupled from all
/// of them.
///
/// Compatibility is plain equality: a peer is compatible iff its token equals
/// this. There is no support window and no range negotiation. The wire grows
/// additively forever under this single token; workflow *behavior* rides on
/// editions and optional *features* on capabilities, so neither forces a bump.
///
/// This is a failsafe, not a version counter: in the normal course it is never
/// bumped. Bumping it to `rlmesh-wire-v2` is a deliberate, public hard pivot — a
/// build that intentionally will not interoperate with `rlmesh-wire-v1` — reserved
/// for a wire break that additive growth, editions, and capabilities genuinely
/// cannot absorb.
pub const PROTOCOL_GENERATION: &str = "rlmesh-wire-v1";

/// Current workflow semantics edition.
///
/// Stable releases use a bare sealed `YYYY.MM` label. Prerelease and local
/// source builds use a suffixed cohort (`YYYY.MM-<semver-prerelease>` or
/// `YYYY.MM-dev.<git-token>`) so moving builds only interoperate with the same
/// cohort unless both sides explicitly advertise a sealed fallback edition.
pub const CURRENT_WORKFLOW_EDITION: &str = env!("RLMESH_CURRENT_WORKFLOW_EDITION");

/// Bare `YYYY.MM` workflow edition base for this build.
pub const WORKFLOW_EDITION_BASE: &str = env!("RLMESH_WORKFLOW_EDITION_BASE");

/// Build cohort used to spell [`CURRENT_WORKFLOW_EDITION`].
pub const BUILD_COHORT: &str = env!("RLMESH_BUILD_COHORT");

/// Source of the build cohort: `release`, `package`, or `git`.
pub const BUILD_SOURCE: &str = env!("RLMESH_BUILD_SOURCE");

/// Workflow editions this crate can operate under, generated by `build.rs` from
/// `rlmesh.toml`'s `[workflow] supported_editions` with
/// [`CURRENT_WORKFLOW_EDITION`] first. Each edition names an immutable
/// behavioral contract documented in `docs/editions/<base>.md` (the spec file is
/// keyed by the bare `YYYY.MM` base, never the suffixed cohort), so a sealed
/// edition is retained here under its bare name for as long as the crate can
/// still drive it. A published crate ships no `rlmesh.toml`; it reads the same
/// list from its shipped `supported_editions.txt`.
pub const SUPPORTED_WORKFLOW_EDITIONS: &[&str] = &SUPPORTED_WORKFLOW_EDITION_ARRAY;

const SUPPORTED_WORKFLOW_EDITION_LIST: &str = env!("RLMESH_SUPPORTED_WORKFLOW_EDITIONS");

const SUPPORTED_WORKFLOW_EDITION_ARRAY: [&str; edition_count(SUPPORTED_WORKFLOW_EDITION_LIST)] =
    split_editions(SUPPORTED_WORKFLOW_EDITION_LIST);

/// Number of comma-separated editions in the build-time list (never zero: the
/// list always carries at least [`CURRENT_WORKFLOW_EDITION`]).
const fn edition_count(list: &str) -> usize {
    let bytes = list.as_bytes();
    let mut count = 1;
    let mut index = 0;
    while index < bytes.len() {
        if bytes[index] == b',' {
            count += 1;
        }
        index += 1;
    }
    count
}

/// Split the build-time list on commas. Edition names are `YYYY.MM` bases with
/// SemVer/git cohort suffixes, so no name can contain the separator.
const fn split_editions<const N: usize>(list: &'static str) -> [&'static str; N] {
    let bytes = list.as_bytes();
    let mut editions = [""; N];
    let mut start = 0;
    let mut index = 0;
    let mut slot = 0;
    while index < bytes.len() {
        if bytes[index] == b',' {
            editions[slot] = edition_at(bytes, start, index);
            slot += 1;
            start = index + 1;
        }
        index += 1;
    }
    editions[slot] = edition_at(bytes, start, bytes.len());
    editions
}

const fn edition_at(bytes: &'static [u8], start: usize, end: usize) -> &'static str {
    let (_, tail) = bytes.split_at(start);
    let (edition, _) = tail.split_at(end - start);
    match std::str::from_utf8(edition) {
        Ok(edition) => edition,
        Err(_) => panic!("RLMESH_SUPPORTED_WORKFLOW_EDITIONS is not UTF-8"),
    }
}

/// A workflow semantics edition this build implements.
///
/// An edition is a sticky behavioral declaration (the NixOS `stateVersion`
/// model), so this enum is deliberately exhaustive and its arms are append-only:
/// a future default change adds an arm here plus a row in [`defaults`] and
/// leaves every sealed arm — and every code path reading it — untouched. A name
/// no arm implements is refused at the boundary by [`Edition::parse`], never
/// folded into a neighbouring arm.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Edition {
    /// `2026.06`, sealed at v0.1.0. Contract: `docs/editions/2026.06.md`.
    E2026_06,
}

impl Edition {
    /// Every arm, in declaration order; append-only, like the enum itself.
    ///
    /// Rust cannot enumerate variants, so this list is written by hand and
    /// guarded by `every_arm_is_listed_and_has_a_row`: its exhaustive match means
    /// a new arm does not compile until it is named there, and the test fails
    /// unless every name this build offers ([`SUPPORTED_WORKFLOW_EDITIONS`]) and
    /// every arm's own base resolve through this list.
    const ALL: &'static [Edition] = &[Edition::E2026_06];

    /// The bare `YYYY.MM` base this edition's contract document is keyed by.
    pub const fn base(&self) -> &'static str {
        match self {
            Edition::E2026_06 => "2026.06",
        }
    }

    /// The arm this build's own [`CURRENT_WORKFLOW_EDITION`] spells: what a
    /// participant runs at when nothing pins it elsewhere.
    pub fn current() -> Edition {
        Edition::parse(CURRENT_WORKFLOW_EDITION)
            .expect("CURRENT_WORKFLOW_EDITION names an arm (every_arm_is_listed_and_has_a_row)")
    }

    /// Resolve a wire edition name to the arm it names.
    ///
    /// The name is split at its first `-` exactly as [`edition_sort_key`] splits
    /// it: the `YYYY.MM` base names the contract, and a cohort suffix only
    /// identifies a moving build of that SAME base (`YYYY.MM-<semver-prerelease>`
    /// or `YYYY.MM-dev.<git-token>`). So every cohort spelling of a known base
    /// resolves to that base's arm — this build's own
    /// ([`CURRENT_WORKFLOW_EDITION`], e.g. `2026.06-0.1.0-rc.12`) and equally a
    /// retained older base's cohort, which this build never spells itself.
    /// Whitespace is trimmed; a base no arm implements is an [`UnknownEdition`].
    ///
    /// This is the ONLY place an edition name is compared as a string. Every
    /// guard and default downstream branches on the typed value.
    pub fn parse(edition: &str) -> Result<Edition, UnknownEdition> {
        let name = edition.trim();
        let (base, _, _) = edition_sort_key(name);
        Edition::ALL
            .iter()
            .copied()
            .find(|edition| edition.base() == base)
            .ok_or_else(|| UnknownEdition(name.to_string()))
    }
}

impl std::fmt::Display for Edition {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.base())
    }
}

/// A wire edition name no [`Edition`] arm implements.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnknownEdition(String);

impl UnknownEdition {
    /// The rejected name, trimmed, exactly as it arrived on the wire.
    pub fn name(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for UnknownEdition {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "unknown workflow edition {:?}", self.0)
    }
}

impl std::error::Error for UnknownEdition {}

/// The edition-governed defaults one [`Edition`] promises: the values a
/// participant would otherwise hardcode, lifted here so a future default change
/// is a new row rather than an edit to a code path a sealed edition already
/// depends on. One row per arm, returned by [`defaults`]; **a sealed row is
/// never edited.**
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EditionDefaults {
    /// Per-episode step bound applied when the session sets no explicit
    /// `max_episode_steps` and the runtime owns resets, so a broken termination
    /// condition surfaces as a truncation instead of hanging the run.
    ///
    /// Not named in the prose of `docs/editions/2026.06.md`, but observable in
    /// the episode ledger the edition's "Episode Accounting" section governs
    /// ("When a sub-environment reports terminated or truncated, its episode
    /// completes"), and pinned by the `05-default-truncation-bound` fingerprint.
    ///
    /// Mirrored by `_MAX_STEPS_PER_EPISODE` in
    /// `python/rlmesh/src/rlmesh/_models/_eval.py`, which bounds the Python
    /// `Session` loop identically. The two loops must agree, so an edition that
    /// changes this row changes that constant in the same commit.
    pub default_max_episode_steps: i64,

    /// The single reserved `ResetRequest.options` key this edition defines.
    ///
    /// `docs/editions/2026.06.md`, Reset: "This edition reserves exactly one,
    /// `trial_index`: the 0-based ordinal of the episode being started, an
    /// integer for a single lane or a per-lane list aligned to the lanes being
    /// reset."
    pub trial_index_option_key: &'static str,

    /// The autoreset modes under which the RUNTIME owns lane restarts (it issues
    /// the `Reset`s), as opposed to the env rolling its own lanes.
    ///
    /// `docs/editions/2026.06.md`, Episode Accounting: "`AUTORESET_MODE_DISABLED`:
    /// only an explicit `Reset` restarts a lane." and "`AUTORESET_MODE_UNSPECIFIED`:
    /// treated as `DISABLED`."
    pub driver_owned_reset_modes: &'static [core::v1::AutoresetMode],

    /// The `final_info` keys this edition reads an episode's task outcome from,
    /// in priority order: the first key present decides, and a numeric value
    /// coerces by truthiness (`1`/`1.0` → true).
    ///
    /// Which keys count is an edition-governed promise — an env that reports its
    /// outcome under one of these names has it surface on the episode summary
    /// and the `EpisodeCompleted` event; one that uses any other name does not.
    /// Gymnasium's `is_success` leads; `success` and `task_success` follow as the
    /// two spellings the ecosystem also ships.
    ///
    /// The Python `Session` loop reads the outcome from the same `info` with
    /// `bool(info[key])` (`_success_from_info` in
    /// `python/rlmesh/src/rlmesh/_models/_eval.py`); an edition that changes this
    /// row revisits that sweep in the same commit.
    pub success_info_keys: &'static [&'static str],

    /// The reserved info-map key a served env reports value-conformance
    /// warnings under.
    ///
    /// `docs/editions/2026.06.md`, Value Conformance: "A conformance warning is
    /// reported in-band under the reserved `rlmesh.conformance.warning` key in
    /// the info map returned by `reset` and `step`, at most once per (deviation
    /// kind, value path) per session."
    pub conformance_warning_info_key: &'static str,
}

/// Edition `2026.06`, sealed at v0.1.0. Never edited: a default change mints a
/// new edition with its own row.
static DEFAULTS_2026_06: EditionDefaults = EditionDefaults {
    default_max_episode_steps: 100_000,
    trial_index_option_key: "trial_index",
    driver_owned_reset_modes: &[
        core::v1::AutoresetMode::Disabled,
        core::v1::AutoresetMode::Unspecified,
    ],
    success_info_keys: &["is_success", "success", "task_success"],
    conformance_warning_info_key: "rlmesh.conformance.warning",
};

/// The defaults row governing a session running at `edition`.
///
/// Exhaustive by construction: a new [`Edition`] arm does not compile until it
/// has a row here.
pub const fn defaults(edition: Edition) -> &'static EditionDefaults {
    match edition {
        Edition::E2026_06 => &DEFAULTS_2026_06,
    }
}

/// Whether this build retains `edition`: it has an arm for it AND
/// [`SUPPORTED_WORKFLOW_EDITIONS`] (generated from `rlmesh.toml`) still names a
/// spelling of it.
///
/// This is the authority for enforcing a runtime-pinned edition. Membership, not
/// equality with [`CURRENT_WORKFLOW_EDITION`]: a retained older edition is a
/// legitimate session floor, and rejecting it would refuse a route this build
/// can actually drive.
pub fn is_retained_edition(edition: Edition) -> bool {
    SUPPORTED_WORKFLOW_EDITIONS
        .iter()
        .any(|retained| Edition::parse(retained) == Ok(edition))
}

/// Resolve a wire edition name to the arm this build both implements AND
/// retains, or say why it cannot.
///
/// The boundary every string-carrying caller goes through — the env client's
/// handshake, the runtime spec, a language binding — so a name this build was not
/// built to drive is refused where it arrives, in one wording, rather than at a
/// later guard that only ever sees the typed value. The message names the string
/// exactly as it arrived (trimmed) and the retained list, so an operator can see
/// both halves of the mismatch.
pub fn parse_retained_edition(edition: &str) -> Result<Edition, String> {
    Edition::parse(edition)
        .ok()
        .filter(|edition| is_retained_edition(*edition))
        .ok_or_else(|| {
            format!(
                "runtime cannot drive workflow edition {:?}; this build implements {:?}",
                edition.trim(),
                SUPPORTED_WORKFLOW_EDITIONS
            )
        })
}

/// Resolve a **declared** edition — a participant's WANT — to the arm it names,
/// or say why this build cannot declare it.
///
/// Stricter than [`parse_retained_edition`], which answers a different question:
/// whether this build can be *pinned* to a value the runtime already selected out
/// of an intersection of CAN sets. A declaration is a ceiling instead
/// ([`want_admits`]), so it is only usable if it admits something this build
/// offers. The bare `YYYY.MM` base of a retained edition always does — it admits
/// every cohort spelling of that base, which is what makes it the value to write
/// down. A cohort spelling that sorts below everything offered (an older
/// prerelease of this build's own base, say) would refuse every session the
/// build takes part in — including a purely local run — so it is refused here,
/// where it was typed.
pub fn parse_declared_edition(edition: &str) -> Result<Edition, String> {
    let parsed = parse_retained_edition(edition)?;
    let want = edition.trim();
    if SUPPORTED_WORKFLOW_EDITIONS
        .iter()
        .any(|can| want_admits(want, can))
    {
        return Ok(parsed);
    }
    Err(format!(
        "workflow edition {want:?} admits none of the editions this build offers \
         ({SUPPORTED_WORKFLOW_EDITIONS:?}), so declaring it would refuse every session; \
         declare one of those, or the bare {:?} base, instead",
        parsed.base()
    ))
}

/// Stable capability names exchanged during handshake.
///
/// Capabilities are advisory. A present key means the named optional feature is
/// available; an absent key means it is not. They cover optional features that
/// preserve interaction semantics. A feature that changes meaning belongs in an
/// edition or generation. House rule: when an older peer would mishandle an
/// absent field, the emitter checks a capability before sending it; if absence
/// changes semantics, use an edition.
pub mod capabilities {
    /// A served model endpoint processes Join-stream requests concurrently
    /// (pipelined predict): responses arrive in completion order rather than
    /// strict arrival order, while per-route lifecycle ordering is preserved.
    ///
    /// Advisory only; this is not an edition change. The wire messages are
    /// identical: every response still mirrors its `request_id`. A client uses
    /// it to decide whether overlapping multiple predicts on one connection will
    /// actually pipeline (capability present) or serialize behind the handler
    /// (capability absent). See `docs/editions/2026.06.md`.
    pub const MODEL_CONCURRENT_PREDICT_V1: &str = "rlmesh.model.concurrent_predict.v1";

    /// A served model endpoint understands observation history: it answers
    /// `ResolveAdapterResponse.history` when the runtime offers
    /// `delivers_history`, ingests `PredictRequest.history` rows, and holds the
    /// producer to consecutive `step`s. Advisory: a runtime that never offers
    /// history sees no wire difference.
    pub const MODEL_OBSERVATION_HISTORY_V1: &str = "rlmesh.model.observation_history.v1";

    /// A served env endpoint steps and resets lanes individually: a `Reset` or
    /// `Step` naming `env_indices` is honored (and answered partial-width)
    /// instead of rejected with `UNSUPPORTED`, and lane-scoped requests on one
    /// Join stream may be processed concurrently, answered in completion order.
    ///
    /// The wire spelling is the bare `subset_step` shipped peers send, so it
    /// carries no `rlmesh.*` prefix. See `docs/editions/2026.06.md`.
    pub const ENV_SUBSET_STEP: &str = "subset_step";
}

/// Whether the given protocol generation is the one this build speaks. Plain
/// equality — there is no support window. Whitespace is trimmed so a padded wire
/// value still matches.
pub fn is_protocol_generation_supported(generation: &str) -> bool {
    generation.trim() == PROTOCOL_GENERATION
}

/// Ordering key for a workflow edition name: `(base, cohort?, suffix)`.
///
/// The name is split at its **first** `-` into a `YYYY.MM` base and an optional
/// cohort suffix:
/// - `base` compares lexicographically — the zero-padded fixed-width `YYYY.MM`
///   makes that chronological, so a newer date always outranks an older one.
/// - `cohort?` is `true` for a suffixed prerelease/dev cohort and `false` for a
///   bare sealed fallback, so an exact matching moving cohort wins over its
///   sealed fallback when both peers support it.
/// - `suffix` is the full cohort as a deterministic third tiebreak; two
///   same-date cohorts are ordered by suffix rather than by iteration order.
///
/// Applies to workflow editions only. Protocol generations are compared by plain
/// equality ([`is_protocol_generation_supported`]), never ordered — there is no
/// generation window to pick a highest from.
pub fn edition_sort_key(edition: &str) -> (&str, bool, &str) {
    match edition.split_once('-') {
        Some((base, suffix)) => (base, true, suffix),
        None => (edition, false, ""),
    }
}

/// Whether a declared WANT admits `edition` as the session's edition.
///
/// A declaration names the contract a participant was authored against, so a
/// bare `YYYY.MM` WANT is a **base-level** ceiling: it admits every spelling of
/// that base or an older one — on a prerelease or dev build, whose CAN set is
/// only its cohort, `2026.06` selects that build's `2026.06-<cohort>`. A WANT
/// that carries a cohort pins to that exact moving build and keeps the full
/// [`edition_sort_key`] order, so a differing cohort of the same base is still
/// excluded.
pub fn want_admits(want: &str, edition: &str) -> bool {
    match edition_sort_key(want) {
        (base, false, _) => edition_sort_key(edition).0 <= base,
        want => edition_sort_key(edition) <= want,
    }
}

/// One participant's bind-time offer: the two sets it brings to negotiation.
///
/// Built for the env, the model, AND this runtime; [`negotiate_session_floor`]
/// reconciles all three (the runtime is a participant because it re-frames
/// env<->model traffic, so the session runs at an edition all three speak). When
/// the model and runtime are the same build (in-process / `run_local`), the floor
/// degenerates to `env ∩ self`. Protocol generation is NOT part of the offer — it
/// is gated by plain equality at each pairwise handshake, so a session that reaches
/// edition negotiation already shares one generation. Capabilities are advisory and
/// pairwise (each peer reads the other's advertised map directly), not negotiated here.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionOffer {
    /// CAN: every workflow edition this participant can operate under — its
    /// retained sealed list, carried on `supported_workflow_editions`.
    pub editions: Vec<String>,
    /// WANT: the one edition this participant declares, carried on
    /// `preferred_workflow_edition`. `None` (an empty wire value) means
    /// undeclared, and negotiation then reads it as `max(editions)` — which is
    /// what every peer built before the field existed means.
    pub preferred: Option<String>,
}

impl SessionOffer {
    /// Build an undeclared offer (no WANT) from edition string slices.
    /// Whitespace is trimmed.
    pub fn new(editions: &[&str]) -> Self {
        Self {
            editions: editions.iter().map(|e| e.trim().to_string()).collect(),
            preferred: None,
        }
    }

    /// This build's own offer: the retained list it CAN drive, plus the edition
    /// it declares — `declared`, or [`CURRENT_WORKFLOW_EDITION`] when the
    /// participant pins nothing.
    pub fn this_build(declared: Option<&str>) -> Self {
        Self {
            editions: supported_workflow_editions(),
            preferred: Some(declared_workflow_edition(declared).to_string()),
        }
    }

    /// The CAN set, trimmed, with empty entries dropped (they never match).
    fn can(&self) -> impl Iterator<Item = &str> {
        self.editions
            .iter()
            .map(|edition| edition.trim())
            .filter(|edition| !edition.is_empty())
    }

    /// The declared WANT, trimmed; `None` when undeclared.
    fn want(&self) -> Option<&str> {
        self.preferred
            .as_deref()
            .map(str::trim)
            .filter(|want| !want.is_empty())
    }
}

/// The edition a participant declares (its WANT): the explicit declaration, or
/// [`CURRENT_WORKFLOW_EDITION`] when it declares none. Whitespace is trimmed and
/// an empty declaration reads as absent.
pub fn declared_workflow_edition(declared: Option<&str>) -> &str {
    declared
        .map(str::trim)
        .filter(|declared| !declared.is_empty())
        .unwrap_or(CURRENT_WORKFLOW_EDITION)
}

/// One tier's position in a refused negotiation, for the diagnostic.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TierOffer {
    /// Which participant this is: `env`, `model`, or `runtime`.
    pub tier: String,
    /// The tier's CAN set as it arrived.
    pub can: Vec<String>,
    /// The tier's declared WANT, or `None` when it declared none.
    pub want: Option<String>,
}

/// No edition satisfies every participant — the negotiation's clean refusal.
///
/// Carries each tier's WANT and CAN so an operator can see which side is the
/// blocker without reading logs from three processes. Displays as one line per
/// the [`Display`](std::fmt::Display) impl; callers wrap it in their own context.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EditionRefusal {
    /// Every participant, in negotiation order.
    pub tiers: Vec<TierOffer>,
}

impl std::fmt::Display for EditionRefusal {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        for (index, tier) in self.tiers.iter().enumerate() {
            if index > 0 {
                f.write_str("; ")?;
            }
            match &tier.want {
                Some(want) => write!(f, "{} wants {want:?} and can {:?}", tier.tier, tier.can)?,
                None => write!(
                    f,
                    "{} declares no edition and can {:?}",
                    tier.tier, tier.can
                )?,
            }
        }
        Ok(())
    }
}

impl std::error::Error for EditionRefusal {}

/// Select the one edition a whole session runs at, from every participant's
/// WANT and CAN — the single rule both [`negotiate_workflow_edition`] (two
/// parties) and [`negotiate_session_floor`] (three) apply.
///
/// ```text
/// E = max { e ∈ ⋂ can(P) : ∀P want(P) admits e }      ordered by edition_sort_key
/// ```
///
/// where a WANT admits an edition per [`want_admits`]: a bare base admits every
/// spelling of that base or older, a cohort spelling admits what sorts at or
/// below it. A participant that declares no WANT is read as wanting `max(can)`
/// — which is what every peer built before the field existed means — and every
/// member of the intersection is already ≤ its own max, so with nothing
/// declared anywhere the ceiling is vacuous and this reduces exactly to the
/// highest mutual edition.
///
/// A WANT is a **ceiling, not an exact demand**: a pin the intersection does not
/// contain still selects the highest mutual edition it admits, and a pin above
/// what another participant can drive never lifts the session above the floor,
/// because the ceiling is applied *after* intersecting the CAN sets.
///
/// `Err` means no edition satisfies everyone; the caller must fail the session
/// before any Join stream opens.
fn select_workflow_edition(tiers: &[(&str, &SessionOffer)]) -> Result<String, EditionRefusal> {
    let refuse = || EditionRefusal {
        tiers: tiers
            .iter()
            .map(|(tier, offer)| TierOffer {
                tier: (*tier).to_string(),
                can: offer.editions.clone(),
                want: offer.want().map(str::to_string),
            })
            .collect(),
    };
    let (_, first) = tiers.first().ok_or_else(&refuse)?;

    first
        .can()
        .filter(|edition| {
            tiers
                .iter()
                .all(|(_, offer)| offer.can().any(|other| other == *edition))
        })
        .filter(|edition| {
            tiers
                .iter()
                .all(|(_, offer)| offer.want().is_none_or(|want| want_admits(want, edition)))
        })
        .max_by_key(|edition| edition_sort_key(edition))
        .map(str::to_string)
        .ok_or_else(&refuse)
}

/// Select the workflow edition governing a session between this runtime and one
/// peer — the co-located floor (`env ∩ self`) the in-process / `run_local` path
/// runs at, with the model and runtime being the same build.
///
/// Applies `select_workflow_edition`'s rule to the two offers. Editions the
/// peer offers that this build does not retain are ignored, never accepted on
/// the assumption they are compatible.
///
/// This is the **env-leg** negotiation only: `peer` is named `env` in the
/// [`EditionRefusal`] diagnostic, which is correct for the one caller (the env
/// client's handshake). A model-leg two-party negotiation would need its own
/// tier label, and a three-party route uses [`negotiate_session_floor`].
pub fn negotiate_workflow_edition(
    peer: &SessionOffer,
    runtime: &SessionOffer,
) -> Result<String, EditionRefusal> {
    select_workflow_edition(&[("env", peer), ("runtime", runtime)])
}

/// Why the runtime capped a session below what its peers would have run at.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeCap {
    /// The runtime's CAN set, not its declaration, is what holds the session
    /// down: undeclared it would still not reach the peers' edition. Upgrading
    /// the runtime unlocks it.
    Capability,
    /// The runtime's declared WANT holds the session below what its own CAN set
    /// would otherwise have reached. This is the sticky model working as
    /// intended, not a defect.
    Declaration,
}

/// The reconciled workflow edition produced by [`negotiate_session_floor`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionFloor {
    /// The edition the session runs at: `select_workflow_edition` across env,
    /// model, AND this runtime.
    pub selected_workflow_edition: String,
    /// The edition env + model alone would have used (always >= `selected`). When
    /// it differs from `selected`, the runtime is the limiting tier — see
    /// [`runtime_cap`](Self::runtime_cap).
    pub desired_workflow_edition: String,
    /// Why the runtime capped the session, or `None` when it did not.
    pub runtime_cap: Option<RuntimeCap>,
}

impl SessionFloor {
    /// Whether the runtime capped the session below what env + model would have
    /// used on their own. The run is still safe at `selected_workflow_edition`
    /// (all three speak it); see [`runtime_cap`](Self::runtime_cap) for whether
    /// that is a limitation or a declaration.
    pub fn runtime_limited(&self) -> bool {
        self.runtime_cap.is_some()
    }
}

/// Reconcile the route's workflow edition across env, model, and this runtime.
///
/// The runtime decode-rebuilds env<->model envelopes (prost drops fields it does
/// not know), so a session runs at an edition all three support — never the
/// env<->model pairwise max — or an edition-gated field would be silently stripped
/// crossing an older runtime. The result also carries the env+model `desired`
/// edition, and why the runtime capped it, so the caller can tell a runtime that
/// *cannot* keep up from one that deliberately declares an older edition.
/// Protocol generation is not reconciled here — it is gated by equality at each
/// pairwise handshake, so all three already share it.
///
/// `Err` when the three share no edition; the caller must then fail the session
/// before any Join stream opens.
pub fn negotiate_session_floor(
    env: &SessionOffer,
    model: &SessionOffer,
    runtime: &SessionOffer,
) -> Result<SessionFloor, EditionRefusal> {
    let selected_workflow_edition =
        select_workflow_edition(&[("env", env), ("model", model), ("runtime", runtime)])?;
    // env+model alone always admit at least `selected` (it clears both their CAN
    // sets and both their ceilings), so this is Ok; fall back defensively.
    let desired_workflow_edition = select_workflow_edition(&[("env", env), ("model", model)])
        .unwrap_or_else(|_| selected_workflow_edition.clone());
    let runtime_cap = (desired_workflow_edition != selected_workflow_edition).then(|| {
        // Attribute the cap to the runtime's declaration only when dropping that
        // declaration would actually lift the session: what its CAN set alone
        // admits. Asking whether its CAN set contains `desired` would blame the
        // CAN set for a runtime that is both older AND declared below its own max,
        // telling the operator to upgrade when the declaration is the real pin.
        let undeclared = SessionOffer {
            editions: runtime.editions.clone(),
            preferred: None,
        };
        // Relaxing one ceiling never refuses what `selected` already cleared.
        let capable =
            select_workflow_edition(&[("env", env), ("model", model), ("runtime", &undeclared)])
                .unwrap_or_else(|_| selected_workflow_edition.clone());
        if edition_sort_key(&selected_workflow_edition) < edition_sort_key(&capable) {
            RuntimeCap::Declaration
        } else {
            RuntimeCap::Capability
        }
    });
    Ok(SessionFloor {
        selected_workflow_edition,
        desired_workflow_edition,
        runtime_cap,
    })
}

/// Whether a client's handshake offer is compatible with this server: the only
/// handshake-level decision is protocol generation (a hard, full-restart break).
///
/// The workflow **edition** is NOT decided here — only the runtime sees every
/// participant, so it is the sole edition authority (via [`negotiate_session_floor`],
/// which degenerates to `env ∩ self` when the model and runtime are the same build).
/// A generation-compatible peer that shares no edition handshakes fine and then
/// fails at the runtime's floor, with a clearer all-tiers diagnostic. Env and model
/// servers use this same function, so the verdict cannot drift between services.
pub fn evaluate_handshake(client_protocol_generation: &str) -> bool {
    is_protocol_generation_supported(client_protocol_generation)
}

/// The core handshake request this build sends as a gRPC client: its protocol
/// generation, the editions it CAN drive, the edition it declares (WANT), its
/// advertised capabilities, and PeerInfo. Shared by the env and model clients
/// (each wraps it in its service-specific request) so the request shape cannot
/// drift between services.
///
/// `declared` is the caller's pinned edition; `None` declares
/// [`CURRENT_WORKFLOW_EDITION`], which is `max(can)` for this build and therefore
/// leaves negotiation exactly where it was before the field existed.
pub fn core_handshake_request(
    component: &str,
    capabilities: &[&str],
    declared: Option<&str>,
) -> core::v1::HandshakeRequest {
    core::v1::HandshakeRequest {
        protocol_generation: PROTOCOL_GENERATION.to_string(),
        peer_info: Some(peer_info(component)),
        capabilities: capability_map(capabilities),
        supported_workflow_editions: supported_workflow_editions(),
        preferred_workflow_edition: declared_workflow_edition(declared).to_string(),
    }
}

/// The human-facing rejection message when a client's protocol generation does not
/// match this server. Generation is the ONLY handshake-level rejection; an edition
/// mismatch is decided — and diagnosed — by the runtime's floor, not here. Shared by
/// the env and model servers so the rejection prose cannot drift.
pub fn generation_mismatch_message(client_protocol_generation: &str) -> String {
    format!(
        "protocol generation {client_protocol_generation} not compatible with server \
         {PROTOCOL_GENERATION}"
    )
}

/// Return supported workflow editions as owned strings for protobuf messages.
pub fn supported_workflow_editions() -> Vec<String> {
    SUPPORTED_WORKFLOW_EDITIONS
        .iter()
        .map(|edition| (*edition).to_string())
        .collect()
}

/// Return a handshake capability map for the given capability names.
///
/// # Value grammar
///
/// A handshake capability map is keyed by capability name; the value is the
/// literal string `"true"` and nothing else is defined by this protocol
/// generation. Every RLMesh emitter goes through this function, so every value
/// RLMesh puts on the wire is `"true"`, and [`has_capability`] is the matching
/// reader: a key whose value is anything else — `"1"`, `"yes"`, `"TRUE"`, or
/// the empty string — reads as NOT advertised, exactly like an absent key. A
/// third-party peer that wants a capability honored must therefore send
/// `"true"` verbatim.
///
/// Capabilities are advisory, so this strictness is safe in both directions:
/// the worst case of an unrecognized value is the feature staying off.
pub fn capability_map(names: &[&str]) -> HashMap<String, String> {
    names
        .iter()
        .map(|name| ((*name).to_string(), "true".to_string()))
        .collect()
}

/// Whether a peer's handshake capability map advertises the named capability.
///
/// Present means the key is mapped to the literal `"true"` (see
/// [`capability_map`] for the value grammar). An absent key, an empty value, or
/// any other value is read as absent — never as present — so an advisory
/// feature a peer spelled differently stays off rather than being negotiated on
/// a guess.
pub fn has_capability(map: &HashMap<String, String>, name: &str) -> bool {
    map.get(name).is_some_and(|value| value == "true")
}

pub mod core {
    pub mod v1 {
        tonic::include_proto!("rlmesh.core.v1");
    }
}

/// Advisory runtime identity supplied by a non-Rust host (e.g. the Python SDK)
/// to enrich the handshake [`PeerInfo`](core::v1::PeerInfo).
///
/// Every field is optional and best-effort. When set process-wide via
/// [`set_peer_info_override`], [`peer_info`] merges these values over the
/// Rust-detected defaults: a non-empty override field wins, an empty/absent one
/// falls back to the Rust-detected value (`os`/`arch`/`package_version`). The
/// `component` passed to [`peer_info`] is always honored; an override
/// `component` is ignored so each call site keeps naming itself.
///
/// This is purely additive diagnostics: PeerInfo never gates compatibility.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PeerInfoOverride {
    /// Implementation language, e.g. `"python"`. Empty leaves the Rust default.
    pub language: String,
    /// Language runtime version, e.g. `"3.11.4"`.
    pub language_version: String,
    /// Package/build version of the host SDK. Empty falls back to this crate's.
    pub package_version: String,
    /// Operating system, e.g. `"linux"`, `"macos"`. Empty falls back to the
    /// Rust-detected [`std::env::consts::OS`].
    pub os: String,
    /// OS version/release.
    pub os_version: String,
    /// CPU architecture, e.g. `"x86_64"`. Empty falls back to the Rust-detected
    /// [`std::env::consts::ARCH`].
    pub arch: String,
    /// High-value framework versions for debugging (e.g. `{"numpy":"1.26.4"}`).
    pub framework_versions: HashMap<String, String>,
    /// Additional advisory key/value diagnostics.
    pub extra: HashMap<String, String>,
}

/// Process-wide host identity override consulted by [`peer_info`].
///
/// `None` (the default) means no override: pure-Rust peers handshake exactly as
/// before. A Python-hosted process sets this once at import; the value applies
/// to every handshake the process performs (it is the only host).
static PEER_INFO_OVERRIDE: RwLock<Option<PeerInfoOverride>> = RwLock::new(None);

/// Install (or replace) the process-wide [`PeerInfoOverride`] consulted by
/// [`peer_info`]. Intended for non-Rust hosts (the Python SDK) to report their
/// real runtime. Idempotent and thread-safe; passing the value again overwrites.
pub fn set_peer_info_override(info: PeerInfoOverride) {
    if let Ok(mut guard) = PEER_INFO_OVERRIDE.write() {
        *guard = Some(info);
    }
}

/// Build advisory [`PeerInfo`](core::v1::PeerInfo) diagnostics for a handshake.
///
/// `component` names the emitting participant (e.g. `"rlmesh-runtime"`,
/// `"rlmesh-env"`, `"rlmesh-model"`). The build version, language, OS and arch
/// default to this crate's compile environment (`language="rust"`, empty
/// `language_version`/`os_version`/`framework_versions`).
///
/// When a process-wide [`PeerInfoOverride`] has been installed via
/// [`set_peer_info_override`] (e.g. by the Python SDK), its non-empty fields win
/// over the Rust defaults, falling back to the Rust-detected
/// `os`/`arch`/`package_version` for any empty override field. The `component`
/// argument is always preserved so each call site keeps naming itself. PeerInfo
/// is advisory only and never gates compatibility.
pub fn peer_info(component: &str) -> core::v1::PeerInfo {
    let mut extra = HashMap::new();
    extra.insert(
        "rlmesh.workflow.base".to_string(),
        WORKFLOW_EDITION_BASE.to_string(),
    );
    extra.insert(
        "rlmesh.workflow.edition".to_string(),
        CURRENT_WORKFLOW_EDITION.to_string(),
    );
    extra.insert("rlmesh.build.cohort".to_string(), BUILD_COHORT.to_string());
    extra.insert("rlmesh.build.source".to_string(), BUILD_SOURCE.to_string());

    let mut info = core::v1::PeerInfo {
        component: component.to_string(),
        package_version: env!("CARGO_PKG_VERSION").to_string(),
        language: "rust".to_string(),
        language_version: String::new(),
        os: std::env::consts::OS.to_string(),
        os_version: String::new(),
        arch: std::env::consts::ARCH.to_string(),
        framework_versions: HashMap::new(),
        extra,
    };

    if let Ok(guard) = PEER_INFO_OVERRIDE.read()
        && let Some(over) = guard.as_ref()
    {
        // Python (or other host) values win when present; empty fields keep the
        // Rust-detected fallback. `component` is never overridden.
        if !over.language.is_empty() {
            info.language = over.language.clone();
        }
        if !over.language_version.is_empty() {
            info.language_version = over.language_version.clone();
        }
        if !over.package_version.is_empty() {
            info.package_version = over.package_version.clone();
        }
        if !over.os.is_empty() {
            info.os = over.os.clone();
        }
        if !over.os_version.is_empty() {
            info.os_version = over.os_version.clone();
        }
        if !over.arch.is_empty() {
            info.arch = over.arch.clone();
        }
        if !over.framework_versions.is_empty() {
            info.framework_versions = over.framework_versions.clone();
        }
        if !over.extra.is_empty() {
            info.extra.extend(over.extra.clone());
        }
    }

    info
}

pub mod env {
    pub mod v1 {
        tonic::include_proto!("rlmesh.env.v1");
    }
}

pub mod spaces {
    pub mod v1 {
        tonic::include_proto!("rlmesh.spaces.v1");
    }
}

pub mod model {
    pub mod v1 {
        tonic::include_proto!("rlmesh.model.v1");
    }
}

/// The endpoint-local split of `JoinResponse.endpoint_total_ns` (ns), plus the
/// pre-handler wait and slot depth a pipelining model endpoint reports.
///
/// A zero duration is "not measured": a peer built before these fields existed
/// sends none of them, and a reader sees this default. The gauges are `Option`
/// because for them a measured zero is a real sample (an even vector has zero
/// skew; an engine that just evicted holds zero episodes) that must stay
/// distinguishable from a peer that never measures. `in_flight` is model-only;
/// `lane_skew_ns` is env-only.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct EndpointPhases {
    /// Decoding the request off the wire.
    pub decode_ns: u64,
    /// The env or model implementation's own work, user code included.
    pub user_ns: u64,
    /// Encoding the response onto the wire.
    pub encode_ns: u64,
    /// Wait before the handler ran. Model: concurrency permit, route gate,
    /// handler lock. Env: the env lock, which serializes ops on one env.
    pub queue_ns: u64,
    /// Adapter work inside `user_ns` (observation assembly + action apply) —
    /// a sub-span, not additive: the model's own forward is `user_ns -
    /// adapter_ns`. Model-only, and zero for a spec-less route.
    pub adapter_ns: u64,
    /// Episodes whose frame-stack windows the endpoint's adapter engine held
    /// when the response was stamped, across all routes. Model-only; `None`
    /// from a handler that keeps no such accounting.
    pub held_episodes: Option<u32>,
    /// Bytes those held frame-stack windows occupy. Model-only; `None` from a
    /// handler that keeps no such accounting.
    pub held_state_bytes: Option<u64>,
    /// Requests holding a concurrency slot when this request was dispatched to
    /// its handler (>= 1): slot occupancy, not parallelism.
    pub in_flight: u32,
    /// Straggler skew across a vector env's lanes for this op — see
    /// [`lane_skew_ns`]. Env-only; `None` from an env that does not time its
    /// lanes, `Some(0)` from one whose lanes were even.
    pub lane_skew_ns: Option<u64>,
}

impl EndpointPhases {
    /// Wire form of a measurement: an unmeasured (zero) phase is left off the
    /// message rather than sent as a `0` a reader cannot tell from silence.
    pub fn reported(ns: u64) -> Option<u64> {
        (ns != 0).then_some(ns)
    }

    /// Build a split from measured phase durations, saturating at `u64::MAX` ns.
    pub fn from_durations(
        decode: std::time::Duration,
        user: std::time::Duration,
        encode: std::time::Duration,
    ) -> Self {
        fn ns(duration: std::time::Duration) -> u64 {
            duration.as_nanos().min(u128::from(u64::MAX)) as u64
        }
        Self {
            decode_ns: ns(decode),
            user_ns: ns(user),
            encode_ns: ns(encode),
            ..Self::default()
        }
    }

    /// Fold an inner layer's own split into this layer's decode/call/encode
    /// measurement. The inner layer's decode and encode accumulate into the
    /// outer ones, and `user_ns` keeps the rest of the call — the inner user
    /// work plus whatever it cost to reach it (GIL wait, adapter overhead) —
    /// so the phases still sum to this layer's whole measurement. An inner
    /// layer that measures nothing, or whose wire share does not even fit
    /// inside the measured call, leaves the whole call as user time.
    pub fn nest(decode_ns: u64, call_ns: u64, encode_ns: u64, inner: Self) -> Self {
        // Lane skew is the inner env's alone; an outer layer never measures it.
        let inner_wire = inner.decode_ns.saturating_add(inner.encode_ns);
        if inner_wire > call_ns {
            return Self {
                decode_ns,
                user_ns: call_ns,
                encode_ns,
                lane_skew_ns: inner.lane_skew_ns,
                ..Self::default()
            };
        }
        Self {
            decode_ns: decode_ns.saturating_add(inner.decode_ns),
            user_ns: call_ns - inner_wire,
            encode_ns: encode_ns.saturating_add(inner.encode_ns),
            lane_skew_ns: inner.lane_skew_ns,
            ..Self::default()
        }
    }

    /// Read the split an env peer stamped; all-zero for a peer that sends none.
    pub fn from_env_response(response: &env::v1::JoinResponse) -> Self {
        Self {
            decode_ns: response.decode_ns.unwrap_or(0),
            user_ns: response.user_ns.unwrap_or(0),
            encode_ns: response.encode_ns.unwrap_or(0),
            queue_ns: response.queue_ns.unwrap_or(0),
            lane_skew_ns: response.lane_skew_ns,
            ..Self::default()
        }
    }

    /// Read the split a model peer stamped; all-zero for a peer that sends none.
    pub fn from_model_response(response: &model::v1::JoinResponse) -> Self {
        Self {
            decode_ns: response.decode_ns.unwrap_or(0),
            user_ns: response.user_ns.unwrap_or(0),
            encode_ns: response.encode_ns.unwrap_or(0),
            queue_ns: response.queue_ns.unwrap_or(0),
            in_flight: response.in_flight.unwrap_or(0),
            adapter_ns: response.adapter_ns.unwrap_or(0),
            held_episodes: response.held_episodes,
            held_state_bytes: response.held_state_bytes,
            ..Self::default()
        }
    }
}

/// Nanoseconds elapsed since `started_at`, saturating instead of wrapping.
pub fn elapsed_ns(started_at: std::time::Instant) -> u64 {
    started_at.elapsed().as_nanos().min(u128::from(u64::MAX)) as u64
}

/// Straggler skew across a vector env's lanes for one op: how much longer the
/// slowest lane took than the median lane, in nanoseconds.
///
/// One scalar per op, so a 32-lane env costs one metric rather than 32 series,
/// and the aggregator's p50/p95/p99 over it answer how often and how badly a
/// single lane gates the batch the other lanes ride in. A shifted-but-even
/// vector reads zero; one bad lane among healthy ones reads its full excess.
///
/// `lane_ns` is reordered in place and never grown — pass a scratch buffer the
/// env keeps across steps. Fewer than two lanes have nothing to compare and
/// read zero. The median is the lower one for an even lane count, so a
/// two-lane vector still reports its straggler.
pub fn lane_skew_ns(lane_ns: &mut [u64]) -> u64 {
    if lane_ns.len() < 2 {
        return 0;
    }
    let (_, median, slower) = lane_ns.select_nth_unstable((lane_ns.len() - 1) / 2);
    let median = *median;
    slower
        .iter()
        .copied()
        .max()
        .unwrap_or(median)
        .saturating_sub(median)
}

/// A completing lane's final info from a step's `infos`, `lane` being its
/// position among the `width` lanes the step covered. An explicit
/// `final_info` entry wins (masked by `_final_info`). Otherwise a single-lane
/// step's info IS the final info, and a vector step's lane is sliced out of the
/// Gymnasium vector-info layout. `None` when the lane has nothing.
pub fn lane_final_info(
    info: Option<&spaces::v1::MetaMap>,
    lane: usize,
    width: usize,
) -> Option<spaces::v1::MetaMap> {
    use spaces::v1::meta_value::Kind;
    let info = info?;
    let Some(final_info) = info.entries.get("final_info") else {
        if width == 1 {
            return Some(info.clone());
        }
        let sliced = vector_info_lane(info, lane, width);
        return (!sliced.entries.is_empty()).then_some(sliced);
    };
    let is_present = match info.entries.get("_final_info") {
        Some(mask) => meta_bool_at(mask, lane).unwrap_or(false),
        None => width == 1,
    };
    if !is_present {
        return None;
    }
    match &final_info.kind {
        Some(Kind::Map(map)) => Some(map.clone()),
        Some(Kind::List(list)) => match &list.items.get(lane)?.kind {
            Some(Kind::Map(map)) => Some(map.clone()),
            _ => None,
        },
        _ => None,
    }
}

/// One lane of a Gymnasium vector info: each `key` holds a per-lane array or a
/// nested info map, and its `_key` mask marks the lanes that actually set it.
fn vector_info_lane(info: &spaces::v1::MetaMap, lane: usize, width: usize) -> spaces::v1::MetaMap {
    use spaces::v1::meta_value::Kind;
    let entries = info
        .entries
        .iter()
        .filter(|(key, _)| {
            !key.strip_prefix('_')
                .is_some_and(|masked| info.entries.contains_key(masked))
        })
        .filter(|(key, _)| {
            info.entries
                .get(&format!("_{key}"))
                .is_none_or(|mask| meta_bool_at(mask, lane) == Some(true))
        })
        .filter_map(|(key, value)| {
            let lane_value = match &value.kind {
                Some(Kind::Map(map)) => spaces::v1::MetaValue {
                    kind: Some(Kind::Map(vector_info_lane(map, lane, width))),
                },
                Some(Kind::List(list)) if list.items.len() == width => list.items[lane].clone(),
                _ => return None,
            };
            Some((key.clone(), lane_value))
        })
        .collect();
    spaces::v1::MetaMap { entries }
}

fn meta_bool_at(value: &spaces::v1::MetaValue, lane: usize) -> Option<bool> {
    use spaces::v1::meta_value::Kind;
    match &value.kind {
        Some(Kind::List(list)) => match list.items.get(lane)?.kind {
            Some(Kind::Bool(flag)) => Some(flag),
            _ => None,
        },
        _ => None,
    }
}

/// The `rlmesh.toml` array scan `build.rs` reads the retained edition list
/// with, compiled into the test build so `cargo test` covers the real parser —
/// a `#[test]` inside a build script never runs.
#[cfg(test)]
#[path = "../build_manifest.rs"]
mod build_manifest;

#[cfg(test)]
mod tests {
    use super::{
        CURRENT_WORKFLOW_EDITION, PROTOCOL_GENERATION, SUPPORTED_WORKFLOW_EDITIONS, SessionOffer,
        evaluate_handshake, is_protocol_generation_supported, negotiate_session_floor,
        negotiate_workflow_edition, supported_workflow_editions,
    };

    /// One tier of a negotiation case: what it CAN drive, and what it WANTs
    /// (`None` = undeclared, which negotiation reads as `max(can)`).
    fn tier(can: &[&str], want: Option<&str>) -> SessionOffer {
        SessionOffer {
            editions: offer(can),
            preferred: want.map(str::to_string),
        }
    }

    #[test]
    fn lane_final_info_reads_each_gymnasium_info_layout() {
        use super::lane_final_info;
        use super::spaces::v1::meta_value::Kind;
        use super::spaces::v1::{MetaList, MetaMap, MetaValue};

        let value = |kind: Kind| MetaValue { kind: Some(kind) };
        let list = |items: Vec<MetaValue>| value(Kind::List(MetaList { items }));
        let map = |entries: Vec<(&str, MetaValue)>| MetaMap {
            entries: entries
                .into_iter()
                .map(|(key, value)| (key.to_string(), value))
                .collect(),
        };
        let success = |flag: bool| map(vec![("is_success", value(Kind::Bool(flag)))]);

        let scalar = success(false);
        assert_eq!(lane_final_info(Some(&scalar), 0, 1), Some(success(false)));
        assert_eq!(lane_final_info(None, 0, 2), None);

        let explicit = map(vec![
            (
                "final_info",
                list(vec![
                    value(Kind::Map(success(true))),
                    value(Kind::Map(MetaMap::default())),
                ]),
            ),
            (
                "_final_info",
                list(vec![value(Kind::Bool(true)), value(Kind::Bool(false))]),
            ),
        ]);
        assert_eq!(lane_final_info(Some(&explicit), 0, 2), Some(success(true)));
        assert_eq!(lane_final_info(Some(&explicit), 1, 2), None);

        // No masks: a lane-wide array is still per-lane.
        let unmasked = map(vec![(
            "is_success",
            list(vec![value(Kind::Bool(false)), value(Kind::Bool(true))]),
        )]);
        assert_eq!(lane_final_info(Some(&unmasked), 1, 2), Some(success(true)));

        let masked_out = map(vec![
            (
                "is_success",
                list(vec![value(Kind::Bool(true)), value(Kind::Bool(false))]),
            ),
            (
                "_is_success",
                list(vec![value(Kind::Bool(true)), value(Kind::Bool(false))]),
            ),
        ]);
        assert_eq!(lane_final_info(Some(&masked_out), 1, 2), None);
    }

    #[test]
    fn phases_from_a_peer_that_stamps_nothing_read_back_as_zero() {
        use super::{EndpointPhases, env, model};

        // An older peer sends neither the split nor the queue scalars.
        let env_response = env::v1::JoinResponse {
            endpoint_total_ns: Some(1_000),
            ..Default::default()
        };
        assert_eq!(
            EndpointPhases::from_env_response(&env_response),
            EndpointPhases::default()
        );

        let model_response = model::v1::JoinResponse {
            endpoint_total_ns: Some(1_000),
            ..Default::default()
        };
        assert_eq!(
            EndpointPhases::from_model_response(&model_response),
            EndpointPhases::default()
        );

        // A peer that does stamp them reads back verbatim.
        let stamped = model::v1::JoinResponse {
            endpoint_total_ns: Some(1_000),
            decode_ns: Some(10),
            user_ns: Some(20),
            encode_ns: Some(30),
            queue_ns: Some(40),
            in_flight: Some(3),
            adapter_ns: Some(15),
            held_episodes: Some(5),
            held_state_bytes: Some(6_000),
            ..Default::default()
        };
        assert_eq!(
            EndpointPhases::from_model_response(&stamped),
            EndpointPhases {
                decode_ns: 10,
                user_ns: 20,
                encode_ns: 30,
                queue_ns: 40,
                in_flight: 3,
                adapter_ns: 15,
                held_episodes: Some(5),
                held_state_bytes: Some(6_000),
                lane_skew_ns: None,
            }
        );

        // Lane skew rides the env response only.
        let skewed = env::v1::JoinResponse {
            endpoint_total_ns: Some(1_000),
            lane_skew_ns: Some(880),
            ..Default::default()
        };
        assert_eq!(
            EndpointPhases::from_env_response(&skewed).lane_skew_ns,
            Some(880)
        );
        // A measured zero (an even vector) is a sample, not an absence.
        let even = env::v1::JoinResponse {
            lane_skew_ns: Some(0),
            ..Default::default()
        };
        assert_eq!(
            EndpointPhases::from_env_response(&even).lane_skew_ns,
            Some(0)
        );

        // An unmeasured phase is left off the wire rather than sent as a zero.
        assert_eq!(EndpointPhases::reported(0), None);
        assert_eq!(EndpointPhases::reported(7), Some(7));
    }

    #[test]
    fn nesting_folds_an_inner_split_into_the_outer_one() {
        use super::EndpointPhases;

        // An inner layer that measures nothing leaves the whole call as user time.
        assert_eq!(
            EndpointPhases::nest(1, 10, 2, EndpointPhases::default()),
            EndpointPhases {
                decode_ns: 1,
                user_ns: 10,
                encode_ns: 2,
                ..EndpointPhases::default()
            }
        );

        // One that does splits the call: its own decode/encode join this layer's,
        // and user keeps the remainder of the call.
        let inner = EndpointPhases {
            decode_ns: 3,
            user_ns: 5,
            encode_ns: 2,
            ..EndpointPhases::default()
        };
        assert_eq!(
            EndpointPhases::nest(1, 10, 2, inner),
            EndpointPhases {
                decode_ns: 4,
                user_ns: 5,
                encode_ns: 4,
                ..EndpointPhases::default()
            }
        );

        // A call longer than the inner layer's own accounting (GIL wait, adapter
        // overhead) keeps the residual in user, so the phases still sum to this
        // layer's whole measurement: 1 + 14 + 2 == 4 + 9 + 4.
        assert_eq!(
            EndpointPhases::nest(1, 14, 2, inner),
            EndpointPhases {
                decode_ns: 4,
                user_ns: 9,
                encode_ns: 4,
                ..EndpointPhases::default()
            }
        );

        // An inner split whose wire share does not fit inside the measured call
        // is inconsistent; folding it would break the sum, so none of it folds.
        assert_eq!(
            EndpointPhases::nest(1, 4, 2, inner),
            EndpointPhases {
                decode_ns: 1,
                user_ns: 4,
                encode_ns: 2,
                ..EndpointPhases::default()
            }
        );
    }

    #[test]
    fn lane_skew_is_the_slowest_lane_over_the_median_one() {
        use super::lane_skew_ns;

        // One straggler among healthy lanes: its full excess over a typical lane.
        assert_eq!(lane_skew_ns(&mut [10, 10, 10, 10, 10, 10, 10, 900]), 890);
        // An evenly slow vector is not a straggler — the whole distribution moved,
        // which `endpoint.user` already shows.
        assert_eq!(lane_skew_ns(&mut [900; 8]), 0);
        // Two lanes compare against the faster one, so the straggler still reads.
        assert_eq!(lane_skew_ns(&mut [10, 100]), 90);
        // A lane faster than its peers is not skew.
        assert_eq!(lane_skew_ns(&mut [1, 100, 100, 100]), 0);
        // Nothing to compare.
        assert_eq!(lane_skew_ns(&mut [42]), 0);
        assert_eq!(lane_skew_ns(&mut []), 0);
    }

    #[test]
    fn nesting_keeps_the_inner_envs_lane_skew() {
        use super::EndpointPhases;

        let measured = EndpointPhases {
            user_ns: 5,
            lane_skew_ns: Some(77),
            ..EndpointPhases::default()
        };
        assert_eq!(
            EndpointPhases::nest(1, 10, 2, measured).lane_skew_ns,
            Some(77)
        );

        // An env that times its lanes but reports no split of its own still gets
        // the skew through.
        let skew_only = EndpointPhases {
            lane_skew_ns: Some(77),
            ..EndpointPhases::default()
        };
        assert_eq!(
            EndpointPhases::nest(1, 10, 2, skew_only).lane_skew_ns,
            Some(77)
        );
    }

    fn offer(editions: &[&str]) -> Vec<String> {
        editions.iter().map(|edition| edition.to_string()).collect()
    }

    #[test]
    fn peer_info_default_then_override_merges_python_with_rust_fallback() {
        use super::{PeerInfoOverride, peer_info, set_peer_info_override};
        use std::collections::HashMap;

        // No override installed yet: a pure-Rust peer reports the Rust defaults.
        let rust_info = peer_info("rlmesh-env");
        assert_eq!(rust_info.component, "rlmesh-env");
        assert_eq!(rust_info.language, "rust");
        assert!(rust_info.language_version.is_empty());
        assert!(rust_info.framework_versions.is_empty());
        let detected_os = rust_info.os.clone();
        let detected_arch = rust_info.arch.clone();
        let detected_pkg = rust_info.package_version.clone();

        // Install a Python-style override with `os`/`package_version` left empty
        // so the Rust-detected fallbacks fill them.
        let mut frameworks = HashMap::new();
        frameworks.insert("numpy".to_string(), "1.26.4".to_string());
        set_peer_info_override(PeerInfoOverride {
            language: "python".to_string(),
            language_version: "3.11.4".to_string(),
            package_version: String::new(),
            os: String::new(),
            os_version: "ubuntu-22.04".to_string(),
            arch: "aarch64".to_string(),
            framework_versions: frameworks,
            extra: HashMap::from([("rlmesh.startup.listen_ms".to_string(), "4200".to_string())]),
        });

        let py_info = peer_info("rlmesh-env");
        // Host-supplied extras ride beside the build keys, never replace them.
        assert_eq!(
            py_info
                .extra
                .get("rlmesh.startup.listen_ms")
                .map(String::as_str),
            Some("4200")
        );
        assert!(py_info.extra.contains_key("rlmesh.build.cohort"));
        // component still names this call site; not taken from the override.
        assert_eq!(py_info.component, "rlmesh-env");
        // Python values win.
        assert_eq!(py_info.language, "python");
        assert_eq!(py_info.language_version, "3.11.4");
        assert_eq!(py_info.os_version, "ubuntu-22.04");
        assert_eq!(py_info.arch, "aarch64");
        assert_eq!(
            py_info.framework_versions.get("numpy").map(String::as_str),
            Some("1.26.4")
        );
        // Empty override fields fall back to the Rust-detected values.
        assert_eq!(py_info.os, detected_os);
        assert_eq!(py_info.package_version, detected_pkg);
        assert_eq!(
            py_info
                .extra
                .get("rlmesh.workflow.edition")
                .map(String::as_str),
            Some(CURRENT_WORKFLOW_EDITION)
        );
        // `arch` was overridden, so it differs from the detected value here.
        let _ = detected_arch;
    }

    #[test]
    fn has_capability_reads_advertised_features() {
        use super::{capabilities, capability_map, has_capability};
        let map = capability_map(&[capabilities::MODEL_CONCURRENT_PREDICT_V1]);
        assert!(has_capability(
            &map,
            capabilities::MODEL_CONCURRENT_PREDICT_V1
        ));
        assert!(!has_capability(&map, "rlmesh.not.advertised.v1"));
        // The documented value grammar: emitters write the literal "true", and
        // any other spelling reads as absent rather than as a guessed "on".
        assert_eq!(map[capabilities::MODEL_CONCURRENT_PREDICT_V1], "true");
        for value in ["1", "yes", "TRUE", ""] {
            let odd =
                std::collections::HashMap::from([("rlmesh.odd.v1".to_string(), value.to_string())]);
            assert!(!has_capability(&odd, "rlmesh.odd.v1"), "{value:?}");
        }
    }

    #[test]
    fn env_subset_step_keeps_its_shipped_wire_spelling() {
        // Peers built against 0.1.0 advertise the bare `subset_step` key; the
        // constant renames the site, never the string on the wire.
        assert_eq!(super::capabilities::ENV_SUBSET_STEP, "subset_step");
    }

    #[test]
    fn protocol_generation_is_plain_equality() {
        // The only generation check is equality with this build's generation —
        // there is no support window. Whitespace is trimmed; anything else is a
        // hard mismatch (a deliberate major break).
        assert!(is_protocol_generation_supported(PROTOCOL_GENERATION));
        assert!(is_protocol_generation_supported(&format!(
            " {PROTOCOL_GENERATION} "
        )));
        assert!(!is_protocol_generation_supported("rlmesh-wire-v2"));
        assert!(!is_protocol_generation_supported(""));
        assert!(!is_protocol_generation_supported("0.1.0"));
    }

    #[test]
    fn split_editions_recovers_every_supported_edition() {
        // The build-time list is comma-separated; today's tree generates one
        // entry, so cover the retained-fallback shape here rather than waiting
        // for a second edition to be sealed.
        const LIST: &str = "2026.09-0.2.0-rc.1,2026.06";
        const EDITIONS: [&str; super::edition_count(LIST)] = super::split_editions(LIST);
        assert_eq!(EDITIONS, ["2026.09-0.2.0-rc.1", "2026.06"]);
        assert_eq!(super::edition_count("2026.06"), 1);
    }

    #[test]
    fn manifest_string_list_reads_both_array_spellings() {
        use super::build_manifest::manifest_string_list;

        // `bump_version.py` writes the single-line spelling, but a hand-edited
        // manifest may spread the array over lines: both must offer the same
        // editions, or a retained edition would silently vanish from the build.
        let single = "[workflow]\nsupported_editions = [\"2026.09-0.2.0-rc.1\", \"2026.06\"]\n";
        let multi = "[workflow]\nsupported_editions = [\n  \"2026.09-0.2.0-rc.1\", # cohort\n  \
                     \"2026.06\",\n]\n";
        let want = ["2026.09-0.2.0-rc.1".to_string(), "2026.06".to_string()];
        assert_eq!(manifest_string_list(single, "supported_editions"), want);
        assert_eq!(manifest_string_list(multi, "supported_editions"), want);
        assert!(manifest_string_list(single, "current_edition").is_empty());
    }

    #[test]
    fn supported_workflow_editions_lead_with_current() {
        // The list is generated from `rlmesh.toml`; whatever it retains, this
        // build's own cohort is first, nothing repeats, and the owned-string
        // form handed to protobuf mirrors it exactly.
        assert!(!SUPPORTED_WORKFLOW_EDITIONS.is_empty());
        assert_eq!(SUPPORTED_WORKFLOW_EDITIONS[0], CURRENT_WORKFLOW_EDITION);
        let unique: std::collections::BTreeSet<&&str> =
            SUPPORTED_WORKFLOW_EDITIONS.iter().collect();
        assert_eq!(unique.len(), SUPPORTED_WORKFLOW_EDITIONS.len());
        assert!(
            SUPPORTED_WORKFLOW_EDITIONS
                .iter()
                .all(|edition| !edition.trim().is_empty())
        );
        assert_eq!(
            supported_workflow_editions(),
            SUPPORTED_WORKFLOW_EDITIONS
                .iter()
                .map(|edition| (*edition).to_string())
                .collect::<Vec<_>>()
        );
    }

    #[test]
    fn negotiation_selects_mutual_edition() {
        let runtime = SessionOffer::this_build(None);
        assert_eq!(
            negotiate_workflow_edition(&SessionOffer::new(&[CURRENT_WORKFLOW_EDITION]), &runtime),
            Ok(CURRENT_WORKFLOW_EDITION.to_string())
        );
        assert_eq!(
            negotiate_workflow_edition(
                &SessionOffer::new(&["2025.01", CURRENT_WORKFLOW_EDITION, "2031.12"]),
                &runtime
            ),
            Ok(CURRENT_WORKFLOW_EDITION.to_string())
        );
    }

    #[test]
    fn negotiation_trims_offered_editions() {
        let padded = SessionOffer {
            editions: vec![format!(" {CURRENT_WORKFLOW_EDITION} ")],
            preferred: None,
        };
        assert_eq!(
            negotiate_workflow_edition(&padded, &SessionOffer::this_build(None)),
            Ok(CURRENT_WORKFLOW_EDITION.to_string())
        );
    }

    #[test]
    fn negotiation_rejects_unknown_or_empty_offers() {
        let runtime = SessionOffer::this_build(None);
        for offered in [
            &[][..],
            &[""][..],
            &["2026"][..],
            &["next"][..],
            &["2026.11", "2027.01"][..],
        ] {
            let refusal = negotiate_workflow_edition(&SessionOffer::new(offered), &runtime)
                .expect_err("no mutual edition");
            // The refusal names both tiers and both of their sets.
            let message = refusal.to_string();
            assert!(message.contains("env"), "{message}");
            assert!(message.contains("runtime"), "{message}");
            assert!(message.contains(CURRENT_WORKFLOW_EDITION), "{message}");
        }
    }

    #[test]
    fn evaluate_handshake_gates_generation_only() {
        // The handshake decides ONE thing: protocol generation. Editions are the
        // runtime's call (the floor), so a generation-ok peer is compatible
        // regardless of editions — even with no mutual edition (it fails later at
        // the floor, with a clearer all-tiers message).
        assert!(evaluate_handshake(PROTOCOL_GENERATION));

        // A protocol mismatch is never compatible.
        assert!(!evaluate_handshake("rlmesh-wire-v2"));
    }

    #[test]
    fn is_retained_edition_matches_the_window() {
        use super::{Edition, is_retained_edition};
        let current = Edition::parse(CURRENT_WORKFLOW_EDITION).expect("the current edition parses");
        assert!(is_retained_edition(current));
        // Trimmed, and this build's cohort spelling names the same arm as its base.
        assert_eq!(
            Edition::parse(&format!("  {CURRENT_WORKFLOW_EDITION}  ")),
            Ok(current)
        );
        assert_eq!(Edition::parse(super::WORKFLOW_EDITION_BASE), Ok(current));
        // An edition no arm implements never resolves, so it is never retained.
        assert!(Edition::parse("2099.01").is_err());
        assert!(Edition::parse("").is_err());
    }

    /// The base names the contract; a cohort suffix only identifies a moving
    /// build of it. So EVERY cohort spelling of a known base resolves to that
    /// base's arm -- including one this build never spells itself, which is what a
    /// policy-legal manifest retaining an older base's cohort would carry.
    #[test]
    fn parse_resolves_any_cohort_spelling_of_a_known_base() {
        use super::{Edition, WORKFLOW_EDITION_BASE, parse_retained_edition};

        let current = Edition::parse(CURRENT_WORKFLOW_EDITION).expect("the current edition parses");
        assert_eq!(Edition::parse(WORKFLOW_EDITION_BASE), Ok(current));
        assert_eq!(Edition::parse(CURRENT_WORKFLOW_EDITION), Ok(current));
        for foreign in [
            format!("{WORKFLOW_EDITION_BASE}-0.0.1-rc.1"),
            format!("{WORKFLOW_EDITION_BASE}-dev.deadbeef"),
        ] {
            assert_eq!(
                Edition::parse(&foreign),
                Ok(current),
                "{foreign} is a cohort of a known base"
            );
        }
        // A base no arm implements is refused, bare or suffixed.
        assert!(Edition::parse("2099.01").is_err());
        assert!(Edition::parse("2099.01-0.1.0-rc.12").is_err());

        // The shared boundary refuses the same names, naming the arrived string
        // and the retained list.
        assert_eq!(
            parse_retained_edition(CURRENT_WORKFLOW_EDITION),
            Ok(current)
        );
        let error = parse_retained_edition(" 2099.01 ").expect_err("no arm implements it");
        assert!(
            error.contains("\"2099.01\"") && error.contains(CURRENT_WORKFLOW_EDITION),
            "expected the arrived name and the retained list, got: {error}"
        );
    }

    /// A declaration is a ceiling, so it is only usable when it admits something
    /// this build offers. Every name this build offers is declarable by
    /// construction, and so is the bare base on EVERY build — it admits every
    /// cohort of itself, which is what makes it the value to write down. A
    /// cohort that would leave the ceiling below the whole CAN set is refused
    /// here rather than deadlocking negotiation.
    #[test]
    fn a_declaration_must_leave_this_build_something_to_run() {
        use super::{Edition, WORKFLOW_EDITION_BASE, parse_declared_edition};

        let current = Edition::parse(CURRENT_WORKFLOW_EDITION).expect("the current edition parses");
        for offered in SUPPORTED_WORKFLOW_EDITIONS {
            assert_eq!(
                parse_declared_edition(offered),
                Ok(current),
                "{offered} is offered, so it is declarable"
            );
        }
        assert_eq!(parse_declared_edition(WORKFLOW_EDITION_BASE), Ok(current));
        assert_eq!(
            parse_declared_edition(&format!(" {WORKFLOW_EDITION_BASE} ")),
            Ok(current)
        );
        // A cohort of this base that sorts below everything offered: on a
        // sealed build the bare base is offered and admitted, on a prerelease
        // or dev build only a higher cohort is, so it is refused naming both
        // halves and the base to declare instead.
        let stale_cohort = format!("{WORKFLOW_EDITION_BASE}-0.0.0");
        match parse_declared_edition(&stale_cohort) {
            Ok(edition) => {
                assert_eq!(edition, current);
                assert!(SUPPORTED_WORKFLOW_EDITIONS.contains(&WORKFLOW_EDITION_BASE));
            }
            Err(error) => {
                assert!(
                    error.contains(&stale_cohort)
                        && error.contains(CURRENT_WORKFLOW_EDITION)
                        && error.contains(&format!("{WORKFLOW_EDITION_BASE:?} base")),
                    "expected both halves of the mismatch, got: {error}"
                );
            }
        }
        // An unimplemented base still fails on the retained check, unchanged.
        assert!(parse_declared_edition("2099.01").is_err());
    }

    /// The two shapes of a WANT: a bare base is a base-level ceiling, a cohort
    /// spelling keeps the exact order.
    #[test]
    fn want_admits_by_base_when_bare_and_by_key_when_suffixed() {
        use super::want_admits;

        assert!(want_admits("2026.06", "2026.06"));
        assert!(want_admits("2026.06", "2026.06-dev.aaa"));
        assert!(want_admits("2026.06", "2026.06-0.1.0-rc.12"));
        assert!(want_admits("2026.06", "2026.01"));
        assert!(!want_admits("2026.06", "2026.08"));
        assert!(!want_admits("2026.06", "2026.08-dev.aaa"));

        assert!(want_admits("2026.06-dev.bbb", "2026.06-dev.bbb"));
        assert!(want_admits("2026.06-dev.bbb", "2026.06-dev.aaa"));
        assert!(want_admits("2026.06-dev.bbb", "2026.06"));
        assert!(!want_admits("2026.06-dev.bbb", "2026.06-dev.ccc"));
        assert!(!want_admits("2026.06-dev.bbb", "2026.08"));
    }

    /// Every arm has a defaults row, and every name this build offers or runs
    /// under resolves back to an arm. `defaults` is an exhaustive match, so the
    /// "has a row" half is enforced by the compiler; this pins the rest.
    #[test]
    fn every_arm_is_listed_and_has_a_row() {
        use super::{Edition, defaults, is_retained_edition};
        for edition in Edition::ALL {
            // Exhaustive: a new arm does not compile until it is named here, so
            // one cannot be added without visiting this test. `Edition::ALL`
            // itself is held complete by the retained-list loop below, which
            // parses every name this build offers back through it.
            match edition {
                Edition::E2026_06 => {
                    assert!(Edition::ALL.contains(&Edition::E2026_06));
                }
            }
            let row = defaults(*edition);
            assert!(!row.trial_index_option_key.is_empty());
            assert!(!row.conformance_warning_info_key.is_empty());
            assert_eq!(Edition::parse(edition.base()), Ok(*edition));
            assert_eq!(edition.to_string(), edition.base());
        }
        assert_eq!(
            Edition::parse(CURRENT_WORKFLOW_EDITION).map(is_retained_edition),
            Ok(true)
        );
        assert_eq!(
            Edition::parse(CURRENT_WORKFLOW_EDITION),
            Ok(Edition::current())
        );
        for retained in SUPPORTED_WORKFLOW_EDITIONS {
            let edition = Edition::parse(retained)
                .unwrap_or_else(|err| panic!("retained edition {retained:?} has no arm: {err}"));
            assert!(is_retained_edition(edition));
        }
    }

    #[test]
    fn session_floor_picks_highest_all_three_support() {
        // Highest edition all three share wins (ranked by edition_sort_key);
        // whitespace is trimmed so a padded edition still matches. When all three
        // reach the same top edition, the runtime is not limiting.
        let env = SessionOffer::new(&["2026.01", " 2026.06 "]);
        let model = SessionOffer::new(&["2026.06", "2026.01"]);
        let runtime = SessionOffer::new(&["2026.06"]);
        let floor = negotiate_session_floor(&env, &model, &runtime).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06");
        assert_eq!(floor.desired_workflow_edition, "2026.06");
        assert!(!floor.runtime_limited());
    }

    #[test]
    fn session_floor_flags_runtime_as_limiting_tier() {
        // env+model both reach 2026.08, but the runtime only speaks 2026.06, so the
        // floor drops to 2026.06 (safe — all three speak it) and the runtime is
        // flagged as the tier holding the session back.
        let env = SessionOffer::new(&["2026.06", "2026.08"]);
        let model = SessionOffer::new(&["2026.06", "2026.08"]);
        let runtime = SessionOffer::new(&["2026.06"]);
        let floor = negotiate_session_floor(&env, &model, &runtime).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06");
        assert_eq!(floor.desired_workflow_edition, "2026.08");
        assert!(floor.runtime_limited());
    }

    #[test]
    fn session_floor_is_none_when_no_common_edition() {
        // env+model agree on 2026.08 but the runtime can't speak it, and they can't
        // speak the runtime's 2026.06 → no edition all three share → None.
        let env = SessionOffer::new(&["2026.08"]);
        let model = SessionOffer::new(&["2026.08"]);
        let runtime = SessionOffer::new(&["2026.06"]);
        assert!(negotiate_session_floor(&env, &model, &runtime).is_err());

        // Empty strings never match, so an offer of only "" has no mutual value.
        let empty = SessionOffer::new(&[""]);
        let ok = SessionOffer::new(&["2026.06"]);
        assert!(negotiate_session_floor(&empty, &ok, &ok).is_err());
    }

    #[test]
    fn edition_ordering_prefers_exact_cohort_then_newer_date() {
        use super::edition_sort_key;

        // Exact moving cohorts beat their own sealed fallback. This lets two
        // matching prerelease/dev peers use the newest cohort while still allowing
        // fallback to the sealed edition when the moving cohorts differ.
        assert!(edition_sort_key("2026.06-0.1.0-rc.1") > edition_sort_key("2026.06"));

        // newer-date-wins: a newer date outranks an older one regardless of
        // cohort status.
        assert!(edition_sort_key("2026.09-0.2.0-beta.1") > edition_sort_key("2026.06"));
        assert!(edition_sort_key("2026.09") > edition_sort_key("2026.06-0.1.0-rc.1"));

        // deterministic suffix tiebreak: two same-date cohorts order by
        // their full suffix, never by iteration order, so two honest builds
        // never disagree on the winner.
        assert!(edition_sort_key("2026.06-0.1.0-rc.2") > edition_sort_key("2026.06-0.1.0-rc.1"));

        // negotiate_workflow_edition applies the same key: offered against a
        // hypothetical multi-edition supported set, the highest by key wins. With
        // the single supported edition this build ships, the current edition is
        // selected when offered alongside older/newer noise.
        assert_eq!(
            negotiate_workflow_edition(
                &SessionOffer::new(&["2025.01", CURRENT_WORKFLOW_EDITION, "2099.12"]),
                &SessionOffer::this_build(None)
            ),
            Ok(CURRENT_WORKFLOW_EDITION.to_string())
        );
    }

    #[test]
    fn session_floor_prefers_exact_edition_cohort_over_sealed_fallback() {
        // The floor uses edition_sort_key: when all three offer the exact moving
        // cohort and its sealed fallback, the exact cohort is selected.
        let editions = &["2026.06", "2026.06-0.1.0-rc.1"];
        let offer = SessionOffer::new(editions);
        let floor = negotiate_session_floor(&offer, &offer, &offer).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06-0.1.0-rc.1");
    }

    /// The WANT/CAN rule, end to end: two- and three-party, declared and not.
    ///
    /// Each case is `(name, [(tier, CAN, WANT)], expected selection)`; `None`
    /// expects a refusal that names every tier. Plain strings throughout —
    /// negotiation is string-level, and `Edition` typing happens at the boundary
    /// the selected name is then parsed through.
    #[test]
    fn want_can_selection_table() {
        type Tier<'a> = (&'a str, &'a [&'a str], Option<&'a str>);
        let cases: &[(&str, &[Tier<'_>], Option<&str>)] = &[
            // --- nothing declared: exactly the pre-WANT highest-mutual rule ---
            (
                "two-party, undeclared: highest mutual",
                &[
                    ("env", &["2026.01", "2026.06"], None),
                    ("runtime", &["2026.06"], None),
                ],
                Some("2026.06"),
            ),
            (
                "three-party, undeclared: highest all three share",
                &[
                    ("env", &["2026.01", " 2026.06 "], None),
                    ("model", &["2026.06", "2026.01"], None),
                    ("runtime", &["2026.06"], None),
                ],
                Some("2026.06"),
            ),
            (
                "three-party, undeclared: the runtime's CAN set caps the floor",
                &[
                    ("env", &["2026.06", "2026.08"], None),
                    ("model", &["2026.06", "2026.08"], None),
                    ("runtime", &["2026.06"], None),
                ],
                Some("2026.06"),
            ),
            (
                "three-party, undeclared: empty intersection refuses",
                &[
                    ("env", &["2026.08"], None),
                    ("model", &["2026.08"], None),
                    ("runtime", &["2026.06"], None),
                ],
                None,
            ),
            // --- a declared WANT is a ceiling ---
            (
                "one peer declares below the mutual max: the declared one wins",
                &[
                    ("env", &["2026.06", "2026.08"], Some("2026.06")),
                    ("model", &["2026.06", "2026.08"], None),
                    ("runtime", &["2026.06", "2026.08"], None),
                ],
                Some("2026.06"),
            ),
            (
                "a WANT above what another peer CAN never lifts the floor",
                &[
                    ("env", &["2026.06", "2026.08", "2026.10"], Some("2026.10")),
                    ("model", &["2026.06", "2026.08"], None),
                    ("runtime", &["2026.06", "2026.08", "2026.10"], None),
                ],
                Some("2026.08"),
            ),
            (
                "a WANT the intersection does not contain still selects the \
                 highest mutual below it: a pin is a ceiling, not an exact demand",
                &[
                    ("env", &["2026.06", "2026.08"], Some("2026.07")),
                    ("model", &["2026.06", "2026.08"], None),
                    ("runtime", &["2026.06", "2026.08"], None),
                ],
                Some("2026.06"),
            ),
            (
                "a WANT below everything mutual refuses rather than running higher",
                &[
                    ("env", &["2026.06"], Some("2025.01")),
                    ("model", &["2026.06"], None),
                    ("runtime", &["2026.06"], None),
                ],
                None,
            ),
            (
                "the lowest WANT wins when several are declared",
                &[
                    ("env", &["2026.06", "2026.08", "2026.10"], Some("2026.10")),
                    ("model", &["2026.06", "2026.08", "2026.10"], Some("2026.08")),
                    (
                        "runtime",
                        &["2026.06", "2026.08", "2026.10"],
                        Some("2026.10"),
                    ),
                ],
                Some("2026.08"),
            ),
            // --- dev cohorts vs their sealed base ---
            (
                "undeclared, matching dev cohorts: the exact cohort beats its base",
                &[
                    ("env", &["2026.06", "2026.06-dev.aaa"], None),
                    ("model", &["2026.06", "2026.06-dev.aaa"], None),
                    ("runtime", &["2026.06", "2026.06-dev.aaa"], None),
                ],
                Some("2026.06-dev.aaa"),
            ),
            (
                "differing dev cohorts do not match: both fall back to the sealed base",
                &[
                    ("env", &["2026.06", "2026.06-dev.aaa"], None),
                    ("model", &["2026.06", "2026.06-dev.bbb"], None),
                    ("runtime", &["2026.06", "2026.06-dev.aaa"], None),
                ],
                Some("2026.06"),
            ),
            (
                "a WANT of the bare base admits every cohort of it: the exact cohort still wins",
                &[
                    ("env", &["2026.06", "2026.06-dev.aaa"], Some("2026.06")),
                    ("model", &["2026.06", "2026.06-dev.aaa"], None),
                    ("runtime", &["2026.06", "2026.06-dev.aaa"], None),
                ],
                Some("2026.06-dev.aaa"),
            ),
            // --- a bare-base WANT against cohort-only builds (the stateVersion value) ---
            (
                "bare base WANT on both sides, cohort-only CANs: selects the cohort",
                &[
                    ("env", &["2026.06-dev.aaa"], Some("2026.06")),
                    ("runtime", &["2026.06-dev.aaa"], Some("2026.06")),
                ],
                Some("2026.06-dev.aaa"),
            ),
            (
                "bare base WANT on one side, three cohort-only tiers: selects the cohort",
                &[
                    ("env", &["2026.06-dev.aaa"], Some("2026.06")),
                    ("model", &["2026.06-dev.aaa"], None),
                    ("runtime", &["2026.06-dev.aaa"], None),
                ],
                Some("2026.06-dev.aaa"),
            ),
            (
                "bare base WANT admits its own cohorts but not a newer base's",
                &[
                    (
                        "env",
                        &["2026.06-dev.aaa", "2026.08-dev.aaa"],
                        Some("2026.06"),
                    ),
                    ("runtime", &["2026.06-dev.aaa", "2026.08-dev.aaa"], None),
                ],
                Some("2026.06-dev.aaa"),
            ),
            (
                "bare base WANT against a newer base only refuses",
                &[
                    ("env", &["2026.08"], Some("2026.06")),
                    ("model", &["2026.08"], None),
                    ("runtime", &["2026.08", "2026.08-dev.aaa"], None),
                ],
                None,
            ),
            (
                "cohort WANT against a differing cohort of the same base refuses",
                &[
                    ("env", &["2026.06-dev.aaa"], Some("2026.06-dev.aaa")),
                    ("runtime", &["2026.06-dev.bbb"], None),
                ],
                None,
            ),
            (
                "cohort WANT keeps the exact order: a higher cohort in the intersection is excluded",
                &[
                    ("env", &["2026.06-dev.bbb"], Some("2026.06-dev.aaa")),
                    ("runtime", &["2026.06-dev.bbb"], None),
                ],
                None,
            ),
            // --- a 0.1.0-shaped peer against a build that retains more ---
            (
                "a 0.1.0-shaped peer (one CAN, no WANT) pins a newer build to 2026.06",
                &[
                    ("env", &["2026.06"], None),
                    ("runtime", &["2026.06", "2099.01"], None),
                ],
                Some("2026.06"),
            ),
            (
                "the same peer against a runtime that declares the newer edition",
                &[
                    ("env", &["2026.06"], None),
                    ("model", &["2026.06"], None),
                    ("runtime", &["2026.06", "2099.01"], Some("2099.01")),
                ],
                Some("2026.06"),
            ),
            // --- degenerate sets ---
            (
                "a tier that CAN nothing refuses",
                &[("env", &[], None), ("runtime", &["2026.06"], None)],
                None,
            ),
        ];

        for (name, tiers, expected) in cases {
            let offers: Vec<(&str, SessionOffer)> = tiers
                .iter()
                .map(|(tier_name, can, want)| (*tier_name, tier(can, *want)))
                .collect();
            let borrowed: Vec<(&str, &SessionOffer)> = offers
                .iter()
                .map(|(tier_name, o)| (*tier_name, o))
                .collect();
            let selected = super::select_workflow_edition(&borrowed);
            match expected {
                Some(edition) => assert_eq!(selected.as_deref(), Ok(*edition), "{name}"),
                None => {
                    let refusal = selected.expect_err(name).to_string();
                    for (tier_name, can, _) in tiers.iter() {
                        assert!(refusal.contains(tier_name), "{name}: {refusal}");
                        for edition in can.iter() {
                            assert!(refusal.contains(edition), "{name}: {refusal}");
                        }
                    }
                }
            }
        }
    }

    /// With nothing declared anywhere, selection is bit-for-bit the pre-WANT
    /// rule: the WANT ceiling is `min(max(can))`, and every member of the
    /// intersection is already <= each tier's own max, so the filter is vacuous.
    ///
    /// `highest_mutual` below is the function this unit replaced, copied
    /// verbatim from `git show HEAD:crates/rlmesh-proto/src/lib.rs` (minus its
    /// generic `key`, which every caller instantiated with `edition_sort_key`).
    #[test]
    fn undeclared_everywhere_is_the_old_highest_mutual() {
        fn highest_mutual(sets: &[&[String]]) -> Option<String> {
            let (first, rest) = sets.split_first()?;
            first
                .iter()
                .map(|value| value.trim())
                .filter(|value| !value.is_empty())
                .filter(|value| {
                    rest.iter()
                        .all(|set| set.iter().any(|other| other.trim() == *value))
                })
                .max_by_key(|value| super::edition_sort_key(value))
                .map(|value| value.to_string())
        }

        let shapes: &[&[&[&str]]] = &[
            &[&["2026.06"], &["2026.06"]],
            &[&["2026.01", " 2026.06 "], &["2026.06", "2026.01"]],
            &[&["2025.01", "2026.06", "2031.12"], &["2026.06"]],
            &[&["2026.11", "2027.01"], &["2026.06"]],
            &[&[""], &["2026.06"]],
            &[&[], &["2026.06"]],
            &[
                &["2026.06", "2026.08"],
                &["2026.06", "2026.08"],
                &["2026.06"],
            ],
            &[&["2026.08"], &["2026.08"], &["2026.06"]],
            &[
                &["2026.06", "2026.06-0.1.0-rc.1"],
                &["2026.06", "2026.06-0.1.0-rc.1"],
                &["2026.06", "2026.06-0.1.0-rc.1"],
            ],
            &[
                &["2026.06", "2026.06-dev.aaa"],
                &["2026.06", "2026.06-dev.bbb"],
                &["2026.06", "2026.06-dev.aaa"],
            ],
            &[&["2026.06"], &["2026.06", "2099.01"]],
        ];

        for shape in shapes {
            let offers: Vec<SessionOffer> = shape.iter().map(|can| tier(can, None)).collect();
            let owned: Vec<Vec<String>> = shape.iter().map(|can| offer(can)).collect();
            let sets: Vec<&[String]> = owned.iter().map(Vec::as_slice).collect();
            let borrowed: Vec<(&str, &SessionOffer)> = offers.iter().map(|o| ("tier", o)).collect();
            assert_eq!(
                super::select_workflow_edition(&borrowed).ok(),
                highest_mutual(&sets),
                "{shape:?}"
            );
        }
    }

    /// The runtime holding the floor down is diagnosed by cause (plan E.3): a
    /// CAN-set limit is a defect the operator can fix by upgrading; a declared
    /// WANT is the sticky model working as intended.
    #[test]
    fn session_floor_reports_why_the_runtime_capped_the_session() {
        use super::RuntimeCap;

        // The runtime cannot drive 2026.08 at all.
        let peers = tier(&["2026.06", "2026.08"], None);
        let floor =
            negotiate_session_floor(&peers, &peers, &tier(&["2026.06"], None)).expect("a floor");
        assert_eq!(floor.runtime_cap, Some(RuntimeCap::Capability));

        // The runtime can drive 2026.08 and declares 2026.06 instead.
        let runtime = tier(&["2026.06", "2026.08"], Some("2026.06"));
        let floor = negotiate_session_floor(&peers, &peers, &runtime).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06");
        assert_eq!(floor.desired_workflow_edition, "2026.08");
        assert_eq!(floor.runtime_cap, Some(RuntimeCap::Declaration));

        // Both older AND declared below its own max: the declaration is what
        // pins the session (undeclared it would have run at 2026.08), so this is
        // not an upgrade-the-runtime diagnostic even though 2026.10 is out of
        // the runtime's reach entirely.
        let wide = tier(&["2026.06", "2026.08", "2026.10"], None);
        let runtime = tier(&["2026.06", "2026.08"], Some("2026.06"));
        let floor = negotiate_session_floor(&wide, &wide, &runtime).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06");
        assert_eq!(floor.desired_workflow_edition, "2026.10");
        assert_eq!(floor.runtime_cap, Some(RuntimeCap::Declaration));

        // A peer, not the runtime, holding the floor down is not a runtime cap.
        let pinned = tier(&["2026.06", "2026.08"], Some("2026.06"));
        let floor = negotiate_session_floor(&pinned, &peers, &peers).expect("a floor");
        assert_eq!(floor.selected_workflow_edition, "2026.06");
        assert_eq!(floor.desired_workflow_edition, "2026.06");
        assert_eq!(floor.runtime_cap, None);
        assert!(!floor.runtime_limited());
    }

    /// An undeclared participant declares this build's current edition, which is
    /// `max(can)` for this build — so this build's own offer never caps a peer.
    #[test]
    fn this_build_declares_its_current_edition_by_default() {
        use super::{core_handshake_request, declared_workflow_edition, edition_sort_key};

        // The no-op argument rests on this: the default WANT is `max(can)`, so a
        // retained edition sorting above the current one would make this build's
        // own offer cap its peers.
        assert_eq!(
            SUPPORTED_WORKFLOW_EDITIONS
                .iter()
                .max_by_key(|edition| edition_sort_key(edition))
                .copied(),
            Some(CURRENT_WORKFLOW_EDITION)
        );

        assert_eq!(declared_workflow_edition(None), CURRENT_WORKFLOW_EDITION);
        assert_eq!(
            declared_workflow_edition(Some("  ")),
            CURRENT_WORKFLOW_EDITION
        );
        assert_eq!(declared_workflow_edition(Some(" 2026.06 ")), "2026.06");

        let offer = SessionOffer::this_build(None);
        assert_eq!(offer.editions, supported_workflow_editions());
        assert_eq!(
            offer.preferred.as_deref(),
            Some(CURRENT_WORKFLOW_EDITION),
            "an undeclared build wants its current edition"
        );

        let request = core_handshake_request("rlmesh-env", &[], None);
        assert_eq!(request.preferred_workflow_edition, CURRENT_WORKFLOW_EDITION);
        assert_eq!(
            core_handshake_request("rlmesh-env", &[], Some("2026.06")).preferred_workflow_edition,
            "2026.06"
        );
    }
}