use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;

use heck::{ToShoutySnakeCase, ToSnakeCase, ToUpperCamelCase};
use indexmap::IndexMap;

use crate::api_plan::{
    ApiPlan, PlannedEnum, PlannedField, PlannedFieldKind, PlannedFieldRole, PlannedFlags,
    PlannedMessageSource, PlannedMessageType, PlannedOperation, PlannedOperationOutput,
    PlannedOperationResourceFieldBinding, PlannedOperationResourceReturn, PlannedResource,
    PlannedResourceMethod, PlannedResourceMethodBindingSpec, PlannedResourceMethodResultKind,
    PlannedScalarType, PlannedSourcedField, PlannedTypeInfo, PlannedValueType, PlannedVariant,
    message_model_name, relative_descriptor_name,
};
use crate::error::{Error, Result};
use crate::generator::{GeneratedFiles, ModelCapabilities};
use crate::language::Language;
use crate::resources::{RequestPlan, ResolvedResourceBindingSource, render_request_plan};
use crate::spec::{
    AuthoredFieldTypeSpec, FunctionArgsSpec, FunctionFieldSpec, FunctionResultSpec,
    LanguageImportSpec, LanguageImportStyle, SupportFragmentSpec, TypeReplacementSpec,
};

const GENERATED_HEADER: &str = "# Generated by nex-gen. DO NOT EDIT!";
const PYTHON_FORMAT_LINE_LENGTH: usize = 88;
const EXPERIMENTAL_WARNING: &str = "This API is experimental and subject to change.";

pub(crate) fn generate(
    api_plan: &ApiPlan,
    support_fragments: &[SupportFragmentSpec],
    language_imports: &[LanguageImportSpec],
) -> Result<GeneratedFiles> {
    let mut enums = IndexMap::new();
    let mut flags = IndexMap::new();
    let mut variants = IndexMap::new();
    let mut models = IndexMap::new();
    let services = api_plan
        .services
        .iter()
        .map(|service| {
            let operations = service
                .operations
                .iter()
                .map(|operation| {
                    resolve_operation(
                        operation,
                        api_plan,
                        &mut enums,
                        &mut flags,
                        &mut variants,
                        &mut models,
                    )
                })
                .collect::<Result<Vec<_>>>()?;

            Ok(RenderedService {
                name: &service.name,
                wire_name: &service.wire_name,
                endpoint: service.endpoint.clone(),
                experimental: service.experimental,
                operations,
                resources: service.resources.clone(),
            })
        })
        .collect::<Result<Vec<_>>>()?;

    for service in &services {
        for resource in &service.resources {
            ensure_resource_field_types(
                resource,
                api_plan,
                &mut enums,
                &mut flags,
                &mut variants,
                &mut models,
            );
        }
    }

    render_package(
        enums.values().collect::<Vec<_>>().as_slice(),
        flags.values().collect::<Vec<_>>().as_slice(),
        variants.values().collect::<Vec<_>>().as_slice(),
        models.values().collect::<Vec<_>>().as_slice(),
        &services,
        support_fragments,
        language_imports,
    )
}

fn ensure_resource_field_types(
    resource: &PlannedResource,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) {
    for field in &resource.fields {
        ensure_resource_field_type(&field.kind, api_plan, enums, flags, variants, models);
    }
    for method in &resource.methods {
        for param in &method.params {
            ensure_resource_field_type(&param.kind, api_plan, enums, flags, variants, models);
        }
    }
}

fn ensure_resource_field_type(
    kind: &PlannedFieldKind,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) {
    match kind {
        PlannedFieldKind::Singular(value) | PlannedFieldKind::Repeated(value) => {
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models);
        }
        PlannedFieldKind::Map { key, value } => {
            resolve_planned_value_type(key, api_plan, enums, flags, variants, models);
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models);
        }
    }
}

pub(crate) fn python_field_name(name: &str) -> String {
    python_ident(&name.to_snake_case())
}

#[derive(Debug)]
struct RenderedService<'a> {
    name: &'a str,
    wire_name: &'a str,
    endpoint: String,
    experimental: bool,
    operations: Vec<RenderedOperation<'a>>,
    resources: Vec<PlannedResource>,
}

#[derive(Debug)]
struct RenderedOperation<'a> {
    name: &'a str,
    wire_name: &'a str,
    attr_name: String,
    experimental: bool,
    doc: Option<String>,
    return_doc: Option<String>,
    input_ref: String,
    input_module_path: Option<String>,
    output_ref: String,
    output_module_path: Option<String>,
    input_annotation: String,
    input_to_proto_expr: String,
    output_annotation: String,
    overload_output_annotation: String,
    output_type_parameters: BTreeSet<String>,
    output_type_expr: String,
    output_transform_expr: Option<String>,
    output_resource_return: Option<PlannedOperationResourceReturn>,
    output_direct_result: bool,
    output_none: bool,
    unpacked_input: Option<RenderedUnpackedInput>,
}

#[derive(Debug, Clone, Copy)]
struct ResourceBoundOperation<'a> {
    operation: &'a RenderedOperation<'a>,
}

#[derive(Debug)]
struct RenderedEnum {
    name: String,
    values: Vec<RenderedEnumValue>,
}

#[derive(Debug)]
struct RenderedEnumValue {
    name: String,
    number: i32,
}

#[derive(Debug)]
struct RenderedFlags {
    name: String,
    flags: Vec<RenderedFlag>,
}

#[derive(Debug)]
struct RenderedFlag {
    name: String,
    bit: usize,
}

#[derive(Debug)]
struct RenderedVariant {
    name: String,
    cases: Vec<RenderedVariantCase>,
}

#[derive(Debug)]
struct RenderedVariantCase {
    name: String,
    payload_annotation: Option<String>,
}

#[derive(Debug)]
struct RenderedModel {
    full_name: String,
    name: String,
    source: PlannedMessageSource,
    proto_ref: Option<String>,
    proto_module_path: Option<String>,
    capabilities: ModelCapabilities,
    experimental: bool,
    fields: Vec<RenderedField>,
    sourced_fields: Vec<RenderedSourcedField>,
}

#[derive(Debug)]
struct RenderedField {
    attr_name: String,
    annotation: String,
    default_kind: PythonFieldDefaultKind,
    default_expr: Option<String>,
    from_proto_setup_lines: Vec<String>,
    from_proto_expr: String,
    to_proto_lines: Vec<String>,
    imports: PythonImports,
}

#[derive(Debug)]
struct RenderedSourcedField {
    to_proto_lines: Vec<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum PythonFieldDefaultKind {
    Required,
    None,
    EmptyList,
    EmptyDict,
    Expression(String),
}

#[derive(Debug)]
struct RenderedUnpackedInput {
    model_name: String,
    parameters: Vec<RenderedUnpackedInputField>,
    request_fields: Vec<RenderedUnpackedRequestField>,
    flattened_messages: Vec<RenderedFlattenedMessage>,
    functions: Vec<RenderedFunctionField>,
}

#[derive(Debug)]
struct RenderedUnpackedInputField {
    attr_name: String,
    annotation: String,
    default_kind: PythonFieldDefaultKind,
    doc: Option<String>,
}

#[derive(Debug)]
struct RenderedUnpackedRequestField {
    attr_name: String,
    value_expr: String,
    default_kind: PythonFieldDefaultKind,
}

#[derive(Debug)]
struct RenderedFlattenedMessage {
    local_name: String,
    model_name: String,
    required: bool,
    fields: Vec<RenderedFlattenedMessageField>,
}

#[derive(Debug)]
struct RenderedFlattenedMessageField {
    attr_name: String,
    annotation: String,
    value_expr: String,
    default_kind: PythonFieldDefaultKind,
    doc: Option<String>,
}

#[derive(Debug, Clone)]
struct RenderedFunctionField {
    callable_field_name: String,
    args_field_name: String,
    args: RenderedFunctionArgs,
    primary: bool,
    callable_annotation: String,
    result_annotation: String,
    erased_result_annotation: String,
    result_type_parameter: Option<String>,
    alternate_annotation: Option<String>,
}

#[derive(Debug, Clone)]
enum RenderedFunctionArgs {
    Varargs {
        prefix: Vec<RenderedFunctionArg>,
    },
    Typed {
        parameters: Vec<RenderedFunctionArg>,
    },
}

#[derive(Debug, Clone)]
struct RenderedFunctionArg {
    name: String,
    annotation: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
enum ResolvedFieldKind {
    Scalar,
    Message,
    Enum,
    Unknown,
}

#[derive(Debug, Clone)]
struct ResolvedFieldType {
    annotation: String,
    imports: PythonImports,
    kind: ResolvedFieldKind,
    message_conversion: Option<MessageValueConversion>,
    enum_conversion: Option<EnumValueConversion>,
}

#[derive(Debug, Clone)]
enum MessageValueConversionKind {
    GeneratedModel {
        model_name: String,
    },
    NativeModel,
    Override {
        from_proto: String,
        to_proto: String,
    },
}

#[derive(Debug, Clone)]
struct MessageValueConversion {
    annotation: String,
    kind: MessageValueConversionKind,
    imports: PythonImports,
}

impl MessageValueConversion {
    fn from_proto_expr(&self, proto_expr: &str) -> String {
        match &self.kind {
            MessageValueConversionKind::GeneratedModel { model_name } => {
                format!("{model_name}.from_proto({proto_expr})")
            }
            MessageValueConversionKind::NativeModel => proto_expr.to_string(),
            MessageValueConversionKind::Override { from_proto, .. } => {
                format!("{from_proto}({proto_expr})")
            }
        }
    }

    fn to_proto_expr(&self, value_expr: &str) -> String {
        match &self.kind {
            MessageValueConversionKind::GeneratedModel { .. } => {
                format!("{value_expr}.to_proto()")
            }
            MessageValueConversionKind::NativeModel => value_expr.to_string(),
            MessageValueConversionKind::Override { to_proto, .. } => {
                format!("{to_proto}({value_expr})")
            }
        }
    }
}

#[derive(Debug, Clone)]
struct EnumValueConversion {
    from_proto: String,
    to_proto: String,
}

impl EnumValueConversion {
    fn from_proto_expr(&self, proto_expr: &str) -> String {
        format!("{}({proto_expr})", self.from_proto)
    }

    fn to_proto_expr(&self, value_expr: &str) -> String {
        format!("{}({value_expr})", self.to_proto)
    }
}

#[derive(Debug, Clone, Default)]
struct PythonImports {
    module_imports: BTreeSet<String>,
}

impl PythonImports {
    fn extend(&mut self, other: &Self) {
        self.module_imports
            .extend(other.module_imports.iter().cloned());
    }
}

fn python_default_from_proto_name(name: &str) -> String {
    format!(
        "{}_from_proto",
        name.rsplit('.')
            .next()
            .expect("converter name source should have a final segment")
            .to_snake_case()
    )
}

fn python_default_to_proto_name(name: &str) -> String {
    format!(
        "{}_to_proto",
        name.rsplit('.')
            .next()
            .expect("converter name source should have a final segment")
            .to_snake_case()
    )
}

fn python_from_proto_converter(name: &str, replacement: &TypeReplacementSpec) -> String {
    replacement
        .from_proto
        .for_language(crate::language::Language::Python)
        .map(str::to_string)
        .unwrap_or_else(|| python_default_from_proto_name(name))
}

fn python_to_proto_converter(name: &str, replacement: &TypeReplacementSpec) -> String {
    replacement
        .to_proto
        .for_language(crate::language::Language::Python)
        .map(str::to_string)
        .unwrap_or_else(|| python_default_to_proto_name(name))
}

fn python_replacement_type_name(replacement: &TypeReplacementSpec) -> Option<String> {
    replacement
        .type_name
        .for_language(Language::Python)
        .map(str::to_string)
}

fn python_function_result_annotation(result: &FunctionResultSpec) -> String {
    match result {
        FunctionResultSpec::Annotation(annotation) => annotation
            .for_language(Language::Python)
            .unwrap_or("typing.Any")
            .to_string(),
        FunctionResultSpec::Wit(wit_type) => python_authored_type_annotation(wit_type),
    }
}

fn python_authored_type_annotation(wit_type: &AuthoredFieldTypeSpec) -> String {
    match wit_type {
        AuthoredFieldTypeSpec::Bool => "bool".to_string(),
        AuthoredFieldTypeSpec::Int => "int".to_string(),
        AuthoredFieldTypeSpec::Float => "float".to_string(),
        AuthoredFieldTypeSpec::String => "str".to_string(),
        AuthoredFieldTypeSpec::Bytes => "bytes".to_string(),
        AuthoredFieldTypeSpec::Option(inner) => {
            format!("{} | None", python_authored_type_annotation(inner))
        }
        AuthoredFieldTypeSpec::List(inner) => {
            format!(
                "collections.abc.Sequence[{}]",
                python_authored_type_annotation(inner)
            )
        }
        AuthoredFieldTypeSpec::Tuple(items) => format!(
            "tuple[{}]",
            items
                .iter()
                .map(python_authored_type_annotation)
                .collect::<Vec<_>>()
                .join(", ")
        ),
        AuthoredFieldTypeSpec::Map(key, value) => format!(
            "collections.abc.Mapping[{}, {}]",
            python_authored_type_annotation(key),
            python_authored_type_annotation(value)
        ),
        AuthoredFieldTypeSpec::Result { ok, err } => python_result_annotation(
            ok.as_ref()
                .map(|ok| python_authored_type_annotation(ok).to_string()),
            err.as_ref()
                .map(|err| python_authored_type_annotation(err).to_string()),
        ),
        AuthoredFieldTypeSpec::Alias {
            type_name, target, ..
        } => type_name
            .for_language(Language::Python)
            .map(str::to_string)
            .unwrap_or_else(|| python_authored_type_annotation(target)),
        AuthoredFieldTypeSpec::Proto(proto_name) => proto_name
            .rsplit('.')
            .next()
            .unwrap_or(proto_name)
            .to_string(),
        AuthoredFieldTypeSpec::Record(name)
        | AuthoredFieldTypeSpec::Enum(name)
        | AuthoredFieldTypeSpec::Flags(name)
        | AuthoredFieldTypeSpec::Variant(name)
        | AuthoredFieldTypeSpec::Resource(name) => name
            .rsplit('.')
            .next()
            .unwrap_or(name)
            .to_upper_camel_case(),
    }
}

fn python_function_args(args: &FunctionArgsSpec) -> RenderedFunctionArgs {
    match args {
        FunctionArgsSpec::Varargs { prefix, .. } => RenderedFunctionArgs::Varargs {
            prefix: prefix
                .iter()
                .map(|arg| RenderedFunctionArg {
                    name: python_field_name(&arg.name),
                    annotation: python_authored_type_annotation(&arg.field_type),
                })
                .collect(),
        },
        FunctionArgsSpec::Fixed(args) => RenderedFunctionArgs::Typed {
            parameters: args
                .iter()
                .map(|arg| RenderedFunctionArg {
                    name: python_field_name(&arg.name),
                    annotation: python_authored_type_annotation(&arg.field_type),
                })
                .collect(),
        },
    }
}

fn python_function_args_field_annotation(args: &FunctionArgsSpec) -> String {
    match python_function_args(args) {
        RenderedFunctionArgs::Varargs { .. } => "list[typing.Any]".to_string(),
        RenderedFunctionArgs::Typed { parameters } => python_function_args_list_annotation(
            parameters
                .iter()
                .map(|parameter| parameter.annotation.as_str())
                .collect::<Vec<_>>()
                .as_slice(),
        ),
    }
}

fn python_function_arg_field_annotation(function: &FunctionFieldSpec, field_name: &str) -> String {
    match &function.args {
        FunctionArgsSpec::Varargs { .. } => "list[typing.Any]".to_string(),
        FunctionArgsSpec::Fixed(args) => args
            .iter()
            .find(|arg| arg.name.to_snake_case() == field_name)
            .map(|arg| python_authored_type_annotation(&arg.field_type))
            .unwrap_or_else(|| python_function_args_field_annotation(&function.args)),
    }
}

fn python_function_args_list_annotation(annotations: &[&str]) -> String {
    let Some(first) = annotations.first() else {
        return "list[typing.Any]".to_string();
    };
    if annotations.iter().all(|annotation| annotation == first) {
        format!("list[{first}]")
    } else {
        "list[typing.Any]".to_string()
    }
}

fn python_result_annotation(ok: Option<String>, err: Option<String>) -> String {
    let ok_case = if let Some(ok) = ok {
        format!("tuple[typing.Literal[\"ok\"], {ok}]")
    } else {
        "tuple[typing.Literal[\"ok\"]]".to_string()
    };
    let err_case = if let Some(err) = err {
        format!("tuple[typing.Literal[\"err\"], {err}]")
    } else {
        "tuple[typing.Literal[\"err\"]]".to_string()
    };
    format!("{ok_case} | {err_case}")
}

fn resolve_operation<'a>(
    operation: &'a PlannedOperation,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> Result<RenderedOperation<'a>> {
    let (input_ref, input_module_path) = operation_service_ref(&operation.input, "models");
    let input_conversion = resolve_message_value_conversion(
        &operation.input,
        None,
        api_plan,
        enums,
        flags,
        variants,
        models,
    );
    let output_transform = operation.output_transform.as_ref();
    let output_resource_return = operation.output_resource_return.clone();
    let (output_ref, output_module_path, output_annotation_default) = match &operation.output {
        PlannedOperationOutput::Message(output) => {
            let (output_ref, output_module_path) = operation_service_ref(output, "models");
            if output_transform.is_none()
                && output_resource_return.is_none()
                && !operation.output_direct_result
            {
                resolve_message_value_conversion(
                    output, None, api_plan, enums, flags, variants, models,
                );
            }
            let annotation =
                resolve_output_annotation(output, api_plan, enums, flags, variants, models);
            (output_ref, output_module_path, annotation)
        }
        PlannedOperationOutput::Resource { type_name } => (
            type_name.clone(),
            Some("._resources".to_string()),
            type_name.clone(),
        ),
        PlannedOperationOutput::None => ("None".to_string(), None, "None".to_string()),
    };
    let overload_output_annotation = output_transform
        .and_then(|transform| {
            transform
                .type_name
                .for_language(crate::language::Language::Python)
                .map(str::to_string)
        })
        .or_else(|| {
            output_resource_return
                .as_ref()
                .map(|resource| resource.resource_type_name.clone())
        })
        .unwrap_or(output_annotation_default);
    let unpacked_input = if matches!(
        input_conversion.kind,
        MessageValueConversionKind::GeneratedModel { .. } | MessageValueConversionKind::NativeModel
    ) {
        Some(build_unpacked_input(
            &operation.input.info.full_name,
            models,
            api_plan,
        )?)
    } else {
        None
    };
    let output_type_parameters = unpacked_input
        .iter()
        .flat_map(|unpacked_input| &unpacked_input.functions)
        .filter_map(|function| function.result_type_parameter.clone())
        .filter(|name| python_annotation_uses_identifier(&overload_output_annotation, name))
        .collect::<BTreeSet<_>>();
    let output_annotation =
        erase_python_type_parameters(&overload_output_annotation, &output_type_parameters);
    let output_type_expr = if output_transform.is_some() || output_resource_return.is_some() {
        output_ref.clone()
    } else {
        output_annotation.clone()
    };
    Ok(RenderedOperation {
        name: operation.name.as_str(),
        wire_name: operation.wire_name.as_str(),
        attr_name: python_ident(&operation.name.to_snake_case()),
        experimental: operation.experimental,
        doc: operation
            .doc
            .for_language(Language::Python)
            .map(str::to_string),
        return_doc: operation
            .return_doc
            .for_language(Language::Python)
            .map(str::to_string),
        input_ref,
        input_module_path,
        output_ref,
        output_module_path,
        input_annotation: input_conversion.annotation.clone(),
        input_to_proto_expr: input_conversion.to_proto_expr("request"),
        output_annotation,
        overload_output_annotation,
        output_type_parameters,
        output_type_expr,
        output_transform_expr: output_transform.and_then(|transform| {
            transform
                .transform
                .for_language(crate::language::Language::Python)
                .map(str::to_string)
        }),
        output_resource_return,
        output_direct_result: operation.output_direct_result,
        output_none: matches!(operation.output, PlannedOperationOutput::None),
        unpacked_input,
    })
}

fn operation_service_ref(
    message: &PlannedMessageType,
    _native_module: &str,
) -> (String, Option<String>) {
    match message.source {
        PlannedMessageSource::Proto => {
            let reference = message_python_ref(&message.info)
                .expect("descriptor messages should have a Python reference");
            (reference.type_ref, Some(reference.module_path))
        }
        PlannedMessageSource::Wit => (
            format!("{_native_module}.{}", message.model_name),
            Some(format!(".{_native_module}")),
        ),
    }
}

fn resolve_output_annotation(
    message: &PlannedMessageType,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> String {
    match message.source {
        PlannedMessageSource::Proto => {
            message_python_ref(&message.info)
                .expect("descriptor messages should have a Python reference")
                .type_ref
        }
        PlannedMessageSource::Wit => {
            let conversion = resolve_message_value_conversion(
                message, None, api_plan, enums, flags, variants, models,
            );
            conversion.annotation
        }
    }
}

fn build_unpacked_input(
    input_full_name: &str,
    models: &IndexMap<String, RenderedModel>,
    api_plan: &ApiPlan,
) -> Result<RenderedUnpackedInput> {
    let model = models
        .get(input_full_name)
        .expect("input model should be rendered before building unpacked input");
    let planned_model = api_plan
        .models
        .get(input_full_name)
        .expect("planned input model should exist");

    let mut parameters = Vec::new();
    let mut request_fields = Vec::new();
    let mut flattened_messages = Vec::new();
    let mut parameter_sources = BTreeMap::<String, String>::new();

    for (planned_field, rendered_field) in planned_model.fields.iter().zip(model.fields.iter()) {
        if let Some(flattened) =
            build_flattened_message(planned_field, rendered_field, models, api_plan)
        {
            request_fields.push(RenderedUnpackedRequestField {
                attr_name: flattened.local_name.clone(),
                value_expr: flattened.local_name.clone(),
                default_kind: PythonFieldDefaultKind::Required,
            });
            for child in &flattened.fields {
                register_unpacked_parameter_name(
                    &mut parameter_sources,
                    &child.value_expr,
                    &format!("{}.{}", flattened.local_name, child.attr_name),
                    &model.name,
                )?;
                parameters.push(RenderedUnpackedInputField {
                    attr_name: child.value_expr.clone(),
                    annotation: child.annotation.clone(),
                    default_kind: child.default_kind.clone(),
                    doc: child.doc.clone(),
                });
            }
            flattened_messages.push(flattened);
            continue;
        }

        register_unpacked_parameter_name(
            &mut parameter_sources,
            &rendered_field.attr_name,
            &rendered_field.attr_name,
            &model.name,
        )?;
        parameters.push(RenderedUnpackedInputField {
            attr_name: rendered_field.attr_name.clone(),
            annotation: rendered_field.annotation.clone(),
            default_kind: rendered_field.default_kind.clone(),
            doc: planned_field
                .doc
                .as_ref()
                .and_then(|doc| doc.for_language(Language::Python))
                .map(str::to_string),
        });
        request_fields.push(RenderedUnpackedRequestField {
            attr_name: rendered_field.attr_name.clone(),
            value_expr: rendered_field.attr_name.clone(),
            default_kind: rendered_field.default_kind.clone(),
        });
    }

    Ok(RenderedUnpackedInput {
        model_name: model.name.clone(),
        parameters,
        request_fields,
        flattened_messages,
        functions: planned_model
            .generated_model
            .functions
            .iter()
            .map(|(field_name, function)| {
                let result_annotation = python_function_result_annotation(&function.result);
                let result_type_parameters = function
                    .result_type_parameter
                    .iter()
                    .cloned()
                    .collect::<BTreeSet<_>>();
                let erased_result_annotation =
                    erase_python_type_parameters(&result_annotation, &result_type_parameters);
                let args = python_function_args(&function.args);
                let alternate_annotation = function
                    .alternate_type
                    .as_ref()
                    .map(python_authored_type_annotation);
                RenderedFunctionField {
                    callable_field_name: planned_model
                        .generated_model
                        .field_name_override(field_name)
                        .map(python_field_name)
                        .unwrap_or_else(|| python_field_name(field_name)),
                    args_field_name: planned_model
                        .generated_model
                        .field_name_override(&function.args_field)
                        .map(python_field_name)
                        .unwrap_or_else(|| python_field_name(&function.args_field)),
                    args: args.clone(),
                    primary: function.primary,
                    callable_annotation: python_function_annotation_from_args(
                        &args,
                        alternate_annotation.as_deref(),
                        &erased_result_annotation,
                    ),
                    result_annotation,
                    erased_result_annotation,
                    result_type_parameter: function.result_type_parameter.clone(),
                    alternate_annotation,
                }
            })
            .collect(),
    })
}

fn register_unpacked_parameter_name(
    parameter_sources: &mut BTreeMap<String, String>,
    parameter_name: &str,
    source: &str,
    model_name: &str,
) -> Result<()> {
    if let Some(conflicting_field) = parameter_sources.get(parameter_name) {
        return Err(Error::FlattenedApiFieldConflict {
            type_name: model_name.to_string(),
            field: parameter_name.to_string(),
            conflicting_field: conflicting_field.clone(),
        });
    }
    parameter_sources.insert(parameter_name.to_string(), source.to_string());
    Ok(())
}

fn build_flattened_message(
    planned_field: &PlannedField,
    rendered_field: &RenderedField,
    models: &IndexMap<String, RenderedModel>,
    api_plan: &ApiPlan,
) -> Option<RenderedFlattenedMessage> {
    let PlannedFieldKind::Singular(PlannedValueType::Message(message_type)) = &planned_field.kind
    else {
        return None;
    };
    let nested_planned_model = api_plan.models.get(&message_type.info.full_name)?;
    if !nested_planned_model.flatten_in_api {
        return None;
    }
    let nested_rendered_model = models.get(&message_type.info.full_name)?;

    Some(RenderedFlattenedMessage {
        local_name: rendered_field.attr_name.clone(),
        model_name: nested_rendered_model.name.clone(),
        required: planned_field.required,
        fields: nested_planned_model
            .fields
            .iter()
            .zip(nested_rendered_model.fields.iter())
            .map(
                |(nested_planned_field, nested_rendered_field)| RenderedFlattenedMessageField {
                    attr_name: nested_rendered_field.attr_name.clone(),
                    annotation: nested_planned_model
                        .generated_model
                        .field_flattened_annotation(&nested_planned_field.proto_name)
                        .and_then(|annotation| annotation.for_language(Language::Python))
                        .map(|annotation| {
                            if nested_planned_field.required {
                                annotation.to_string()
                            } else {
                                format!("{annotation} | None")
                            }
                        })
                        .unwrap_or_else(|| nested_rendered_field.annotation.clone()),
                    value_expr: nested_rendered_field.attr_name.clone(),
                    default_kind: if nested_planned_field.required {
                        PythonFieldDefaultKind::Required
                    } else {
                        nested_rendered_field.default_kind.clone()
                    },
                    doc: nested_planned_field
                        .doc
                        .as_ref()
                        .and_then(|doc| doc.for_language(Language::Python))
                        .map(str::to_string),
                },
            )
            .collect(),
    })
}

fn resolve_message_value_conversion(
    message: &PlannedMessageType,
    proto_ref_override: Option<&str>,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> MessageValueConversion {
    if let Some(language_override) = &message.replacement
        && let Some(type_name) = python_replacement_type_name(language_override)
    {
        return MessageValueConversion {
            annotation: type_name,
            kind: MessageValueConversionKind::Override {
                from_proto: python_from_proto_converter(&message.info.full_name, language_override),
                to_proto: python_to_proto_converter(&message.info.full_name, language_override),
            },
            imports: PythonImports::default(),
        };
    }
    if let Some(authored_type) = &message.authored_type {
        return MessageValueConversion {
            annotation: python_authored_type_annotation(authored_type),
            kind: MessageValueConversionKind::Override {
                from_proto: python_default_from_proto_name(&message.info.full_name),
                to_proto: python_default_to_proto_name(&message.info.full_name),
            },
            imports: PythonImports::default(),
        };
    }

    ensure_rendered_model(
        message,
        proto_ref_override,
        api_plan,
        enums,
        flags,
        variants,
        models,
    );
    let kind = if message.source == PlannedMessageSource::Wit {
        MessageValueConversionKind::NativeModel
    } else {
        MessageValueConversionKind::GeneratedModel {
            model_name: message.model_name.clone(),
        }
    };

    MessageValueConversion {
        annotation: models
            .get(&message.info.full_name)
            .expect("planned model should be rendered")
            .name
            .clone(),
        kind,
        imports: PythonImports::default(),
    }
}

fn ensure_rendered_model(
    message: &PlannedMessageType,
    proto_ref_override: Option<&str>,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) {
    let planned_model = api_plan
        .models
        .get(&message.info.full_name)
        .unwrap_or_else(|| panic!("planned model should exist for {}", message.info.full_name));

    if let Some(existing) = models.get_mut(&message.info.full_name) {
        if let Some(proto_ref) = proto_ref_override {
            existing.proto_ref = Some(proto_ref.to_string());
            existing.proto_module_path = Some(module_path(proto_ref).to_string());
        }
        return;
    }

    let (proto_ref, proto_module_path) = if message.source == PlannedMessageSource::Proto {
        (
            Some(
                proto_ref_override
                    .map(str::to_string)
                    .or_else(|| {
                        message_python_ref(&planned_model.info).map(|reference| reference.type_ref)
                    })
                    .expect("descriptor messages should have a Python reference"),
            ),
            Some(
                proto_ref_override
                    .map(module_path)
                    .map(str::to_string)
                    .or_else(|| {
                        message_python_ref(&planned_model.info)
                            .map(|reference| reference.module_path)
                    })
                    .expect("descriptor messages should have a Python module path"),
            ),
        )
    } else {
        (None, None)
    };

    models.insert(
        message.info.full_name.clone(),
        RenderedModel {
            full_name: message.info.full_name.clone(),
            name: planned_model.name.clone(),
            source: message.source,
            proto_ref,
            proto_module_path,
            capabilities: planned_model.capabilities,
            experimental: planned_model.experimental,
            fields: Vec::new(),
            sourced_fields: Vec::new(),
        },
    );

    let fields = planned_model
        .fields
        .iter()
        .map(|field| build_field(field, api_plan, enums, flags, variants, models))
        .collect();

    let sourced_fields = planned_model
        .sourced_fields
        .iter()
        .map(|field| build_sourced_field(field, api_plan, enums, flags, variants, models))
        .collect();

    models
        .get_mut(&message.info.full_name)
        .expect("model should be inserted before recursive field resolution")
        .fields = fields;
    models
        .get_mut(&message.info.full_name)
        .expect("model should be inserted before recursive field resolution")
        .sourced_fields = sourced_fields;
}

fn ensure_rendered_enum(planned_enum: &PlannedEnum, enums: &mut IndexMap<String, RenderedEnum>) {
    enums
        .entry(planned_enum.info.full_name.clone())
        .or_insert_with(|| RenderedEnum {
            name: planned_enum.name.clone(),
            values: planned_enum
                .values
                .iter()
                .map(|value| RenderedEnumValue {
                    name: value.name.clone(),
                    number: value.number,
                })
                .collect(),
        });
}

fn ensure_rendered_flags(
    planned_flags: &PlannedFlags,
    flags: &mut IndexMap<String, RenderedFlags>,
) {
    flags
        .entry(planned_flags.info.full_name.clone())
        .or_insert_with(|| RenderedFlags {
            name: planned_flags.name.clone(),
            flags: planned_flags
                .flags
                .iter()
                .map(|flag| RenderedFlag {
                    name: flag.name.clone(),
                    bit: flag.bit,
                })
                .collect(),
        });
}

fn ensure_rendered_variant(
    planned_variant: &PlannedVariant,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) {
    if variants.contains_key(&planned_variant.info.full_name) {
        return;
    }

    let cases = planned_variant
        .cases
        .iter()
        .map(|case| RenderedVariantCase {
            name: case.name.clone(),
            payload_annotation: case.payload.as_ref().map(|payload| {
                resolve_planned_value_type(payload, api_plan, enums, flags, variants, models)
                    .annotation
            }),
        })
        .collect();
    variants.insert(
        planned_variant.info.full_name.clone(),
        RenderedVariant {
            name: planned_variant.name.clone(),
            cases,
        },
    );
}

fn build_field(
    field: &PlannedField,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> RenderedField {
    let attr_name = python_field_name(&field.authored_name);

    if let PlannedFieldKind::Map { key, value } = &field.kind {
        let key_type = resolve_planned_value_type(key, api_plan, enums, flags, variants, models);
        let value_type =
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models);
        let mut imports = key_type.imports.clone();
        imports.extend(&value_type.imports);
        return RenderedField {
            attr_name: attr_name.clone(),
            annotation: python_field_annotation(
                field,
                format!("dict[{}, {}]", key_type.annotation, value_type.annotation),
                false,
            ),
            default_kind: PythonFieldDefaultKind::EmptyDict,
            default_expr: Some("dataclasses.field(default_factory=dict)".to_string()),
            from_proto_setup_lines: Vec::new(),
            from_proto_expr: map_value_from_proto_expr(&value_type, &field.proto_name),
            to_proto_lines: map_value_to_proto_lines(&value_type, &attr_name, &field.proto_name),
            imports,
        };
    }

    let (resolved_type, repeated) = match &field.kind {
        PlannedFieldKind::Singular(value) => (
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models),
            false,
        ),
        PlannedFieldKind::Repeated(value) => (
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models),
            true,
        ),
        PlannedFieldKind::Map { .. } => unreachable!("handled above"),
    };
    let model_name = field.owner_name.clone();

    if repeated {
        return RenderedField {
            attr_name: attr_name.clone(),
            annotation: python_field_annotation(
                field,
                format!("list[{}]", resolved_type.annotation),
                false,
            ),
            default_kind: PythonFieldDefaultKind::EmptyList,
            default_expr: Some("dataclasses.field(default_factory=list)".to_string()),
            from_proto_setup_lines: Vec::new(),
            from_proto_expr: repeated_from_proto_expr(&resolved_type, &field.proto_name),
            to_proto_lines: repeated_to_proto_lines(&resolved_type, &attr_name, &field.proto_name),
            imports: resolved_type.imports,
        };
    }

    if let Some(default_value) = &field.default_value {
        let default_expr = enum_default_expr(&resolved_type, &default_value.enum_case);
        return RenderedField {
            attr_name: attr_name.clone(),
            annotation: python_field_annotation(field, resolved_type.annotation.clone(), false),
            default_kind: PythonFieldDefaultKind::Expression(default_expr.clone()),
            default_expr: Some(default_expr),
            from_proto_setup_lines: Vec::new(),
            from_proto_expr: defaulted_enum_from_proto_expr(
                &resolved_type,
                &field.proto_name,
                field,
                &default_value.enum_case,
            ),
            to_proto_lines: required_to_proto_lines(&attr_name, &field.proto_name, &resolved_type),
            imports: resolved_type.imports,
        };
    }

    if field.required {
        return RenderedField {
            attr_name: attr_name.clone(),
            annotation: python_field_annotation(field, resolved_type.annotation.clone(), false),
            default_kind: PythonFieldDefaultKind::Required,
            default_expr: None,
            from_proto_setup_lines: required_from_proto_setup_lines(
                &model_name,
                &attr_name,
                &field.proto_name,
                field,
                &resolved_type,
            ),
            from_proto_expr: attr_name.clone(),
            to_proto_lines: match &field.role {
                PlannedFieldRole::Function(function_field)
                    if function_field.converter.is_some() =>
                {
                    required_function_to_proto_lines(
                        &attr_name,
                        &field.proto_name,
                        &resolved_type,
                        function_field
                            .converter
                            .as_deref()
                            .expect("function converter should be present"),
                    )
                }
                _ => required_to_proto_lines(&attr_name, &field.proto_name, &resolved_type),
            },
            imports: resolved_type.imports,
        };
    }

    RenderedField {
        attr_name: attr_name.clone(),
        annotation: python_field_annotation(field, resolved_type.annotation.clone(), true),
        default_kind: PythonFieldDefaultKind::None,
        default_expr: Some("None".to_string()),
        from_proto_setup_lines: Vec::new(),
        from_proto_expr: optional_from_proto_expr(&resolved_type, &field.proto_name, field),
        to_proto_lines: match &field.role {
            PlannedFieldRole::Function(function_field) if function_field.converter.is_some() => {
                optional_function_to_proto_lines(
                    &attr_name,
                    &field.proto_name,
                    &resolved_type,
                    function_field
                        .converter
                        .as_deref()
                        .expect("function converter should be present"),
                )
            }
            _ => optional_to_proto_lines(&resolved_type, &attr_name, &field.proto_name),
        },
        imports: resolved_type.imports,
    }
}

fn build_sourced_field(
    field: &PlannedSourcedField,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> RenderedSourcedField {
    let source_expr = field.source_expr.to_string();
    if let PlannedFieldKind::Map { value, .. } = &field.kind {
        let value_type =
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models);
        return RenderedSourcedField {
            to_proto_lines: sourced_map_value_to_proto_lines(
                &value_type,
                &source_expr,
                &field.proto_name,
            ),
        };
    }

    let (resolved_type, repeated) = match &field.kind {
        PlannedFieldKind::Singular(value) => (
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models),
            false,
        ),
        PlannedFieldKind::Repeated(value) => (
            resolve_planned_value_type(value, api_plan, enums, flags, variants, models),
            true,
        ),
        PlannedFieldKind::Map { .. } => unreachable!("handled above"),
    };
    let source_expr = field.source_expr.to_string();
    if repeated {
        return RenderedSourcedField {
            to_proto_lines: sourced_repeated_to_proto_lines(
                &resolved_type,
                &source_expr,
                &field.proto_name,
            ),
        };
    }

    RenderedSourcedField {
        to_proto_lines: sourced_optional_to_proto_lines(
            &resolved_type,
            &source_expr,
            &field.proto_name,
        ),
    }
}

fn python_field_annotation(
    field: &PlannedField,
    default_base_annotation: String,
    is_optional: bool,
) -> String {
    if let PlannedFieldRole::FunctionArgs(function) = &field.role {
        let args_annotation = python_function_arg_field_annotation(function, &field.proto_name);
        return if is_optional {
            format!("{args_annotation} | None")
        } else {
            args_annotation
        };
    }

    if let Some(annotation) = &field.annotation_override {
        if let Some(annotation) = annotation.for_language(Language::Python) {
            return annotation.to_string();
        }
    }

    if let PlannedFieldRole::Function(function) = &field.role {
        let result_annotation = erase_python_type_parameters(
            &python_function_result_annotation(&function.result),
            &function
                .result_type_parameter
                .iter()
                .cloned()
                .collect::<BTreeSet<_>>(),
        );
        return python_function_annotation_from_args(
            &python_function_args(&function.args),
            function
                .alternate_type
                .as_ref()
                .map(python_authored_type_annotation)
                .as_deref(),
            &result_annotation,
        );
    }

    if is_optional {
        format!("{default_base_annotation} | None")
    } else {
        default_base_annotation
    }
}

fn python_function_field_annotation(
    function: &FunctionFieldSpec,
    alternate_annotation: Option<String>,
) -> String {
    let result_annotation = erase_python_type_parameters(
        &python_function_result_annotation(&function.result),
        &function
            .result_type_parameter
            .iter()
            .cloned()
            .collect::<BTreeSet<_>>(),
    );
    python_function_annotation_from_args(
        &python_function_args(&function.args),
        alternate_annotation.as_deref(),
        &result_annotation,
    )
}

fn python_function_annotation_from_args(
    args: &RenderedFunctionArgs,
    alternate_annotation: Option<&str>,
    result_annotation: &str,
) -> String {
    let callable_annotation = match args {
        RenderedFunctionArgs::Varargs { .. } => {
            format!("collections.abc.Callable[..., {}]", result_annotation)
        }
        RenderedFunctionArgs::Typed { parameters } => format!(
            "collections.abc.Callable[[{}], {}]",
            parameters
                .iter()
                .map(|parameter| parameter.annotation.clone())
                .collect::<Vec<_>>()
                .join(", "),
            result_annotation
        ),
    };
    if let Some(alternate_annotation) = alternate_annotation {
        format!("{alternate_annotation} | {callable_annotation}")
    } else {
        callable_annotation
    }
}

fn resolve_planned_value_type(
    value_type: &PlannedValueType,
    api_plan: &ApiPlan,
    enums: &mut IndexMap<String, RenderedEnum>,
    flags: &mut IndexMap<String, RenderedFlags>,
    variants: &mut IndexMap<String, RenderedVariant>,
    models: &mut IndexMap<String, RenderedModel>,
) -> ResolvedFieldType {
    match value_type {
        PlannedValueType::Scalar(PlannedScalarType::Float) => ResolvedFieldType {
            annotation: "float".to_string(),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Scalar,
            message_conversion: None,
            enum_conversion: None,
        },
        PlannedValueType::Scalar(PlannedScalarType::Int32 | PlannedScalarType::Int64) => {
            ResolvedFieldType {
                annotation: "int".to_string(),
                imports: PythonImports::default(),
                kind: ResolvedFieldKind::Scalar,
                message_conversion: None,
                enum_conversion: None,
            }
        }
        PlannedValueType::Scalar(PlannedScalarType::Bool) => ResolvedFieldType {
            annotation: "bool".to_string(),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Scalar,
            message_conversion: None,
            enum_conversion: None,
        },
        PlannedValueType::Scalar(PlannedScalarType::String) => ResolvedFieldType {
            annotation: "str".to_string(),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Scalar,
            message_conversion: None,
            enum_conversion: None,
        },
        PlannedValueType::Scalar(PlannedScalarType::Bytes) => ResolvedFieldType {
            annotation: "bytes".to_string(),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Scalar,
            message_conversion: None,
            enum_conversion: None,
        },
        PlannedValueType::Enum(enum_type) => {
            if let Some(replacement) = &enum_type.replacement
                && let Some(type_name) = python_replacement_type_name(replacement)
            {
                ResolvedFieldType {
                    annotation: type_name,
                    imports: PythonImports::default(),
                    kind: ResolvedFieldKind::Enum,
                    message_conversion: None,
                    enum_conversion: enum_type.info.as_ref().map(|info| EnumValueConversion {
                        from_proto: python_from_proto_converter(&info.full_name, replacement),
                        to_proto: python_to_proto_converter(&info.full_name, replacement),
                    }),
                }
            } else if let (Some(info), Some(name)) = (&enum_type.info, &enum_type.name) {
                if let Some(planned_enum) = api_plan.enums.get(&info.full_name) {
                    ensure_rendered_enum(planned_enum, enums);
                }
                let mut imports = PythonImports::default();
                if let Some(reference) = enum_python_ref(info) {
                    imports.module_imports.insert(reference.module_path);
                }
                ResolvedFieldType {
                    annotation: name.clone(),
                    imports,
                    kind: ResolvedFieldKind::Enum,
                    message_conversion: None,
                    enum_conversion: None,
                }
            } else {
                ResolvedFieldType {
                    annotation: "int".to_string(),
                    imports: PythonImports::default(),
                    kind: ResolvedFieldKind::Enum,
                    message_conversion: None,
                    enum_conversion: None,
                }
            }
        }
        PlannedValueType::Flags(flags_type) => {
            if let Some(planned_flags) = api_plan.flags.get(&flags_type.info.full_name) {
                ensure_rendered_flags(planned_flags, flags);
            }
            ResolvedFieldType {
                annotation: flags_type.name.clone(),
                imports: PythonImports::default(),
                kind: ResolvedFieldKind::Scalar,
                message_conversion: None,
                enum_conversion: None,
            }
        }
        PlannedValueType::Variant(variant_type) => {
            if let Some(planned_variant) = api_plan.variants.get(&variant_type.info.full_name) {
                ensure_rendered_variant(planned_variant, api_plan, enums, flags, variants, models);
            }
            ResolvedFieldType {
                annotation: variant_type.name.clone(),
                imports: PythonImports::default(),
                kind: ResolvedFieldKind::Scalar,
                message_conversion: None,
                enum_conversion: None,
            }
        }
        PlannedValueType::Message(message_type) => {
            let conversion = if let Some(replacement) = &message_type.replacement
                && let Some(type_name) = python_replacement_type_name(replacement)
            {
                MessageValueConversion {
                    annotation: type_name,
                    kind: MessageValueConversionKind::Override {
                        from_proto: python_from_proto_converter(
                            &message_type.info.full_name,
                            replacement,
                        ),
                        to_proto: python_to_proto_converter(
                            &message_type.info.full_name,
                            replacement,
                        ),
                    },
                    imports: PythonImports::default(),
                }
            } else if let Some(authored_type) = &message_type.authored_type {
                MessageValueConversion {
                    annotation: python_authored_type_annotation(authored_type),
                    kind: MessageValueConversionKind::Override {
                        from_proto: python_default_from_proto_name(&message_type.info.full_name),
                        to_proto: python_default_to_proto_name(&message_type.info.full_name),
                    },
                    imports: PythonImports::default(),
                }
            } else {
                ensure_rendered_model(message_type, None, api_plan, enums, flags, variants, models);
                let model = models
                    .get(&message_type.info.full_name)
                    .expect("planned model should be rendered");
                MessageValueConversion {
                    annotation: model.name.clone(),
                    kind: MessageValueConversionKind::GeneratedModel {
                        model_name: message_type.model_name.clone(),
                    },
                    imports: PythonImports::default(),
                }
            };
            ResolvedFieldType {
                annotation: conversion.annotation.clone(),
                imports: conversion.imports.clone(),
                kind: ResolvedFieldKind::Message,
                message_conversion: Some(conversion),
                enum_conversion: None,
            }
        }
        PlannedValueType::Tuple(items) => ResolvedFieldType {
            annotation: format!(
                "tuple[{}]",
                items
                    .iter()
                    .map(|item| {
                        resolve_planned_value_type(item, api_plan, enums, flags, variants, models)
                            .annotation
                    })
                    .collect::<Vec<_>>()
                    .join(", ")
            ),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Scalar,
            message_conversion: None,
            enum_conversion: None,
        },
        PlannedValueType::Result { ok, err } => {
            let ok = ok
                .as_ref()
                .map(|ok| resolve_planned_value_type(ok, api_plan, enums, flags, variants, models));
            let err = err.as_ref().map(|err| {
                resolve_planned_value_type(err, api_plan, enums, flags, variants, models)
            });
            let mut imports = PythonImports::default();
            if let Some(ok) = &ok {
                imports.extend(&ok.imports);
            }
            if let Some(err) = &err {
                imports.extend(&err.imports);
            }
            ResolvedFieldType {
                annotation: python_result_annotation(
                    ok.as_ref().map(|ok| ok.annotation.clone()),
                    err.as_ref().map(|err| err.annotation.clone()),
                ),
                imports,
                kind: ResolvedFieldKind::Scalar,
                message_conversion: None,
                enum_conversion: None,
            }
        }
        PlannedValueType::External {
            type_name,
            fallback,
        } => {
            let mut resolved =
                resolve_planned_value_type(fallback, api_plan, enums, flags, variants, models);
            if let Some(annotation) = type_name.for_language(Language::Python) {
                resolved.annotation = annotation.to_string();
            }
            resolved
        }
        PlannedValueType::Unknown => ResolvedFieldType {
            annotation: "object".to_string(),
            imports: PythonImports::default(),
            kind: ResolvedFieldKind::Unknown,
            message_conversion: None,
            enum_conversion: None,
        },
    }
}

fn repeated_from_proto_expr(resolved_type: &ResolvedFieldType, proto_name: &str) -> String {
    match resolved_type.kind {
        ResolvedFieldKind::Message => format!(
            "[{} for value in proto.{proto_name}]",
            resolved_type
                .message_conversion
                .as_ref()
                .expect("message conversion should be present")
                .from_proto_expr("value")
        ),
        ResolvedFieldKind::Enum => format!(
            "[{} for value in proto.{proto_name}]",
            enum_from_proto_expr(resolved_type, "value")
        ),
        _ => format!("list(proto.{proto_name})"),
    }
}

fn repeated_to_proto_lines(
    resolved_type: &ResolvedFieldType,
    attr_name: &str,
    proto_name: &str,
) -> Vec<String> {
    repeated_to_proto_lines_from_expr(resolved_type, &format!("self.{attr_name}"), proto_name)
}

fn repeated_to_proto_lines_from_expr(
    resolved_type: &ResolvedFieldType,
    value_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match resolved_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("if {value_expr}:"),
            format!("    for value in {value_expr}:"),
            format!("        item = message.{proto_name}.add()"),
            format!(
                "        item.CopyFrom({})",
                resolved_type
                    .message_conversion
                    .as_ref()
                    .expect("message conversion should be present")
                    .to_proto_expr("value")
            ),
        ],
        ResolvedFieldKind::Enum => vec![
            format!("if {value_expr}:"),
            format!(
                "    message.{proto_name}.extend({} for value in {value_expr})",
                enum_to_proto_expr(resolved_type, "value")
            ),
        ],
        _ => vec![
            format!("if {value_expr}:"),
            format!("    message.{proto_name}.extend({value_expr})"),
        ],
    }
}

fn map_value_from_proto_expr(map_value_type: &ResolvedFieldType, proto_name: &str) -> String {
    match map_value_type.kind {
        ResolvedFieldKind::Message => format!(
            "{{key: {} for key, value in proto.{proto_name}.items()}}",
            map_value_type
                .message_conversion
                .as_ref()
                .expect("message conversion should be present")
                .from_proto_expr("value")
        ),
        ResolvedFieldKind::Enum => format!(
            "{{key: {} for key, value in proto.{proto_name}.items()}}",
            enum_from_proto_expr(map_value_type, "value")
        ),
        _ => format!("{{key: value for key, value in proto.{proto_name}.items()}}"),
    }
}

fn map_value_to_proto_lines(
    map_value_type: &ResolvedFieldType,
    attr_name: &str,
    proto_name: &str,
) -> Vec<String> {
    map_value_to_proto_lines_from_expr(map_value_type, &format!("self.{attr_name}"), proto_name)
}

fn map_value_to_proto_lines_from_expr(
    map_value_type: &ResolvedFieldType,
    value_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match map_value_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("if {value_expr}:"),
            format!("    for key, value in {value_expr}.items():"),
            format!(
                "        message.{proto_name}[key].CopyFrom({})",
                map_value_type
                    .message_conversion
                    .as_ref()
                    .expect("message conversion should be present")
                    .to_proto_expr("value")
            ),
        ],
        ResolvedFieldKind::Enum => vec![
            format!("if {value_expr}:"),
            format!(
                "    message.{proto_name}.update({{key: {} for key, value in {value_expr}.items()}})",
                enum_to_proto_expr(map_value_type, "value")
            ),
        ],
        _ => vec![
            format!("if {value_expr}:"),
            format!("    message.{proto_name}.update({value_expr})"),
        ],
    }
}

fn required_from_proto_setup_lines(
    model_name: &str,
    attr_name: &str,
    proto_name: &str,
    field: &PlannedField,
    resolved_type: &ResolvedFieldType,
) -> Vec<String> {
    let error = python_string_literal(&format!("missing required field {model_name}.{attr_name}"));
    let proto_expr = format!("proto.{proto_name}");
    let mut lines = Vec::new();
    if field.has_presence {
        lines.push(format!("if not proto.HasField(\"{proto_name}\"):"));
        lines.push(format!("    raise ValueError({error})"));
    } else if matches!(
        &field.kind,
        PlannedFieldKind::Singular(PlannedValueType::Scalar(
            PlannedScalarType::String | PlannedScalarType::Bytes
        ))
    ) {
        lines.push(format!("if not proto.{proto_name}:"));
        lines.push(format!("    raise ValueError({error})"));
    }
    lines.push(format!(
        "{attr_name} = {}",
        required_from_proto_value_expr(resolved_type, &proto_expr)
    ));
    lines
}

fn required_from_proto_value_expr(resolved_type: &ResolvedFieldType, proto_expr: &str) -> String {
    match resolved_type.kind {
        ResolvedFieldKind::Message => resolved_type
            .message_conversion
            .as_ref()
            .expect("message conversion should be present")
            .from_proto_expr(proto_expr),
        ResolvedFieldKind::Enum => enum_from_proto_expr(resolved_type, proto_expr),
        _ => proto_expr.to_string(),
    }
}

fn required_to_proto_lines(
    attr_name: &str,
    proto_name: &str,
    resolved_type: &ResolvedFieldType,
) -> Vec<String> {
    let mut lines = Vec::new();
    match resolved_type.kind {
        ResolvedFieldKind::Message => lines.push(format!(
            "message.{proto_name}.CopyFrom({})",
            resolved_type
                .message_conversion
                .as_ref()
                .expect("message conversion should be present")
                .to_proto_expr(&format!("self.{attr_name}"))
        )),
        ResolvedFieldKind::Enum => lines.push(format!(
            "message.{proto_name} = {}",
            enum_to_proto_expr(resolved_type, &format!("self.{attr_name}"))
        )),
        _ => lines.push(format!(
            "message.{proto_name} = {}",
            format!("self.{attr_name}")
        )),
    }
    lines
}

fn required_function_to_proto_lines(
    attr_name: &str,
    proto_name: &str,
    resolved_type: &ResolvedFieldType,
    converter: &str,
) -> Vec<String> {
    let converted_value = format!("{converter}(self.{attr_name})");
    match resolved_type.kind {
        ResolvedFieldKind::Message => {
            vec![format!("message.{proto_name}.CopyFrom({converted_value})")]
        }
        _ => vec![format!("message.{proto_name} = {converted_value}")],
    }
}

fn optional_from_proto_expr(
    resolved_type: &ResolvedFieldType,
    proto_name: &str,
    field: &PlannedField,
) -> String {
    let has_presence = field.has_presence;
    match resolved_type.kind {
        ResolvedFieldKind::Message => {
            let from_proto = resolved_type
                .message_conversion
                .as_ref()
                .expect("message conversion should be present")
                .from_proto_expr(&format!("proto.{proto_name}"));
            if has_presence {
                format!("{from_proto} if proto.HasField(\"{proto_name}\") else None")
            } else {
                from_proto
            }
        }
        ResolvedFieldKind::Enum => {
            let from_proto = enum_from_proto_expr(resolved_type, &format!("proto.{proto_name}"));
            if has_presence {
                format!("{from_proto} if proto.HasField(\"{proto_name}\") else None")
            } else {
                from_proto
            }
        }
        _ => {
            if has_presence {
                format!("proto.{proto_name} if proto.HasField(\"{proto_name}\") else None")
            } else {
                format!("proto.{proto_name}")
            }
        }
    }
}

fn defaulted_enum_from_proto_expr(
    resolved_type: &ResolvedFieldType,
    proto_name: &str,
    field: &PlannedField,
    enum_case: &str,
) -> String {
    let default_expr = enum_default_expr(resolved_type, enum_case);
    let from_proto = enum_from_proto_expr(resolved_type, &format!("proto.{proto_name}"));
    if field.has_presence {
        format!("{from_proto} if proto.HasField(\"{proto_name}\") else {default_expr}")
    } else {
        from_proto
    }
}

fn optional_to_proto_lines(
    resolved_type: &ResolvedFieldType,
    attr_name: &str,
    proto_name: &str,
) -> Vec<String> {
    optional_to_proto_lines_from_expr(resolved_type, &format!("self.{attr_name}"), proto_name)
}

fn optional_to_proto_lines_from_expr(
    resolved_type: &ResolvedFieldType,
    value_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match resolved_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("if {value_expr} is not None:"),
            format!(
                "    message.{proto_name}.CopyFrom({})",
                resolved_type
                    .message_conversion
                    .as_ref()
                    .expect("message conversion should be present")
                    .to_proto_expr(value_expr)
            ),
        ],
        ResolvedFieldKind::Enum => vec![
            format!("if {value_expr} is not None:"),
            format!(
                "    message.{proto_name} = {}",
                enum_to_proto_expr(resolved_type, value_expr)
            ),
        ],
        _ => vec![
            format!("if {value_expr} is not None:"),
            format!("    message.{proto_name} = {value_expr}"),
        ],
    }
}

fn optional_function_to_proto_lines(
    attr_name: &str,
    proto_name: &str,
    resolved_type: &ResolvedFieldType,
    converter: &str,
) -> Vec<String> {
    let value_expr = format!("self.{attr_name}");
    let converted_value = format!("{converter}({value_expr})");
    match resolved_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("if {value_expr} is not None:"),
            format!("    message.{proto_name}.CopyFrom({converted_value})"),
        ],
        _ => vec![
            format!("if {value_expr} is not None:"),
            format!("    message.{proto_name} = {converted_value}"),
        ],
    }
}

fn sourced_repeated_to_proto_lines(
    resolved_type: &ResolvedFieldType,
    source_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match resolved_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("for value in {source_expr}:"),
            format!("    item = message.{proto_name}.add()"),
            format!(
                "    item.CopyFrom({})",
                resolved_type
                    .message_conversion
                    .as_ref()
                    .expect("message conversion should be present")
                    .to_proto_expr("value")
            ),
        ],
        ResolvedFieldKind::Enum => vec![format!(
            "message.{proto_name}.extend({} for value in {source_expr})",
            enum_to_proto_expr(resolved_type, "value")
        )],
        _ => vec![format!("message.{proto_name}.extend({source_expr})")],
    }
}

fn sourced_map_value_to_proto_lines(
    map_value_type: &ResolvedFieldType,
    source_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match map_value_type.kind {
        ResolvedFieldKind::Message => vec![
            format!("for key, value in {source_expr}.items():"),
            format!(
                "    message.{proto_name}[key].CopyFrom({})",
                map_value_type
                    .message_conversion
                    .as_ref()
                    .expect("message conversion should be present")
                    .to_proto_expr("value")
            ),
        ],
        ResolvedFieldKind::Enum => vec![format!(
            "message.{proto_name}.update({{key: {} for key, value in {source_expr}.items()}})",
            enum_to_proto_expr(map_value_type, "value")
        )],
        _ => vec![format!("message.{proto_name}.update({source_expr})")],
    }
}

fn sourced_optional_to_proto_lines(
    resolved_type: &ResolvedFieldType,
    source_expr: &str,
    proto_name: &str,
) -> Vec<String> {
    match resolved_type.kind {
        ResolvedFieldKind::Message => vec![format!(
            "message.{proto_name}.CopyFrom({})",
            resolved_type
                .message_conversion
                .as_ref()
                .expect("message conversion should be present")
                .to_proto_expr(source_expr)
        )],
        ResolvedFieldKind::Enum => vec![format!(
            "message.{proto_name} = {}",
            enum_to_proto_expr(resolved_type, source_expr)
        )],
        _ => vec![format!("message.{proto_name} = {source_expr}")],
    }
}

fn enum_from_proto_expr(resolved_type: &ResolvedFieldType, expr: &str) -> String {
    if let Some(enum_conversion) = &resolved_type.enum_conversion {
        enum_conversion.from_proto_expr(expr)
    } else {
        format!("{}({expr})", resolved_type.annotation)
    }
}

fn enum_default_expr(resolved_type: &ResolvedFieldType, enum_case: &str) -> String {
    let value_name = if resolved_type.enum_conversion.is_some() {
        enum_case.to_shouty_snake_case()
    } else {
        enum_case.to_upper_camel_case()
    };
    format!("{}.{}", resolved_type.annotation, value_name)
}

fn enum_to_proto_expr(resolved_type: &ResolvedFieldType, expr: &str) -> String {
    if let Some(enum_conversion) = &resolved_type.enum_conversion {
        enum_conversion.to_proto_expr(expr)
    } else {
        format!("int({expr})")
    }
}

#[derive(Debug)]
struct PythonReference {
    module_path: String,
    type_ref: String,
}

fn message_python_ref(type_info: &PlannedTypeInfo) -> Option<PythonReference> {
    let module_path =
        python_module_path_for_file_name(type_info.file_name.as_deref(), &type_info.package)?;
    let relative_name = relative_descriptor_name(&type_info.full_name, &type_info.package);
    Some(PythonReference {
        type_ref: format!("{module_path}.{relative_name}"),
        module_path,
    })
}

fn enum_python_ref(type_info: &PlannedTypeInfo) -> Option<PythonReference> {
    let module_path =
        python_module_path_for_file_name(type_info.file_name.as_deref(), &type_info.package)?;
    let relative_name = relative_descriptor_name(&type_info.full_name, &type_info.package);
    Some(PythonReference {
        type_ref: format!("{module_path}.{relative_name}.ValueType"),
        module_path,
    })
}

fn python_module_path_for_file_name(file_name: Option<&str>, package: &str) -> Option<String> {
    if let Some(file_name) = file_name {
        let mut module_path = file_name.replace('/', ".");
        if let Some(stripped) = module_path.strip_suffix(".proto") {
            module_path = format!("{stripped}_pb2");
        }
        if let Some(suffix) = module_path.strip_prefix("temporal.") {
            module_path = format!("temporalio.{suffix}");
        }
        return Some(module_path);
    }

    if package.is_empty() {
        None
    } else if let Some(suffix) = package.strip_prefix("temporal.") {
        Some(format!("temporalio.{suffix}"))
    } else {
        Some(package.to_string())
    }
}

fn render_package(
    enums: &[&RenderedEnum],
    flags: &[&RenderedFlags],
    variants: &[&RenderedVariant],
    models: &[&RenderedModel],
    services: &[RenderedService<'_>],
    support_fragments: &[SupportFragmentSpec],
    language_imports: &[LanguageImportSpec],
) -> Result<GeneratedFiles> {
    let mut files = BTreeMap::new();
    render_support_package(&mut files, support_fragments)?;
    let model_names = enums
        .iter()
        .map(|enumeration| enumeration.name.clone())
        .chain(flags.iter().map(|flag_set| flag_set.name.clone()))
        .chain(variants.iter().map(|variant| variant.name.clone()))
        .chain(models.iter().map(|model| model.name.clone()))
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();
    let resource_operation_owners = resource_operation_owners(services);
    let resource_names = services
        .iter()
        .flat_map(|service| {
            service
                .resources
                .iter()
                .map(|resource| resource.type_name.clone())
        })
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();
    let support_names = support_export_names(support_fragments);

    insert_generated_file(
        &mut files,
        "__init__.py",
        render_package_init(services, &resource_names, &resource_operation_owners),
    )?;
    insert_generated_file(
        &mut files,
        "models.py",
        render_models_module(
            enums,
            flags,
            variants,
            models,
            &support_names,
            language_imports,
        ),
    )?;
    insert_generated_file(
        &mut files,
        "_resources/__init__.py",
        render_resources_package_init(services),
    )?;
    insert_generated_file(&mut files, "service.py", render_service_module(services))?;
    insert_generated_file(
        &mut files,
        "operations/__init__.py",
        render_operations_package_init(),
    )?;

    for service in services {
        for resource in &service.resources {
            let bound_operations = resource_bound_operations(service, resource);
            insert_generated_file(
                &mut files,
                format!("_resources/{}.py", resource_module_name(resource)),
                render_resource_module_file(
                    service,
                    resource,
                    &bound_operations,
                    &model_names,
                    &resource_names,
                    &support_names,
                    language_imports,
                ),
            )?;
        }
        for operation in &service.operations {
            if resource_operation_owners.contains_key(&operation_key(service.name, operation.name))
            {
                continue;
            }
            insert_generated_file(
                &mut files,
                format!("operations/{}.py", operation.attr_name),
                render_operation_module(
                    service,
                    operation,
                    &model_names,
                    &resource_names,
                    &support_names,
                    language_imports,
                ),
            )?;
        }
    }

    Ok(GeneratedFiles::directory(files))
}

fn operation_key(service_name: &str, operation_name: &str) -> String {
    format!("{service_name}::{operation_name}")
}

fn python_resource_type_id(service: &RenderedService<'_>, resource: &PlannedResource) -> String {
    format!("{}::resource::{}", service.name, resource.name)
}

fn resource_module_name(resource: &PlannedResource) -> String {
    python_ident(&resource.name.to_snake_case())
}

fn resource_operation_owners(services: &[RenderedService<'_>]) -> BTreeMap<String, String> {
    let mut owners = BTreeMap::new();
    for service in services {
        for resource in &service.resources {
            let resource_module = resource_module_name(resource);
            for method in &resource.methods {
                if let PlannedResourceMethodBindingSpec::Operation { operation_name, .. } =
                    &method.binding
                {
                    owners.insert(
                        operation_key(service.name, operation_name),
                        resource_module.clone(),
                    );
                }
            }
        }
    }
    owners
}

fn insert_generated_file(
    files: &mut BTreeMap<std::path::PathBuf, String>,
    path: impl Into<std::path::PathBuf>,
    contents: String,
) -> Result<()> {
    let path = path.into();
    if files.insert(path.clone(), contents).is_some() {
        return Err(Error::GeneratedFileConflict { path });
    }
    Ok(())
}

fn render_support_package(
    files: &mut BTreeMap<std::path::PathBuf, String>,
    support_fragments: &[SupportFragmentSpec],
) -> Result<Vec<String>> {
    let mut module_names = Vec::new();
    for fragment in support_fragments {
        let module_name = support_module_name(fragment)?;
        insert_generated_file(
            files,
            format!("_support/{module_name}.py"),
            fragment.contents.clone(),
        )?;
        module_names.push(module_name);
    }

    if !module_names.is_empty() {
        let mut output = String::new();
        render_generated_file_header(&mut output);
        output.push('\n');
        for module_name in &module_names {
            output.push_str("from .");
            output.push_str(module_name);
            output.push_str(" import *  # noqa: F401,F403\n");
        }
        insert_generated_file(files, "_support/__init__.py", output)?;
    }

    Ok(module_names)
}

fn support_module_name(fragment: &SupportFragmentSpec) -> Result<String> {
    let path = Path::new(&fragment.path);
    let Some(file_name) = path.file_name() else {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "support path must have a file name".to_string(),
        });
    };
    let Some(file_name) = file_name.to_str() else {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "support path must be valid UTF-8".to_string(),
        });
    };
    let Some(module_name) = Path::new(file_name).file_stem() else {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "support path must have a module stem".to_string(),
        });
    };
    let Some(module_name) = module_name.to_str() else {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "support path must be valid UTF-8".to_string(),
        });
    };
    if Path::new(file_name)
        .extension()
        .and_then(|extension| extension.to_str())
        != Some("py")
    {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "Python support files must end with `.py`".to_string(),
        });
    }
    if module_name.is_empty() {
        return Err(Error::InvalidGeneratedPath {
            path: path.to_path_buf(),
            reason: "support module name cannot be empty".to_string(),
        });
    }
    Ok(module_name.to_string())
}

fn render_generated_file_header(output: &mut String) {
    output.push_str(GENERATED_HEADER);
    output.push_str("\n# pyright: reportAny=false");
    output.push_str("\n\n");
    output.push_str("from __future__ import annotations\n");
}

fn render_python_module_imports(output: &mut String, module_imports: &BTreeSet<String>) {
    render_python_module_imports_with_indent(output, module_imports, "");
}

fn render_python_module_imports_with_indent(
    output: &mut String,
    module_imports: &BTreeSet<String>,
    indent: &str,
) {
    for import in module_imports {
        output.push_str(indent);
        if let Some(relative_import) = import.strip_prefix('.') {
            output.push_str("from . import ");
            output.push_str(relative_import.trim_start_matches('.'));
            output.push('\n');
        } else {
            output.push_str("import ");
            output.push_str(import);
            output.push('\n');
        }
    }
}

fn body_uses_python_module_path(body: &str, module_path: &str) -> bool {
    let dotted_module = format!("{module_path}.");
    body.match_indices(&dotted_module).any(|(index, _)| {
        let previous = body[..index].chars().next_back();
        !previous.is_some_and(|character| is_python_identifier_char(character) || character == '.')
    })
}

fn used_python_module_imports(body: &str, module_imports: &BTreeSet<String>) -> BTreeSet<String> {
    module_imports
        .iter()
        .filter(|module_import| body_uses_python_module_path(body, module_import))
        .cloned()
        .collect()
}

fn is_python_identifier_char(character: char) -> bool {
    character.is_ascii_alphanumeric() || character == '_'
}

fn body_uses_python_symbol(body: &str, symbol: &str) -> bool {
    body.match_indices(symbol).any(|(index, _)| {
        let previous = body[..index].chars().next_back();
        let next = body[index + symbol.len()..].chars().next();
        !previous.is_some_and(|character| is_python_identifier_char(character) || character == '.')
            && !next.is_some_and(is_python_identifier_char)
    })
}

fn used_python_symbol_imports(body: &str, candidates: &[String]) -> Vec<String> {
    candidates
        .iter()
        .filter(|candidate| body_uses_python_symbol(body, candidate))
        .cloned()
        .collect()
}

fn render_named_python_import(output: &mut String, module: &str, names: &[String]) {
    if names.is_empty() {
        return;
    }

    if names.len() == 1 {
        output.push_str("from ");
        output.push_str(module);
        output.push_str(" import ");
        output.push_str(&names[0]);
        output.push('\n');
        return;
    }

    output.push_str("from ");
    output.push_str(module);
    output.push_str(" import (\n");
    for name in names {
        output.push_str("    ");
        output.push_str(name);
        output.push_str(",\n");
    }
    output.push_str(")\n");
}

fn top_level_python_symbols(source: &str) -> BTreeSet<String> {
    let mut symbols = BTreeSet::new();

    for line in source.lines() {
        let line = line.trim_end();
        if line.is_empty()
            || line.starts_with(' ')
            || line.starts_with('\t')
            || line.starts_with('#')
            || line.starts_with('@')
            || line.starts_with("import ")
            || line.starts_with("from ")
        {
            continue;
        }

        if let Some(name) = line
            .strip_prefix("def ")
            .and_then(|line| line.split('(').next())
        {
            symbols.insert(name.trim().to_string());
            continue;
        }
        if let Some(name) = line
            .strip_prefix("class ")
            .and_then(|line| line.split(['(', ':']).next())
        {
            symbols.insert(name.trim().to_string());
            continue;
        }
        if let Some((name, _)) = line.split_once('=') {
            let name = name.trim();
            if !name.is_empty() && name.chars().all(is_python_identifier_char) {
                symbols.insert(name.to_string());
            }
        }
    }

    symbols
}

fn support_export_names(support_fragments: &[SupportFragmentSpec]) -> Vec<String> {
    support_fragments
        .iter()
        .flat_map(|fragment| top_level_python_symbols(&fragment.contents))
        .filter(|name| !name.starts_with('_'))
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect()
}

fn render_optional_python_imports(
    output: &mut String,
    body: &str,
    module_imports: &BTreeSet<String>,
    language_imports: &[LanguageImportSpec],
) -> bool {
    let simple_imports = [
        ("import collections.abc\n", "collections.abc"),
        ("import dataclasses\n", "dataclasses"),
        ("import enum\n", "enum"),
        ("import typing\n", "typing"),
        ("import typing_extensions\n", "typing_extensions"),
    ];

    let mut wrote_any = false;
    for (import_line, module_path) in simple_imports {
        if body_uses_python_module_path(body, module_path) {
            output.push_str(import_line);
            wrote_any = true;
        }
    }
    let used_module_imports = used_python_module_imports(body, module_imports);
    wrote_any |=
        render_language_python_imports(output, body, language_imports, &used_module_imports);

    if !used_module_imports.is_empty() {
        render_python_module_imports(output, &used_module_imports);
        wrote_any = true;
    }

    wrote_any
}

fn render_language_python_imports(
    output: &mut String,
    body: &str,
    language_imports: &[LanguageImportSpec],
    skipped_module_imports: &BTreeSet<String>,
) -> bool {
    let mut module_imports = BTreeSet::new();
    let mut named_imports = BTreeMap::<String, BTreeSet<String>>::new();
    for import in language_imports {
        let used = match import.import_style {
            LanguageImportStyle::Module | LanguageImportStyle::Namespace => {
                body_uses_python_module_path(body, &import.reference)
            }
            LanguageImportStyle::Named => body_uses_python_module_path(body, &import.reference),
        };
        if !used {
            continue;
        }

        match import.import_style {
            LanguageImportStyle::Module | LanguageImportStyle::Namespace => {
                if !skipped_module_imports.contains(&import.module) {
                    module_imports.insert(import.module.clone());
                }
            }
            LanguageImportStyle::Named => {
                named_imports
                    .entry(import.module.clone())
                    .or_default()
                    .insert(
                        import
                            .name
                            .clone()
                            .unwrap_or_else(|| import.reference.clone()),
                    );
            }
        }
    }

    if module_imports.is_empty() && named_imports.is_empty() {
        return false;
    }

    render_python_module_imports(output, &module_imports);
    for (module, names) in named_imports {
        render_named_python_import(
            &mut *output,
            &module,
            &names.into_iter().collect::<Vec<_>>(),
        );
    }
    true
}

fn render_models_module(
    enums: &[&RenderedEnum],
    flags: &[&RenderedFlags],
    variants: &[&RenderedVariant],
    models: &[&RenderedModel],
    support_names: &[String],
    language_imports: &[LanguageImportSpec],
) -> String {
    let mut module_imports = BTreeSet::new();
    if models
        .iter()
        .any(|model| model.source == PlannedMessageSource::Wit)
    {
        module_imports.insert("nex_gen_runtime".to_string());
    }
    for model in models {
        if let Some(proto_module_path) = &model.proto_module_path {
            module_imports.insert(proto_module_path.clone());
        }
        for field in &model.fields {
            module_imports.extend(field.imports.module_imports.iter().cloned());
        }
    }
    let mut body = String::new();
    if !enums.is_empty() {
        if !body.is_empty() {
            body.push_str("\n\n");
        }
        for (index, enumeration) in enums.iter().enumerate() {
            render_enum(&mut body, enumeration);
            if index + 1 != enums.len() {
                body.push_str("\n\n");
            }
        }
    }

    if !flags.is_empty() {
        if !body.is_empty() {
            body.push_str("\n\n");
        }
        for (index, flag_set) in flags.iter().enumerate() {
            render_flags(&mut body, flag_set);
            if index + 1 != flags.len() {
                body.push_str("\n\n");
            }
        }
    }

    if !models.is_empty() {
        if !body.is_empty() {
            body.push_str("\n\n");
        }
        for (index, model) in models.iter().enumerate() {
            render_model(&mut body, model);
            if index + 1 != models.len() {
                body.push_str("\n\n");
            }
        }
    }

    if !variants.is_empty() {
        if !body.is_empty() {
            body.push_str("\n\n");
        }
        for (index, variant) in variants.iter().enumerate() {
            render_variant(&mut body, variant);
            if index + 1 != variants.len() {
                body.push_str("\n\n");
            }
        }
    }

    render_nexus_type_registrations(&mut body, models);

    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    let wrote_imports =
        render_optional_python_imports(&mut output, &body, &module_imports, language_imports);
    let used_support_names = used_python_symbol_imports(&body, support_names);
    if !used_support_names.is_empty() {
        if wrote_imports {
            output.push('\n');
        }
        render_named_python_import(&mut output, "._support", &used_support_names);
    }
    if !body.is_empty() {
        output.push('\n');
        output.push('\n');
        output.push_str(&body);
    }

    output
}

fn render_nexus_type_registrations(output: &mut String, models: &[&RenderedModel]) {
    if !models
        .iter()
        .any(|model| model.source == PlannedMessageSource::Wit)
    {
        return;
    }
    if !output.is_empty() {
        output.push_str("\n\n");
    }
    for model in models
        .iter()
        .filter(|model| model.source == PlannedMessageSource::Wit)
    {
        output.push_str("nex_gen_runtime.register_nexus_type(");
        output.push_str(&model.name);
        output.push_str(", ");
        output.push_str(&python_string_literal(&model.full_name));
        output.push_str(")\n");
    }
}

fn render_resources_package_init(services: &[RenderedService<'_>]) -> String {
    let resources = services
        .iter()
        .flat_map(|service| service.resources.iter())
        .collect::<Vec<_>>();
    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    for resource in &resources {
        render_named_python_import(
            &mut output,
            &format!(".{}", resource_module_name(resource)),
            std::slice::from_ref(&resource.type_name),
        );
    }
    output.push_str("\n__all__ = [\n");
    for resource in &resources {
        output.push_str("    ");
        output.push_str(&python_string_literal(&resource.type_name));
        output.push_str(",\n");
    }
    output.push_str("]\n");
    output
}

fn resource_bound_operations<'a>(
    service: &'a RenderedService<'a>,
    resource: &'a PlannedResource,
) -> Vec<ResourceBoundOperation<'a>> {
    resource
        .methods
        .iter()
        .filter_map(|method| match &method.binding {
            PlannedResourceMethodBindingSpec::Operation { operation_name, .. } => {
                Some(ResourceBoundOperation {
                    operation: service
                        .operations
                        .iter()
                        .find(|operation| operation.name == operation_name)
                        .expect("bound resource operation should exist on the service"),
                })
            }
            PlannedResourceMethodBindingSpec::Stub => None,
        })
        .collect()
}

fn render_resource_module_file(
    service: &RenderedService<'_>,
    resource: &PlannedResource,
    bound_operations: &[ResourceBoundOperation<'_>],
    model_names: &[String],
    resource_names: &[String],
    support_names: &[String],
    language_imports: &[LanguageImportSpec],
) -> String {
    let mut module_imports = bound_operations
        .iter()
        .flat_map(|bound_operation| {
            [
                bound_operation.operation.input_module_path.clone(),
                bound_operation.operation.output_module_path.clone(),
            ]
        })
        .flatten()
        .collect::<BTreeSet<_>>();
    module_imports.insert("nex_gen_runtime".to_string());

    let mut body = String::new();
    let function_fields = bound_operations
        .iter()
        .filter_map(|bound_operation| bound_operation.operation.unpacked_input.as_ref())
        .flat_map(|unpacked_input| unpacked_input.functions.iter().cloned())
        .collect::<Vec<_>>();
    let output_type_parameters = bound_operations
        .iter()
        .flat_map(|bound_operation| bound_operation.operation.output_type_parameters.iter())
        .cloned()
        .collect::<BTreeSet<_>>();
    if !function_fields.is_empty() {
        render_function_type_parameter_definitions(
            &mut body,
            &function_fields,
            &output_type_parameters,
        );
        body.push_str("\n\n");
    }
    render_resource(&mut body, resource, bound_operations);
    if !bound_operations.is_empty() {
        body.push_str("\n\n");
        for (index, bound_operation) in bound_operations.iter().enumerate() {
            render_operation_functions(&mut body, service, bound_operation.operation);
            if index + 1 != bound_operations.len() {
                body.push_str("\n\n");
            }
        }
    }
    module_imports.insert("temporalio.workflow".to_string());
    body.push_str("\n\n");
    body.push_str("nex_gen_runtime.register_nexus_type(");
    body.push_str(&resource.type_name);
    body.push_str(", ");
    body.push_str(&python_string_literal(&python_resource_type_id(
        service, resource,
    )));
    body.push_str(")\n");

    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    let wrote_imports =
        render_optional_python_imports(&mut output, &body, &module_imports, language_imports);
    if wrote_imports {
        output.push('\n');
    }
    let used_model_names = used_python_symbol_imports(&body, model_names);
    if !used_model_names.is_empty() {
        render_named_python_import(&mut output, "..models", &used_model_names);
    }
    let used_resource_names = used_python_symbol_imports(&body, resource_names)
        .into_iter()
        .filter(|name| name != &resource.type_name)
        .collect::<Vec<_>>();
    if !used_resource_names.is_empty() {
        render_named_python_import(&mut output, ".", &used_resource_names);
    }
    let used_support_names = used_python_symbol_imports(&body, support_names);
    if !used_support_names.is_empty() {
        render_named_python_import(&mut output, ".._support", &used_support_names);
    }
    if !body.is_empty() {
        output.push('\n');
        output.push('\n');
        output.push_str(&body);
    }
    output
}

fn render_service_module(services: &[RenderedService<'_>]) -> String {
    let mut module_imports = BTreeSet::new();
    let mut model_type_imports = BTreeSet::new();
    let resource_modules = services
        .iter()
        .flat_map(|service| service.resources.iter())
        .map(|resource| (resource.type_name.clone(), resource_module_name(resource)))
        .collect::<BTreeMap<_, _>>();
    let mut resource_type_imports = BTreeMap::new();
    for service in services {
        for operation in &service.operations {
            if let Some(input_module_path) = &operation.input_module_path {
                if input_module_path == ".models" {
                    if let Some(type_name) = operation.input_ref.strip_prefix("models.") {
                        model_type_imports.insert(type_name.to_string());
                    }
                } else {
                    module_imports.insert(input_module_path.clone());
                }
            }
            if let Some(output_module_path) = &operation.output_module_path {
                if output_module_path == "._resources" {
                    if let Some(module_name) = resource_modules.get(&operation.output_ref) {
                        resource_type_imports
                            .insert(operation.output_ref.clone(), module_name.clone());
                    }
                } else if output_module_path == ".models" {
                    if let Some(type_name) = operation.output_ref.strip_prefix("models.") {
                        model_type_imports.insert(type_name.to_string());
                    }
                } else {
                    module_imports.insert(output_module_path.clone());
                }
            }
        }
    }

    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    output.push_str("from nexusrpc import Operation, service\n");
    let uses_typing = services
        .iter()
        .flat_map(|service| service.operations.iter())
        .any(|operation| {
            operation.input_ref.contains("typing.") || operation.output_ref.contains("typing.")
        });
    if uses_typing {
        output.push_str("import typing\n");
    }
    if !module_imports.is_empty()
        || !model_type_imports.is_empty()
        || !resource_type_imports.is_empty()
    {
        if uses_typing {
            output.push('\n');
        }
        render_python_module_imports(&mut output, &module_imports);
        render_named_python_import(
            &mut output,
            ".models",
            &model_type_imports.into_iter().collect::<Vec<_>>(),
        );
        for (resource_type_name, module_name) in &resource_type_imports {
            render_named_python_import(
                &mut output,
                &format!("._resources.{module_name}"),
                std::slice::from_ref(resource_type_name),
            );
        }
    }
    if !services.is_empty() {
        output.push_str("\n\n");
        for (index, service) in services.iter().enumerate() {
            render_service_definition(&mut output, service);
            if index + 1 != services.len() {
                output.push_str("\n\n");
            }
        }
    }

    output
}

fn render_operation_module(
    service: &RenderedService<'_>,
    operation: &RenderedOperation<'_>,
    model_names: &[String],
    resource_names: &[String],
    support_names: &[String],
    language_imports: &[LanguageImportSpec],
) -> String {
    let mut module_imports = BTreeSet::new();
    if let Some(input_module_path) = &operation.input_module_path {
        module_imports.insert(input_module_path.clone());
    }
    if let Some(output_module_path) = &operation.output_module_path {
        module_imports.insert(output_module_path.clone());
    }

    let mut body = String::new();
    if let Some(unpacked_input) = &operation.unpacked_input
        && !unpacked_input.functions.is_empty()
    {
        render_function_type_parameter_definitions(
            &mut body,
            &unpacked_input.functions,
            &operation.output_type_parameters,
        );
        body.push_str("\n\n");
    }
    render_operation_functions(&mut body, service, operation);
    module_imports.insert("temporalio.workflow".to_string());

    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    let wrote_imports =
        render_optional_python_imports(&mut output, &body, &module_imports, language_imports);
    if wrote_imports {
        output.push('\n');
    }
    let used_model_names = used_python_symbol_imports(&body, model_names);
    render_named_python_import(&mut output, "..models", &used_model_names);
    let used_resource_names = used_python_symbol_imports(&body, resource_names);
    if !used_resource_names.is_empty() {
        render_named_python_import(&mut output, ".._resources", &used_resource_names);
    }
    let used_support_names = used_python_symbol_imports(&body, support_names);
    if !used_support_names.is_empty() {
        render_named_python_import(&mut output, ".._support", &used_support_names);
    }
    output.push('\n');
    output.push('\n');
    output.push_str(&body);
    output
}

fn render_operations_package_init() -> String {
    let mut output = String::new();
    render_generated_file_header(&mut output);
    output
}

fn function_type_parameters(
    functions: &[RenderedFunctionField],
    output_type_parameters: &BTreeSet<String>,
) -> Vec<PythonTypeParameter> {
    let mut parameters = BTreeSet::new();
    for function in functions {
        for case in function_overload_cases(function, output_type_parameters) {
            parameters.extend(case.type_parameters);
        }
    }
    parameters.into_iter().collect()
}

fn render_function_type_parameter_definitions(
    output: &mut String,
    functions: &[RenderedFunctionField],
    output_type_parameters: &BTreeSet<String>,
) {
    let parameters = function_type_parameters(functions, output_type_parameters);
    if parameters.is_empty() {
        return;
    }

    for parameter in parameters {
        match parameter {
            PythonTypeParameter::TypeVar(name) => {
                output.push_str(&name);
                output.push_str(" = typing.TypeVar(");
                output.push_str(&python_string_literal(&name));
                output.push_str(")\n");
            }
            PythonTypeParameter::TypeVarTuple(name) => {
                output.push_str(&name);
                output.push_str(" = typing_extensions.TypeVarTuple(");
                output.push_str(&python_string_literal(&name));
                output.push_str(")\n");
            }
        }
    }
}

fn render_package_init(
    services: &[RenderedService<'_>],
    _resource_names: &[String],
    resource_operation_owners: &BTreeMap<String, String>,
) -> String {
    let operation_function_names = services
        .iter()
        .flat_map(|service| {
            service
                .operations
                .iter()
                .map(|operation| operation.attr_name.clone())
        })
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();
    let mut output = String::new();
    render_generated_file_header(&mut output);
    output.push('\n');
    if !services.is_empty() {
        output.push_str("from . import service as _service\n");
    }
    if !services.is_empty() {
        for service in services {
            for operation in &service.operations {
                let imports = vec![operation.attr_name.clone()];
                let module = if let Some(resource_owner) =
                    resource_operation_owners.get(&operation_key(service.name, operation.name))
                {
                    format!("._resources.{resource_owner}")
                } else {
                    format!(".operations.{}", operation.attr_name)
                };
                render_named_python_import(&mut output, &module, &imports);
            }
        }
    }
    output.push_str("\n__all__ = [\n");
    for name in &operation_function_names {
        output.push_str("    ");
        output.push_str(&python_string_literal(name));
        output.push_str(",\n");
    }
    output.push_str("]\n");
    output.push_str("\n\n__nexus_operation_registry__ = {\n");
    for service in services {
        for operation in &service.operations {
            output.push_str("    (\n");
            output.push_str("        \"");
            output.push_str(service.name);
            output.push_str("\",\n");
            output.push_str("        \"");
            output.push_str(operation.wire_name);
            output.push_str("\",\n");
            output.push_str("    ): ");
            output.push_str("_service.");
            output.push_str(service.name);
            output.push('.');
            output.push_str(&operation.attr_name);
            output.push_str(",\n");
        }
    }
    output.push_str("}\n");
    output
}

fn render_service_definition(output: &mut String, service: &RenderedService<'_>) {
    if service.wire_name == service.name {
        output.push_str("@service\n");
    } else {
        output.push_str("@service(name=");
        output.push_str(&python_string_literal(service.wire_name));
        output.push_str(")\n");
    }
    output.push_str("class ");
    output.push_str(service.name);
    output.push_str(":\n");
    render_python_docstring(output, "    ", None, &[], None, service.experimental);

    if service.operations.is_empty() {
        if !service.experimental {
            output.push_str("    pass\n");
        }
        return;
    }

    for (operation_index, operation) in service.operations.iter().enumerate() {
        if operation.experimental {
            output.push_str("    # ");
            output.push_str(".. warning:: ");
            output.push_str(EXPERIMENTAL_WARNING);
            output.push('\n');
        }
        output.push_str("    ");
        output.push_str(&operation.attr_name);
        output.push_str(": Operation[\n");
        output.push_str("        ");
        output.push_str(&service_type_ref(&operation.input_ref));
        output.push_str(",\n");
        output.push_str("        ");
        output.push_str(&service_type_ref(&operation.output_ref));
        output.push_str(",\n");
        output.push_str("    ] = Operation(name=");
        output.push_str(&python_string_literal(operation.wire_name));
        output.push_str(")\n");

        if operation_index + 1 != service.operations.len() {
            output.push('\n');
        }
    }
}

fn service_type_ref(type_ref: &str) -> String {
    type_ref
        .strip_prefix("models.")
        .unwrap_or(type_ref)
        .to_string()
}

fn render_enum(output: &mut String, enumeration: &RenderedEnum) {
    output.push_str("class ");
    output.push_str(&enumeration.name);
    output.push_str("(enum.IntEnum):\n");

    if enumeration.values.is_empty() {
        output.push_str("    pass\n");
        return;
    }

    for value in &enumeration.values {
        output.push_str("    ");
        output.push_str(&value.name);
        output.push_str(" = ");
        output.push_str(&value.number.to_string());
        output.push('\n');
    }
}

fn render_flags(output: &mut String, flags: &RenderedFlags) {
    output.push_str("class ");
    output.push_str(&flags.name);
    output.push_str("(enum.IntFlag):\n");

    if flags.flags.is_empty() {
        output.push_str("    pass\n");
        return;
    }

    for flag in &flags.flags {
        output.push_str("    ");
        output.push_str(&flag.name);
        output.push_str(" = 1 << ");
        output.push_str(&flag.bit.to_string());
        output.push('\n');
    }
}

fn render_variant(output: &mut String, variant: &RenderedVariant) {
    output.push_str(variant.name.as_str());
    output.push_str(" = ");
    if variant.cases.is_empty() {
        output.push_str("typing.Never\n");
        return;
    }
    if variant.cases.len() > 1 {
        output.push_str("(\n    ");
    }
    for (index, case) in variant.cases.iter().enumerate() {
        if index > 0 {
            output.push_str("\n    | ");
        }
        output.push_str("tuple[typing.Literal[");
        output.push_str(&python_string_literal(&case.name));
        output.push(']');
        if let Some(payload_annotation) = &case.payload_annotation {
            output.push_str(", ");
            output.push_str(payload_annotation);
        }
        output.push(']');
    }
    if variant.cases.len() > 1 {
        output.push_str("\n)");
    }
    output.push('\n');
}

fn render_model(output: &mut String, model: &RenderedModel) {
    if model_needs_keyword_only_dataclass(model) {
        output.push_str("@dataclasses.dataclass(slots=True, kw_only=True)\n");
    } else {
        output.push_str("@dataclasses.dataclass(slots=True)\n");
    }
    output.push_str("class ");
    output.push_str(&model.name);
    output.push_str(":\n");
    render_python_docstring(output, "    ", None, &[], None, model.experimental);

    if model.fields.is_empty() {
        let mut wrote_method = false;
        if model.capabilities.from_proto {
            let proto_ref = model
                .proto_ref
                .as_deref()
                .expect("from_proto models should have a proto reference");
            output.push_str("    @classmethod\n");
            output.push_str("    def from_proto(\n");
            output.push_str("        cls,\n");
            output.push_str("        _proto: ");
            output.push_str(proto_ref);
            output.push_str(",\n");
            output.push_str("    ) -> ");
            output.push_str(&model.name);
            output.push_str(":\n");
            output.push_str("        return cls()\n");
            wrote_method = true;
        }
        if model.capabilities.to_proto {
            let proto_ref = model
                .proto_ref
                .as_deref()
                .expect("to_proto models should have a proto reference");
            if wrote_method {
                output.push('\n');
            }
            output.push_str("    def to_proto(self) -> ");
            output.push_str(proto_ref);
            output.push_str(":\n");
            output.push_str("        message = ");
            output.push_str(proto_ref);
            output.push_str("()\n");
            for field in &model.sourced_fields {
                for line in &field.to_proto_lines {
                    output.push_str("        ");
                    output.push_str(line);
                    output.push('\n');
                }
            }
            output.push_str("        return message\n");
        }
        if !wrote_method && !model.capabilities.to_proto {
            output.push_str("    pass\n");
        }
        return;
    }

    for field in &model.fields {
        output.push_str("    ");
        output.push_str(&field.attr_name);
        output.push_str(": ");
        output.push_str(&python_parameter_annotation(
            &field.annotation,
            &field.default_kind,
        ));
        if let Some(default_expr) = &field.default_expr {
            render_python_default_expr(output, default_expr, "    ");
        }
        output.push('\n');
    }

    let mut wrote_method = false;
    if model.capabilities.from_proto {
        let proto_ref = model
            .proto_ref
            .as_deref()
            .expect("from_proto models should have a proto reference");
        output.push_str("\n");
        output.push_str("    @classmethod\n");
        output.push_str("    def from_proto(\n");
        output.push_str("        cls,\n");
        output.push_str("        proto: ");
        output.push_str(proto_ref);
        output.push_str(",\n");
        output.push_str("    ) -> ");
        output.push_str(&model.name);
        output.push_str(":\n");

        for field in &model.fields {
            for line in &field.from_proto_setup_lines {
                output.push_str("        ");
                output.push_str(line);
                output.push('\n');
            }
        }

        output.push_str("        return cls(\n");
        for field in &model.fields {
            output.push_str("            ");
            output.push_str(&field.attr_name);
            output.push_str("=");
            output.push_str(&field.from_proto_expr);
            output.push_str(",\n");
        }
        output.push_str("        )\n");
        wrote_method = true;
    }
    if model.capabilities.to_proto {
        let proto_ref = model
            .proto_ref
            .as_deref()
            .expect("to_proto models should have a proto reference");
        output.push_str("\n");
        if wrote_method {
            output.push('\n');
        }
        output.push_str("    def to_proto(self) -> ");
        output.push_str(proto_ref);
        output.push_str(":\n");
        output.push_str("        message = ");
        output.push_str(proto_ref);
        output.push_str("()\n");
        for field in &model.fields {
            for line in &field.to_proto_lines {
                output.push_str("        ");
                output.push_str(line);
                output.push('\n');
            }
        }
        for field in &model.sourced_fields {
            for line in &field.to_proto_lines {
                output.push_str("        ");
                output.push_str(line);
                output.push('\n');
            }
        }
        output.push_str("        return message\n");
    }
}

fn model_needs_keyword_only_dataclass(model: &RenderedModel) -> bool {
    let mut saw_defaulted_field = false;
    for field in &model.fields {
        if field.default_expr.is_some() {
            saw_defaulted_field = true;
        } else if saw_defaulted_field {
            return true;
        }
    }
    false
}

fn render_resource(
    output: &mut String,
    resource: &PlannedResource,
    bound_operations: &[ResourceBoundOperation<'_>],
) {
    output.push_str("@dataclasses.dataclass\n");
    output.push_str("class ");
    output.push_str(&resource.type_name);
    output.push_str(":\n");
    for field in &resource.fields {
        output.push_str("    ");
        output.push_str(&python_field_name(&field.name));
        output.push_str(": ");
        output.push_str(&python_resource_field_annotation(
            &field.kind,
            field.optional,
            field.function.as_ref(),
        ));
        output.push('\n');
    }

    if resource.methods.is_empty() {
        output.push_str("\n    pass\n");
        return;
    }

    for method in &resource.methods {
        output.push_str("\n");
        render_resource_class_method(output, resource, method, bound_operations);
    }
}

fn render_resource_class_method(
    output: &mut String,
    resource: &PlannedResource,
    method: &PlannedResourceMethod,
    bound_operations: &[ResourceBoundOperation<'_>],
) {
    let result_annotation = python_resource_method_result_annotation(method);
    output.push_str("    async def ");
    output.push_str(&python_field_name(&method.name));
    output.push_str("(\n");
    output.push_str("        self,\n");
    for param in &method.params {
        output.push_str("        ");
        output.push_str(&python_field_name(&param.name));
        output.push_str(": ");
        output.push_str(&python_resource_field_annotation(
            &param.kind,
            param.optional,
            param.function.as_ref(),
        ));
        if param.optional {
            output.push_str(" = None");
        }
        output.push_str(",\n");
    }
    output.push_str("    ) -> ");
    output.push_str(&result_annotation);
    output.push_str(":\n");

    match &method.binding {
        PlannedResourceMethodBindingSpec::Operation { operation_name, .. } => {
            let operation = bound_operations
                .iter()
                .find(|bound_operation| bound_operation.operation.name == operation_name)
                .map(|bound_operation| bound_operation.operation)
                .expect("bound resource operation should be rendered in the same module");
            render_resource_method_operation_body(output, method, operation);
        }
        PlannedResourceMethodBindingSpec::Stub => {
            output.push_str("        raise NotImplementedError(");
            output.push_str(&python_string_literal(&format!(
                "{}.{} is not yet implemented",
                resource.name,
                python_field_name(&method.name)
            )));
            output.push_str(")\n");
        }
    }
}

fn render_resource_method_operation_body(
    output: &mut String,
    method: &PlannedResourceMethod,
    operation: &RenderedOperation<'_>,
) {
    let result_annotation = python_resource_method_result_annotation(method);
    let PlannedResourceMethodBindingSpec::Operation {
        request_plan,
        direct_return,
        ..
    } = &method.binding
    else {
        return;
    };

    let operation_attr_name = request_operation_function_name(operation);
    let request_expr = render_request_plan_python(request_plan);
    output.push_str("        request = ");
    if let Some(unpacked_input) = &operation.unpacked_input {
        output.push_str(&unpacked_input.model_name);
    } else {
        output.push_str(&operation.input_annotation);
    }
    output.push_str("(\n");
    match request_plan {
        RequestPlan::Construct { fields, .. } => {
            for field in fields {
                output.push_str("            ");
                output.push_str(&python_field_name(&field.field_name));
                output.push('=');
                output.push_str(&render_request_plan_python(&field.value));
                output.push_str(",\n");
            }
        }
        RequestPlan::Source(_) => {
            output.push_str("            value=");
            output.push_str(&request_expr);
            output.push_str(",\n");
        }
    }
    output.push_str("        )\n");
    if *direct_return {
        if result_annotation == "None" {
            output.push_str("        await ");
            output.push_str(&operation_attr_name);
            output.push_str("(request)\n");
        } else {
            output.push_str("        return await ");
            output.push_str(&operation_attr_name);
            output.push_str("(request)\n");
        }
    } else {
        output.push_str("        handle = await ");
        output.push_str(&operation_attr_name);
        output.push_str("(request)\n");
        if result_annotation == "None" {
            output.push_str("        await handle\n");
        } else {
            output.push_str("        return await handle\n");
        }
    }
}

fn python_resource_method_result_annotation(method: &PlannedResourceMethod) -> String {
    let Some(result) = &method.result else {
        return "None".to_string();
    };
    let annotation = match &result.kind {
        PlannedResourceMethodResultKind::Resource { type_name } => type_name.clone(),
        PlannedResourceMethodResultKind::Value(kind) => {
            python_resource_field_annotation(kind, result.optional, None)
        }
    };
    if result.optional && !annotation.contains("| None") {
        format!("{annotation} | None")
    } else {
        annotation
    }
}

fn python_resource_field_annotation(
    kind: &PlannedFieldKind,
    optional: bool,
    function: Option<&FunctionFieldSpec>,
) -> String {
    let base = if let Some(function) = function {
        python_function_field_annotation(
            function,
            function
                .alternate_type
                .as_ref()
                .map(python_authored_type_annotation),
        )
    } else {
        match kind {
            PlannedFieldKind::Singular(value) => python_resource_value_annotation(value),
            PlannedFieldKind::Repeated(value) => {
                format!(
                    "collections.abc.Sequence[{}]",
                    python_resource_value_annotation(value)
                )
            }
            PlannedFieldKind::Map { key, value } => format!(
                "collections.abc.Mapping[{}, {}]",
                python_resource_value_annotation(key),
                python_resource_value_annotation(value)
            ),
        }
    };
    if optional {
        format!("{base} | None")
    } else {
        base
    }
}

fn python_resource_value_annotation(value: &PlannedValueType) -> String {
    match value {
        PlannedValueType::Scalar(PlannedScalarType::Float) => "float".to_string(),
        PlannedValueType::Scalar(PlannedScalarType::Int32 | PlannedScalarType::Int64) => {
            "int".to_string()
        }
        PlannedValueType::Scalar(PlannedScalarType::Bool) => "bool".to_string(),
        PlannedValueType::Scalar(PlannedScalarType::String) => "str".to_string(),
        PlannedValueType::Scalar(PlannedScalarType::Bytes) => "bytes".to_string(),
        PlannedValueType::Enum(enum_type) => enum_type
            .replacement
            .as_ref()
            .and_then(python_replacement_type_name)
            .or_else(|| enum_type.name.clone())
            .unwrap_or_else(|| "int".to_string()),
        PlannedValueType::Flags(flags_type) => flags_type.name.clone(),
        PlannedValueType::Variant(variant_type) => variant_type.name.clone(),
        PlannedValueType::Message(message_type) => message_type
            .replacement
            .as_ref()
            .and_then(python_replacement_type_name)
            .unwrap_or_else(|| message_type.model_name.clone()),
        PlannedValueType::Tuple(items) => format!(
            "tuple[{}]",
            items
                .iter()
                .map(python_resource_value_annotation)
                .collect::<Vec<_>>()
                .join(", ")
        ),
        PlannedValueType::Result { ok, err } => python_result_annotation(
            ok.as_ref()
                .map(|ok| python_resource_value_annotation(ok).to_string()),
            err.as_ref()
                .map(|err| python_resource_value_annotation(err).to_string()),
        ),
        PlannedValueType::External {
            type_name,
            fallback,
        } => type_name
            .for_language(Language::Python)
            .map(str::to_string)
            .unwrap_or_else(|| python_resource_value_annotation(fallback)),
        PlannedValueType::Unknown => "object".to_string(),
    }
}

fn render_request_plan_python(plan: &RequestPlan) -> String {
    render_request_plan(
        plan,
        python_field_name,
        |name, value| format!("{name}={value}"),
        |message_name, fields| {
            let model_name = message_model_name(message_name);
            if fields.is_empty() {
                format!("{model_name}()")
            } else {
                format!("{model_name}({})", fields.join(", "))
            }
        },
        |name| format!("self.{}", python_field_name(name)),
        python_field_name,
    )
}

fn resource_return_binding_expr_python(binding: &PlannedOperationResourceFieldBinding) -> String {
    match &binding.source {
        ResolvedResourceBindingSource::RequestField {
            field_name,
            proto_field_name,
            hidden,
        } => {
            if *hidden {
                let expr = format!("request_proto.{proto_field_name}");
                if binding.optional {
                    format!("{expr} or None")
                } else {
                    expr
                }
            } else {
                format!("request.{}", python_field_name(field_name))
            }
        }
        ResolvedResourceBindingSource::ResultField {
            proto_field_name, ..
        } => {
            let expr = format!("result.{proto_field_name}");
            if binding.optional {
                format!("{expr} or None")
            } else {
                expr
            }
        }
    }
}

#[derive(Debug, Clone)]
struct RenderedFunctionOverloadCase {
    callable_field_name: String,
    args_field_name: String,
    callable_annotation: String,
    positional_args: Vec<RenderedFunctionOverloadPositionalArg>,
    args_annotation: Option<String>,
    args_optional: bool,
    result_type_parameter: Option<String>,
    type_parameters: Vec<PythonTypeParameter>,
}

#[derive(Debug, Clone)]
struct RenderedFunctionOverloadPositionalArg {
    name: String,
    annotation: String,
    variadic: bool,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
enum PythonTypeParameter {
    TypeVar(String),
    TypeVarTuple(String),
}

fn render_function_unpacked_overloads(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    unpacked_input: &RenderedUnpackedInput,
) {
    let overload_combinations = collect_function_overload_combinations(
        &unpacked_input.functions,
        &operation.output_type_parameters,
    );

    for (index, overload_cases) in overload_combinations.iter().enumerate() {
        render_function_unpacked_overload(output, operation, unpacked_input, overload_cases);
        if index + 1 != overload_combinations.len() {
            output.push('\n');
        }
    }
}

fn collect_function_overload_combinations(
    functions: &[RenderedFunctionField],
    output_type_parameters: &BTreeSet<String>,
) -> Vec<Vec<RenderedFunctionOverloadCase>> {
    let per_function_cases = functions
        .iter()
        .map(|function| function_overload_cases(function, output_type_parameters))
        .collect::<Vec<_>>();
    let mut combinations = Vec::new();
    let mut current = Vec::new();
    collect_function_overload_combinations_recursive(
        &per_function_cases,
        0,
        &mut current,
        &mut combinations,
    );
    combinations
}

fn collect_function_overload_combinations_recursive(
    per_function_cases: &[Vec<RenderedFunctionOverloadCase>],
    index: usize,
    current: &mut Vec<RenderedFunctionOverloadCase>,
    combinations: &mut Vec<Vec<RenderedFunctionOverloadCase>>,
) {
    if index == per_function_cases.len() {
        combinations.push(current.clone());
        return;
    }

    for case in &per_function_cases[index] {
        current.push(case.clone());
        collect_function_overload_combinations_recursive(
            per_function_cases,
            index + 1,
            current,
            combinations,
        );
        current.pop();
    }
}

fn function_overload_cases(
    function: &RenderedFunctionField,
    output_type_parameters: &BTreeSet<String>,
) -> Vec<RenderedFunctionOverloadCase> {
    let result_type_parameter = function
        .result_type_parameter
        .clone()
        .filter(|name| output_type_parameters.contains(name));
    let result_annotation = if result_type_parameter.is_some() {
        &function.result_annotation
    } else {
        &function.erased_result_annotation
    };
    let mut cases = Vec::new();
    if let Some(alternate_annotation) = &function.alternate_annotation {
        cases.push(RenderedFunctionOverloadCase {
            callable_field_name: function.callable_field_name.clone(),
            args_field_name: function.args_field_name.clone(),
            callable_annotation: alternate_annotation.clone(),
            positional_args: python_alternate_function_positional_args(function),
            args_annotation: python_alternate_function_args_annotation(function),
            args_optional: true,
            result_type_parameter: None,
            type_parameters: Vec::new(),
        });
    }

    match &function.args {
        RenderedFunctionArgs::Varargs { prefix } if function.primary => {
            let type_parameter_prefix = function_overload_parameter_prefix(function);
            let function_args = format!("{type_parameter_prefix}Args");
            let callable_args = python_varargs_callable_arg_list(prefix, &function_args);
            let mut type_parameters = Vec::new();
            if let Some(result_type_parameter) = &result_type_parameter {
                type_parameters.push(PythonTypeParameter::TypeVar(result_type_parameter.clone()));
            }
            type_parameters.push(PythonTypeParameter::TypeVarTuple(function_args.clone()));
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: format!(
                    "collections.abc.Callable[[{}], {}]",
                    callable_args, result_annotation
                ),
                positional_args: vec![RenderedFunctionOverloadPositionalArg {
                    name: "positional_args".to_string(),
                    annotation: format!("typing_extensions.Unpack[{function_args}]"),
                    variadic: true,
                }],
                args_annotation: None,
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
            let type_parameters = result_type_parameter
                .iter()
                .map(|name| PythonTypeParameter::TypeVar(name.clone()))
                .collect();
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: format!(
                    "collections.abc.Callable[..., {}]",
                    result_annotation
                ),
                positional_args: Vec::new(),
                args_annotation: Some("list[typing.Any]".to_string()),
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
        }
        RenderedFunctionArgs::Varargs { prefix } => {
            let type_parameter_prefix = function_overload_parameter_prefix(function);
            let first_arg = format!("{type_parameter_prefix}Arg");
            let callable_prefix = prefix
                .iter()
                .map(|parameter| parameter.annotation.clone())
                .collect::<Vec<_>>();
            let no_arg_callable_args = python_callable_arg_list(&callable_prefix);
            let mut one_arg_callable_prefix = callable_prefix;
            one_arg_callable_prefix.push(first_arg.clone());
            let one_arg_callable_args = python_callable_arg_list(&one_arg_callable_prefix);
            let type_parameters = result_type_parameter
                .iter()
                .map(|name| PythonTypeParameter::TypeVar(name.clone()))
                .collect();
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: format!(
                    "collections.abc.Callable[[{}], {}]",
                    no_arg_callable_args, result_annotation
                ),
                positional_args: Vec::new(),
                args_annotation: None,
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
            let mut type_parameters = Vec::new();
            if let Some(result_type_parameter) = &result_type_parameter {
                type_parameters.push(PythonTypeParameter::TypeVar(result_type_parameter.clone()));
            }
            type_parameters.push(PythonTypeParameter::TypeVar(first_arg.clone()));
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: format!(
                    "collections.abc.Callable[[{}], {}]",
                    one_arg_callable_args, result_annotation
                ),
                positional_args: Vec::new(),
                args_annotation: Some(first_arg.clone()),
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
            let type_parameters = result_type_parameter
                .iter()
                .map(|name| PythonTypeParameter::TypeVar(name.clone()))
                .collect();
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: format!(
                    "collections.abc.Callable[..., {}]",
                    result_annotation
                ),
                positional_args: Vec::new(),
                args_annotation: Some("list[typing.Any]".to_string()),
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
        }
        RenderedFunctionArgs::Typed { parameters } => {
            let args_annotation = python_function_args_list_annotation(
                parameters
                    .iter()
                    .map(|parameter| parameter.annotation.as_str())
                    .collect::<Vec<_>>()
                    .as_slice(),
            );
            let type_parameters = result_type_parameter
                .iter()
                .map(|name| PythonTypeParameter::TypeVar(name.clone()))
                .collect();
            cases.push(RenderedFunctionOverloadCase {
                callable_field_name: function.callable_field_name.clone(),
                args_field_name: function.args_field_name.clone(),
                callable_annotation: python_rendered_function_callable_annotation_with_result(
                    function,
                    result_annotation,
                ),
                positional_args: if function.primary {
                    parameters
                        .iter()
                        .map(|parameter| RenderedFunctionOverloadPositionalArg {
                            name: parameter.name.clone(),
                            annotation: parameter.annotation.clone(),
                            variadic: false,
                        })
                        .collect()
                } else {
                    Vec::new()
                },
                args_annotation: if function.primary {
                    None
                } else if parameters.len() == 1 {
                    Some(parameters[0].annotation.clone())
                } else {
                    Some(args_annotation.clone())
                },
                args_optional: false,
                result_type_parameter: result_type_parameter.clone(),
                type_parameters,
            });
        }
    }

    cases
}

fn python_alternate_function_positional_args(
    function: &RenderedFunctionField,
) -> Vec<RenderedFunctionOverloadPositionalArg> {
    if !function.primary {
        return Vec::new();
    }
    match &function.args {
        RenderedFunctionArgs::Typed { parameters } => parameters
            .iter()
            .map(|parameter| RenderedFunctionOverloadPositionalArg {
                name: parameter.name.clone(),
                annotation: parameter.annotation.clone(),
                variadic: false,
            })
            .collect(),
        RenderedFunctionArgs::Varargs { .. } => {
            vec![RenderedFunctionOverloadPositionalArg {
                name: "positional_args".to_string(),
                annotation: "object".to_string(),
                variadic: true,
            }]
        }
    }
}

fn python_alternate_function_args_annotation(function: &RenderedFunctionField) -> Option<String> {
    match (&function.args, function.primary) {
        (RenderedFunctionArgs::Typed { .. }, true) => None,
        _ => Some("list[typing.Any] | None".to_string()),
    }
}

fn python_rendered_function_callable_annotation_with_result(
    function: &RenderedFunctionField,
    result_annotation: &str,
) -> String {
    match &function.args {
        RenderedFunctionArgs::Varargs { .. } => {
            format!("collections.abc.Callable[..., {}]", result_annotation)
        }
        RenderedFunctionArgs::Typed { parameters } => format!(
            "collections.abc.Callable[[{}], {}]",
            parameters
                .iter()
                .map(|parameter| parameter.annotation.clone())
                .collect::<Vec<_>>()
                .join(", "),
            result_annotation
        ),
    }
}

fn python_varargs_callable_arg_list(prefix: &[RenderedFunctionArg], varargs_name: &str) -> String {
    let mut args = prefix
        .iter()
        .map(|parameter| parameter.annotation.clone())
        .collect::<Vec<_>>();
    args.push(format!("typing_extensions.Unpack[{varargs_name}]"));
    args.join(", ")
}

fn python_callable_arg_list(args: &[String]) -> String {
    if args.is_empty() {
        String::new()
    } else {
        args.join(", ")
    }
}

fn function_overload_parameter_prefix(function: &RenderedFunctionField) -> String {
    function.callable_field_name.to_upper_camel_case()
}

fn primary_function(functions: &[RenderedFunctionField]) -> Option<&RenderedFunctionField> {
    functions.iter().find(|function| function.primary)
}

fn python_function_positional_args_annotation(function: &RenderedFunctionField) -> String {
    match &function.args {
        RenderedFunctionArgs::Typed { parameters } => python_function_args_list_annotation(
            parameters
                .iter()
                .map(|parameter| parameter.annotation.as_str())
                .collect::<Vec<_>>()
                .as_slice(),
        )
        .trim_start_matches("list[")
        .trim_end_matches(']')
        .to_string(),
        RenderedFunctionArgs::Varargs { .. } => "object".to_string(),
    }
}

fn python_function_fixed_positional_args(
    function: &RenderedFunctionField,
) -> Option<&[RenderedFunctionArg]> {
    match &function.args {
        RenderedFunctionArgs::Typed { parameters } if function.primary => Some(parameters),
        _ => None,
    }
}

fn python_function_args_parameter_annotation(
    function: &RenderedFunctionField,
    primary_args: bool,
) -> String {
    match (&function.args, primary_args) {
        (RenderedFunctionArgs::Typed { parameters }, true) => {
            let annotations = parameters
                .iter()
                .map(|parameter| parameter.annotation.as_str())
                .collect::<Vec<_>>();
            format!(
                "{} | None",
                python_function_args_list_annotation(annotations.as_slice())
            )
        }
        (RenderedFunctionArgs::Typed { parameters }, false) => {
            if parameters.len() == 1 {
                format!("{} | None", parameters[0].annotation)
            } else {
                let annotations = parameters
                    .iter()
                    .map(|parameter| parameter.annotation.as_str())
                    .collect::<Vec<_>>();
                format!(
                    "{} | None",
                    python_function_args_list_annotation(annotations.as_slice())
                )
            }
        }
        (RenderedFunctionArgs::Varargs { .. }, true) => "list[typing.Any] | None".to_string(),
        (RenderedFunctionArgs::Varargs { .. }, false) => {
            "object | list[typing.Any] | None".to_string()
        }
    }
}

fn python_function_normalized_args_annotation(function: &RenderedFunctionField) -> String {
    match &function.args {
        RenderedFunctionArgs::Typed { parameters } => {
            let annotations = parameters
                .iter()
                .map(|parameter| parameter.annotation.as_str())
                .collect::<Vec<_>>();
            python_function_args_list_annotation(annotations.as_slice())
        }
        RenderedFunctionArgs::Varargs { .. } => "list[typing.Any]".to_string(),
    }
}

fn overload_case_for_callable<'a>(
    overload_cases: &'a [RenderedFunctionOverloadCase],
    callable_field_name: &str,
) -> Option<&'a RenderedFunctionOverloadCase> {
    overload_cases
        .iter()
        .find(|function_case| function_case.callable_field_name == callable_field_name)
}

fn render_function_unpacked_overload(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    unpacked_input: &RenderedUnpackedInput,
    overload_cases: &[RenderedFunctionOverloadCase],
) {
    output.push_str("@typing.overload\n");
    output.push_str("async def ");
    output.push_str(&operation.attr_name);
    output.push_str("(\n");

    let primary_function = primary_function(&unpacked_input.functions);
    let primary_case = primary_function.and_then(|function| {
        overload_case_for_callable(overload_cases, &function.callable_field_name)
    });
    if let (Some(function), Some(function_case)) = (primary_function, primary_case) {
        output.push_str("    ");
        output.push_str(&function.callable_field_name);
        output.push_str(": ");
        output.push_str(&function_case.callable_annotation);
        output.push_str(",\n");
        for positional_arg in &function_case.positional_args {
            output.push_str("    ");
            if positional_arg.variadic {
                output.push('*');
            }
            output.push_str(&positional_arg.name);
            output.push_str(": ");
            output.push_str(&positional_arg.annotation);
            output.push_str(",\n");
        }
    }

    if !primary_case.is_some_and(|function_case| !function_case.positional_args.is_empty()) {
        output.push_str("    *,\n");
    }

    for field in &unpacked_input.parameters {
        if primary_function.is_some_and(|function| function.callable_field_name == field.attr_name)
        {
            continue;
        }
        if primary_function
            .and_then(python_function_fixed_positional_args)
            .is_some_and(|parameters| {
                parameters
                    .iter()
                    .any(|parameter| parameter.name == field.attr_name)
            })
        {
            continue;
        }
        if primary_function.is_some_and(|function| function.args_field_name == field.attr_name) {
            if let Some(function_case) = primary_case {
                if let Some(args_annotation) = &function_case.args_annotation {
                    output.push_str("    ");
                    output.push_str(&field.attr_name);
                    output.push_str(": ");
                    output.push_str(args_annotation);
                    if function_case.args_optional {
                        output.push_str(" = ...");
                    }
                    output.push_str(",\n");
                }
            }
            continue;
        }

        if let Some(function_case) = overload_cases
            .iter()
            .find(|function_case| function_case.callable_field_name == field.attr_name)
        {
            output.push_str("    ");
            output.push_str(&field.attr_name);
            output.push_str(": ");
            output.push_str(&function_case.callable_annotation);
            if field.default_kind != PythonFieldDefaultKind::Required {
                output.push_str(" = ...");
            }
            output.push_str(",\n");
            continue;
        }

        if let Some(function_case) = overload_cases
            .iter()
            .find(|function_case| function_case.args_field_name == field.attr_name)
        {
            if let Some(args_annotation) = &function_case.args_annotation {
                output.push_str("    ");
                output.push_str(&field.attr_name);
                output.push_str(": ");
                output.push_str(args_annotation);
                if function_case.args_optional {
                    output.push_str(" = ...");
                }
                output.push_str(",\n");
            }
            continue;
        }

        output.push_str("    ");
        output.push_str(&field.attr_name);
        output.push_str(": ");
        output.push_str(&python_parameter_annotation(
            &field.annotation,
            &field.default_kind,
        ));
        if field.default_kind != PythonFieldDefaultKind::Required {
            output.push_str(" = ...");
        }
        output.push_str(",\n");
    }

    output.push_str(") -> ");
    output.push_str(&function_unpacked_overload_return_annotation(
        operation,
        primary_case,
    ));
    output.push_str(": ...\n");
}

fn function_unpacked_overload_return_annotation(
    operation: &RenderedOperation<'_>,
    primary_case: Option<&RenderedFunctionOverloadCase>,
) -> String {
    let active_type_parameters = primary_case
        .and_then(|function_case| function_case.result_type_parameter.as_ref())
        .into_iter()
        .cloned()
        .collect::<BTreeSet<_>>();
    let output_annotation = erase_inactive_python_type_parameters(
        &operation.overload_output_annotation,
        &operation.output_type_parameters,
        &active_type_parameters,
    );
    if operation.output_transform_expr.is_some()
        || operation.output_resource_return.is_some()
        || operation.output_direct_result
    {
        output_annotation
    } else {
        format!("temporalio.workflow.NexusOperationHandle[{output_annotation}]")
    }
}

fn erase_inactive_python_type_parameters(
    annotation: &str,
    type_parameters: &BTreeSet<String>,
    active_type_parameters: &BTreeSet<String>,
) -> String {
    let inactive_type_parameters = type_parameters
        .difference(active_type_parameters)
        .cloned()
        .collect::<BTreeSet<_>>();
    erase_python_type_parameters(annotation, &inactive_type_parameters)
}

fn erase_python_type_parameters(annotation: &str, type_parameters: &BTreeSet<String>) -> String {
    type_parameters
        .iter()
        .fold(annotation.to_string(), |current, name| {
            replace_python_identifier(&current, name, "typing.Any")
        })
}

fn python_annotation_uses_identifier(annotation: &str, identifier: &str) -> bool {
    annotation != replace_python_identifier(annotation, identifier, "typing.Any")
}

fn replace_python_identifier(source: &str, identifier: &str, replacement: &str) -> String {
    if identifier.is_empty() {
        return source.to_string();
    }

    let mut output = String::with_capacity(source.len());
    let mut index = 0;
    while let Some(relative_match_start) = source[index..].find(identifier) {
        let match_start = index + relative_match_start;
        let match_end = match_start + identifier.len();
        let before = source[..match_start].chars().next_back();
        let after = source[match_end..].chars().next();
        if before.is_some_and(|character| is_python_identifier_char(character) || character == '.')
            || after.is_some_and(is_python_identifier_char)
        {
            output.push_str(&source[index..match_end]);
        } else {
            output.push_str(&source[index..match_start]);
            output.push_str(replacement);
        }
        index = match_end;
    }
    output.push_str(&source[index..]);
    output
}

fn render_operation_functions(
    output: &mut String,
    service: &RenderedService<'_>,
    operation: &RenderedOperation<'_>,
) {
    render_request_only_operation_function(output, service, operation);
    if let Some(unpacked_input) = &operation.unpacked_input {
        output.push('\n');
        if function_unpacked_needs_overloads(&unpacked_input.functions) {
            render_function_unpacked_overloads(output, operation, unpacked_input);
            output.push('\n');
        }
        render_unpacked_operation_function(output, operation, unpacked_input);
    }
}

fn function_unpacked_needs_overloads(functions: &[RenderedFunctionField]) -> bool {
    functions.iter().any(|function| {
        function.alternate_annotation.is_some()
            || matches!(function.args, RenderedFunctionArgs::Varargs { .. })
    })
}

fn request_operation_function_name(operation: &RenderedOperation<'_>) -> String {
    if operation.unpacked_input.is_some() {
        format!("_{}", operation.attr_name)
    } else {
        operation.attr_name.clone()
    }
}

fn render_request_only_operation_function(
    output: &mut String,
    service: &RenderedService<'_>,
    operation: &RenderedOperation<'_>,
) {
    let function_name = request_operation_function_name(operation);
    let render_request_doc = operation.unpacked_input.is_none();
    output.push_str("async def ");
    output.push_str(&function_name);
    output.push_str("(\n");
    output.push_str("    request: ");
    output.push_str(&operation.input_annotation);
    output.push_str(",\n");
    if operation.output_transform_expr.is_some()
        || operation.output_resource_return.is_some()
        || operation.output_direct_result
    {
        output.push_str(") -> ");
        output.push_str(&operation.output_annotation);
        output.push_str(":\n");
    } else {
        output.push_str(") -> temporalio.workflow.NexusOperationHandle[\n");
        output.push_str("    ");
        output.push_str(&operation.output_annotation);
        output.push_str(",\n");
        output.push_str("]:\n");
        if render_request_doc {
            render_operation_docstring(output, operation, &[], None);
        }
        render_inline_nexus_client(output, service, "    ");
        output.push_str("    return await nexus_client.start_operation(\n");
        output.push_str("        operation=");
        output.push_str(&python_string_literal(operation.wire_name));
        output.push_str(",\n");
        output.push_str("        ");
        output.push_str("input=");
        output.push_str(&operation.input_to_proto_expr);
        output.push_str(",\n");
        if !operation.output_none {
            output.push_str("        output_type=");
            output.push_str(&operation.output_type_expr);
            output.push_str(",\n");
        }
        output.push_str("    )\n");
        return;
    }

    if render_request_doc {
        render_operation_docstring(output, operation, &[], None);
    }

    if operation.output_direct_result {
        render_inline_nexus_client(output, service, "    ");
        output.push_str("    handle = await nexus_client.start_operation(\n");
        output.push_str("        operation=");
        output.push_str(&python_string_literal(operation.wire_name));
        output.push_str(",\n");
        output.push_str("        ");
        output.push_str("input=");
        output.push_str(&operation.input_to_proto_expr);
        output.push_str(",\n");
        output.push_str("        output_type=");
        output.push_str(&operation.output_type_expr);
        output.push_str(",\n");
        output.push_str("    )\n");
        output.push_str("    return await handle\n");
        return;
    }

    output.push_str("    request_proto = ");
    output.push_str(&operation.input_to_proto_expr);
    output.push('\n');
    render_inline_nexus_client(output, service, "    ");
    output.push_str("    handle = await nexus_client.start_operation(\n");
    output.push_str("        operation=");
    output.push_str(&python_string_literal(operation.wire_name));
    output.push_str(",\n");
    output.push_str("        input=request_proto,\n");
    if !operation.output_none {
        output.push_str("        output_type=");
        output.push_str(&operation.output_type_expr);
        output.push_str(",\n");
    }
    output.push_str("    )\n");
    if let Some(transform_expr) = &operation.output_transform_expr {
        output.push_str("    result = await handle\n");
        output.push_str("    return ");
        output.push_str(transform_expr);
        output.push('\n');
    } else if let Some(resource_return) = &operation.output_resource_return {
        output.push_str("    result = await handle\n");
        output.push_str("    return ");
        output.push_str(&resource_return.resource_type_name);
        output.push_str("(\n");
        for binding in &resource_return.bindings {
            output.push_str("        ");
            output.push_str(&python_field_name(&binding.field_name));
            output.push('=');
            output.push_str(&resource_return_binding_expr_python(binding));
            output.push_str(",\n");
        }
        output.push_str("    )\n");
    }
}

fn render_unpacked_operation_docstring(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    unpacked_input: &RenderedUnpackedInput,
    primary_function: Option<&RenderedFunctionField>,
) {
    let mut args = Vec::<(String, String)>::new();
    for field in &unpacked_input.parameters {
        if let Some(function) = primary_function
            && field.attr_name == function.callable_field_name
        {
            if let Some(doc) = &field.doc {
                args.push((field.attr_name.clone(), doc.clone()));
            }
            if let Some(parameters) = python_function_fixed_positional_args(function) {
                for parameter in parameters {
                    args.push((
                        parameter.name.clone(),
                        format!("Argument for {}.", function.callable_field_name),
                    ));
                }
            } else {
                args.push((
                    "positional_args".to_string(),
                    format!(
                        "Positional arguments for {}. Cannot be set if args is set.",
                        function.callable_field_name
                    ),
                ));
            }
            continue;
        }

        if let Some(function) = unpacked_input
            .functions
            .iter()
            .find(|function| function.args_field_name == field.attr_name)
        {
            let generated_doc = if function.primary {
                match &function.args {
                    RenderedFunctionArgs::Varargs { .. } => format!(
                        "List-form arguments for {}. Cannot be set if positional_args are set. \
                        For typed {} callables, list contents are not statically typechecked; \
                        pass {} arguments positionally for precise typechecking.",
                        function.callable_field_name,
                        function.callable_field_name,
                        function.callable_field_name
                    ),
                    _ => format!(
                        "List-form arguments for {}. Cannot be set if positional_args are set.",
                        function.callable_field_name
                    ),
                }
            } else {
                match &function.args {
                    RenderedFunctionArgs::Varargs { .. } => format!(
                        "Argument value, or list of argument values, for {}. For typed \
                        single-argument signals, scalar signal_args values are statically \
                        typechecked. List-form signal_args values are not precisely typechecked. \
                        To pass a single signal argument that is itself a list, wrap it in \
                        another list; otherwise the list is interpreted as multiple signal \
                        arguments.",
                        function.callable_field_name
                    ),
                    _ => format!(
                        "Argument value, or list of argument values, for {}. To pass a single \
                        argument that is itself a list, wrap it in another list; otherwise the \
                        list is interpreted as multiple arguments.",
                        function.callable_field_name
                    ),
                }
            };
            args.push((
                field.attr_name.clone(),
                field.doc.clone().unwrap_or(generated_doc),
            ));
            continue;
        }

        if let Some(doc) = &field.doc {
            args.push((field.attr_name.clone(), doc.clone()));
        }
    }

    render_operation_docstring(output, operation, &args, operation.return_doc.as_deref());
}

fn render_operation_docstring(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    args: &[(String, String)],
    returns: Option<&str>,
) {
    render_python_docstring(
        output,
        "    ",
        operation.doc.as_deref(),
        args,
        returns,
        operation.experimental,
    );
}

fn render_python_docstring(
    output: &mut String,
    indent: &str,
    summary: Option<&str>,
    args: &[(String, String)],
    returns: Option<&str>,
    experimental: bool,
) {
    let mut lines = Vec::<String>::new();
    let docstring_width = PYTHON_FORMAT_LINE_LENGTH.saturating_sub(indent.chars().count());
    let has_summary = summary.is_some_and(|summary| !summary.trim().is_empty());
    if let Some(summary) = summary.map(str::trim).filter(|summary| !summary.is_empty()) {
        for line in summary.lines() {
            push_wrapped_python_docstring_line(&mut lines, "", "", line.trim(), docstring_width);
        }
    }
    if experimental {
        if !lines.is_empty() {
            lines.push(String::new());
        }
        lines.push(".. warning::".to_string());
        push_wrapped_python_docstring_line(
            &mut lines,
            "    ",
            "    ",
            EXPERIMENTAL_WARNING,
            docstring_width,
        );
    }
    if !args.is_empty() {
        if !lines.is_empty() {
            lines.push(String::new());
        }
        lines.push("Args:".to_string());
        for (name, doc) in args {
            let mut doc_lines = doc.trim().lines();
            let first_prefix = format!("    {name}: ");
            let continuation_prefix = "        ";
            let first = doc_lines.next().unwrap_or_default().trim();
            push_wrapped_python_docstring_line(
                &mut lines,
                &first_prefix,
                continuation_prefix,
                first,
                docstring_width,
            );
            for line in doc_lines {
                push_wrapped_python_docstring_line(
                    &mut lines,
                    continuation_prefix,
                    continuation_prefix,
                    line.trim(),
                    docstring_width,
                );
            }
        }
    }
    if let Some(returns) = returns.map(str::trim).filter(|returns| !returns.is_empty()) {
        if !lines.is_empty() {
            lines.push(String::new());
        }
        lines.push("Returns:".to_string());
        let mut return_lines = returns.lines();
        if let Some(first) = return_lines.next() {
            push_wrapped_python_docstring_line(
                &mut lines,
                "    ",
                "    ",
                first.trim(),
                docstring_width,
            );
        }
        for line in return_lines {
            push_wrapped_python_docstring_line(
                &mut lines,
                "    ",
                "    ",
                line.trim(),
                docstring_width,
            );
        }
    }
    if lines.is_empty() {
        return;
    }

    output.push_str(indent);
    output.push_str("\"\"\"");
    if !has_summary {
        output.push('\n');
        for line in &lines {
            if !line.is_empty() {
                output.push_str(indent);
                output.push_str(&python_docstring_literal_text(line));
            }
            output.push('\n');
        }
        output.push_str(indent);
        output.push_str("\"\"\"\n");
        return;
    }
    if lines.len() == 1 {
        output.push_str(&python_docstring_literal_text(&lines[0]));
        output.push_str("\"\"\"\n");
        return;
    }

    output.push_str(&python_docstring_literal_text(&lines[0]));
    output.push('\n');
    for line in lines.iter().skip(1) {
        if !line.is_empty() {
            output.push_str(indent);
            output.push_str(&python_docstring_literal_text(line));
        }
        output.push('\n');
    }
    output.push_str(indent);
    output.push_str("\"\"\"\n");
}

fn push_wrapped_python_docstring_line(
    lines: &mut Vec<String>,
    first_prefix: &str,
    continuation_prefix: &str,
    text: &str,
    max_width: usize,
) {
    if text.is_empty() {
        lines.push(first_prefix.trim_end().to_string());
        return;
    }

    let mut prefix = first_prefix;
    let mut current = String::new();
    for word in text.split_whitespace() {
        let prefix_width = prefix.chars().count();
        let current_width = current.chars().count();
        let word_width = word.chars().count();
        let separator_width = usize::from(!current.is_empty());
        if current_width > 0
            && prefix_width + current_width + separator_width + word_width > max_width
        {
            lines.push(format!("{prefix}{current}"));
            prefix = continuation_prefix;
            current.clear();
        }
        if !current.is_empty() {
            current.push(' ');
        }
        current.push_str(word);
    }

    lines.push(format!("{prefix}{current}"));
}

fn python_docstring_literal_text(value: &str) -> String {
    value
        .replace('\\', "\\\\")
        .replace("\"\"\"", "\\\"\\\"\\\"")
}

fn render_unpacked_operation_function(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    unpacked_input: &RenderedUnpackedInput,
) {
    if !unpacked_input.functions.is_empty() {
        render_function_unpacked_implementation(output, operation, unpacked_input);
        return;
    }

    output.push_str("async def ");
    output.push_str(&operation.attr_name);
    output.push_str("(\n");
    output.push_str("    *,\n");
    for field in &unpacked_input.parameters {
        output.push_str("    ");
        output.push_str(&field.attr_name);
        output.push_str(": ");
        output.push_str(&python_parameter_annotation(
            &field.annotation,
            &field.default_kind,
        ));
        if let Some(default_expr) = python_parameter_default_expr(&field.default_kind) {
            render_python_default_expr(output, &default_expr, "    ");
        }
        output.push_str(",\n");
    }
    if operation.output_transform_expr.is_some()
        || operation.output_resource_return.is_some()
        || operation.output_direct_result
    {
        output.push_str(") -> ");
        output.push_str(&operation.output_annotation);
        output.push_str(":\n");
    } else {
        output.push_str(") -> temporalio.workflow.NexusOperationHandle[\n");
        output.push_str("    ");
        output.push_str(&operation.output_annotation);
        output.push_str(",\n");
        output.push_str("]:\n");
    }
    render_unpacked_operation_docstring(output, operation, unpacked_input, None);
    for flattened_message in &unpacked_input.flattened_messages {
        render_flattened_message_setup(output, flattened_message);
    }
    output.push_str("    request = ");
    output.push_str(&unpacked_input.model_name);
    output.push_str("(\n");
    for field in &unpacked_input.request_fields {
        output.push_str("        ");
        output.push_str(&field.attr_name);
        output.push('=');
        output.push_str(&unpacked_request_field_value_expr(field));
        output.push_str(",\n");
    }
    output.push_str("    )\n");
    output.push_str("    return await ");
    output.push_str(&request_operation_function_name(operation));
    output.push_str("(request)\n");
}

fn render_function_unpacked_implementation(
    output: &mut String,
    operation: &RenderedOperation<'_>,
    unpacked_input: &RenderedUnpackedInput,
) {
    output.push_str("async def ");
    output.push_str(&operation.attr_name);
    output.push_str("(\n");

    let primary_function = primary_function(&unpacked_input.functions);
    if let Some(function) = primary_function {
        output.push_str("    ");
        output.push_str(&function.callable_field_name);
        output.push_str(": ");
        output.push_str(&function.callable_annotation);
        output.push_str(",\n");
        if let Some(parameters) = python_function_fixed_positional_args(function) {
            for parameter in parameters {
                output.push_str("    ");
                output.push_str(&parameter.name);
                output.push_str(": ");
                output.push_str(&parameter.annotation);
                output.push_str(",\n");
            }
        } else {
            output.push_str("    *positional_args: ");
            output.push_str(&python_function_positional_args_annotation(function));
            output.push_str(",\n");
        }
    }

    let has_keyword_only_parameters = unpacked_input
        .parameters
        .iter()
        .any(|field| rendered_function_implementation_renders_field(field, primary_function));
    if primary_function.is_none()
        || (primary_function
            .and_then(python_function_fixed_positional_args)
            .is_some()
            && has_keyword_only_parameters)
    {
        output.push_str("    *,\n");
    }

    for field in &unpacked_input.parameters {
        if !rendered_function_implementation_renders_field(field, primary_function) {
            continue;
        }
        output.push_str("    ");
        output.push_str(&field.attr_name);
        output.push_str(": ");
        if let Some(function) = unpacked_input
            .functions
            .iter()
            .find(|function| function.callable_field_name == field.attr_name)
        {
            output.push_str(&function.callable_annotation);
        } else if unpacked_input
            .functions
            .iter()
            .any(|function| function.args_field_name == field.attr_name)
        {
            let function = unpacked_input
                .functions
                .iter()
                .find(|function| function.args_field_name == field.attr_name)
                .expect("function args field should have a function");
            let primary_args = primary_function
                .is_some_and(|function| function.args_field_name == field.attr_name);
            output.push_str(&python_function_args_parameter_annotation(
                function,
                primary_args,
            ));
        } else {
            output.push_str(&python_parameter_annotation(
                &field.annotation,
                &field.default_kind,
            ));
        }
        if let Some(default_expr) = python_parameter_default_expr(&field.default_kind) {
            render_python_default_expr(output, &default_expr, "    ");
        }
        output.push_str(",\n");
    }

    if operation.output_transform_expr.is_some()
        || operation.output_resource_return.is_some()
        || operation.output_direct_result
    {
        output.push_str(") -> ");
        output.push_str(&operation.output_annotation);
        output.push_str(":\n");
    } else {
        output.push_str(") -> temporalio.workflow.NexusOperationHandle[\n");
        output.push_str("    ");
        output.push_str(&operation.output_annotation);
        output.push_str(",\n");
        output.push_str("]:\n");
    }
    render_unpacked_operation_docstring(output, operation, unpacked_input, primary_function);

    for function in &unpacked_input.functions {
        if function.primary && matches!(function.args, RenderedFunctionArgs::Typed { .. }) {
            continue;
        }
        let normalized_args_name = format!("normalized_{}", function.args_field_name);
        if function.primary {
            output.push_str("    if positional_args and ");
            output.push_str(&function.args_field_name);
            output.push_str(" is not None:\n");
            output.push_str(
                "        raise TypeError(\"cannot specify both positional arguments and ",
            );
            output.push_str(&function.args_field_name);
            output.push_str("\")\n");
            output.push_str("    ");
            output.push_str(&normalized_args_name);
            output.push_str(": ");
            output.push_str(&python_function_normalized_args_annotation(function));
            output.push_str(" | None = (\n");
            output.push_str("        list(positional_args)\n");
            output.push_str("        if positional_args\n");
            output.push_str("        else ");
            output.push_str(&function.args_field_name);
            output.push('\n');
            output.push_str("    )\n");
        } else {
            output.push_str("    ");
            output.push_str(&normalized_args_name);
            output.push_str(": ");
            output.push_str(&python_function_normalized_args_annotation(function));
            output.push_str(" | None\n");
            output.push_str("    if ");
            output.push_str(&function.args_field_name);
            output.push_str(" is None:\n");
            output.push_str("        ");
            output.push_str(&normalized_args_name);
            output.push_str(" = None\n");
            output.push_str("    elif isinstance(");
            output.push_str(&function.args_field_name);
            output.push_str(", list):\n");
            output.push_str("        ");
            output.push_str(&normalized_args_name);
            match &function.args {
                RenderedFunctionArgs::Varargs { .. } => {
                    output.push_str(" = typing.cast(list[typing.Any], ");
                    output.push_str(&function.args_field_name);
                    output.push_str(")\n");
                }
                _ => {
                    output.push_str(" = ");
                    output.push_str(&function.args_field_name);
                    output.push('\n');
                }
            }
            output.push_str("    else:\n");
            output.push_str("        ");
            output.push_str(&normalized_args_name);
            output.push_str(" = [");
            output.push_str(&function.args_field_name);
            output.push_str("]\n");
        }
    }
    for flattened_message in &unpacked_input.flattened_messages {
        render_flattened_message_setup(output, flattened_message);
    }

    output.push_str("    request = ");
    output.push_str(&unpacked_input.model_name);
    output.push_str("(\n");
    for field in &unpacked_input.request_fields {
        let value_expr = unpacked_input
            .functions
            .iter()
            .find(|function| function.args_field_name == field.value_expr)
            .map(|function| format!("normalized_{}", function.args_field_name))
            .unwrap_or_else(|| unpacked_request_field_value_expr(field));
        output.push_str("        ");
        output.push_str(&field.attr_name);
        output.push('=');
        output.push_str(&value_expr);
        output.push_str(",\n");
    }
    output.push_str("    )\n");
    output.push_str("    return await ");
    output.push_str(&request_operation_function_name(operation));
    output.push_str("(request)\n");
}

fn rendered_function_implementation_renders_field(
    field: &RenderedUnpackedInputField,
    primary_function: Option<&RenderedFunctionField>,
) -> bool {
    if primary_function.is_some_and(|function| function.callable_field_name == field.attr_name) {
        return false;
    }
    if primary_function
        .and_then(python_function_fixed_positional_args)
        .is_some_and(|parameters| {
            parameters
                .iter()
                .any(|parameter| parameter.name == field.attr_name)
        })
    {
        return false;
    }
    true
}

fn render_flattened_message_setup(
    output: &mut String,
    flattened_message: &RenderedFlattenedMessage,
) {
    let always_construct = flattened_message.required
        || flattened_message
            .fields
            .iter()
            .any(|field| field.default_kind == PythonFieldDefaultKind::Required);

    output.push_str("    ");
    output.push_str(&flattened_message.local_name);
    output.push_str(" = ");
    if always_construct {
        output.push_str(&flattened_message.model_name);
        output.push_str("(\n");
        for field in &flattened_message.fields {
            output.push_str("        ");
            output.push_str(&field.attr_name);
            output.push('=');
            output.push_str(&field.value_expr);
            output.push_str(",\n");
        }
        output.push_str("    )\n");
        return;
    }

    output.push_str("(\n");
    output.push_str("        None\n");
    output.push_str("        if ");
    for (index, field) in flattened_message.fields.iter().enumerate() {
        if index > 0 {
            output.push_str(" and ");
        }
        output.push_str(&field.value_expr);
        output.push_str(" is None");
    }
    output.push_str("\n");
    output.push_str("        else ");
    output.push_str(&flattened_message.model_name);
    output.push_str("(\n");
    for field in &flattened_message.fields {
        output.push_str("            ");
        output.push_str(&field.attr_name);
        output.push('=');
        output.push_str(&field.value_expr);
        output.push_str(",\n");
    }
    output.push_str("        )\n");
    output.push_str("    )\n");
}

fn render_inline_nexus_client(output: &mut String, service: &RenderedService<'_>, indent: &str) {
    output.push_str(indent);
    output.push_str("nexus_client = temporalio.workflow.create_nexus_client(\n");
    output.push_str(indent);
    output.push_str("    service=");
    output.push_str(&python_string_literal(service.wire_name));
    output.push_str(",\n");
    output.push_str(indent);
    output.push_str("    endpoint=");
    output.push_str(&python_string_literal(&service.endpoint));
    output.push_str(",\n");
    output.push_str(indent);
    output.push_str(")\n");
}

fn python_parameter_default_expr(default_kind: &PythonFieldDefaultKind) -> Option<String> {
    match default_kind {
        PythonFieldDefaultKind::Required => None,
        PythonFieldDefaultKind::None => Some("None".to_string()),
        PythonFieldDefaultKind::EmptyList | PythonFieldDefaultKind::EmptyDict => {
            Some("None".to_string())
        }
        PythonFieldDefaultKind::Expression(expr) => Some(expr.clone()),
    }
}

fn render_python_default_expr(output: &mut String, default_expr: &str, indent: &str) {
    if default_expr.chars().count() <= 48 {
        output.push_str(" = ");
        output.push_str(default_expr);
        return;
    }

    output.push_str(" = (\n");
    output.push_str(indent);
    output.push_str("    ");
    output.push_str(default_expr);
    output.push('\n');
    output.push_str(indent);
    output.push(')');
}

fn python_parameter_annotation(annotation: &str, default_kind: &PythonFieldDefaultKind) -> String {
    if matches!(
        default_kind,
        PythonFieldDefaultKind::EmptyList | PythonFieldDefaultKind::EmptyDict
    ) && !annotation.contains("| None")
    {
        format!("{annotation} | None")
    } else {
        annotation.to_string()
    }
}

fn unpacked_request_field_value_expr(field: &RenderedUnpackedRequestField) -> String {
    match &field.default_kind {
        PythonFieldDefaultKind::EmptyList => format!("{} or []", field.value_expr),
        PythonFieldDefaultKind::EmptyDict => format!("{} or {{}}", field.value_expr),
        PythonFieldDefaultKind::Required
        | PythonFieldDefaultKind::None
        | PythonFieldDefaultKind::Expression(_) => field.value_expr.clone(),
    }
}

fn module_path(reference: &str) -> &str {
    reference
        .rsplit_once('.')
        .map(|(module, _)| module)
        .unwrap_or(reference)
}

fn python_ident(name: &str) -> String {
    if is_python_keyword(name) {
        format!("{name}_")
    } else {
        name.to_string()
    }
}

fn python_string_literal(value: &str) -> String {
    format!("{value:?}")
}

fn is_python_keyword(name: &str) -> bool {
    matches!(
        name,
        "False"
            | "None"
            | "True"
            | "and"
            | "as"
            | "assert"
            | "async"
            | "await"
            | "break"
            | "class"
            | "continue"
            | "def"
            | "del"
            | "elif"
            | "else"
            | "except"
            | "finally"
            | "for"
            | "from"
            | "global"
            | "if"
            | "import"
            | "in"
            | "is"
            | "lambda"
            | "nonlocal"
            | "not"
            | "or"
            | "pass"
            | "raise"
            | "return"
            | "try"
            | "while"
            | "with"
            | "yield"
            | "match"
            | "case"
    )
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::process::Command;
    use std::time::{SystemTime, UNIX_EPOCH};

    use crate::SupportFiles;
    use crate::descriptors::DescriptorIndex;
    use crate::error::Error;
    use crate::generator::{GeneratedOutputLayout, generate_files, generate_source};
    use crate::language::Language;
    use crate::spec::{ApiSpec, LanguageImportSpec, LanguageImportStyle};

    fn sample_input_path(root: &std::path::Path) -> PathBuf {
        root.join("examples/inputs/workflow-service.wit")
    }

    fn start_workflow_input_path(root: &std::path::Path) -> PathBuf {
        root.join("examples/inputs/start-workflow.wit")
    }

    fn type_roundtrip_input_path(root: &std::path::Path) -> PathBuf {
        root.join("examples/inputs/type-roundtrip.wit")
    }

    fn linked_inputs_path(root: &std::path::Path) -> PathBuf {
        root.join("examples/inputs/deps")
    }

    fn example_input_paths(root: &std::path::Path, input_path: PathBuf) -> Vec<PathBuf> {
        vec![input_path, linked_inputs_path(root)]
    }

    fn sample_python_output_path(root: &std::path::Path) -> PathBuf {
        root.join("examples/python/workflow_service")
    }

    fn unique_temp_dir(label: &str) -> PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        std::env::temp_dir().join(format!("nex-gen-python-{label}-{unique}"))
    }

    fn write_package(dir: &Path, files: &BTreeMap<PathBuf, String>) {
        for (relative_path, contents) in files {
            let path = dir.join(relative_path);
            if let Some(parent) = path.parent() {
                fs::create_dir_all(parent).unwrap();
            }
            fs::write(path, contents).unwrap();
        }
    }

    #[test]
    fn optional_imports_are_selected_from_generic_references() {
        let mut module_imports = std::collections::BTreeSet::new();
        module_imports.insert("temporalio.api.common.v1.message_pb2".to_string());
        module_imports.insert("temporalio.api.unused.v1.message_pb2".to_string());

        let body = r#"
@dataclasses.dataclass
class Example(enum.Enum):
    value: collections.abc.Mapping[str, typing.Any]
    retry_policy: temporalio.common.RetryPolicy
    request: temporalio.api.common.v1.message_pb2.WorkflowExecution
    deadline: datetime.timedelta

    def run(self) -> typing_extensions.Self:
        temporalio.workflow.create_nexus_client(service="svc", endpoint="endpoint")
        return self
"#;
        let language_imports = [
            LanguageImportSpec {
                language: Language::Python,
                reference: "temporalio.common".to_string(),
                module: "temporalio.common".to_string(),
                name: None,
                type_only: false,
                import_style: LanguageImportStyle::Module,
            },
            LanguageImportSpec {
                language: Language::Python,
                reference: "datetime".to_string(),
                module: "datetime".to_string(),
                name: None,
                type_only: false,
                import_style: LanguageImportStyle::Module,
            },
            LanguageImportSpec {
                language: Language::Python,
                reference: "temporalio.workflow".to_string(),
                module: "temporalio.workflow".to_string(),
                name: None,
                type_only: false,
                import_style: LanguageImportStyle::Module,
            },
        ];

        let mut output = String::new();
        let wrote_imports = super::render_optional_python_imports(
            &mut output,
            body,
            &module_imports,
            &language_imports,
        );

        assert!(wrote_imports);
        assert!(output.contains("import collections.abc\n"));
        assert!(output.contains("import dataclasses\n"));
        assert!(output.contains("import enum\n"));
        assert!(output.contains("import typing\n"));
        assert!(output.contains("import typing_extensions\n"));
        assert!(output.contains("import temporalio.common\n"));
        assert!(output.contains("import datetime\n"));
        assert!(output.contains("import temporalio.workflow\n"));
        assert!(output.contains("import temporalio.api.common.v1.message_pb2\n"));
        assert!(!output.contains("temporalio.api.unused.v1.message_pb2"));

        let mut output = String::new();
        super::render_optional_python_imports(
            &mut output,
            "async def call(workflow: str) -> None:\n    pass\n",
            &std::collections::BTreeSet::new(),
            &language_imports,
        );
        assert!(!output.contains("import temporalio.workflow\n"));
    }

    fn read_python_package_files(dir: &Path) -> BTreeMap<PathBuf, String> {
        fn visit(root: &Path, dir: &Path, files: &mut BTreeMap<PathBuf, String>) {
            let mut entries = fs::read_dir(dir)
                .unwrap()
                .map(|entry| entry.unwrap().path())
                .collect::<Vec<_>>();
            entries.sort();
            for path in entries {
                if path.is_dir() {
                    visit(root, &path, files);
                } else if path.extension().and_then(|extension| extension.to_str()) == Some("py") {
                    if path
                        .file_name()
                        .and_then(|file_name| file_name.to_str())
                        .is_some_and(|file_name| file_name.starts_with("test_"))
                    {
                        continue;
                    }
                    files.insert(
                        path.strip_prefix(root).unwrap().to_path_buf(),
                        fs::read_to_string(&path).unwrap(),
                    );
                }
            }
        }

        let mut files = BTreeMap::new();
        visit(dir, dir, &mut files);
        files
    }

    fn format_python_output(
        root: &std::path::Path,
        files: &BTreeMap<PathBuf, String>,
    ) -> BTreeMap<PathBuf, String> {
        let temp_dir = unique_temp_dir("format");
        fs::create_dir_all(&temp_dir).unwrap();
        write_package(&temp_dir, files);
        let status = Command::new("uv")
            .current_dir(root.join("examples/python"))
            .args([
                "run",
                "ruff",
                "format",
                "--line-length",
                "88",
                temp_dir.to_str().unwrap(),
            ])
            .status()
            .unwrap();
        assert!(status.success());
        let formatted = read_python_package_files(&temp_dir);
        fs::remove_dir_all(&temp_dir).unwrap();
        formatted
    }

    #[test]
    fn renders_sample_output() {
        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let spec = ApiSpec::load_for_language_with_inputs(
            Language::Python,
            &example_input_paths(&root, sample_input_path(&root)),
        )
        .unwrap();
        let descriptors =
            DescriptorIndex::load(&root.join("examples/descriptors/temporal_api.bin")).unwrap();
        let generated = generate_files(
            Language::Python,
            &spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap();
        assert_eq!(generated.layout, GeneratedOutputLayout::Directory);
        let output = format_python_output(&root, &generated.files);
        let expected = read_python_package_files(&sample_python_output_path(&root));

        assert_eq!(output, expected);
    }

    #[test]
    fn renders_required_fields_and_custom_message_types() {
        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let spec = ApiSpec::load_for_language_with_inputs(
            Language::Python,
            &example_input_paths(&root, sample_input_path(&root)),
        )
        .unwrap();
        let descriptors =
            DescriptorIndex::load(&root.join("examples/descriptors/temporal_api.bin")).unwrap();
        let output = generate_source(
            Language::Python,
            &spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap();

        assert!(output.contains("from ._support import ("));
        assert!(output.contains("retry_policy_to_proto,"));
        assert!(output.contains("def retry_policy_from_proto("));
        assert!(output.contains(
            "workflow: str | collections.abc.Callable[..., collections.abc.Awaitable[typing.Any]]"
        ));
        assert!(output.contains("id: str"));
        assert!(output.contains("task_queue: str"));
        assert!(output.contains(
            "signal: str | collections.abc.Callable[..., None | collections.abc.Awaitable[None]]"
        ));
        assert!(output.contains("class SignalWithStartWorkflowRequest:"));
        assert!(output.contains("args: list[typing.Any] | None = None"));
        assert!(!output.contains("namespace: str | None = None"));
        assert!(output.contains("message.namespace = workflow_namespace()"));
        assert!(output.contains("result = await handle"));
        assert!(output.contains("return temporalio.workflow.get_external_workflow_handle("));
        assert!(output.contains("run_id=result.run_id"));
        assert!(output.contains("signal_args: list[typing.Any] | None = None"));
        assert!(output.contains("temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE"));
        assert!(output.contains(
            "id_conflict_policy: temporalio.common.WorkflowIDConflictPolicy | None = None"
        ));
        assert!(!output.contains("identity: str | None"));
        assert!(output.contains("memo: collections.abc.Mapping[str, typing.Any] | None = None"));
        assert!(output.contains("user_metadata: UserMetadata | None = None"));
        assert!(output.contains("static_summary: typing.Any | None = None"));
        assert!(output.contains("static_details: typing.Any | None = None"));
        assert!(output.contains("retry_policy: temporalio.common.RetryPolicy"));
        assert!(
            output.contains(
                "search_attributes: temporalio.common.TypedSearchAttributes | None = None"
            )
        );
        assert!(output.contains("priority: temporalio.common.Priority | None = None"));
        assert!(
            output.contains(
                "versioning_override: temporalio.common.VersioningOverride | None = None"
            )
        );
        assert!(output.contains("workflow_id_reuse_policy_to_proto(self.id_reuse_policy)"));
        assert!(output.contains("if self.id_conflict_policy is not None:"));
        assert!(output.contains("workflow_id_conflict_policy_to_proto(self.id_conflict_policy)"));
        assert!(output.contains("message.input.CopyFrom(payloads_to_proto(self.args))"));
        assert!(!output.contains("header:"));
        assert!(!output.contains("header_to_proto("));
        assert!(!output.contains("links:"));
        assert!(!output.contains("link_to_proto("));
        assert!(output.contains("async def _signal_with_start_workflow("));
        assert!(output.contains("request: SignalWithStartWorkflowRequest"));
        assert!(output.contains(") -> temporalio.workflow.ExternalWorkflowHandle[typing.Any]:"));
        assert!(!output.contains("(typing.TypedDict, total=False):"));
        assert!(!output.contains("typing.Unpack["));
        assert!(!output.contains("args: tuple[typing.Any, ...] | None = None"));
        assert!(!output.contains("signal_args: tuple[typing.Any, ...] | None = None"));
        assert!(output.contains("@typing.overload"));
        assert!(output.contains("workflow: str,"));
        assert!(output.contains("*positional_args: object,"));
        assert!(output.contains("args: list[typing.Any] | None = ...,"));
        assert!(output.contains(
            "workflow: collections.abc.Callable[[typing.Any, typing_extensions.Unpack[WorkflowArgs]], collections.abc.Awaitable[WorkflowResult]],"
        ));
        assert!(output.contains("WorkflowArgs = typing_extensions.TypeVarTuple(\"WorkflowArgs\")"));
        assert!(output.contains("WorkflowResult = typing.TypeVar(\"WorkflowResult\")"));
        assert!(
            output.contains(") -> temporalio.workflow.ExternalWorkflowHandle[WorkflowResult]:")
        );
        assert!(output.contains("async def signal_with_start_workflow("));
        assert!(output.contains("*positional_args: typing_extensions.Unpack[WorkflowArgs],"));
        assert!(output.contains("args: list[typing.Any],"));
        assert!(!output.contains("tuple[FirstWorkflowArg"));
        assert!(output.contains(
            "signal: collections.abc.Callable[[typing.Any, SignalArg], None | collections.abc.Awaitable[None]],"
        ));
        assert!(output.contains("SignalArg = typing.TypeVar(\"SignalArg\")"));
        assert!(output.contains("signal_args: SignalArg,"));
        assert!(output.contains("signal_args: list[typing.Any],"));
        assert!(output.contains("async def signal_with_start_workflow("));
        assert!(output.contains(
            "async def signal_with_start_workflow(\n    workflow: str | collections.abc.Callable[..., collections.abc.Awaitable[typing.Any]],\n    *positional_args: object,\n    args: list[typing.Any] | None = None,"
        ));
        assert!(output.contains(
            "signal: str | collections.abc.Callable[..., None | collections.abc.Awaitable[None]],"
        ));
        assert!(output.contains("signal_args: object | list[typing.Any] | None = None,"));
        assert_eq!(
            output
                .matches("Signal a workflow, starting it first if needed.")
                .count(),
            1
        );
        assert!(output.contains(
            "\"\"\"Signal a workflow, starting it first if needed.\n\n    .. warning::\n        This API is experimental and subject to change.\n\n    Args:\n        workflow: Workflow type name or callable identifying the workflow to start.\n        positional_args: Positional arguments for workflow. Cannot be set if args is\n            set.\n        args: List-form arguments for workflow. Cannot be set if positional_args are\n            set. For typed workflow callables, list contents are not statically\n            typechecked; pass workflow arguments positionally for precise typechecking.\n        id: Unique identifier for the workflow execution.\n        task_queue: Task queue to run the workflow on.\n        signal: Signal name or callable to send with the start request.\n        signal_args: Argument value, or list of argument values, for signal. For typed\n            single-argument signals, scalar signal_args values are statically\n            typechecked. List-form signal_args values are not precisely typechecked. To\n            pass a single signal argument that is itself a list, wrap it in another\n            list; otherwise the list is interpreted as multiple signal arguments."
        ));
        assert!(output.contains(
            "cron_schedule: Cron schedule for recurring workflow executions. See\n            https://docs.temporal.io/cron-job."
        ));
        assert!(output.contains(
            "static_summary: Single-line fixed summary for the workflow execution that may\n            appear in UI and CLI. This can be in single-line Temporal Markdown format."
        ));
        assert!(output.contains(
            "\n\n    Returns:\n        A workflow handle to the started workflow.\n    \"\"\""
        ));
        assert!(output.contains("static_summary: str | None = None,"));
        assert!(output.contains("static_details: str | None = None,"));
        assert!(!output.contains("user_metadata_static_summary:"));
        assert!(!output.contains("user_metadata_static_details:"));
        assert!(!output.contains("def _nexus_is_function_args_list("));
        assert!(!output.contains("def _nexus_normalize_function_args("));
        assert!(!output.contains("_nexus_arg_unset = object()"));
        assert!(output.contains(
            "normalized_signal_args: list[typing.Any] | None\n    if signal_args is None:\n        normalized_signal_args = None\n    elif isinstance(signal_args, list):\n        normalized_signal_args = typing.cast(list[typing.Any], signal_args)\n    else:\n        normalized_signal_args = [signal_args]"
        ));
        assert!(output.contains("if positional_args and args is not None:"));
        assert!(
            output
                .contains("raise TypeError(\"cannot specify both positional arguments and args\")")
        );
        assert!(output.contains("normalized_args: list[typing.Any] | None = ("));
        assert!(output.contains("list(positional_args)"));
        assert!(output.contains("else args"));
        assert!(output.contains("user_metadata = ("));
        assert!(output.contains("if static_summary is None and static_details is None"));
        assert!(output.contains("static_summary=static_summary,"));
        assert!(output.contains("static_details=static_details,"));
        assert!(output.contains("args=normalized_args,"));
        assert!(output.contains("signal_args=normalized_signal_args,"));
        assert!(output.contains("user_metadata=user_metadata,"));
        assert!(output.contains("request = SignalWithStartWorkflowRequest("));
        assert!(output.contains("workflow=workflow,"));
        assert!(output.contains("args=normalized_args,"));
        assert!(output.contains("return await _signal_with_start_workflow(request)"));
        assert!(!output.contains("SignalWithStartWorkflowRequest.from_proto"));
        assert!(!output.contains(
            "proto: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest,\n    ) -> SignalWithStartWorkflowRequest:"
        ));
        assert!(!output.contains("class SignalWithStartWorkflowRequestTyped["));
        assert!(!output.contains("async def signal_with_start_workflow_typed("));
        assert!(!output.contains("class RetryPolicy:"));
        assert!(!output.contains("class WorkflowType:"));
        assert!(!output.contains("class TaskQueue:"));
        assert!(!output.contains("class WorkflowIdReusePolicy(enum.IntEnum):"));
        assert!(!output.contains("class WorkflowIdConflictPolicy(enum.IntEnum):"));
        assert!(!output.contains("class Payload:"));
        assert!(!output.contains("class ExternalPayloadDetails:"));
        assert!(!output.contains("class Payloads:"));
        assert!(!output.contains("class Memo:"));
        assert!(!output.contains("class Header:"));
        assert!(!output.contains("class SearchAttributes:"));
        assert!(output.contains("class UserMetadata:"));
        assert!(!output.contains("class Link:"));
        assert!(!output.contains("class Priority:"));
        assert!(!output.contains("class VersioningOverride:"));
        assert!(!output.contains("build_signal_with_start_workflow_request"));
        assert!(!output.contains("from model_overrides import"));

        let type_roundtrip_spec = ApiSpec::load_for_language_with_inputs(
            Language::Python,
            &example_input_paths(&root, type_roundtrip_input_path(&root)),
        )
        .unwrap();
        let type_roundtrip_output = generate_source(
            Language::Python,
            &type_roundtrip_spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap();
        assert!(
            type_roundtrip_output.contains(
                "raise ValueError(\"missing required field ActivityOptions.retry_policy\")"
            )
        );
        assert!(type_roundtrip_output.contains("if not proto.HasField(\"retry_policy\"):\n            raise ValueError(\"missing required field ActivityOptions.retry_policy\")"));
        assert!(
            type_roundtrip_output
                .contains("retry_policy = retry_policy_from_proto(proto.retry_policy)")
        );
        assert!(type_roundtrip_output.contains("request: temporalio.common.RetryPolicy"));
        assert!(type_roundtrip_output.contains("retry_policy_to_proto(request)"));
        assert!(type_roundtrip_output.contains("async def activity_options_operation("));
        assert!(type_roundtrip_output.contains("task_queue: str | None = None,"));
        assert!(type_roundtrip_output.contains("retry_policy: temporalio.common.RetryPolicy,"));
        assert!(type_roundtrip_output.contains("request = ActivityOptions("));
    }

    #[test]
    fn renders_experimental_annotations() {
        let wit = r#"
package temporal:nexus@1.0.0;

world system {
  export example-service;
}

/// @nexus.endpoint "example"
/// @nexus.experimental
interface example-service {
  /// @nexus.experimental
  record request {
    id: string,
  }

  /// @nexus.experimental
  record response {
    ok: bool,
  }

  /// @nexus.experimental
  /// @nexus.doc "Runs example."
  request-op: func(request: request) -> response;
}
"#;
        let spec = ApiSpec::parse_for_language(Language::Python, wit, PathBuf::from("inline.wit"))
            .unwrap();
        let descriptors = DescriptorIndex::load(
            &PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("examples/descriptors/temporal_api.bin"),
        )
        .unwrap();
        let output = generate_source(
            Language::Python,
            &spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap();

        assert!(output.contains(
            "class Request:\n    \"\"\"\n    .. warning::\n        This API is experimental and subject to change.\n    \"\"\""
        ));
        assert!(output.contains(
            "class ExampleService:\n    \"\"\"\n    .. warning::\n        This API is experimental and subject to change.\n    \"\"\""
        ));
        assert!(output.contains(
            "    # .. warning:: This API is experimental and subject to change.\n    request_op: Operation"
        ));
        assert!(output.contains(
            "    \"\"\"Runs example.\n\n    .. warning::\n        This API is experimental and subject to change."
        ));
    }

    #[test]
    fn renders_resource_bound_operations_inside_resource_modules() {
        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let spec = ApiSpec::load_for_language_with_inputs(
            Language::Python,
            &example_input_paths(&root, start_workflow_input_path(&root)),
        )
        .unwrap();
        let descriptors =
            DescriptorIndex::load(&root.join("examples/descriptors/temporal_api.bin")).unwrap();
        let output = generate_source(
            Language::Python,
            &spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap();

        assert!(output.contains("### _resources/started_workflow.py"));
        assert!(output.contains("### operations/start_workflow.py"));
        assert!(output.contains("async def cancel_workflow("));
        assert!(output.contains("async def restart_workflow("));
        assert!(output.contains("async def cancel("));
        assert!(output.contains("async def restart_workflow("));
        assert!(!output.contains("should be replaced during package initialization"));
    }

    #[test]
    fn rejects_flattened_api_field_name_conflicts() {
        let wit = r#"
package temporal:nexus@1.0.0;

world system {
  export workflow-service;
}

/// @nexus.endpoint "__temporal_system"
interface workflow-service {
  use nexus:temporal-types/model@1.0.0.{
    duration,
    placeholder,
    signal-function,
    task-queue,
    user-metadata,
    workflow-function,
  };

  /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionRequest"
  record signal-with-start-workflow-request {
    /// @nexus.proto-field "workflow_type"
    workflow: workflow-function,
    workflow-id: string,
    task-queue: task-queue,
    /// @nexus.proto-field "signal_name"
    signal: signal-function,
    /// @nexus.proto-field "request_id"
    static-summary: string,
    user-metadata: option<user-metadata>,
    /// @nexus.source "workflow_namespace"
    namespace: option<string>,
    /// @nexus.omit
    workflow-execution-timeout: placeholder,
    /// @nexus.omit
    workflow-run-timeout: placeholder,
    /// @nexus.omit
    workflow-task-timeout: placeholder,
    /// @nexus.omit
    identity: placeholder,
    /// @nexus.omit
    workflow-id-reuse-policy: placeholder,
    /// @nexus.omit
    workflow-id-conflict-policy: placeholder,
    /// @nexus.omit
    control: placeholder,
    /// @nexus.omit
    retry-policy: placeholder,
    /// @nexus.omit
    cron-schedule: placeholder,
    /// @nexus.omit
    memo: placeholder,
    /// @nexus.omit
    search-attributes: placeholder,
    /// @nexus.omit
    header: placeholder,
    workflow-start-delay: option<duration>,
    /// @nexus.omit
    links: placeholder,
    /// @nexus.omit
    versioning-override: placeholder,
    /// @nexus.omit
    priority: placeholder,
    /// @nexus.omit
    time-skipping-config: placeholder,
  }

  /// @nexus.proto "temporal.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse"
  record signal-with-start-workflow-response {
    run-id: option<string>,
    started: option<bool>,
    /// @nexus.omit
    signal-link: placeholder,
  }

  signal-with-start-workflow-execution: func(
    request: signal-with-start-workflow-request,
  ) -> signal-with-start-workflow-response;
}
"#;

        let temp_dir = unique_temp_dir("flatten-conflict");
        fs::create_dir_all(&temp_dir).unwrap();
        let input_path = temp_dir.join("conflict.wit");
        fs::write(&input_path, wit).unwrap();

        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let spec = ApiSpec::load_for_language_with_inputs(
            Language::Python,
            &example_input_paths(&root, input_path.clone()),
        )
        .unwrap();
        let descriptors =
            DescriptorIndex::load(&root.join("examples/descriptors/temporal_api.bin")).unwrap();
        let error = generate_source(
            Language::Python,
            &spec,
            &descriptors,
            &crate::SupportFiles::default(),
        )
        .unwrap_err();

        fs::remove_dir_all(temp_dir).unwrap();

        assert!(matches!(
            error,
            Error::FlattenedApiFieldConflict {
                type_name,
                field,
                conflicting_field,
            } if type_name == "SignalWithStartWorkflowRequest"
                && field == "static_summary"
                && conflicting_field.contains("static_summary")
        ));
    }

    #[test]
    fn rejects_missing_service_endpoint() {
        let wit = r#"
package temporal:nexus@1.0.0;

world system {
  export example-service;
}

interface example-service {
  use nexus:temporal-types/model@1.0.0.{retry-policy};

  example-operation: func(request: retry-policy) -> retry-policy;
}
"#;
        let spec = ApiSpec::parse_for_language_with_inputs(
            Language::Python,
            wit,
            PathBuf::from("inline.wit"),
            &[linked_inputs_path(&PathBuf::from(env!(
                "CARGO_MANIFEST_DIR"
            )))],
        )
        .unwrap();
        let descriptors = DescriptorIndex::load(
            &PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("examples/descriptors/temporal_api.bin"),
        )
        .unwrap();

        let error = generate_source(
            Language::Python,
            &spec,
            &descriptors,
            &SupportFiles::default(),
        )
        .unwrap_err();
        assert_eq!(
            error.to_string(),
            "service `ExampleService` is missing an endpoint"
        );
    }
}
