olai_codegen/codegen/

mod.rs

//! Code generation module for REST API handlers and language bindings.
//!
//! This module provides the core functionality for generating Rust code from
//! protobuf metadata extracted from service definitions.
//!
//! ## Architecture
//!
//! The code generation process follows these phases:
//! 1. **Analysis**: Process collected metadata to understand service structure
//! 2. **Planning**: Determine what code needs to be generated
//! 3. **Generation**: Create Rust code using templates and metadata
//! 4. **Output**: Write generated code to appropriate files
//!
//! ## Generated Code Types
//!
//! - **Handler Traits**: Async trait definitions for service operations
//! - **Request Extractors**: Axum FromRequest/FromRequestParts implementations
//! - **Route Handlers**: Axum handler functions that delegate to traits
//! - **Client Code**: HTTP client implementations for services
//! - **Type Mappings**: Conversions between protobuf and Rust types

use std::collections::{BTreeMap, BTreeSet, HashMap};

use convert_case::{Case, Casing};
use itertools::Itertools;
use proc_macro2::TokenStream;
use quote::{format_ident, quote};
use syn::Ident;

use crate::error::{Error, Result};

use crate::analysis::{
    BodyField, GenerationPlan, ManagedResource, MethodPlan, RequestParam, RequestType, ServicePlan,
    analyze_metadata, split_body_fields,
};
use crate::output;
use crate::parsing::types::{self, RenderContext, UnifiedType};
use crate::parsing::{CodeGenMetadata, MessageField, MessageInfo};

mod aggregate;
mod bindings;
mod builder;
mod client;
mod config;
mod connect_client;
mod handler;
pub(crate) mod node;
mod python;
mod resource_client;
mod resources;
mod server;
mod tokens;
mod wasm;

pub use config::{
    BindingsConfig, ClientProtocol, ClientProtocols, CodeGenConfig, CodeGenOutput,
    DEFAULT_TRANSPORT_TYPE_PATH, ModelsPath, PythonApi, Runtime,
};
pub(crate) use tokens::{doc_tokens, format_tokens, format_tokens_static};

/// How a language binding lowers a method's call into the underlying Rust client.
///
/// Resource services expose a scoped client (e.g. `CatalogClient`) that captures path params;
/// resource-less services have no such client, so their methods live on the root client and must
/// pass every param themselves (the "flat" shape, treated like collection methods).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BindingMode {
    /// Method lives on a resource-scoped client that already holds the path params.
    Scoped,
    /// Method lives on the root client and receives all params, including path params.
    Flat,
}

/// Language-agnostic description of a service's resource accessor method on the aggregate client.
///
/// Computed once from the IR (`ServicePlan::resource_accessor_params`) and rendered by each language
/// emitter, so the param list and nested-vs-flat decision are shared by construction.
pub(crate) struct AccessorSpec {
    /// The resource singular, used as the accessor method name (e.g. `catalog`).
    pub(crate) singular: String,
    /// The ordered accessor params (ancestors + `<singular>_name` leaf), e.g.
    /// `["catalog_name", "schema_name"]` or `["name"]`.
    pub(crate) params: Vec<String>,
    /// Whether the resource is nested (has parent components); its full name is the dot-joined
    /// params and the accessor delegates through a `<singular>_from_full_name` helper.
    pub(crate) nested: bool,
}

impl AccessorSpec {
    /// The `format!` template for joining the params into a composite full name
    /// (e.g. `"{}.{}"` for two params). Only meaningful when [`Self::nested`] is true.
    pub(crate) fn join_format(&self) -> String {
        std::iter::repeat_n("{}", self.params.len())
            .collect::<Vec<_>>()
            .join(".")
    }
}

impl MethodPlan {
    pub(crate) fn resource_client_method(&self) -> Ident {
        match &self.scoped_verb {
            Some(verb) => format_ident!("{}", verb),
            None => format_ident!("{}", self.handler_function_name),
        }
    }

    pub(crate) fn base_method_ident(&self) -> Ident {
        format_ident!("{}", self.handler_function_name)
    }
}

/// Generate all code described by `config` from `metadata`.
///
/// Writes the following outputs, depending on which [`CodeGenOutput`] fields are `Some`:
///
/// | Field | Contents |
/// |-------|----------|
/// | `output.common` | Axum extractor code, per-service `mod.rs` (always written) |
/// | `output.models_gen` | `labels.rs` resource-enum file (falls back to `common` if `None`) |
/// | `output.server` | Handler trait + Axum route wiring per service |
/// | `output.client` | HTTP client structs and request builders per service |
/// | `output.python` | PyO3 binding wrappers + `.pyi` typings stub |
/// | `output.node` | NAPI binding wrappers |
/// | `output.node_ts` | TypeScript client (`client.ts`) |
///
/// # Required fields
///
/// - `output.common` is always required.
/// - `bindings` must be `Some` when any of `output.python`, `output.node`, or `output.node_ts`
///   is `Some`; otherwise returns [`Error::MissingBindingsConfig`].
/// - `models_path_template` and `models_path_crate_template` must be valid Rust path templates
///   (containing `{service}`); invalid templates return [`Error::InvalidModelsPathTemplate`].
///
/// # Optional fields
///
/// Setting `generate_resource_enum` to `false` skips `labels.rs` generation.
/// Setting `bindings` to `None` skips all language-binding output.
///
/// Call [`CodeGenConfig::validate`] before this function to surface config errors at
/// construction time rather than mid-generation.
///
/// # Errors
///
/// Returns:
///
/// - [`Error::InvalidModelsPathTemplate`] if `models_path_template` or `models_path_crate_template`
///   is not a valid Rust path template.
/// - [`Error::InvalidConfigPath`] if `context_type_path` or `result_type_path` is not a valid Rust
///   path.
/// - [`Error::MissingBindingsConfig`] if any of the `python`, `node`, or `node_ts` outputs is
///   requested without a `bindings` config.
/// - [`Error::MissingAnnotation`] or [`Error::Build`] surfaced while analyzing the metadata (e.g. a
///   required `google.api` annotation is missing, or a service cannot be planned).
/// - [`Error::InvalidRustPath`] if a proto message or field maps to an invalid Rust path, and
///   [`Error::InvalidErrorTypePath`] if the configured `error_type_path` is not a valid Rust path.
/// - [`Error::GeneratedParse`] if an internal generator produces tokens that do not parse as a Rust
///   file.
/// - [`Error::Io`] if creating the models directory or writing any generated file fails.
pub fn generate_code(metadata: &CodeGenMetadata, config: &CodeGenConfig) -> Result<()> {
    // Validate templates early so callers get a clean error before any generation starts.
    let models_path = ModelsPath::new(&config.models_path_template)?;
    let models_path_crate = ModelsPath::new(&config.models_path_crate_template)?;

    // The context/result type paths come from user config and are parsed as `syn::Path` deep in
    // the per-method generators (handler/client/server/builder). Validate them once here so a
    // typo surfaces as a clean error instead of panicking mid-generation.
    for (field, path) in [
        ("context_type", &config.context_type_path),
        ("result_type", &config.result_type_path),
    ] {
        syn::parse_str::<syn::Path>(path).map_err(|source| Error::InvalidConfigPath {
            field,
            path: path.clone(),
            source,
        })?;
    }

    // Validate that bindings config is present when language-binding output is requested.
    if (config.output.python.is_some()
        || config.output.node.is_some()
        || config.output.node_ts.is_some())
        && config.bindings.is_none()
    {
        return Err(Error::MissingBindingsConfig);
    }

    // The Connect client emitter needs the connect-rust client's import path.
    if config.client_protocols.connect
        && config.output.client.is_some()
        && config.connect_client_path.is_none()
    {
        return Err(Error::MissingConnectClientPath);
    }

    let plan = analyze_metadata(metadata)?;

    // Validate every service's `{service}` substitution up front so a base path that isn't a
    // valid Rust path segment fails cleanly here rather than panicking inside per-service codegen.
    for service in &plan.services {
        models_path.validate_for(&service.base_path)?;
        models_path_crate.validate_for(&service.base_path)?;
    }

    let common_code = generate_common_code(&plan, metadata, config)?;
    output::write_generated_code(&common_code, &config.output.common)?;

    if config.output.models.is_some() {
        let subdir = config
            .output
            .models_subdir_path()
            .expect("models is Some so subdir is Some");
        std::fs::create_dir_all(&subdir).map_err(Error::Io)?;

        if config.generate_resource_enum {
            let resource_enum = resources::generate_resource_enum(
                &plan,
                metadata,
                config,
                config.error_type_path.as_deref(),
            )?;
            let mut models_files = GeneratedCode {
                files: std::collections::HashMap::new(),
            };
            models_files
                .files
                .insert("labels.rs".to_string(), resource_enum);
            output::write_generated_code(&models_files, &subdir)?;
        }

        // When Python bindings are generated, the bindings pass model types
        // across the FFI boundary by their bare type, so the model crate must
        // carry `FromPyObject`/`IntoPyObject` impls. Emit them here (orphan rule:
        // the model types are local to this crate). Gated on the `python` feature
        // in the generated `mod.rs`.
        let mut include_pyo3_impls = false;
        if config.output.python.is_some() {
            let pyo3_impls = python::generate_pyo3_impls(&plan, metadata, config.runtime)?;
            if !pyo3_impls.is_empty() {
                let mut pyo3_files = GeneratedCode {
                    files: std::collections::HashMap::new(),
                };
                pyo3_files
                    .files
                    .insert("pyo3_impls.rs".to_string(), pyo3_impls);
                output::write_generated_code(&pyo3_files, &subdir)?;
                include_pyo3_impls = true;
            }
        }

        let include_labels = config.generate_resource_enum;
        // The Axum extractors are co-located in this same directory (the common
        // generator wrote `<svc>/server.rs` + `<svc>/mod.rs` here), and this
        // `mod.rs` overwrites the common generator's top-level `mod.rs` — so it
        // re-declares the per-service extractor submodules it emitted.
        let mod_content = generate_models_mod(
            &plan.services,
            include_labels,
            include_pyo3_impls,
            metadata,
            config.runtime,
        );
        // Route through write_generated_code so the `// @generated` header is prepended like
        // every other emitted file, rather than writing it raw.
        let mut mod_files = GeneratedCode {
            files: std::collections::HashMap::new(),
        };
        mod_files.files.insert("mod.rs".to_string(), mod_content);
        output::write_generated_code(&mod_files, &subdir)?;
    }

    if let Some(ref server_dir) = config.output.server {
        let server_code = generate_server_code(&plan, metadata, config)?;
        output::write_generated_code(&server_code, server_dir)?;
    }

    if let Some(ref client_dir) = config.output.client {
        let client_code = generate_client_code(&plan, metadata, config)?;
        output::write_generated_code(&client_code, client_dir)?;
    }

    if let Some(ref python_dir) = config.output.python {
        let python_code = generate_python_code(&plan, metadata, config)?;
        output::write_generated_code(&python_code, python_dir)?;
    }

    if let Some(ref node_dir) = config.output.node {
        let node_code = generate_node_code(&plan, metadata, config)?;
        output::write_generated_code(&node_code, node_dir)?;
    }

    if let Some(ref node_ts_dir) = config.output.node_ts {
        let node_ts_code = generate_node_ts_code(&plan, metadata, config)?;
        output::write_generated_code(&node_ts_code, node_ts_dir)?;
    }

    if let Some(ref wasm_dir) = config.output.wasm {
        let wasm_code = generate_wasm_code(&plan, metadata, config)?;
        output::write_generated_code(&wasm_code, wasm_dir)?;
    }

    Ok(())
}

fn generate_common_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let mut files = HashMap::new();

    for service in &plan.services {
        let handler = ServiceHandler {
            plan: service,
            metadata,
            config,
        };
        let server_code = server::generate_common(&handler)?;
        // Nest the extractors under an `extractors/` subdir so their per-service
        // module names can't collide with the co-located model package modules
        // (e.g. a `unitycatalog.agent_skills.v0alpha1` package yields a model
        // module `agent_skills` that would otherwise clash with the `agent_skills`
        // service extractor module).
        files.insert(
            format!("{EXTRACTORS_SUBDIR}/{}/server.rs", service.base_path),
            server_code,
        );
        let module_code = generate_common_module();
        files.insert(
            format!("{EXTRACTORS_SUBDIR}/{}/mod.rs", service.base_path),
            module_code,
        );
    }

    // The `extractors/mod.rs` that declares the per-service extractor submodules.
    files.insert(
        format!("{EXTRACTORS_SUBDIR}/mod.rs"),
        generate_extractors_module(&plan.services),
    );

    let module_code = main_module(&plan.services);
    files.insert("mod.rs".to_string(), module_code);

    Ok(GeneratedCode { files })
}

fn generate_server_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let mut files = HashMap::new();

    for service in &plan.services {
        let handler = ServiceHandler {
            plan: service,
            metadata,
            config,
        };
        let trait_code = handler::generate(&handler)?;
        files.insert(format!("{}/handler.rs", service.base_path), trait_code);
        let server_code = server::generate_server(&handler)?;
        files.insert(format!("{}/server.rs", service.base_path), server_code);
        let module_code = generate_server_module(service);
        files.insert(format!("{}/mod.rs", service.base_path), module_code);
    }

    let module_code = main_module(&plan.services);
    files.insert("mod.rs".to_string(), module_code);

    Ok(GeneratedCode { files })
}

fn generate_python_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let mut files = HashMap::new();

    let handlers = plan
        .services
        .iter()
        .map(|service| ServiceHandler {
            plan: service,
            metadata,
            config,
        })
        .collect_vec();

    // Per-service scoped client modules are only meaningful for services with a scoped binding.
    // The methods of resource-less and flat-lowered services live on the root client (emitted by
    // `main_module`), so they don't get a scoped module of their own.
    for service in handlers.iter().filter(|s| s.emits_scoped_binding()) {
        let python_code = python::generate(service)?;
        files.insert(format!("{}.rs", service.plan.base_path), python_code);
    }

    let module_code = python::main_module(&handlers)?;
    files.insert("mod.rs".to_string(), module_code);

    let python_api = config
        .bindings
        .as_ref()
        .map(|b| b.python_api)
        .unwrap_or_default();
    files.insert(
        "runtime.rs".to_string(),
        python::generate_runtime(python_api),
    );

    let python_typings_code = python::generate_typings(&handlers);
    files.insert(
        config.output.python_typings_filename.clone(),
        python_typings_code,
    );

    Ok(GeneratedCode { files })
}

fn generate_node_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let mut files = HashMap::new();

    let handlers = plan
        .services
        .iter()
        .map(|service| ServiceHandler {
            plan: service,
            metadata,
            config,
        })
        .collect_vec();

    // See `generate_python_code`: only services with a scoped binding get a per-service scoped
    // module; resource-less and flat-lowered services' methods live on the root client.
    for service in handlers.iter().filter(|s| s.emits_scoped_binding()) {
        let napi_code = node::generate(service)?;
        files.insert(format!("{}.rs", service.plan.base_path), napi_code);
    }

    let module_code = node::main_module(&handlers)?;
    files.insert("mod.rs".to_string(), module_code);

    Ok(GeneratedCode { files })
}

fn generate_node_ts_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let handlers = plan
        .services
        .iter()
        .map(|service| ServiceHandler {
            plan: service,
            metadata,
            config,
        })
        .collect_vec();

    let ts_code = node::typescript::generate_client_ts(&handlers);
    let mut files = HashMap::new();
    files.insert("client.ts".to_string(), ts_code);

    // The `models/index.ts` barrel re-exports every service's generated protobuf-es modules so
    // `client.ts`'s `from "./models"` imports (message and service request/response types) resolve.
    let models_barrel = node::typescript::generate_models_barrel(&handlers);
    files.insert("models/index.ts".to_string(), models_barrel);

    Ok(GeneratedCode { files })
}

/// Generate the WASM/browser `#[wasm_bindgen]` binding layer plus its `.d.ts`.
///
/// Emits `bindings.rs` (a `#[wasm_bindgen]` wrapper per service client, exchanging request/response
/// values as plain JS objects via `serde-wasm-bindgen`) and `client.d.ts` (TypeScript declarations
/// for JS/TS consumers). The wrapped clients are the generated low-level clients, which must be
/// generated with the WASM transport (`transport_type_path = "olai_http_wasm::WasmClient"`).
fn generate_wasm_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let handlers = plan
        .services
        .iter()
        .map(|service| ServiceHandler {
            plan: service,
            metadata,
            config,
        })
        .collect_vec();

    let mut files = HashMap::new();
    // Emit as `mod.rs` so the client crate mounts the bindings with a plain
    // `mod wasm;` (no `#[path]` alias) over the `wasm/` convention subdir.
    files.insert("mod.rs".to_string(), wasm::generate_bindings(&handlers)?);
    files.insert("client.d.ts".to_string(), wasm::generate_dts(&handlers));

    Ok(GeneratedCode { files })
}

fn generate_client_code(
    plan: &GenerationPlan,
    metadata: &CodeGenMetadata,
    config: &CodeGenConfig,
) -> Result<GeneratedCode> {
    let mut files = HashMap::new();
    let protocols = config.client_protocols.selected();
    let multi = config.client_protocols.is_multi();

    for service in &plan.services {
        let handler = ServiceHandler {
            plan: service,
            metadata,
            config,
        };

        // Emit each selected protocol's low-level client + builders. With a single protocol the
        // files are flat under `<service>/`; with both they go under `<service>/<protocol>/` to
        // avoid the two `client.rs`/`builders.rs` colliding. The shared builder layer binds to the
        // protocol's low-level client and only calls `client.<method>(&request)`.
        let mut emitted_resource_client = false;
        for &protocol in &protocols {
            let prefix = if multi {
                format!("{}/{}", service.base_path, protocol.subdir())
            } else {
                service.base_path.clone()
            };

            let client_code = match protocol {
                ClientProtocol::Rest => client::generate(&handler)?,
                ClientProtocol::Connect => connect_client::generate(&handler)?,
            };
            files.insert(format!("{prefix}/client.rs"), client_code);
            files.insert(
                format!("{prefix}/builders.rs"),
                builder::generate(&handler, protocol)?,
            );

            // Ergonomic resource-scoped clients are a REST-only feature (driven by
            // `google.api.resource` hierarchies). ConnectRPC-only protos rarely carry those, so the
            // Connect client emits just the flat per-service client + builders.
            let has_resource_client = protocol == ClientProtocol::Rest
                && config.output.generate_resource_clients
                && resource_client::generate(&handler, plan)?
                    .map(|code| {
                        files.insert(format!("{prefix}/resource.rs"), code);
                    })
                    .is_some();

            files.insert(
                format!("{prefix}/mod.rs"),
                generate_client_module(has_resource_client),
            );
            emitted_resource_client |= has_resource_client;
        }

        // When both protocols are emitted under subdirs, the `<service>/mod.rs` mounts and re-exports
        // them. (Single-protocol output already wrote a flat `<service>/mod.rs` above.)
        if multi {
            files.insert(
                format!("{}/mod.rs", service.base_path),
                generate_protocol_split_module(&protocols),
            );
        }
        let _ = emitted_resource_client;
    }

    // Top-level aggregate root client (e.g. `UnityCatalogClient`), emitted only when a bindings
    // config (which carries the aggregate name) is present.
    let has_aggregate = aggregate::generate(plan, metadata, config)?
        .map(|aggregate_code| {
            files.insert("client.rs".to_string(), aggregate_code);
        })
        .is_some();

    let module_code = generate_client_main_module(&plan.services, has_aggregate);
    files.insert("mod.rs".to_string(), module_code);

    Ok(GeneratedCode { files })
}

fn generate_common_module() -> String {
    let tokens = quote! {
        #[cfg(feature = "axum")]
        pub mod server;
    };
    format_tokens_static(tokens)
}

/// Subdirectory (under the co-located models `_gen/` dir) that holds the Axum
/// request extractors, one `<service>/server.rs` per service. Nesting them keeps
/// their module names from colliding with the model package modules that share
/// the dir.
pub(crate) const EXTRACTORS_SUBDIR: &str = "extractors";

/// The `extractors/mod.rs` that declares one `pub mod <service>;` per service.
/// Axum-coupled, so gated on the `axum` feature.
fn generate_extractors_module(services: &[ServicePlan]) -> String {
    let decls: Vec<TokenStream> = services
        .iter()
        .map(|s| format_ident!("{}", s.base_path))
        // Dedupe: multiple services can share a base_path.
        .collect::<BTreeSet<_>>()
        .into_iter()
        .map(|m| quote! { #[cfg(feature = "axum")] pub mod #m; })
        .collect();
    format_tokens_static(quote! { #(#decls)* })
}

fn generate_server_module(service: &ServicePlan) -> String {
    let handler_ident = format_ident!("{}", service.handler_name);
    let tokens = quote! {
        pub use handler::#handler_ident;

        mod handler;
        #[cfg(feature = "axum")]
        pub mod server;
    };
    format_tokens_static(tokens)
}

fn generate_client_module(has_resource_client: bool) -> String {
    // Emit the scoped resource client module only when it was generated for this service.
    // Generated client modules structurally trip a few lints: the full client
    // surface is emitted whether or not a given consumer calls every builder
    // (`dead_code`), flat methods can exceed the arg-count threshold
    // (`too_many_arguments`), and proto comment prose may wrap list items
    // (`doc_lazy_continuation`). Allow these on the module so consumers' own
    // `-D warnings` builds stay clean; genuine issues are fixed in the generator.
    let allow = quote! {
        #[allow(dead_code, clippy::too_many_arguments, clippy::doc_lazy_continuation)]
    };
    // Glob re-exports of the per-service submodules. Not every submodule
    // contributes a name that the parent's consumers use directly, so a
    // `pub use <mod>::*` can trip `unused_imports` under `-D warnings` (notably
    // under buffa, where some helper types are unreferenced); allow it.
    let reexport_allow = quote! { #[allow(unused_imports)] };
    let resource_module = if has_resource_client {
        quote! {
            #reexport_allow
            pub use resource::*;
            #allow
            pub mod resource;
        }
    } else {
        quote! {}
    };
    let tokens = quote! {
        #reexport_allow
        pub use client::*;
        #reexport_allow
        pub use builders::*;
        #resource_module

        #allow
        pub mod client;
        #allow
        pub mod builders;
    };
    format_tokens_static(tokens)
}

/// The `<service>/mod.rs` for a service that emits **both** protocols: mount each protocol's client
/// as its own submodule (`pub mod rest;` / `pub mod connect;`).
///
/// They are *not* glob-re-exported into the service module: both protocols define identically-named
/// types (`GetThingBuilder`, …), so a flat re-export would collide. Callers disambiguate by module
/// (`read::rest::GetThingBuilder` vs `read::connect::GetThingBuilder`).
fn generate_protocol_split_module(protocols: &[ClientProtocol]) -> String {
    let allow = quote! {
        #[allow(dead_code, clippy::too_many_arguments, clippy::doc_lazy_continuation)]
    };
    let mods: Vec<TokenStream> = protocols
        .iter()
        .map(|p| {
            let m = format_ident!("{}", p.subdir());
            quote! { #allow pub mod #m; }
        })
        .collect();
    format_tokens_static(quote! { #(#mods)* })
}

pub fn main_module(services: &[ServicePlan]) -> String {
    let service_modules: Vec<TokenStream> = services
        .iter()
        .map(|s| {
            let module_name = format_ident!("{}", s.base_path);
            quote! { pub mod #module_name; }
        })
        .collect();

    let tokens = quote! {
        #(#service_modules)*
    };
    format_tokens_static(tokens)
}

fn generate_client_main_module(services: &[ServicePlan], has_aggregate: bool) -> String {
    let service_modules: Vec<TokenStream> = services
        .iter()
        .map(|s| {
            let module_name = format_ident!("{}", s.base_path);
            quote! { pub mod #module_name; }
        })
        .collect();

    // Export the aggregate root client (`client.rs`) when it was generated. Per-service files live
    // under `<base_path>/`, so a top-level `client` module never collides with them.
    // The aggregate client has many flat methods (arg-count) and renders proto
    // comment prose (doc list wrapping), so allow those structural lints.
    let aggregate_module = if has_aggregate {
        quote! {
            #[allow(clippy::too_many_arguments, clippy::doc_lazy_continuation)]
            pub mod client;
            #[allow(unused_imports)]
            pub use client::*;
        }
    } else {
        quote! {}
    };

    // The `stream_paginated` helper (and its `futures::Future` import) is used
    // only by `List` builders' `into_stream`. Emit it solely when some service
    // has a List method, so it isn't dead code otherwise.
    let has_list = services
        .iter()
        .flat_map(|s| s.methods.iter())
        .any(|m| matches!(m.request_type, RequestType::List));
    let pagination = if has_list {
        quote! {
            use futures::Future;

            pub(super) fn stream_paginated<F, Fut, S, T>(
                state: S,
                op: F,
            ) -> impl futures::Stream<Item = crate::Result<T>>
            where
                F: Fn(S, Option<String>) -> Fut + Copy,
                Fut: Future<Output = crate::Result<(T, S, Option<String>)>>,
            {
                enum PaginationState<T> {
                    Start(T),
                    HasMore(T, String),
                    Done,
                }

                futures::stream::unfold(
                    PaginationState::Start(state),
                    move |state| async move {
                        let (s, page_token) = match state {
                            PaginationState::Start(s) => (s, None),
                            PaginationState::HasMore(s, page_token) if !page_token.is_empty() => {
                                (s, Some(page_token))
                            }
                            _ => {
                                return None;
                            }
                        };

                        let (resp, s, continuation) = match op(s, page_token).await {
                            Ok(resp) => resp,
                            Err(e) => return Some((Err(e), PaginationState::Done)),
                        };

                        let next_state = match continuation {
                            Some(token) => PaginationState::HasMore(s, token),
                            None => PaginationState::Done,
                        };

                        Some((Ok(resp), next_state))
                    },
                )
            }
        }
    } else {
        quote! {}
    };

    let tokens = quote! {
        #(#service_modules)*

        #aggregate_module

        #pagination
    };
    format_tokens_static(tokens)
}

/// Generate the `mod.rs` for `crates/common/src/models/`.
///
/// Emits `pub mod <service> { pub mod <version> { include!("./<pkg>.rs") } }` blocks
/// for every service in the plan, plus static re-exports and module declarations.
/// The model files are co-located in this directory (written by the buf plugin), so
/// the includes are plain sibling references.
///
/// When `include_labels` is `true`, also emits `pub mod labels; pub use labels::{ObjectLabel, Resource};`.
///
/// `metadata` is used to discover all resource-annotated messages (even those not returned
/// directly by an RPC) so they can be included in `pub use` re-exports.
pub fn generate_models_mod(
    services: &[ServicePlan],
    include_labels: bool,
    include_pyo3_impls: bool,
    metadata: &CodeGenMetadata,
    runtime: Runtime,
) -> String {
    // Services are keyed by proto package, not by service: multiple services can share a
    // package (e.g. `CatalogService` and `SchemaService` both in `example.catalog.v1`), and
    // emitting one `pub mod`/`pub use` block per service would produce duplicate modules and
    // imports (E0428). Group by package first, then emit once per package.
    //
    // package = "unitycatalog.catalogs.v1"
    // parts   = ["unitycatalog", "catalogs", "v1"]
    // service module = second-to-last segment ("catalogs"); version = last segment ("v1").
    let mod_segments = |svc: &ServicePlan| -> (String, String) {
        let parts: Vec<&str> = svc.package.split('.').collect();
        if parts.len() >= 2 {
            (
                parts[parts.len() - 2].to_string(),
                parts[parts.len() - 1].to_string(),
            )
        } else {
            (svc.base_path.clone(), "v1".to_string())
        }
    };

    // One entry per package, in sorted order, with the union of re-exported type names.
    let mut packages: BTreeMap<String, (String, String, BTreeSet<String>)> = BTreeMap::new();
    for svc in services {
        let (svc_seg, ver_seg) = mod_segments(svc);
        let entry = packages
            .entry(svc.package.clone())
            .or_insert_with(|| (svc_seg, ver_seg, BTreeSet::new()));
        let type_names = &mut entry.2;

        // Re-export managed resources for this service...
        type_names.extend(svc.managed_resources.iter().map(|r| r.type_name.clone()));

        // ...plus every resource-annotated message in the same package (catches nested types
        // like `Column` that aren't direct RPC return types but still carry annotations).
        let fq_prefix = format!(".{}.", svc.package);
        let fq_pkg = format!(".{}", svc.package);
        for (fq_name, msg_info) in &metadata.messages {
            if msg_info.resource_descriptor.is_some()
                && (fq_name.starts_with(&fq_prefix) || fq_name.starts_with(&fq_pkg))
            {
                let simple = fq_name
                    .rfind('.')
                    .map(|i| &fq_name[i + 1..])
                    .unwrap_or(fq_name.as_str());
                type_names.insert(simple.to_string());
            }
        }
    }

    let mut service_mods: Vec<TokenStream> = Vec::new();
    let mut reexports: Vec<TokenStream> = Vec::new();
    for (package, (svc_seg, ver_seg, type_names)) in &packages {
        let svc_mod = format_ident!("{}", svc_seg);
        let ver_mod = format_ident!("{}", ver_seg);

        // Models are co-located: the buf plugin writes `<package>.rs` (and, for
        // prost, `<package>.serde.rs` / `<package>.tonic.rs`) into this same
        // directory, so the include is a plain sibling reference.
        let main_include = format!("./{package}.rs");
        // The separate `<package>.tonic.rs` (grpc service stubs over the prost
        // `ProstCodec`) is only emitted by the prost model plugin. The buffa plugin
        // produces no such file, so including it would break the build whenever the
        // `grpc` feature is enabled (e.g. CI's `--all-features`). Gate the include on
        // the runtime, not just the `grpc` cfg.
        let tonic_include_tokens = match runtime {
            Runtime::Prost => {
                let tonic_include = format!("./{package}.tonic.rs");
                quote! {
                    #[cfg(feature = "grpc")]
                    include!(#tonic_include);
                }
            }
            Runtime::Buffa => quote! {},
        };
        service_mods.push(quote! {
            pub mod #svc_mod {
                pub mod #ver_mod {
                    include!(#main_include);
                    #tonic_include_tokens
                }
            }
        });

        for type_name in type_names {
            let type_ident = format_ident!("{}", type_name);
            reexports.push(quote! {
                pub use #svc_mod::#ver_mod::#type_ident;
            });
        }
    }

    let labels_decl: TokenStream = if include_labels {
        quote! {
            pub mod labels;
            pub use labels::{ObjectLabel, Resource};
        }
    } else {
        quote! {}
    };

    // PyO3 boundary conversions for the model types (see `python::generate_pyo3_impls`).
    // Gated on the `python` feature so non-Python consumers never pull in pyo3.
    let pyo3_impls_decl: TokenStream = if include_pyo3_impls {
        quote! {
            // pyo3's `#[pyclass(from_py_object)]` extracts via `Clone::clone`, which trips
            // `clippy::clone_on_copy` on the (`Copy`) enum wrappers. The call lives in the
            // macro-generated impl, so only a module-level allow can reach it.
            #[cfg(feature = "python")]
            #[allow(clippy::clone_on_copy)]
            pub mod pyo3_impls;
            // Re-export the `Py*` wrapper `#[pyclass]` types at the models root so the
            // generated language bindings (and consumers) can name them without
            // reaching into the `pyo3_impls` submodule.
            #[cfg(feature = "python")]
            pub use pyo3_impls::*;
        }
    } else {
        quote! {}
    };

    // The Axum extractors are always co-located in this directory under an
    // `extractors/` subdir (the common generator wrote `extractors/<svc>/...`
    // here), and this `mod.rs` overwrites the top-level `mod.rs` the common
    // generator emitted — so it must re-declare the `extractors` module.
    // Axum-coupled, so gate it on the `axum` feature. Without this, the
    // `FromRequest`/`FromRequestParts` impls never compile and the generated
    // route fns can't be mounted. Nesting under `extractors` also avoids the
    // per-service module names clashing with the model package modules above.
    let extractors_mod = format_ident!("{EXTRACTORS_SUBDIR}");
    let common_mods: TokenStream = quote! {
        #[cfg(feature = "axum")]
        pub mod #extractors_mod;
    };

    // Lints to silence across this module and every `include!`d model file below.
    // These are inner attributes, so they cover the descendant `pub mod <svc>::<v1>`
    // modules the per-package files are pasted into.
    let lint_allows = match runtime {
        // The buffa model plugin emits enum variants with their proto SCREAMING_SNAKE
        // names (`non_camel_case_types`) and explicit `Default` impls a `#[derive]`
        // could replace (`clippy::derivable_impls`). The generated files carry no
        // allows of their own, so silence both here to keep `clippy -D warnings` green.
        Runtime::Buffa => quote! {
            #![allow(non_camel_case_types)]
            #![allow(clippy::derivable_impls)]
        },
        Runtime::Prost => quote! {},
    };

    let tokens = quote! {
        // The per-package modules below gate a `tonic.rs` include on
        // `#[cfg(feature = "grpc")]`. Consumers that don't define a `grpc`
        // feature would otherwise see `unexpected_cfgs`; allow it here.
        #![allow(unexpected_cfgs)]
        // The grpc-gated `tonic.rs` includes carry empty (`///`) doc comments on
        // generated service methods; allow them so `clippy -D warnings` stays
        // green when the `grpc` feature is enabled.
        #![allow(clippy::empty_docs)]
        // The prost-serde plugin emits `write!(f, "...", &FIELDS)` for the
        // deserializer's expected-fields message; newer clippy flags the `&` as a
        // useless borrow in a formatting macro. Allow it here so plugin output
        // stays `-D warnings` clean without post-processing each generated serde file.
        #![allow(clippy::useless_borrows_in_formatting)]
        #lint_allows

        use std::collections::HashMap;

        #labels_decl

        #pyo3_impls_decl

        #(#reexports)*

        pub type PropertyMap = HashMap<String, serde_json::Value>;

        #(#service_mods)*

        #common_mods
    };

    format_tokens_static(tokens)
}

/// An in-memory bundle of generated source files awaiting output.
///
/// Each generator phase (common, server, client, bindings, …) returns one of these; the
/// [`crate::output`] writer then materializes the [`files`](Self::files) onto disk, prepending the
/// `// @generated` header to each. Keeping generation and writing separate lets generation stay
/// pure (no I/O) and makes the produced files easy to assert on in tests.
#[derive(Debug)]
pub struct GeneratedCode {
    /// Generated files keyed by path relative to the phase's output directory.
    pub files: HashMap<String, String>,
}

impl CodeGenMetadata {
    fn get_message_meta(&self, message_name: &str) -> Option<MessageMeta<'_>> {
        self.messages.get(message_name).map(|info| MessageMeta {
            info,
            metadata: self,
        })
    }
}

/// A protobuf message paired with the metadata needed to resolve its references.
///
/// Wraps a single [`MessageInfo`](crate::parsing::MessageInfo) together with a borrow of the whole
/// [`CodeGenMetadata`] it came from, so the generators can look up a message's fields while still
/// being able to resolve types that point at other messages in the same descriptor set. Obtained
/// via `CodeGenMetadata::get_message_meta`.
pub(crate) struct MessageMeta<'a> {
    info: &'a MessageInfo,
    #[allow(dead_code)]
    metadata: &'a CodeGenMetadata,
}

/// The per-service generation context shared by every code generator.
///
/// Bundles the three borrows each generator phase needs for one service: its analyzed
/// [`ServicePlan`](crate::analysis::ServicePlan), the full [`CodeGenMetadata`] (for resolving
/// types referenced from other services), and the active [`CodeGenConfig`]. The various
/// `codegen::*` submodules take a `&ServiceHandler` and read these plus the derived accessors on
/// this type (client type names, models paths, [`Self::methods`]) rather than threading the three
/// borrows separately. Methods are exposed as [`MethodHandler`]s via [`Self::methods`].
pub(crate) struct ServiceHandler<'a> {
    pub(crate) plan: &'a ServicePlan,
    pub(crate) metadata: &'a CodeGenMetadata,
    pub(crate) config: &'a CodeGenConfig,
}

impl ServiceHandler<'_> {
    pub(crate) fn resource(&self) -> Option<&ManagedResource> {
        self.plan.managed_resources.first()
    }

    /// Whether this service manages a `google.api.resource` and therefore gets a resource-scoped
    /// client (e.g. `CatalogClient` bound to a name).
    ///
    /// Resource-less services (entity tag assignments, temporary credentials, delta commits) have
    /// no scoped client; their methods are emitted on the root client and pass every param
    /// directly ([`BindingMode::Flat`]). See [`Self::binding_mode`].
    pub(crate) fn is_resource_scoped(&self) -> bool {
        self.resource().is_some()
    }

    /// The [`BindingMode`] used to lower this service's methods in language bindings.
    ///
    /// A resource service whose identity can't be modeled by a string-component scoped client
    /// (see [`Self::supports_scoped_client`]) is lowered flat, exactly like a resource-less
    /// service: every method lives on the root client and passes all params directly.
    pub(crate) fn binding_mode(&self) -> BindingMode {
        if self.emits_scoped_binding() {
            BindingMode::Scoped
        } else {
            BindingMode::Flat
        }
    }

    /// Whether language bindings emit a per-service scoped client module + accessor for this
    /// service. True only for resource services whose identity a string-component scoped client
    /// can express; others are lowered flat onto the root client. See [`Self::binding_mode`].
    pub(crate) fn emits_scoped_binding(&self) -> bool {
        self.is_resource_scoped() && self.supports_scoped_client()
    }

    /// The language-agnostic spec for this service's resource accessor on the aggregate client, or
    /// `None` for resource-less services.
    ///
    /// All four accessor emitters (Rust aggregate, Python, Node, TypeScript) render from this single
    /// spec rather than re-deriving the param list and nesting decision, which previously diverged
    /// between the Rust aggregate and the bindings.
    pub(crate) fn accessor_spec(&self) -> Option<AccessorSpec> {
        let resource = self.resource()?;
        let params = self.plan.resource_accessor_params.clone()?;
        // A resource is nested when its accessor params include parent components beyond its own
        // name; its full name is then the dot-joined components. This is the single, converged
        // nesting rule — previously the Rust aggregate used `len > 1` while Python/Node also gated
        // on `name_field`, which could disagree.
        let nested = params.len() > 1;
        Some(AccessorSpec {
            singular: resource.descriptor.singular.clone(),
            params,
            nested,
        })
    }

    pub(crate) fn methods(&self) -> impl Iterator<Item = MethodHandler<'_>> {
        self.plan.methods.iter().map(|plan| MethodHandler {
            plan,
            metadata: self.metadata,
            config: self.config,
        })
    }

    /// The methods that have a `google.api.http` route — the ones the REST emitters (client,
    /// handler, server, aggregate) can generate. Routeless methods (no HTTP annotation) are emitted
    /// only by the ConnectRPC client, which iterates [`Self::methods`] directly.
    pub(crate) fn rest_methods(&self) -> impl Iterator<Item = MethodHandler<'_>> {
        self.methods().filter(|m| m.plan.has_http_route)
    }

    /// The ergonomic, public-facing client type for this service.
    ///
    /// For a resource-scoped service this is the **scoped resource client** (e.g. `CatalogClient`,
    /// bound to a name) — the type callers and language bindings use. For a resource-less service it
    /// is the single flat client (e.g. `TagAssignmentClient`). The low-level transport client (one
    /// async method per RPC) is [`Self::low_level_client_type`].
    pub(crate) fn client_type(&self) -> Ident {
        if let Some(resource) = self.resource() {
            format_ident!(
                "{}",
                format!("{} client", resource.descriptor.singular).to_case(Case::Pascal)
            )
        } else {
            format_ident!(
                "{}Client",
                self.plan
                    .service_name
                    .trim_end_matches("Service")
                    .trim_end_matches('s')
            )
        }
    }

    /// The low-level transport client type (`{ client: CloudClient, base_url: Url }` with one async
    /// method per RPC), emitted by [`client`](super::client).
    ///
    /// For a resource-scoped service this is `<Singular>ServiceClient` (e.g. `CatalogServiceClient`),
    /// distinct from the ergonomic scoped [`Self::client_type`] (`CatalogClient`). For a
    /// resource-less service there is no scoped client, so the low-level client *is* the
    /// [`Self::client_type`] (e.g. `TagAssignmentClient`).
    pub(crate) fn low_level_client_type(&self, protocol: crate::codegen::ClientProtocol) -> Ident {
        // ConnectRPC: a single adapter type per service, named deterministically from the proto
        // service so it never collides with (nor is confused for) the connect-rust `{Service}Client`
        // it wraps. Connect has no scoped resource client, so this is always the low-level type.
        if protocol == crate::codegen::ClientProtocol::Connect {
            return format_ident!(
                "{}ConnectClient",
                self.plan.service_name.to_case(Case::Pascal)
            );
        }
        if let Some(resource) = self.resource() {
            format_ident!(
                "{}ServiceClient",
                resource.descriptor.singular.to_case(Case::Pascal)
            )
        } else {
            self.client_type()
        }
    }

    /// The connect-rust generated client type for this service, as a fully-qualified path.
    ///
    /// Resolves to `<connect_client_path>::<ServiceName>Client` (e.g.
    /// `headwaters_proto::connect_gen::headwaters::read::v1::ReadServiceClient`). Used only in
    /// [`ClientProtocol::Connect`](crate::codegen::ClientProtocol) mode; errors if
    /// `connect_client_path` is unset.
    pub(crate) fn connect_inner_client_path(&self) -> crate::Result<syn::Path> {
        let base = self.config.connect_client_path.as_deref().ok_or_else(|| {
            crate::Error::Build(
                "client_protocol = Connect requires `connect_client_path` to be set".to_string(),
            )
        })?;
        let full = format!(
            "{base}::{}Client",
            self.plan.service_name.to_case(Case::Pascal)
        );
        syn::parse_str(&full).map_err(|source| crate::error::Error::InvalidRustPath {
            path: full.clone(),
            message: "connect_client_path + service client name".to_string(),
            source,
        })
    }

    /// The ergonomic scoped resource client type, or `None` for a resource-less service.
    ///
    /// Same as [`Self::client_type`] but `Option`-typed to make "resource-scoped only" explicit at
    /// call sites that generate the scoped client / its accessor.
    pub(crate) fn scoped_client_type(&self) -> Option<Ident> {
        self.resource().map(|_| self.client_type())
    }

    /// Whether this service's resource can be modeled by an ergonomic resource-scoped client.
    ///
    /// The scoped client (and the aggregate client's `<singular>(…)` accessor) captures a
    /// resource's identity as an ordered list of *string* name components and fills each method's
    /// path parameters from them. A resource whose scoped instance methods carry a non-string path
    /// parameter — e.g. a model version addressed by its parent's composite `{full_name}` plus an
    /// integer `{version}` — cannot be expressed that way, so no scoped client is emitted for it
    /// (its methods remain available on the low-level service client). Returns `false` in that case.
    pub(crate) fn supports_scoped_client(&self) -> bool {
        !self
            .methods()
            .filter(|m| m.is_scoped_instance_method())
            .any(|m| {
                m.required_parameters().any(|p| {
                    p.is_path_param()
                        && !matches!(
                            p.field_type().base_type,
                            crate::parsing::types::BaseType::String
                        )
                })
            })
    }

    /// The module segment under which this service's generated models live.
    ///
    /// Models are emitted per proto *package* (e.g. `unitycatalog.tags.v1` → module `tags`),
    /// which is not always the same as the service's `base_path` (derived from the service
    /// *name*, e.g. `TagPoliciesService` → `tag_policies`). Use the package's leaf-before-version
    /// segment so the models import path matches where the proto plugin actually wrote the types.
    /// Falls back to `base_path` when the package is empty or unparsable.
    fn models_segment(&self) -> String {
        let segs: Vec<&str> = self
            .plan
            .package
            .split('.')
            .filter(|s| !s.is_empty())
            .collect();
        match segs.as_slice() {
            // `unitycatalog.tags.v1` → `tags`
            [.., name, version] if is_version_segment(version) => name.to_string(),
            // `unitycatalog.tags` → `tags`
            [.., name] => name.to_string(),
            // No package info: fall back to the service-name-derived base path.
            _ => self.plan.base_path.clone(),
        }
    }

    /// The version segment under which this service's generated models live
    /// (e.g. `unitycatalog.tags.v1` → `v1`, `unitycatalog.agents.v0alpha1` → `v0alpha1`).
    ///
    /// Matches where the proto plugin actually wrote the types. Falls back to `v1` when the
    /// package carries no recognizable version segment.
    fn models_version(&self) -> String {
        let segs: Vec<&str> = self
            .plan
            .package
            .split('.')
            .filter(|s| !s.is_empty())
            .collect();
        match segs.as_slice() {
            [.., _name, version] if is_version_segment(version) => version.to_string(),
            _ => "v1".to_string(),
        }
    }

    pub(crate) fn models_path(&self) -> syn::Path {
        // Templates (and each service's substitution) are validated by `generate_code` before
        // any `ServiceHandler` is used, so skip the redundant re-validation `new` would do.
        ModelsPath::from_template(&self.config.models_path_template)
            .resolve(&self.models_segment(), &self.models_version())
    }

    pub(crate) fn models_path_crate(&self) -> syn::Path {
        ModelsPath::from_template(&self.config.models_path_crate_template)
            .resolve(&self.models_segment(), &self.models_version())
    }

    /// The models *root* module path — the prefix of [`Self::models_path`] before the
    /// per-service/version segments (e.g. `example_common::models` for the template
    /// `example_common::models::{service}::v1`).
    ///
    /// The `Py*` wrapper `#[pyclass]` types are re-exported here (see
    /// `generate_models_mod`), so the Python bindings import them from this path.
    /// Returns `None` if the template has no `{service}` placeholder to split on.
    pub(crate) fn models_root_path(&self) -> Option<syn::Path> {
        let template = &self.config.models_path_template;
        let prefix = template.split("{service}").next()?;
        let trimmed = prefix.trim_end_matches(':');
        if trimmed.is_empty() {
            return None;
        }
        syn::parse_str(trimmed).ok()
    }
}

/// Whether a proto package segment is a version marker like `v1`, `v2beta1`, etc.
fn is_version_segment(seg: &str) -> bool {
    seg.strip_prefix('v')
        .is_some_and(|rest| rest.starts_with(|c: char| c.is_ascii_digit()))
}

/// The per-method counterpart of [`ServiceHandler`], scoped to a single RPC.
///
/// Pairs one analyzed [`MethodPlan`](crate::analysis::MethodPlan) with the same [`CodeGenMetadata`]
/// and [`CodeGenConfig`] borrows, and exposes the per-method derivations the generators rely on:
/// input/output message lookup, builder and binding method names, parameter plans
/// ([`Self::param_plan`]), and the language-agnostic [`EmitShape`](crate::analysis::EmitShape).
/// Yielded by [`ServiceHandler::methods`]; each generator iterates these to emit one unit of code
/// (a trait method, a client method, a builder, …) per RPC.
pub(crate) struct MethodHandler<'a> {
    plan: &'a MethodPlan,
    metadata: &'a CodeGenMetadata,
    config: &'a CodeGenConfig,
}

impl MethodHandler<'_> {
    pub(crate) fn is_collection_method(&self) -> bool {
        crate::analysis::is_collection_method(
            &self.plan.request_type,
            &self.plan.metadata.method_name,
        )
    }

    /// The language-agnostic [`EmitShape`] for this method, derived once from its request type and
    /// name. The Python/Node/TypeScript emitters branch on this instead of each re-matching
    /// `RequestType`.
    pub(crate) fn emit_shape(&self) -> crate::analysis::EmitShape {
        crate::analysis::emit_shape(&self.plan.request_type, self.is_collection_method())
    }

    /// Whether this method is emitted as an instance method on a resource-scoped client
    /// (`get`/`update`/`delete` or a resource-targeted custom `POST`/`PATCH`).
    ///
    /// This is the exact set the bindings previously matched in their `resource_client_method`
    /// dispatch: standard `Get`/`Update`/`Delete`, plus a non-collection custom `POST`/`PATCH`
    /// (e.g. `POST /catalogs/{name}:rotateToken`). Other custom verbs (a custom `GET`/`DELETE`) are
    /// deliberately **not** surfaced on the scoped client, matching prior behavior.
    pub(crate) fn is_scoped_instance_method(&self) -> bool {
        // Standard instance verbs are always scoped.
        if matches!(
            self.plan.request_type,
            RequestType::Get | RequestType::Update | RequestType::Delete
        ) {
            return true;
        }
        // A *resource-targeted* custom RPC (any verb) is a scoped instance method: it has at least
        // one path param (so it operates on a specific resource) and is not a collection method
        // (those — list / create / factory — live on the root client). This surfaces custom reads
        // like `GetTableExists` / `GetPermissions` (`GET …/{name}/exists`) and resource-targeted
        // custom POST/PATCH (`…/{name}:rotateToken`) on the scoped client, rather than leaving their
        // generated builders orphaned/unreachable.
        matches!(self.plan.request_type, RequestType::Custom(_))
            && !self.is_collection_method()
            && self.plan.path_parameters().next().is_some()
    }

    /// The ordered binding parameter list for `mode`, with shared filtering applied once.
    ///
    /// Required params come before optional ones (matching the previous
    /// `required_parameters().chain(optional_parameters())` order). Filtering reproduces exactly what
    /// the old per-language `collection_method_parameters` / `resource_method_parameters` did,
    /// keyed on [`EmitShape`]:
    ///
    /// - **collection shapes** (`List`, `Create`): never drop path params (collection methods live
    ///   on the root client and pass everything); drop `page_token` for `List` (streaming handles
    ///   pagination).
    /// - **instance shapes** (`GetUpdate`, `Delete`): drop path params in [`BindingMode::Scoped`]
    ///   (the scoped client already holds them); never filter `page_token`.
    ///
    /// Per-language emitters render the returned `&RequestParam`s (applying their own type mapping /
    /// capability filters) instead of each re-deriving this filtered list.
    pub(crate) fn param_plan(&self, mode: BindingMode) -> Vec<&RequestParam> {
        let (required, optional) = self.param_plan_split(mode);
        required.into_iter().chain(optional).collect()
    }

    /// Like [`Self::param_plan`] but keeps the required/optional partition, for emitters (e.g.
    /// TypeScript) that render the two groups differently (required positional args vs an optional
    /// `options` object). Both groups have the same shared path/`page_token` filtering applied.
    pub(crate) fn param_plan_split(
        &self,
        mode: BindingMode,
    ) -> (Vec<&RequestParam>, Vec<&RequestParam>) {
        use crate::analysis::EmitShape;
        let shape = self.emit_shape();
        let is_collection_shape = matches!(shape, EmitShape::List | EmitShape::Create);
        let drop_path = !is_collection_shape && mode == BindingMode::Scoped;
        let drop_page_token = shape == EmitShape::List;
        let keep = |p: &&RequestParam| {
            let is_dropped_path = drop_path && p.is_path_param();
            let is_dropped_page_token = drop_page_token && p.name() == "page_token";
            !is_dropped_path && !is_dropped_page_token
        };
        let required = self.required_parameters().filter(keep).collect();
        let optional = self.optional_parameters().filter(keep).collect();
        (required, optional)
    }

    pub(crate) fn output_message(&self) -> Option<MessageMeta<'_>> {
        if self.plan.metadata.output_type.ends_with("Empty") {
            return None;
        }
        self.metadata
            .get_message_meta(&self.plan.metadata.output_type)
    }

    pub(crate) fn output_type(&self) -> Option<Ident> {
        self.output_message()
            .map(|t| extract_type_ident(&t.info.name))
    }

    pub(crate) fn list_output_field(&self) -> Option<&MessageField> {
        self.output_message()?
            .info
            .fields
            .iter()
            .find(|f| !f.name.contains("page_token"))
    }

    pub(crate) fn input_message(&self) -> Option<MessageMeta<'_>> {
        if self.plan.metadata.input_type == "Empty" {
            return None;
        }
        self.metadata
            .get_message_meta(&self.plan.metadata.input_type)
    }

    pub(crate) fn input_type(&self) -> Option<Ident> {
        self.input_message()
            .map(|t| extract_type_ident(&t.info.name))
    }

    pub(crate) fn builder_type(&self) -> Ident {
        format_ident!("{}Builder", self.plan.metadata.method_name)
    }

    /// The method name to expose in language bindings.
    ///
    /// Prefers the gnostic `operation_id` annotation (`gnostic.openapi.v3.operation`) when present,
    /// snake-cased, so proto authors can give bindings a hand-tuned name. Falls back to the
    /// proto-derived name ([`MethodPlan::base_method_ident`]).
    ///
    /// This affects only the *emitted* binding method name — the call into the underlying Rust
    /// client still uses [`MethodPlan::base_method_ident`], so the handler/client API is unchanged.
    pub(crate) fn binding_method_name(&self) -> Ident {
        format_ident!("{}", self.binding_method_name_str())
    }

    /// The snake_case binding method name (see [`Self::binding_method_name`]).
    ///
    /// Provided as a `String` so the string-templated TypeScript generator can camel-case it and
    /// stay consistent with the snake→camel name NAPI derives for the native method.
    pub(crate) fn binding_method_name_str(&self) -> String {
        match self.plan.metadata.operation.as_ref() {
            Some(op) if !op.operation_id.is_empty() => op.operation_id.to_case(Case::Snake),
            _ => self.plan.handler_function_name.clone(),
        }
    }

    /// Get type representation for rust depending on context
    ///
    /// Depending on context we may want concrete types (e.g. 'String') or more flexible types (e.g. 'Into<String d>')
    pub(crate) fn field_type(&self, field_type: &UnifiedType, ctx: RenderContext) -> syn::Type {
        let rust_type = types::unified_to_rust(field_type, ctx);
        // `rust_type` comes from this crate's own proto→Rust mapping, not user input, so an
        // unparsable result is a bug in `unified_to_rust`. Panic with the offending string so
        // the broken mapping is obvious (the generated-code parse test in
        // `tests/generation_integration.rs` is the broader guard for output validity).
        syn::parse_str(&rust_type).unwrap_or_else(|e| {
            panic!("internal: unified_to_rust produced an invalid type `{rust_type}`: {e}")
        })
    }

    /// Get field assignment TokenStream for constructor
    pub(crate) fn field_assignment(
        &self,
        field_type: &UnifiedType,
        field_ident: &proc_macro2::Ident,
        ctx: &RenderContext,
    ) -> TokenStream {
        types::field_assignment(field_type, field_ident, ctx, self.config.runtime)
    }

    pub(crate) fn required_parameters(&self) -> impl Iterator<Item = &RequestParam> {
        self.plan
            .parameters
            .iter()
            .filter(|param| !param.is_optional())
    }

    pub(crate) fn optional_parameters(&self) -> impl Iterator<Item = &RequestParam> {
        self.plan
            .parameters
            .iter()
            .filter(|param| param.is_optional())
    }

    /// Split body fields into required and optional subsets.
    pub(crate) fn split_body_fields(&self) -> (Vec<&BodyField>, Vec<&BodyField>) {
        split_body_fields(self.plan)
    }
}

/// Extract the final type name from a fully qualified protobuf type and convert to Ident.
///
/// Thin `Ident` adapter over [`crate::utils::extract_simple_type_name`] — the single source of
/// truth for last-segment extraction. (Not [`crate::utils::extract_qualified_type_name`], which
/// parent-prefixes nested types — different behavior.)
pub(crate) fn extract_type_ident(full_type: &str) -> Ident {
    format_ident!("{}", crate::utils::extract_simple_type_name(full_type))
}