olai-codegen 0.0.14

//! PyO3 wrapper `#[pyclass]` generation for the protobuf-defined model types.
//!
//! ## Why wrappers
//!
//! Models are generated by `buf` (the buffa or prost plugin), not by trestle, so
//! we can neither add `#[pyclass]` to the struct definition nor a `#[derive]`.
//! An earlier approach implemented PyO3's [`FromPyObject`]/[`IntoPyObject`]
//! directly on the model types via [`pythonize`], so a model crossed the FFI
//! boundary as a plain Python **dict**. That compiled, but the Python side never
//! got *real objects*: no attribute access (`catalog.name`), no `isinstance`, no
//! `__repr__`, and enums arrived as bare proto-name strings.
//!
//! Instead we emit, for every project message, a frozen newtype `#[pyclass]` wrapper
//! (`PyCatalog(Catalog)`) with native, typed getters, a keyword constructor on the
//! models that a request takes, and `__repr__`/`__eq__`. The wrapper is *local* to
//! the model crate (orphan rule satisfied), and `From`/`Into` conversions bridge it
//! to the bare model type so the generated client bindings can convert at the
//! boundary. Enums are `enum.StrEnum` classes in the Python package; their wrapper
//! converts a member or its name (see [`str_enum_wrapper`]).
//!
//! Everything is emitted into a sibling `pyo3_impls.rs` of the generated models
//! module and gated behind `#[cfg(feature = "python")]`, so pure-Rust /
//! server-only / WASM consumers never pull `pyo3` into their dependency tree.
//!
//! ## Runtime differences
//!
//! The wrapper *bodies* are runtime-aware: buffa
//! stores singular messages as `MessageField<T>` and enums as `EnumValue<E>`,
//! while prost uses `Option<T>` (or `Option<Box<T>>` for fields prost boxes to
//! break a recursion cycle) and bare `i32`. The accessor expressions are selected
//! by [`Runtime`] and, for messages, by per-field boxing (see [`boxed_fields`]);
//! the *public* wrapper surface (class + method names + Python-visible types) is
//! identical across runtimes.
//!
//! [`FromPyObject`]: pyo3::FromPyObject
//! [`IntoPyObject`]: pyo3::IntoPyObject
//! [`pythonize`]: https://docs.rs/pythonize

use std::collections::BTreeSet;

use convert_case::{Case, Casing};
use proc_macro2::TokenStream;
use quote::{format_ident, quote};

use super::super::{doc_tokens, format_tokens};
use crate::analysis::{GenerationPlan, ServicePlan};
use crate::codegen::{CodeGenConfig, PythonOptions, Runtime};
use crate::parsing::types::{BaseType, JsonType, UnifiedType};
use crate::parsing::{
    CodeGenMetadata, EnumInfo, EnumValue, MessageField, MessageInfo, OneofVariant,
};
use crate::utils::{extract_qualified_type_name, extract_simple_type_name, metadata_key};

/// Generate the `pyo3_impls.rs` content: a `#[pyclass]` wrapper for every message
/// and enum defined in the project's own packages, plus their `From`/`Into`
/// bridges to the bare model types.
///
/// Returns an empty string when there are no project messages/enums to emit (e.g.
/// a service-less descriptor), so the caller can skip writing the file.
pub(crate) fn generate_pyo3_impls(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> crate::error::Result<String> {
    let runtime = config.runtime;
    let options = config.python_options();
    let services: Vec<&ServicePlan> = plan.services.iter().collect();
    let classes = ModelClasses::new(&services, metadata);
    let ctx = Ctx {
        options,
        classes: &classes,
        enums_module: format!("{}._enums", options.package.name),
    };

    let mut items: Vec<TokenStream> = Vec::new();

    // Enums first so message getters can reference the enum wrappers.
    for name in &classes.enums {
        let Some(rust_path) = type_rust_path(name, &classes.packages) else {
            continue;
        };
        let info = &metadata.enums[*name];
        items.push(str_enum_wrapper(name, info, &rust_path, runtime, &ctx)?);
    }

    for name in &classes.messages {
        let Some(rust_path) = type_rust_path(name, &classes.packages) else {
            continue;
        };
        let info = &metadata.messages[*name];
        items.push(message_wrapper(
            name, info, &rust_path, runtime, metadata, &ctx,
        )?);
    }

    if items.is_empty() {
        return Ok(String::new());
    }
    items.push(python_repr_fn());
    if runtime == Runtime::Buffa && crate::codegen::metadata_has_json_fields(metadata) {
        items.push(json_conversions());
    }

    // No file-level inner attributes (`#![..]`) or inner doc comments (`//!`):
    // this file is mounted via `#[cfg(feature = "python")] mod pyo3_impls;` in the
    // generated `mod.rs` *and* `include!`d directly in some setups; inner
    // attributes are illegal inside an `include!`d module body, so every allow is
    // applied as an outer attribute on the individual emitted items instead.
    let tokens = quote! {
        // PyO3 wrapper `#[pyclass]` types for the generated models.
        //
        // Each model message gets a frozen newtype wrapper exposing native, typed
        // getters and, for an input model, a keyword constructor; each enum a wrapper
        // of its `enum.StrEnum` class; plus `From`/`Into` bridges to the bare model
        // type. Mounted under the `python` feature; see the model module's `mod.rs`.

        #(#items)*
    };

    format_tokens(tokens)
}

/// The generated function converting a Python object into the free-form JSON type `json`.
pub(crate) fn json_from_py_fn(json: JsonType) -> proc_macro2::Ident {
    format_ident!("json_{}_from_py", json.fn_suffix())
}

/// Convert the Python object `value` (a `Bound<PyAny>`) into the free-form JSON field type `ty`.
/// For a method argument (`arg`), a list or map is collected into a `Vec`, as the client takes
/// an `impl IntoIterator` that leaves nothing to infer the collection from.
pub(crate) fn json_from_py(
    json: JsonType,
    ty: &UnifiedType,
    value: TokenStream,
    arg: bool,
) -> TokenStream {
    let from_py = json_from_py_fn(json);
    let collection = if arg {
        quote! { ::std::vec::Vec<_> }
    } else {
        quote! { _ }
    };
    if ty.is_repeated {
        quote! {{
            let items: ::std::vec::Vec<::pyo3::Bound<'_, ::pyo3::PyAny>> =
                ::pyo3::types::PyAnyMethods::extract(&#value)?;
            items.iter().map(#from_py).collect::<::pyo3::PyResult<#collection>>()?
        }}
    } else if matches!(ty.base_type, BaseType::Map(..)) {
        quote! {{
            let items: ::std::collections::HashMap<::std::string::String, ::pyo3::Bound<'_, ::pyo3::PyAny>> =
                ::pyo3::types::PyAnyMethods::extract(&#value)?;
            items
                .iter()
                .map(|(key, item)| ::core::result::Result::Ok((key.clone(), #from_py(item)?)))
                .collect::<::pyo3::PyResult<#collection>>()?
        }}
    } else {
        quote! { ::buffa::MessageField::some(#from_py(&#value)?) }
    }
}

/// Conversions between Python objects and buffa's `google.protobuf.Value`/`Struct`/`ListValue`,
/// and `PyJson`, the typed return value of a JSON getter.
fn json_conversions() -> TokenStream {
    quote! {
        /// The type hint of a JSON getter.
        pub trait JsonTypeHint {
            /// The hint.
            const HINT: ::pyo3::inspect::PyStaticExpr;
        }

        /// `typing.Any`.
        pub struct JsonAny;

        impl JsonTypeHint for JsonAny {
            const HINT: ::pyo3::inspect::PyStaticExpr = ::pyo3::type_hint_identifier!("typing", "Any");
        }

        /// `dict[str, V]`.
        pub struct JsonDict<V>(::core::marker::PhantomData<V>);

        impl<V: JsonTypeHint> JsonTypeHint for JsonDict<V> {
            const HINT: ::pyo3::inspect::PyStaticExpr = ::pyo3::type_hint_subscript!(
                ::pyo3::type_hint_identifier!("builtins", "dict"),
                ::pyo3::type_hint_identifier!("builtins", "str"),
                V::HINT
            );
        }

        /// `list[V]`.
        pub struct JsonList<V>(::core::marker::PhantomData<V>);

        impl<V: JsonTypeHint> JsonTypeHint for JsonList<V> {
            const HINT: ::pyo3::inspect::PyStaticExpr = ::pyo3::type_hint_subscript!(
                ::pyo3::type_hint_identifier!("builtins", "list"),
                V::HINT
            );
        }

        /// The Python object of a JSON getter, typed `H` in the stubs.
        pub struct PyJson<H>(::pyo3::Py<::pyo3::PyAny>, ::core::marker::PhantomData<H>);

        impl<H> PyJson<H> {
            fn new(object: ::pyo3::Py<::pyo3::PyAny>) -> Self {
                Self(object, ::core::marker::PhantomData)
            }
        }

        impl<'py, H: JsonTypeHint> ::pyo3::IntoPyObject<'py> for PyJson<H> {
            type Target = ::pyo3::PyAny;
            type Output = ::pyo3::Bound<'py, ::pyo3::PyAny>;
            type Error = ::core::convert::Infallible;
            const OUTPUT_TYPE: ::pyo3::inspect::PyStaticExpr = H::HINT;

            fn into_pyobject(self, py: ::pyo3::Python<'py>) -> ::core::result::Result<Self::Output, Self::Error> {
                ::core::result::Result::Ok(self.0.into_bound(py))
            }
        }

        /// A JSON value as a Python object: `None`, `bool`, `int` for a whole number, `float`,
        /// `str`, `list` or `dict`.
        pub fn json_value_to_py(
            py: ::pyo3::Python<'_>,
            value: &::buffa_types::google::protobuf::Value,
        ) -> ::pyo3::PyResult<::pyo3::Py<::pyo3::PyAny>> {
            use ::buffa_types::google::protobuf::value::Kind;
            use ::pyo3::IntoPyObjectExt;
            match &value.kind {
                ::core::option::Option::None | ::core::option::Option::Some(Kind::NullValue(_)) => {
                    ::core::result::Result::Ok(py.None())
                }
                ::core::option::Option::Some(Kind::BoolValue(b)) => b.into_py_any(py),
                ::core::option::Option::Some(Kind::NumberValue(n))
                    if n.fract() == 0.0 && n.abs() <= 9_007_199_254_740_992.0 =>
                {
                    (*n as i64).into_py_any(py)
                }
                ::core::option::Option::Some(Kind::NumberValue(n)) => n.into_py_any(py),
                ::core::option::Option::Some(Kind::StringValue(s)) => s.into_py_any(py),
                ::core::option::Option::Some(Kind::StructValue(s)) => json_struct_to_py(py, s),
                ::core::option::Option::Some(Kind::ListValue(l)) => json_list_to_py(py, l),
            }
        }

        /// A JSON object as a Python `dict`.
        pub fn json_struct_to_py(
            py: ::pyo3::Python<'_>,
            value: &::buffa_types::google::protobuf::Struct,
        ) -> ::pyo3::PyResult<::pyo3::Py<::pyo3::PyAny>> {
            let dict = ::pyo3::types::PyDict::new(py);
            for (key, item) in &value.fields {
                ::pyo3::types::PyDictMethods::set_item(&dict, key, json_value_to_py(py, item)?)?;
            }
            ::core::result::Result::Ok(dict.into_any().unbind())
        }

        /// A JSON array as a Python `list`.
        pub fn json_list_to_py(
            py: ::pyo3::Python<'_>,
            value: &::buffa_types::google::protobuf::ListValue,
        ) -> ::pyo3::PyResult<::pyo3::Py<::pyo3::PyAny>> {
            let items = value
                .values
                .iter()
                .map(|item| json_value_to_py(py, item))
                .collect::<::pyo3::PyResult<::std::vec::Vec<_>>>()?;
            ::core::result::Result::Ok(::pyo3::types::PyList::new(py, items)?.into_any().unbind())
        }

        /// A Python object as a JSON value. It takes `None`, `bool`, `int`, `float`, `str`, and
        /// `list`, `tuple` and `dict` of those.
        pub fn json_value_from_py(
            obj: &::pyo3::Bound<'_, ::pyo3::PyAny>,
        ) -> ::pyo3::PyResult<::buffa_types::google::protobuf::Value> {
            use ::buffa_types::google::protobuf::Value;
            use ::pyo3::types::{PyAnyMethods, PyTypeMethods};
            if obj.is_none() {
                return ::core::result::Result::Ok(Value::null());
            }
            // `bool` before `int`: a Python `bool` is an `int`.
            if obj.is_instance_of::<::pyo3::types::PyBool>() {
                return ::core::result::Result::Ok(Value::from(obj.extract::<bool>()?));
            }
            if obj.is_instance_of::<::pyo3::types::PyInt>()
                || obj.is_instance_of::<::pyo3::types::PyFloat>()
            {
                let number = obj.extract::<f64>()?;
                if !number.is_finite() {
                    return ::core::result::Result::Err(::pyo3::exceptions::PyValueError::new_err(
                        ::std::format!("not a JSON number: {number}"),
                    ));
                }
                return ::core::result::Result::Ok(Value::from(number));
            }
            if obj.is_instance_of::<::pyo3::types::PyString>() {
                return ::core::result::Result::Ok(Value::from(obj.extract::<::std::string::String>()?));
            }
            if obj.is_instance_of::<::pyo3::types::PyDict>() {
                return ::core::result::Result::Ok(Value::from(json_struct_from_py(obj)?));
            }
            if obj.is_instance_of::<::pyo3::types::PyList>()
                || obj.is_instance_of::<::pyo3::types::PyTuple>()
            {
                return ::core::result::Result::Ok(Value::from(json_list_from_py(obj)?));
            }
            ::core::result::Result::Err(::pyo3::exceptions::PyTypeError::new_err(::std::format!(
                "not a JSON value: {}",
                obj.get_type().name()?
            )))
        }

        /// A Python `dict` with `str` keys as a JSON object.
        pub fn json_struct_from_py(
            obj: &::pyo3::Bound<'_, ::pyo3::PyAny>,
        ) -> ::pyo3::PyResult<::buffa_types::google::protobuf::Struct> {
            let items: ::std::collections::HashMap<::std::string::String, ::pyo3::Bound<'_, ::pyo3::PyAny>> =
                ::pyo3::types::PyAnyMethods::extract(obj)?;
            let mut value = ::buffa_types::google::protobuf::Struct::default();
            for (key, item) in &items {
                value.fields.insert(key.clone(), json_value_from_py(item)?);
            }
            ::core::result::Result::Ok(value)
        }

        /// A Python `list` or `tuple` as a JSON array.
        pub fn json_list_from_py(
            obj: &::pyo3::Bound<'_, ::pyo3::PyAny>,
        ) -> ::pyo3::PyResult<::buffa_types::google::protobuf::ListValue> {
            let items: ::std::vec::Vec<::pyo3::Bound<'_, ::pyo3::PyAny>> =
                ::pyo3::types::PyAnyMethods::extract(obj)?;
            let mut value = ::buffa_types::google::protobuf::ListValue::default();
            for item in &items {
                value.values.push(json_value_from_py(item)?);
            }
            ::core::result::Result::Ok(value)
        }
    }
}

/// The `Py`-prefixed wrapper ident for a fully-qualified proto type name, using
/// the parent-qualified emitted name (collision-free for nested types).
fn wrapper_ident(fq_name: &str) -> proc_macro2::Ident {
    format_ident!("Py{}", extract_qualified_type_name(fq_name))
}

/// The Python-visible class name (no `Py` prefix) for a fully-qualified proto name.
pub(crate) fn python_class_name(fq_name: &str) -> String {
    extract_qualified_type_name(fq_name)
}

/// The models that the bindings wrap, shared by `pyo3_impls.rs`, the `#[pymodule]` and
/// `_enums.py`.
pub(crate) struct ModelClasses<'a> {
    /// The project's own packages (those of the plan's services). Types of other packages
    /// (google, gnostic, ...) get no wrapper: orphan rule, and no in-crate path.
    packages: BTreeSet<&'a str>,
    /// The project enums that have values, sorted by name.
    pub(crate) enums: Vec<&'a String>,
    /// The project messages that Python sees (see [`surfaced_messages`]), sorted by name.
    messages: Vec<&'a String>,
    /// Messages that a request takes; see [`input_messages`].
    inputs: BTreeSet<String>,
}

impl<'a> ModelClasses<'a> {
    pub(crate) fn new(services: &[&'a ServicePlan], metadata: &'a CodeGenMetadata) -> Self {
        let packages: BTreeSet<&str> = services.iter().map(|s| s.package.as_str()).collect();
        let mut enums: Vec<&String> = metadata
            .enums
            .iter()
            .filter(|(name, info)| {
                !info.values.is_empty() && type_rust_path(name, &packages).is_some()
            })
            .map(|(name, _)| name)
            .collect();
        enums.sort();
        let inputs = input_messages(services, metadata);
        let surfaced = surfaced_messages(services, metadata, &inputs);
        let mut messages: Vec<&String> = metadata
            .messages
            .keys()
            .filter(|name| type_rust_path(name, &packages).is_some() && surfaced.contains(*name))
            .collect();
        messages.sort();
        Self {
            packages,
            enums,
            messages,
            inputs,
        }
    }

    /// The `#[pyclass]` wrappers with their package segment: the messages (the enums are `str`
    /// classes).
    pub(crate) fn pyclasses(&self) -> impl Iterator<Item = (&'a str, proc_macro2::Ident)> {
        self.messages.iter().filter_map(|name| {
            let package = package_of(name, &self.packages)?;
            Some((super::package_segment(package), wrapper_ident(name)))
        })
    }

    /// The Python class names with their package segment and whether they are enums.
    pub(crate) fn public_names(&self) -> impl Iterator<Item = (&'a str, String, bool)> {
        let enums = self.enums.iter().map(|name| (*name, true));
        let messages = self.messages.iter().map(|name| (*name, false));
        enums.chain(messages).filter_map(|(name, is_enum)| {
            let package = package_of(name, &self.packages)?;
            Some((
                super::package_segment(package),
                python_class_name(name),
                is_enum,
            ))
        })
    }
}

/// Whether an enum value has no Python member: the zero `*_UNSPECIFIED` value, which a field
/// holding it reads as `None`.
pub(crate) fn hidden_value(value: &EnumValue) -> bool {
    value.number == 0 && value.name.ends_with("_UNSPECIFIED")
}

fn has_hidden_zero(info: &EnumInfo) -> bool {
    info.values.iter().any(hidden_value)
}

/// The longest project package that prefixes `fq_name`.
fn package_of<'a>(fq_name: &str, project_packages: &BTreeSet<&'a str>) -> Option<&'a str> {
    let stripped = fq_name.strip_prefix('.').unwrap_or(fq_name);
    project_packages
        .iter()
        .filter(|pkg| {
            stripped
                .strip_prefix(**pkg)
                .is_some_and(|rest| rest.starts_with('.'))
        })
        .max_by_key(|pkg| pkg.len())
        .copied()
}

/// The messages that a request takes: the message types of the RPC input fields and everything
/// reachable from them. The input messages themselves become method parameters, not classes.
fn input_messages(services: &[&ServicePlan], metadata: &CodeGenMetadata) -> BTreeSet<String> {
    let mut queue = Vec::new();
    for method in services.iter().flat_map(|s| &s.methods) {
        if let Some(info) = metadata
            .messages
            .get(&metadata_key(&method.metadata.input_type))
        {
            push_field_messages(info, &mut queue);
        }
    }
    reachable(queue, metadata)
}

/// The messages that Python sees: the `inputs`, the RPC outputs, and everything reachable from
/// those outputs.
fn surfaced_messages(
    services: &[&ServicePlan],
    metadata: &CodeGenMetadata,
    inputs: &BTreeSet<String>,
) -> BTreeSet<String> {
    let outputs = services
        .iter()
        .flat_map(|s| &s.methods)
        .map(|m| metadata_key(&m.metadata.output_type))
        .filter(|name| !name.ends_with(".Empty"))
        .collect();
    let mut surfaced = reachable(outputs, metadata);
    surfaced.extend(inputs.iter().cloned());
    surfaced
}

/// `start` and every message reachable from it.
fn reachable(start: Vec<String>, metadata: &CodeGenMetadata) -> BTreeSet<String> {
    let mut queue = start;
    let mut seen = BTreeSet::new();
    while let Some(name) = queue.pop() {
        if !seen.insert(name.clone()) {
            continue;
        }
        if let Some(info) = metadata.messages.get(&name) {
            push_field_messages(info, &mut queue);
        }
    }
    seen
}

/// Push the message types of the fields of `info` (including map values and oneof variants).
fn push_field_messages(info: &MessageInfo, queue: &mut Vec<String>) {
    for field in &info.fields {
        let types = std::iter::once(&field.unified_type)
            .chain(field.oneof_variants.iter().flatten().map(|v| &v.field_type));
        for ty in types {
            let ty = match &ty.base_type {
                BaseType::Map(_, value) => value,
                _ => ty,
            };
            if let BaseType::Message(name) = &ty.base_type {
                queue.push(metadata_key(name));
            }
        }
    }
}

/// What the wrappers of one generation run share.
struct Ctx<'a> {
    options: &'a PythonOptions,
    classes: &'a ModelClasses<'a>,
    /// Import name of the `_enums` module (`<package>._enums`).
    enums_module: String,
}

impl Ctx<'_> {
    /// The `module = "…",` argument of `#[pyclass]` for a type, with its trailing comma.
    fn module_arg(&self, fq_name: &str) -> Option<TokenStream> {
        let package = package_of(fq_name, &self.classes.packages)?;
        Some(super::module_arg(
            self.options,
            Some(super::package_segment(package)),
        ))
    }

    /// Whether the class of message `fq_name` gets a keyword constructor: the models that a
    /// request takes do; a model that only comes back in responses does not.
    fn has_constructor(&self, fq_name: &str) -> bool {
        self.classes.inputs.contains(&metadata_key(fq_name))
    }
}

/// The shared helper of the Python-style `__repr__`.
fn python_repr_fn() -> TokenStream {
    quote! {
        /// `Name(field=value, …)` without the `None` fields; a redacted field shows `'***'`.
        fn __python_repr(
            obj: &::pyo3::Bound<'_, ::pyo3::PyAny>,
            fields: &[(&str, bool)],
        ) -> ::pyo3::PyResult<::std::string::String> {
            use ::pyo3::types::{PyAnyMethods, PyTypeMethods};
            let mut shown = ::std::vec::Vec::with_capacity(fields.len());
            for (field, redact) in fields {
                let value = obj.getattr(*field)?;
                if value.is_none() {
                    continue;
                }
                if *redact {
                    shown.push(::std::format!("{field}='***'"));
                } else {
                    shown.push(::std::format!("{field}={}", value.repr()?));
                }
            }
            Ok(::std::format!("{}({})", obj.get_type().name()?, shown.join(", ")))
        }
    }
}

// ---------------------------------------------------------------------------
// Enum wrappers
// ---------------------------------------------------------------------------

/// Emit the bridge of a proto enum to its `enum.StrEnum` class in `_enums.py`: a newtype that
/// extracts from a member or a name and converts back into a member.
fn str_enum_wrapper(
    fq_name: &str,
    info: &EnumInfo,
    rust_path: &str,
    runtime: Runtime,
    ctx: &Ctx<'_>,
) -> crate::error::Result<TokenStream> {
    let wrapper = wrapper_ident(fq_name);
    let class_name = python_class_name(fq_name);
    let inner: syn::Path = parse_path(rust_path, fq_name)?;
    let enums_module = ctx.enums_module.as_str();

    let shown: Vec<&EnumValue> = info.values.iter().filter(|v| !hidden_value(v)).collect();
    let names: Vec<&str> = shown.iter().map(|v| v.name.as_str()).collect();
    let numbers: Vec<proc_macro2::Literal> = shown
        .iter()
        .map(|v| proc_macro2::Literal::i32_suffixed(v.number))
        .collect();
    let invalid = format!(
        "invalid {class_name}: {{:?}}; expected one of: {}",
        names.join(", ")
    );
    let no_member = format!("{class_name} value {{}} has no Python member");

    let (inner_to_i32, i32_to_inner): (TokenStream, TokenStream) = match runtime {
        Runtime::Buffa => (
            quote! { <#inner as ::buffa::Enumeration>::to_i32(&self.0) },
            quote! { <#inner as ::buffa::Enumeration>::from_i32(n).unwrap_or_default() },
        ),
        Runtime::Prost => (
            quote! { self.0 as i32 },
            quote! { <#inner as ::core::convert::TryFrom<i32>>::try_from(n).unwrap_or_default() },
        ),
    };

    let doc = format!(" A member of `{enums_module}.{class_name}`, or its name.");
    Ok(quote! {
        #[doc = #doc]
        #[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
        pub struct #wrapper(pub #inner);

        impl #wrapper {
            /// The name of the value, or `None` for a value without a Python member.
            fn __name(&self) -> ::core::option::Option<&'static str> {
                match #inner_to_i32 {
                    #(#numbers => ::core::option::Option::Some(#names),)*
                    _ => ::core::option::Option::None,
                }
            }
        }

        impl ::core::convert::From<#inner> for #wrapper {
            fn from(value: #inner) -> Self {
                Self(value)
            }
        }

        impl ::core::convert::From<#wrapper> for #inner {
            fn from(value: #wrapper) -> Self {
                value.0
            }
        }

        impl<'a, 'py> ::pyo3::FromPyObject<'a, 'py> for #wrapper {
            type Error = ::pyo3::PyErr;
            const INPUT_TYPE: ::pyo3::inspect::PyStaticExpr = {
                // The macro calls itself without a path.
                use ::pyo3::type_hint_union;
                type_hint_union!(
                    ::pyo3::type_hint_identifier!(#enums_module, #class_name),
                    ::pyo3::type_hint_identifier!("builtins", "str")
                )
            };

            fn extract(obj: ::pyo3::Borrowed<'a, 'py, ::pyo3::PyAny>) -> ::pyo3::PyResult<Self> {
                let name: ::pyo3::pybacked::PyBackedStr = ::pyo3::types::PyAnyMethods::extract(&*obj)?;
                let n: i32 = match &*name {
                    #(#names => #numbers,)*
                    other => {
                        return ::core::result::Result::Err(
                            ::pyo3::exceptions::PyValueError::new_err(::std::format!(#invalid, other)),
                        );
                    }
                };
                ::core::result::Result::Ok(Self(#i32_to_inner))
            }
        }

        impl<'py> ::pyo3::IntoPyObject<'py> for #wrapper {
            type Target = ::pyo3::PyAny;
            type Output = ::pyo3::Bound<'py, ::pyo3::PyAny>;
            type Error = ::pyo3::PyErr;
            const OUTPUT_TYPE: ::pyo3::inspect::PyStaticExpr =
                ::pyo3::type_hint_identifier!(#enums_module, #class_name);

            fn into_pyobject(self, py: ::pyo3::Python<'py>) -> ::pyo3::PyResult<Self::Output> {
                static CLASS: ::pyo3::sync::PyOnceLock<::pyo3::Py<::pyo3::types::PyType>> =
                    ::pyo3::sync::PyOnceLock::new();
                let class = CLASS.import(py, #enums_module, #class_name)?;
                let name = self.__name().ok_or_else(|| {
                    ::pyo3::exceptions::PyValueError::new_err(::std::format!(#no_member, #inner_to_i32))
                })?;
                ::pyo3::types::PyAnyMethods::getattr(class.as_any(), name)
            }
        }
    })
}

// ---------------------------------------------------------------------------
// Message wrappers
// ---------------------------------------------------------------------------

/// Emit a frozen newtype `#[pyclass]` wrapping a model message: a keyword constructor for an
/// input model, native typed getters per field, `__repr__`/`__eq__`, and the `From`/`Into`
/// bridges to the bare model type.
fn message_wrapper(
    fq_name: &str,
    info: &MessageInfo,
    rust_path: &str,
    runtime: Runtime,
    metadata: &CodeGenMetadata,
    ctx: &Ctx<'_>,
) -> crate::error::Result<TokenStream> {
    let wrapper = wrapper_ident(fq_name);
    let class_name = python_class_name(fq_name);
    let inner: syn::Path = parse_path(rust_path, fq_name)?;

    // Plain (non-oneof) fields get one `FieldAccessor` each; oneof fields are
    // handled separately by `OneofAccessor`, which flattens each oneof into one
    // optional keyword arg / per-variant setter+getter (see issue #99).
    let plain_fields: Vec<&MessageField> = info
        .fields
        .iter()
        .filter(|f| f.oneof_variants.is_none())
        .collect();
    let oneof_fields: Vec<&MessageField> = info
        .fields
        .iter()
        .filter(|f| f.oneof_variants.is_some())
        .collect();

    let packages = metadata_packages(metadata);
    let project_pkg_refs: BTreeSet<&str> = packages.iter().map(String::as_str).collect();
    let boxed = boxed_fields(metadata, runtime);
    let accessors: Vec<FieldAccessor> = plain_fields
        .iter()
        .map(|f| {
            let is_boxed = boxed.contains(&(fq_name.to_string(), f.name.clone()));
            let hidden_zero = match &f.unified_type.base_type {
                BaseType::Enum(n) => metadata
                    .enums
                    .get(&metadata_key(n))
                    .is_some_and(has_hidden_zero),
                _ => false,
            };
            FieldAccessor::new(
                f,
                runtime,
                &packages,
                &project_pkg_refs,
                is_boxed,
                hidden_zero,
            )
        })
        .collect();

    // Resolve each oneof into its variants. The oneof enum lives in the message's
    // parent module (buffa exposes it there via a natural-path re-export, prost
    // nests it directly); every variant payload is resolved to its Python-visible
    // wrapper type just like a plain field.
    let oneofs: Vec<OneofAccessor> = oneof_fields
        .iter()
        .map(|f| OneofAccessor::new(f, &inner, runtime, &packages))
        .collect();

    let getters = accessors.iter().map(FieldAccessor::getter);
    let oneof_getters = oneofs.iter().flat_map(OneofAccessor::variant_getters);
    let ctor = ctx
        .has_constructor(fq_name)
        .then(|| constructor(&accessors, &oneofs, &inner, ctx.options.keyword_only));

    // `r#type` is `type` in Python.
    let repr_fields = accessors
        .iter()
        .map(|a| (a.ident.clone(), a.sensitive))
        .chain(
            oneofs
                .iter()
                .flat_map(OneofAccessor::variant_param_idents)
                .map(|ident| (ident, false)),
        )
        .map(|(ident, redact)| {
            let name = syn::ext::IdentExt::unraw(&ident).to_string();
            quote! { (#name, #redact) }
        });

    let module_arg = ctx.module_arg(fq_name);
    let doc = doc_tokens(info.documentation.as_deref());

    Ok(quote! {
        #doc
        // `from_py_object`: these wrappers are `Clone`, and pyo3 0.28 made the auto
        // `FromPyObject` derive for `Clone` `#[pyclass]` types opt-in. Keep the pre-0.28
        // behavior (derive it) explicitly so Python callers can still pass a wrapper into
        // Rust-side functions and to silence the deprecation under `-D warnings`.
        #[::pyo3::pyclass(name = #class_name, eq, frozen, #module_arg from_py_object)]
        #[derive(Clone, Debug, PartialEq)]
        pub struct #wrapper(pub #inner);

        #[allow(clippy::too_many_arguments, clippy::useless_conversion)]
        #[::pyo3::pymethods]
        impl #wrapper {
            #ctor

            #(#getters)*
            #(#oneof_getters)*

            fn __repr__(slf: &::pyo3::Bound<'_, Self>) -> ::pyo3::PyResult<::std::string::String> {
                __python_repr(slf.as_any(), &[#(#repr_fields),*])
            }
        }

        impl ::core::convert::From<#inner> for #wrapper {
            fn from(value: #inner) -> Self {
                Self(value)
            }
        }

        impl ::core::convert::From<#wrapper> for #inner {
            fn from(value: #wrapper) -> Self {
                value.0
            }
        }
    })
}

/// Project package set, recomputed lazily for `FieldAccessor` so it can decide
/// whether a referenced message/enum is in-project (gets a wrapper) or foreign.
fn metadata_packages(metadata: &CodeGenMetadata) -> BTreeSet<String> {
    metadata
        .services
        .values()
        .map(|s| s.package.clone())
        .collect()
}

/// The set of `(message_fq_name, field_name)` pairs that the prost model plugin
/// stores as `Option<Box<T>>` rather than `Option<T>`.
///
/// prost only boxes a singular message field when leaving it unboxed would make
/// the struct infinitely sized — i.e. the field's message type can transitively
/// reach back to the containing message through a chain of singular (non-repeated,
/// non-map) message fields. Repeated and map fields already introduce an
/// indirection (`Vec`/`HashMap`), so they break the cycle and are never boxed.
///
/// We replicate that rule here so the PyO3 read/write accessors deref/box exactly
/// the fields prost boxed. Returns an empty set for the buffa runtime, which uses
/// `MessageField<T>` uniformly and never boxes.
fn boxed_fields(metadata: &CodeGenMetadata, runtime: Runtime) -> BTreeSet<(String, String)> {
    let mut boxed = BTreeSet::new();
    if runtime != Runtime::Prost {
        return boxed;
    }
    for (msg_name, info) in &metadata.messages {
        for field in &info.fields {
            // Only singular message fields can be boxed; repeated/map fields carry
            // their own indirection, and scalars/enums are never boxed.
            if field.unified_type.is_repeated
                || matches!(field.unified_type.base_type, BaseType::Map(_, _))
            {
                continue;
            }
            let BaseType::Message(target) = &field.unified_type.base_type else {
                continue;
            };
            // Box iff the field's type can reach back to the containing message.
            if reaches(metadata, target, msg_name) {
                boxed.insert((msg_name.clone(), field.name.clone()));
            }
        }
    }
    boxed
}

/// Whether message `from` can transitively reach message `target` by following
/// singular (non-repeated, non-map) message fields. Used to detect the recursion
/// cycles that drive prost's automatic field boxing.
fn reaches(metadata: &CodeGenMetadata, from: &str, target: &str) -> bool {
    let mut stack = vec![from.to_string()];
    let mut seen = BTreeSet::new();
    while let Some(cur) = stack.pop() {
        if cur == target {
            return true;
        }
        if !seen.insert(cur.clone()) {
            continue;
        }
        let Some(info) = metadata.messages.get(&cur) else {
            continue;
        };
        for field in &info.fields {
            if field.unified_type.is_repeated
                || matches!(field.unified_type.base_type, BaseType::Map(_, _))
            {
                continue;
            }
            if let BaseType::Message(next) = &field.unified_type.base_type {
                stack.push(next.clone());
            }
        }
    }
    false
}

/// A single message field, resolved into the Rust expressions needed to expose it
/// across the PyO3 boundary as a native, typed Python attribute.
struct FieldAccessor {
    /// The Python/Rust-visible field identifier (snake_case, keyword-sanitized).
    ident: proc_macro2::Ident,
    /// The original field, for cardinality flags and base type.
    ty: UnifiedType,
    runtime: Runtime,
    /// Whether the referenced message/enum is in the project (gets a wrapper).
    in_project: bool,
    /// In-crate Rust path of a referenced message/enum (e.g. `super::catalog::v1::Color`),
    /// when in-project; used by enum/message read expressions. `None` otherwise.
    inner_rust_path: Option<TokenStream>,
    /// Whether the prost model stores this singular message field as
    /// `Option<Box<T>>` (it sits on a recursion cycle). Drives whether the read
    /// path derefs the box and the write path re-boxes. Always `false` for buffa.
    is_boxed: bool,
    /// Whether the field is an enum whose zero value has no Python member (see
    /// [`hidden_value`]).
    hidden_zero: bool,
    /// The proto field comment, for the getter.
    documentation: Option<String>,
    /// The free-form JSON type of the field (buffa only).
    json: Option<JsonType>,
    /// Whether the proto marks the field `REQUIRED` (a positional constructor argument with
    /// [`PythonOptions::keyword_only`](crate::PythonOptions::keyword_only)).
    required: bool,
    /// Whether the proto marks the field `debug_redact` (hidden in the Python repr).
    sensitive: bool,
}

impl FieldAccessor {
    fn new(
        field: &MessageField,
        runtime: Runtime,
        packages: &BTreeSet<String>,
        project_pkg_refs: &BTreeSet<&str>,
        is_boxed: bool,
        hidden_zero: bool,
    ) -> Self {
        let ident = crate::utils::field_ident(&super::sanitize_python_field_name(&field.name));
        // A map is in the project when its values are.
        let element = match &field.unified_type.base_type {
            BaseType::Map(_, value) => &value.base_type,
            base => base,
        };
        let in_project = match element {
            BaseType::Message(n) | BaseType::Enum(n) => is_in_project(n, packages),
            _ => false,
        };
        let inner_rust_path = match &field.unified_type.base_type {
            BaseType::Message(n) | BaseType::Enum(n) if in_project => {
                type_rust_path(n, project_pkg_refs)
                    .and_then(|p| syn::parse_str::<syn::Path>(&p).ok())
                    .map(|p| quote! { #p })
            }
            _ => None,
        };
        Self {
            ident,
            ty: field.unified_type.clone(),
            runtime,
            in_project,
            inner_rust_path,
            is_boxed,
            hidden_zero,
            documentation: field.documentation.clone(),
            json: JsonType::of_field(&field.unified_type).filter(|_| runtime == Runtime::Buffa),
            required: field
                .field_behavior
                .contains(&crate::google::api::FieldBehavior::Required),
            sensitive: field.is_sensitive,
        }
    }

    /// The `#[getter]` of a free-form JSON field, which returns a plain Python object typed by
    /// a [`json_conversions`] hint: `typing.Any`, `dict[str, typing.Any]` or `list[typing.Any]`,
    /// in a `list` or `dict` for a repeated or map field.
    fn json_getter(&self, json: JsonType) -> TokenStream {
        let ident = &self.ident;
        let doc = doc_tokens(self.documentation.as_deref());
        let to_py = format_ident!("json_{}_to_py", json.fn_suffix());
        let item = match json {
            JsonType::Value => quote! { JsonAny },
            JsonType::Struct => quote! { JsonDict<JsonAny> },
            JsonType::ListValue => quote! { JsonList<JsonAny> },
        };
        let (ret, body) = if self.ty.is_repeated {
            (
                quote! { PyJson<JsonList<#item>> },
                quote! {
                    let items = self.0.#ident
                        .iter()
                        .map(|value| #to_py(py, value))
                        .collect::<::pyo3::PyResult<::std::vec::Vec<_>>>()?;
                    ::core::result::Result::Ok(PyJson::new(
                        ::pyo3::types::PyList::new(py, items)?.into_any().unbind(),
                    ))
                },
            )
        } else if matches!(self.ty.base_type, BaseType::Map(..)) {
            (
                quote! { PyJson<JsonDict<#item>> },
                quote! {
                    let dict = ::pyo3::types::PyDict::new(py);
                    for (key, value) in &self.0.#ident {
                        ::pyo3::types::PyDictMethods::set_item(&dict, key, #to_py(py, value)?)?;
                    }
                    ::core::result::Result::Ok(PyJson::new(dict.into_any().unbind()))
                },
            )
        } else if json == JsonType::Value {
            // A JSON value is `typing.Any`, `None` included.
            (
                quote! { PyJson<#item> },
                quote! {
                    let value = self.0.#ident.as_option().map(|value| #to_py(py, value)).transpose()?;
                    ::core::result::Result::Ok(PyJson::new(value.unwrap_or_else(|| py.None())))
                },
            )
        } else {
            (
                quote! { ::core::option::Option<PyJson<#item>> },
                quote! {
                    self.0.#ident
                        .as_option()
                        .map(|value| #to_py(py, value).map(PyJson::new))
                        .transpose()
                },
            )
        };
        quote! {
            #doc
            #[getter]
            fn #ident(&self, py: ::pyo3::Python<'_>) -> ::pyo3::PyResult<#ret> {
                #body
            }
        }
    }

    /// Convert the Python object `value` into the field's free-form JSON type.
    fn json_store(&self, json: JsonType) -> TokenStream {
        json_from_py(json, &self.ty, quote! { value }, false)
    }

    /// The Python-visible Rust type for this field (the getter return / constructor
    /// arg). Messages/enums map to their `Py`-prefixed wrapper; collections wrap
    /// in `Vec`/`HashMap`; optional non-collection fields wrap in `Option`.
    fn py_type(&self) -> TokenStream {
        let base = self.base_py_type();
        if self.ty.is_repeated {
            return quote! { ::std::vec::Vec<#base> };
        }
        if let BaseType::Map(_, v) = &self.ty.base_type {
            let val = py_type_for(v, self.in_project);
            return quote! { ::std::collections::HashMap<::std::string::String, #val> };
        }
        // Singular (non-repeated) message fields carry proto presence semantics, so
        // they are always Python-visible as `Optional[...]` — the read path returns
        // `Option<Wrapper>` for both runtimes (buffa `MessageField::into_option`,
        // prost `Option<T>` or `Option<Box<T>>` for recursion-boxed fields).
        if self.is_singular_message() {
            return quote! { ::core::option::Option<#base> };
        }
        // Explicit-presence scalars/enums (proto3 `optional`) are stored as
        // `Option<T>` by both the buffa and prost model plugins, so the wrapper
        // surfaces them as `Optional[...]` too.
        if self.is_optional_scalar() || self.is_hidden_zero_enum() {
            return quote! { ::core::option::Option<#base> };
        }
        base
    }

    /// A singular, non-`optional` enum field with a hidden zero value: `Optional[...]` in Python.
    fn is_hidden_zero_enum(&self) -> bool {
        self.hidden_zero && !self.ty.is_repeated && !self.ty.is_optional
    }

    /// The `.filter(..)` that drops a hidden zero value from an `Option` or iterator of the
    /// bare Rust enum.
    fn hidden_zero_filter(&self) -> Option<TokenStream> {
        let inner = self.inner_rust_path.as_ref();
        self.hidden_zero.then(|| match self.runtime {
            Runtime::Buffa => {
                quote! { .filter(|e| <#inner as ::buffa::Enumeration>::to_i32(e) != 0) }
            }
            Runtime::Prost => quote! { .filter(|e| *e as i32 != 0) },
        })
    }

    fn base_py_type(&self) -> TokenStream {
        py_type_for(&self.ty, self.in_project)
    }

    fn is_singular_message(&self) -> bool {
        !self.ty.is_repeated
            && matches!(self.ty.base_type, BaseType::Message(_) | BaseType::OneOf(_))
    }

    /// Whether [`Self::py_type`] is itself an `Option<..>` — i.e. the field is a
    /// singular message or an explicit-presence scalar/enum. Such fields must not be
    /// re-wrapped in the keyword constructor (would yield ambiguous `Option<Option<T>>`).
    fn py_type_is_optional(&self) -> bool {
        self.is_singular_message() || self.is_optional_scalar() || self.is_hidden_zero_enum()
    }

    /// An explicit-presence scalar or enum (proto3 `optional`): stored as
    /// `Option<T>` (scalar) or `Option<EnumValue<E>>`/`Option<i32>` (enum) by the
    /// model plugins, as opposed to messages whose presence is intrinsic.
    fn is_optional_scalar(&self) -> bool {
        self.ty.is_optional
            && !self.ty.is_repeated
            && !matches!(
                self.ty.base_type,
                BaseType::Message(_) | BaseType::OneOf(_) | BaseType::Map(_, _)
            )
    }

    /// `#[getter]` returning the field as its Python-visible type.
    fn getter(&self) -> TokenStream {
        if let Some(json) = self.json {
            return self.json_getter(json);
        }
        let ident = &self.ident;
        let (ret, body) = match self.ty.is_singular_bytes() {
            true => self.bytes_getter(),
            false => (self.py_type(), self.read_expr()),
        };
        let doc = doc_tokens(self.documentation.as_deref());
        quote! {
            #doc
            #[getter]
            fn #ident(&self) -> #ret {
                #body
            }
        }
    }

    /// The return type and body of a singular `bytes` getter: a borrowed `Cow`, which PyO3 copies
    /// into `bytes` once.
    fn bytes_getter(&self) -> (TokenStream, TokenStream) {
        let ident = &self.ident;
        match self.is_optional_scalar() {
            true => (
                quote! { ::core::option::Option<::std::borrow::Cow<'_, [u8]>> },
                quote! { self.0.#ident.as_deref().map(::std::borrow::Cow::Borrowed) },
            ),
            false => (
                quote! { ::std::borrow::Cow<'_, [u8]> },
                quote! { ::std::borrow::Cow::Borrowed(&self.0.#ident) },
            ),
        }
    }

    /// The type a constructor argument takes: [`Self::py_type`], but `PyBackedBytes` for `bytes`.
    fn input_type(&self) -> TokenStream {
        match self.ty.is_singular_bytes() {
            true => super::bindings::py_bytes_input(self.is_optional_scalar()),
            false => self.py_type(),
        }
    }

    /// [`Self::input_type`] without its `Option`, for a required constructor argument.
    fn input_base_type(&self) -> TokenStream {
        match self.ty.is_singular_bytes() {
            true => super::bindings::py_bytes_input(false),
            false => self.base_py_type(),
        }
    }

    /// Expression that reads `self.0.<field>` and converts it to the Python type.
    fn read_expr(&self) -> TokenStream {
        let ident = &self.ident;
        let access = quote! { self.0.#ident };

        if self.ty.is_repeated {
            return self.read_repeated(&access);
        }
        if let BaseType::Map(_, v) = &self.ty.base_type {
            return read_map(&access, v, self.in_project);
        }
        if self.is_optional_scalar() {
            return self.read_optional_scalar(&access);
        }

        match &self.ty.base_type {
            BaseType::Int32
            | BaseType::Int64
            | BaseType::Bool
            | BaseType::Float32
            | BaseType::Float64 => quote! { #access },
            BaseType::Enum(_) => self.read_enum(&access),
            BaseType::Message(_) | BaseType::OneOf(_) => self.read_message(&access),
            BaseType::String | BaseType::Bytes | BaseType::Unit | BaseType::Map(_, _) => {
                quote! { #access.clone() }
            }
        }
    }

    fn read_repeated(&self, access: &TokenStream) -> TokenStream {
        let base = self.base_py_type();
        let inner = self.inner_rust_path.as_ref();
        match &self.ty.base_type {
            BaseType::Enum(_) => {
                let hidden = self.hidden_zero_filter();
                match self.runtime {
                    // buffa repeated enum: Vec<EnumValue<E>>; keep known values only.
                    Runtime::Buffa => quote! {
                        #access.iter().filter_map(|e| e.as_known()) #hidden .map(#base::from).collect()
                    },
                    // prost repeated enum: Vec<i32>; map through the bare Rust enum.
                    Runtime::Prost => quote! {
                        #access.iter()
                            .filter_map(|&v| <#inner as ::core::convert::TryFrom<i32>>::try_from(v).ok())
                            #hidden
                            .map(#base::from)
                            .collect()
                    },
                }
            }
            BaseType::Message(_) => quote! {
                #access.iter().cloned().map(#base::from).collect()
            },
            _ => quote! { #access.clone() },
        }
    }

    fn read_enum(&self, access: &TokenStream) -> TokenStream {
        let base = self.base_py_type();
        let inner = self.inner_rust_path.as_ref();
        if self.is_hidden_zero_enum() {
            let hidden = self.hidden_zero_filter();
            return match self.runtime {
                Runtime::Buffa => quote! { #access.as_known() #hidden .map(#base::from) },
                Runtime::Prost => quote! {
                    <#inner as ::core::convert::TryFrom<i32>>::try_from(#access).ok() #hidden .map(#base::from)
                },
            };
        }
        match self.runtime {
            // buffa: EnumValue<E> -> Option<E> via as_known(); fall back to the inner
            // enum's Default (the zero/unspecified variant), then to the wrapper.
            Runtime::Buffa => quote! {
                #base::from(#access.as_known().unwrap_or_default())
            },
            // prost: i32 -> E via TryFrom, fall back to the inner enum's Default.
            Runtime::Prost => quote! {
                #base::from(
                    <#inner as ::core::convert::TryFrom<i32>>::try_from(#access)
                        .unwrap_or_default()
                )
            },
        }
    }

    fn read_message(&self, access: &TokenStream) -> TokenStream {
        let base = self.base_py_type();
        match self.runtime {
            // buffa: MessageField<T> -> Option<T>
            Runtime::Buffa => quote! {
                #access.clone().into_option().map(#base::from)
            },
            // prost recursion-boxed field: Option<Box<T>> — deref the box.
            Runtime::Prost if self.is_boxed => quote! {
                #access.clone().map(|b| #base::from(*b))
            },
            // prost plain singular message: Option<T>.
            Runtime::Prost => quote! {
                #access.clone().map(#base::from)
            },
        }
    }

    /// Read an explicit-presence (proto3 `optional`) scalar/enum field, whose
    /// stored type is `Option<..>` — surfaced to Python as `Option<base>`.
    fn read_optional_scalar(&self, access: &TokenStream) -> TokenStream {
        let base = self.base_py_type();
        let inner = self.inner_rust_path.as_ref();
        match &self.ty.base_type {
            // buffa: Option<EnumValue<E>>; prost: Option<i32>.
            BaseType::Enum(_) => {
                let hidden = self.hidden_zero_filter();
                match self.runtime {
                    Runtime::Buffa => quote! {
                        #access.as_ref().and_then(|e| e.as_known()) #hidden .map(#base::from)
                    },
                    Runtime::Prost => quote! {
                        #access.and_then(|v| <#inner as ::core::convert::TryFrom<i32>>::try_from(v).ok())
                            #hidden
                            .map(#base::from)
                    },
                }
            }
            // Option<{int,bool,float}>: the inner scalar is `Copy`, so the `Option`
            // copies by value — `.clone()` here would trip `clippy::clone_on_copy`.
            BaseType::Int32
            | BaseType::Int64
            | BaseType::Bool
            | BaseType::Float32
            | BaseType::Float64 => quote! { #access },
            // Option<String>/Option<Vec<u8>>: non-Copy, clone through.
            _ => quote! { #access.clone() },
        }
    }

    /// Expression converting a Python-typed `value` back into the inner field
    /// representation for storage.
    fn write_expr(&self) -> TokenStream {
        if self.ty.is_repeated {
            return self.write_repeated();
        }
        if self.ty.is_singular_bytes() {
            return super::bindings::bytes_to_vec(quote! { value }, self.is_optional_scalar());
        }
        if let BaseType::Map(_, value) = &self.ty.base_type {
            // Message values arrive as their wrappers.
            return if self.in_project && matches!(value.base_type, BaseType::Message(_)) {
                quote! { value.into_iter().map(|(k, v)| (k, ::core::convert::Into::into(v))).collect() }
            } else {
                quote! { value }
            };
        }
        if self.is_optional_scalar() {
            return self.write_optional_scalar();
        }
        match &self.ty.base_type {
            BaseType::Enum(_) => self.write_enum(),
            BaseType::Message(_) | BaseType::OneOf(_) => self.write_message(),
            _ => quote! { value },
        }
    }

    /// Store an explicit-presence scalar/enum: `value` is `Option<base>`.
    fn write_optional_scalar(&self) -> TokenStream {
        match &self.ty.base_type {
            // buffa: Option<EnumValue<E>>; prost: Option<i32>.
            BaseType::Enum(_) => {
                let store = self.enum_store(quote! { e });
                quote! { value.map(|e| #store) }
            }
            // Option<String> etc. store directly.
            _ => quote! { value },
        }
    }

    fn write_repeated(&self) -> TokenStream {
        match &self.ty.base_type {
            BaseType::Enum(_) => {
                let store = self.enum_store(quote! { e });
                quote! { value.into_iter().map(|e| #store).collect() }
            }
            BaseType::Message(_) => quote! {
                value.into_iter().map(::core::convert::Into::into).collect()
            },
            _ => quote! { value },
        }
    }

    fn write_enum(&self) -> TokenStream {
        // A hidden-zero field takes `Option<base>`; `None` stores the zero value.
        if self.is_hidden_zero_enum() {
            let store = self.enum_store(quote! { e });
            let zero = match self.runtime {
                Runtime::Buffa => quote! { unwrap_or_default() },
                Runtime::Prost => quote! { unwrap_or(0) },
            };
            return quote! { value.map(|e| #store).#zero };
        }
        self.enum_store(quote! { value })
    }

    /// The stored form of the bare enum `value`: buffa's `EnumValue<E>`, prost's `i32`.
    fn enum_store(&self, value: TokenStream) -> TokenStream {
        let inner = self.inner_rust_path.as_ref();
        match self.runtime {
            Runtime::Buffa => quote! {
                ::buffa::EnumValue::Known(<#inner as ::core::convert::From<_>>::from(#value))
            },
            Runtime::Prost => quote! {
                <#inner as ::core::convert::From<_>>::from(#value) as i32
            },
        }
    }

    /// The constructor parameter of a required field: the bare type, without the `Option` of a
    /// singular message or explicit-presence scalar.
    fn required_ctor_param(&self) -> TokenStream {
        let ident = &self.ident;
        if self.json.is_some() {
            return quote! { #ident: ::pyo3::Bound<'_, ::pyo3::PyAny> };
        }
        let ty = if self.py_type_is_optional() {
            self.input_base_type()
        } else {
            self.input_type()
        };
        quote! { #ident: #ty }
    }

    /// The constructor parameter of an optional field. A field whose Python type is already
    /// `Option<..>` is not wrapped again: pyo3 cannot coerce an `Option<Option<T>>` argument.
    fn optional_ctor_param(&self) -> TokenStream {
        let ident = &self.ident;
        if self.json.is_some() {
            return quote! { #ident: ::core::option::Option<::pyo3::Bound<'_, ::pyo3::PyAny>> };
        }
        let ty = self.input_type();
        if self.py_type_is_optional() {
            quote! { #ident: #ty }
        } else {
            quote! { #ident: ::core::option::Option<#ty> }
        }
    }

    /// The constructor statement storing a required argument.
    fn required_ctor_assign(&self) -> TokenStream {
        let ident = &self.ident;
        let store = match self.json {
            Some(json) => self.json_store(json),
            None => self.write_expr(),
        };
        // The store expression of an `Option` field reads an `Option`.
        if self.json.is_none() && self.py_type_is_optional() {
            quote! {
                {
                    let value = ::core::option::Option::Some(#ident);
                    inner.#ident = #store;
                }
            }
        } else {
            let bind = bind_value(ident);
            quote! {
                {
                    #bind
                    inner.#ident = #store;
                }
            }
        }
    }

    /// The constructor statement storing an optional argument.
    fn optional_ctor_assign(&self) -> TokenStream {
        let ident = &self.ident;
        if let Some(json) = self.json {
            let store = self.json_store(json);
            return quote! {
                if let ::core::option::Option::Some(value) = #ident {
                    inner.#ident = #store;
                }
            };
        }
        // Bind `value` for the store expression.
        let store = self.write_expr();
        if self.py_type_is_optional() {
            // The whole `Option` argument is stored; an omitted one is `None`.
            let bind = bind_value(ident);
            quote! {
                {
                    #bind
                    inner.#ident = #store;
                }
            }
        } else {
            quote! {
                if let ::core::option::Option::Some(value) = #ident {
                    inner.#ident = #store;
                }
            }
        }
    }

    /// Singular message write: `value` is always `Option<Wrapper>` (see
    /// [`Self::is_singular_message`]); store into the runtime's optional-message
    /// representation.
    fn write_message(&self) -> TokenStream {
        match self.runtime {
            // buffa: MessageField<T>
            Runtime::Buffa => quote! {
                value.map(|w| ::buffa::MessageField::some(w.into())).unwrap_or_default()
            },
            // prost recursion-boxed field: Option<Box<T>> — re-box on store.
            Runtime::Prost if self.is_boxed => quote! {
                value.map(|w| ::std::boxed::Box::new(w.into()))
            },
            // prost plain singular message: Option<T>.
            Runtime::Prost => quote! {
                value.map(|w| w.into())
            },
        }
    }
}

// ---------------------------------------------------------------------------
// Oneof accessors
// ---------------------------------------------------------------------------

/// A proto `oneof`, resolved into the tokens needed to expose it across the PyO3
/// boundary. Each oneof is *flattened*: one optional keyword arg + one `<variant>`
/// getter per member (issue #99).
///
/// The inner model stores a oneof as `Option<Enum>` where `Enum` has one tuple
/// variant per member. buffa reaches the enum through a natural-path re-export in
/// the message's snake_case module and boxes *every* message/group payload
/// (`Variant(Box<T>)`); prost nests the enum in the same module and stores message
/// payloads inline (`Variant(T)`), boxing only recursion-cycle payloads. The
/// message write/read paths branch on [`Runtime`] to add/strip that `Box`
/// (`Box::new`/`**` for buffa, direct/`*` for prost); scalar and enum payloads are
/// inline for both. Payloads cross the boundary via the wrapper's `From`/`Into`.
struct OneofAccessor {
    /// The oneof field ident on the inner struct (e.g. `provider`).
    field_ident: proc_macro2::Ident,
    /// In-crate path to the oneof enum (e.g. `super::catalog::v1::storage_config::Provider`).
    enum_path: TokenStream,
    runtime: Runtime,
    variants: Vec<OneofVariantAccessor>,
}

/// One member of a oneof, resolved to its Python-visible payload type and the
/// Rust variant it maps to.
struct OneofVariantAccessor {
    /// The Rust enum variant ident (PascalCase proto field name, e.g. `S3`).
    variant_ident: proc_macro2::Ident,
    /// The flattened keyword-arg / accessor ident (the proto field name, e.g. `s3`).
    param_ident: proc_macro2::Ident,
    /// The Python-visible payload wrapper type (e.g. `PyS3Config`, `i64`).
    payload_ty: TokenStream,
    /// Whether the payload is an in-project message (needs `.into()`/`From` bridge)
    /// vs. a scalar (stored/read directly).
    is_message: bool,
    /// Whether the payload is an in-project enum (crosses via the enum wrapper's
    /// `From`, and is stored inline — never boxed).
    is_enum: bool,
    /// Documentation carried from the proto variant, for the getter.
    documentation: Option<String>,
}

impl OneofAccessor {
    fn new(
        field: &MessageField,
        inner: &syn::Path,
        runtime: Runtime,
        packages: &BTreeSet<String>,
    ) -> Self {
        let field_ident =
            crate::utils::field_ident(&super::sanitize_python_field_name(&field.name));

        // The oneof enum's name in the IR is already `<msg_snake>::<OneofPascal>`
        // (built in `parsing/message.rs`); it lives in the message's parent module.
        // Splice it onto the message's own module path: drop the message leaf from
        // `inner` (e.g. `super::catalog::v1::StorageConfig`) and append the enum name.
        let BaseType::OneOf(enum_name) = &field.unified_type.base_type else {
            // Only ever constructed for oneof fields; fall back defensively.
            unreachable!("OneofAccessor::new called on a non-oneof field");
        };
        let enum_path = oneof_enum_path(inner, enum_name);

        let variants = field
            .oneof_variants
            .as_deref()
            .unwrap_or(&[])
            .iter()
            .map(|v| OneofVariantAccessor::new(v, packages))
            .collect();

        Self {
            field_ident,
            enum_path,
            runtime,
            variants,
        }
    }

    /// The flattened keyword-arg idents (proto field names), for the `#[pyo3(signature)]`.
    fn variant_param_idents(&self) -> Vec<proc_macro2::Ident> {
        self.variants
            .iter()
            .map(|v| v.param_ident.clone())
            .collect()
    }

    /// One `Option<PyPayload>` constructor param per variant.
    fn ctor_params(&self) -> Vec<TokenStream> {
        self.variants
            .iter()
            .map(|v| {
                let ident = &v.param_ident;
                let ty = &v.payload_ty;
                quote! { #ident: ::core::option::Option<#ty> }
            })
            .collect()
    }

    /// Constructor body: set the oneof from whichever variant kwarg is `Some`.
    /// Emitted in declaration order, so a later variant overrides an earlier one.
    fn ctor_assigns(&self) -> Vec<TokenStream> {
        let field_ident = &self.field_ident;
        self.variants
            .iter()
            .map(|v| {
                let param = &v.param_ident;
                let store = v.write_variant(&self.enum_path, self.runtime);
                quote! {
                    if let ::core::option::Option::Some(value) = #param {
                        inner.#field_ident = ::core::option::Option::Some(#store);
                    }
                }
            })
            .collect()
    }

    /// A `#[getter]` per variant returning `Option<PyPayload>` — `Some` only when
    /// that variant is the active one.
    fn variant_getters(&self) -> Vec<TokenStream> {
        let field_ident = &self.field_ident;
        self.variants
            .iter()
            .map(|v| {
                let getter = &v.param_ident;
                let ty = &v.payload_ty;
                let variant = &v.variant_ident;
                let enum_path = &self.enum_path;
                let read = v.read_variant(self.runtime);
                let doc = v
                    .documentation
                    .as_deref()
                    .map(|d| quote! { #[doc = #d] })
                    .unwrap_or_default();
                quote! {
                    #doc
                    #[getter]
                    fn #getter(&self) -> ::core::option::Option<#ty> {
                        match &self.0.#field_ident {
                            ::core::option::Option::Some(#enum_path::#variant(value)) => {
                                ::core::option::Option::Some(#read)
                            }
                            _ => ::core::option::Option::None,
                        }
                    }
                }
            })
            .collect()
    }
}

impl OneofVariantAccessor {
    fn new(variant: &OneofVariant, packages: &BTreeSet<String>) -> Self {
        let param_ident =
            crate::utils::field_ident(&super::sanitize_python_field_name(&variant.field_name));
        let variant_ident = format_ident!("{}", variant.variant_name);

        let (is_message, is_enum) = match &variant.field_type.base_type {
            BaseType::Message(n) => (is_in_project(n.as_str(), packages), false),
            BaseType::Enum(n) => (false, is_in_project(n.as_str(), packages)),
            _ => (false, false),
        };
        let in_project = is_message || is_enum;
        let payload_ty = py_type_for(&variant.field_type, in_project);

        Self {
            variant_ident,
            param_ident,
            payload_ty,
            is_message,
            is_enum,
            documentation: variant.documentation.clone(),
        }
    }

    /// Expression producing `<EnumPath>::<Variant>(payload)` from a bound `value`
    /// of the Python-visible payload type.
    fn write_variant(&self, enum_path: &TokenStream, runtime: Runtime) -> TokenStream {
        let variant = &self.variant_ident;
        if self.is_message {
            // Message payloads bridge via `From<PyWrapper>` for the bare model type.
            // buffa boxes *every* message variant; prost stores it inline (matching
            // the builder's `generate_oneof_variant_methods`). A prost oneof whose
            // variant is recursive would be boxed by prost and mismatch here — the
            // same untested corner the builder shares; no IR flag distinguishes it.
            match runtime {
                Runtime::Buffa => quote! {
                    #enum_path::#variant(::std::boxed::Box::new(value.into()))
                },
                Runtime::Prost => quote! {
                    #enum_path::#variant(value.into())
                },
            }
        } else if self.is_enum {
            // Enum payloads: the wrapper converts to the bare model enum via `From`.
            quote! {
                #enum_path::#variant(value.into())
            }
        } else {
            // Scalar payloads are stored directly.
            quote! {
                #enum_path::#variant(value)
            }
        }
    }

    /// Expression converting the matched variant payload — bound as `value` by ref
    /// in the getter's `match` arm — into the Python-visible type.
    fn read_variant(&self, runtime: Runtime) -> TokenStream {
        let ty = &self.payload_ty;
        if self.is_message {
            // buffa boxes message variants (`value: &Box<T>`), prost does not
            // (`value: &T`). Deref to `T` accordingly, clone, then bridge to the
            // wrapper via `From<T>`.
            match runtime {
                Runtime::Buffa => quote! { <#ty>::from((**value).clone()) },
                Runtime::Prost => quote! { <#ty>::from((*value).clone()) },
            }
        } else if self.is_enum {
            // `value: &BareEnum` (`Copy`); the wrapper bridges via `From<BareEnum>`.
            quote! { <#ty>::from(*value) }
        } else {
            // Scalar payloads: clone the referenced value out.
            quote! { ::core::clone::Clone::clone(value) }
        }
    }
}

/// Build the in-crate Rust path to a oneof enum, given the containing message's
/// path (`inner`) and the enum's IR name (`<msg_snake>::<OneofPascal>`).
///
/// The enum sits in the message's *parent* module — buffa via a natural-path
/// re-export, prost via direct nesting — so we drop the message's own leaf
/// segment from `inner` (e.g. `super::catalog::v1::StorageConfig` →
/// `super::catalog::v1`) and append the enum name (`storage_config::Provider`).
fn oneof_enum_path(inner: &syn::Path, enum_name: &str) -> TokenStream {
    let parent_segments = inner
        .segments
        .iter()
        .take(inner.segments.len().saturating_sub(1))
        .map(|s| s.ident.clone());
    let tail: syn::Path =
        syn::parse_str(enum_name).expect("oneof enum IR name is a valid `snake::Pascal` path");
    quote! { #(#parent_segments::)* #tail }
}

/// The Python-visible base type (no cardinality wrapping) for a unified type.
fn py_type_for(ty: &UnifiedType, in_project: bool) -> TokenStream {
    match &ty.base_type {
        BaseType::String => quote! { ::std::string::String },
        BaseType::Bytes => quote! { ::std::vec::Vec<u8> },
        BaseType::Int32 => quote! { i32 },
        BaseType::Int64 => quote! { i64 },
        BaseType::Bool => quote! { bool },
        BaseType::Float32 => quote! { f32 },
        BaseType::Float64 => quote! { f64 },
        BaseType::Unit => quote! { () },
        BaseType::Message(n) | BaseType::Enum(n) if in_project => {
            let w = wrapper_ident(n);
            quote! { #w }
        }
        // A foreign type (no wrapper) or a oneof enum: the bare simple name.
        BaseType::Message(n) | BaseType::Enum(n) | BaseType::OneOf(n) => {
            let id = format_ident!("{}", extract_simple_type_name(n));
            quote! { #id }
        }
        BaseType::Map(_, v) => {
            let val = py_type_for(v, in_project);
            quote! { ::std::collections::HashMap<::std::string::String, #val> }
        }
    }
}

/// Read a `map<string, V>` field into a `HashMap`, converting message values to
/// their wrappers when in-project.
fn read_map(access: &TokenStream, value: &UnifiedType, in_project: bool) -> TokenStream {
    match &value.base_type {
        BaseType::Message(_) if in_project => {
            let w = py_type_for(value, true);
            quote! {
                #access.iter().map(|(k, v)| (k.clone(), #w::from(v.clone()))).collect()
            }
        }
        _ => quote! { #access.clone() },
    }
}

/// `let value = <argument>;` for the store expressions; nothing for an argument named `value`.
fn bind_value(ident: &proc_macro2::Ident) -> Option<TokenStream> {
    (ident != "value").then(|| quote! { let value = #ident; })
}

/// The keyword `#[new]` constructor: every plain field is an optional keyword argument, and each
/// oneof is flattened into one optional argument per variant (the last one given wins). With
/// `keyword_only`, the `REQUIRED` fields come first as positional arguments.
fn constructor(
    accessors: &[FieldAccessor],
    oneofs: &[OneofAccessor],
    inner: &syn::Path,
    keyword_only: bool,
) -> TokenStream {
    if accessors.is_empty() && oneofs.is_empty() {
        return quote! {
            #[new]
            fn new() -> Self {
                Self(::core::default::Default::default())
            }
        };
    }

    let is_required = |a: &&FieldAccessor| keyword_only && a.required;
    let (required, optional): (Vec<&FieldAccessor>, Vec<&FieldAccessor>) =
        accessors.iter().partition(is_required);

    let required_params = required.iter().map(|a| a.required_ctor_param());
    let optional_params = optional.iter().map(|a| a.optional_ctor_param());
    let oneof_params = oneofs.iter().flat_map(OneofAccessor::ctor_params);
    let params = required_params.chain(optional_params).chain(oneof_params);

    // Signature: the required arguments, `*` with `keyword_only`, then `<name> = None` for the
    // other fields and the oneof variants.
    let optional_idents: Vec<proc_macro2::Ident> = optional
        .iter()
        .map(|a| a.ident.clone())
        .chain(oneofs.iter().flat_map(OneofAccessor::variant_param_idents))
        .collect();
    let mut sig_parts: Vec<TokenStream> = required
        .iter()
        .map(|a| {
            let ident = &a.ident;
            quote! { #ident }
        })
        .collect();
    if keyword_only && !optional_idents.is_empty() {
        sig_parts.push(quote! { * });
    }
    sig_parts.extend(optional_idents.iter().map(|ident| quote! { #ident = None }));
    let sig = quote! { #(#sig_parts),* };

    let required_assigns = required.iter().map(|a| a.required_ctor_assign());
    let optional_assigns = optional.iter().map(|a| a.optional_ctor_assign());
    let oneof_assigns = oneofs.iter().flat_map(OneofAccessor::ctor_assigns);
    // A free-form JSON argument can fail to convert.
    let (ret, result) = if accessors.iter().any(|a| a.json.is_some()) {
        (
            quote! { ::pyo3::PyResult<Self> },
            quote! { ::core::result::Result::Ok(Self(inner)) },
        )
    } else {
        (quote! { Self }, quote! { Self(inner) })
    };

    quote! {
        #[new]
        #[pyo3(signature = (#sig))]
        fn new(#(#params,)*) -> #ret {
            let mut inner = <#inner as ::core::default::Default>::default();
            #(#required_assigns)*
            #(#optional_assigns)*
            #(#oneof_assigns)*
            #result
        }
    }
}

// ---------------------------------------------------------------------------
// Path helpers
// ---------------------------------------------------------------------------

fn parse_path(rust_path: &str, fq_name: &str) -> crate::error::Result<syn::Path> {
    syn::parse_str(rust_path).map_err(|e| crate::error::Error::InvalidRustPath {
        path: rust_path.to_string(),
        message: fq_name.to_string(),
        source: e,
    })
}

/// Whether a fully-qualified type name belongs to one of the project's packages.
fn is_in_project(fq_name: &str, packages: &BTreeSet<String>) -> bool {
    let stripped = fq_name.strip_prefix('.').unwrap_or(fq_name);
    packages.iter().any(|pkg| {
        stripped == pkg
            || stripped
                .strip_prefix(pkg.as_str())
                .is_some_and(|rest| rest.starts_with('.'))
    })
}

/// Map a fully-qualified proto type name to its in-crate Rust path, as exposed by
/// the model module tree `generate_models_mod` emits.
///
/// `pyo3_impls.rs` is a sibling of the per-package model modules inside the models
/// subdir, so the path is `super::<svc_seg>::<ver_seg>::<...nested>::<Name>` where
/// `(svc_seg, ver_seg)` are the second-to-last and last segments of the owning
/// package — exactly the module nesting `generate_models_mod` produces.
///
/// Returns `None` for types outside the project's own packages (e.g. imported
/// google/gnostic types), which have no in-crate path and must be skipped.
fn type_rust_path(fq_name: &str, project_packages: &BTreeSet<&str>) -> Option<String> {
    let stripped = fq_name.strip_prefix('.').unwrap_or(fq_name);
    let package = package_of(fq_name, project_packages)?;
    let remainder = stripped.strip_prefix(package)?.trim_start_matches('.');
    if remainder.is_empty() {
        return None;
    }

    let pkg_parts: Vec<&str> = package.split('.').collect();
    let module_path = if pkg_parts.len() >= 2 {
        format!(
            "{}::{}",
            pkg_parts[pkg_parts.len() - 2],
            pkg_parts[pkg_parts.len() - 1]
        )
    } else {
        pkg_parts[0].to_string()
    };

    // `remainder` is the dot-joined type path within the package, e.g.
    // `GenerateTableCredentialsRequest.Operation` for a message-nested enum. Both
    // the prost and buffa model trees nest a type defined inside a message under a
    // *snake_case module* named after that parent message, with only the leaf type
    // keeping its proto (PascalCase) name. So snake_case every segment except the
    // last (`generate_table_credentials_request::Operation`), not a flat `::` join.
    let segments: Vec<&str> = remainder.split('.').collect();
    let (leaf, parents) = segments.split_last()?;
    let type_path = parents
        .iter()
        .map(|seg| seg.to_case(Case::Snake))
        .chain(std::iter::once((*leaf).to_string()))
        .collect::<Vec<_>>()
        .join("::");
    Some(format!("super::{module_path}::{type_path}"))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn pkgs(list: &[&'static str]) -> BTreeSet<&'static str> {
        list.iter().copied().collect()
    }

    #[test]
    fn oneof_enum_path_splices_enum_name_onto_message_parent_module() {
        // The oneof enum lives in the message's *parent* module: drop the message
        // leaf from `inner` and append the IR enum name (`<msg_snake>::<OneofPascal>`).
        let inner: syn::Path = syn::parse_str("super::catalog::v1::StorageConfig").unwrap();
        let path = oneof_enum_path(&inner, "storage_config::Provider");
        assert_eq!(
            path.to_string().replace(' ', ""),
            "super::catalog::v1::storage_config::Provider"
        );
    }

    #[test]
    fn oneof_enum_path_handles_top_level_message() {
        // A message at the crate root (single-segment `inner`) still resolves: the
        // parent module is empty, so the enum name is the whole path.
        let inner: syn::Path = syn::parse_str("Shape").unwrap();
        let path = oneof_enum_path(&inner, "shape::Kind");
        assert_eq!(path.to_string().replace(' ', ""), "shape::Kind");
    }

    #[test]
    fn three_segment_package_maps_to_service_version_module() {
        let p = pkgs(&["example.catalog.v1"]);
        assert_eq!(
            type_rust_path(".example.catalog.v1.Catalog", &p).as_deref(),
            Some("super::catalog::v1::Catalog")
        );
    }

    #[test]
    fn two_segment_package_keeps_first_segment_as_module() {
        let p = pkgs(&["demo.v1"]);
        assert_eq!(
            type_rust_path(".demo.v1.Widget", &p).as_deref(),
            Some("super::demo::v1::Widget")
        );
    }

    #[test]
    fn nested_type_lives_in_snake_cased_parent_module() {
        let p = pkgs(&["example.catalog.v1"]);
        // A type defined inside a message is nested under a snake_case module named
        // after the parent message; only the leaf keeps its proto name.
        assert_eq!(
            type_rust_path(".example.catalog.v1.Catalog.Nested", &p).as_deref(),
            Some("super::catalog::v1::catalog::Nested")
        );
    }

    #[test]
    fn deeply_nested_type_snake_cases_every_parent() {
        let p = pkgs(&["example.creds.v1"]);
        assert_eq!(
            type_rust_path(
                ".example.creds.v1.GenerateTableCredentialsRequest.Operation",
                &p
            )
            .as_deref(),
            Some("super::creds::v1::generate_table_credentials_request::Operation")
        );
    }

    #[test]
    fn foreign_package_is_skipped() {
        let p = pkgs(&["example.catalog.v1"]);
        assert_eq!(type_rust_path(".google.protobuf.Any", &p), None);
    }

    #[test]
    fn wrapper_ident_is_py_prefixed_qualified_name() {
        assert_eq!(
            wrapper_ident(".example.catalog.v1.Catalog").to_string(),
            "PyCatalog"
        );
        assert_eq!(
            wrapper_ident(".example.catalog.v1.Catalog.Nested").to_string(),
            "PyCatalogNested"
        );
    }

    // --- boxing detection -------------------------------------------------

    fn msg_field(name: &str, ty: UnifiedType) -> MessageField {
        MessageField {
            name: name.to_string(),
            json_name: None,
            response_header: None,
            request_header: None,
            emit_unpopulated: false,
            raw_body: false,
            unified_type: ty,
            documentation: None,
            oneof_variants: None,
            field_behavior: vec![],
            is_sensitive: false,
            resource_reference: None,
        }
    }

    fn singular_message(target: &str) -> UnifiedType {
        UnifiedType {
            base_type: BaseType::Message(target.to_string()),
            is_optional: true,
            is_repeated: false,
        }
    }

    fn repeated_message(target: &str) -> UnifiedType {
        UnifiedType {
            base_type: BaseType::Message(target.to_string()),
            is_optional: false,
            is_repeated: true,
        }
    }

    fn metadata_with(messages: Vec<(&str, Vec<MessageField>)>) -> CodeGenMetadata {
        let mut meta = CodeGenMetadata::default();
        for (name, fields) in messages {
            meta.messages.insert(
                name.to_string(),
                MessageInfo {
                    name: name.to_string(),
                    fields,
                    resource_descriptor: None,
                    documentation: None,
                },
            );
        }
        meta
    }

    #[test]
    fn non_recursive_singular_message_is_not_boxed() {
        // A.f: B, B has no message fields -> no cycle -> unboxed.
        let meta = metadata_with(vec![
            (".pkg.A", vec![msg_field("f", singular_message(".pkg.B"))]),
            (".pkg.B", vec![]),
        ]);
        assert!(boxed_fields(&meta, Runtime::Prost).is_empty());
    }

    #[test]
    fn direct_self_recursion_boxes_the_field() {
        // A.next: A -> cycle -> boxed.
        let meta = metadata_with(vec![(
            ".pkg.A",
            vec![msg_field("next", singular_message(".pkg.A"))],
        )]);
        let boxed = boxed_fields(&meta, Runtime::Prost);
        assert!(boxed.contains(&(".pkg.A".to_string(), "next".to_string())));
        assert_eq!(boxed.len(), 1);
    }

    #[test]
    fn mutual_recursion_boxes_both_back_edges() {
        // A.b: B and B.a: A -> A<->B cycle -> both boxed.
        let meta = metadata_with(vec![
            (".pkg.A", vec![msg_field("b", singular_message(".pkg.B"))]),
            (".pkg.B", vec![msg_field("a", singular_message(".pkg.A"))]),
        ]);
        let boxed = boxed_fields(&meta, Runtime::Prost);
        assert!(boxed.contains(&(".pkg.A".to_string(), "b".to_string())));
        assert!(boxed.contains(&(".pkg.B".to_string(), "a".to_string())));
    }

    #[test]
    fn repeated_field_breaks_the_cycle_and_is_not_boxed() {
        // A.items: repeated A -> Vec breaks the size cycle -> never boxed.
        let meta = metadata_with(vec![(
            ".pkg.A",
            vec![msg_field("items", repeated_message(".pkg.A"))],
        )]);
        assert!(boxed_fields(&meta, Runtime::Prost).is_empty());
    }

    #[test]
    fn buffa_runtime_never_boxes() {
        let meta = metadata_with(vec![(
            ".pkg.A",
            vec![msg_field("next", singular_message(".pkg.A"))],
        )]);
        assert!(boxed_fields(&meta, Runtime::Buffa).is_empty());
    }
}