etdl_target_python/

lib.rs

//! The `python` [`CodeGenerator`] target.
//!
//! Consumes the exact same validated [`EtlDocument`] + resolved fault-tree
//! probabilities + [`AsyncApiRegistry`] every other target consumes.
//! Nothing here re-parses `.etdl`, re-validates ECEL conditions, or
//! re-evaluates fault trees. Generated Python is a thin, `ctypes`-based
//! binding to the same compiled `etdl-runtime-ffi` C ABI the Java target
//! binds to via `java.lang.foreign` — branch/SLA accounting, retry
//! backoff, and ECEL `matches`/`in` semantics stay in Rust. See
//! `docs/architecture/targets.md`.
//!
//! Building this crate never requires Python — it only emits Python
//! source text. Python (3.9+, for the `list[T]`/`X | None` type-hint
//! syntax generated code uses) is needed only to *run* the generated
//! output, and `etdl-runtime-ffi`'s compiled shared library is needed to
//! run anything that touches `BranchMonitor`/`RetryPolicy`/`Condition`.
//! `ctypes` is Python's standard library — no `pip install` is needed
//! either.

use etdl_compiler::{CodeGenerator, Diagnostic, GeneratedFile};
use etdl_parser::ast::{
    BackoffStrategy as AstBackoffStrategy, ChannelRef, Condition, EtlDocument, EventTree,
    MessageRef, Node,
};
use etdl_parser::asyncapi::AsyncApiRegistry;
use etdl_parser::ecel;
use std::collections::BTreeMap;

pub struct PythonCodeGenerator {
    pub version: String,
    /// Python package for document-specific output. `None` derives it
    /// (snake_case) from `info.domain`.
    pub package: Option<String>,
}

impl Default for PythonCodeGenerator {
    fn default() -> Self {
        Self::new()
    }
}

impl PythonCodeGenerator {
    pub fn new() -> Self {
        PythonCodeGenerator {
            version: env!("CARGO_PKG_VERSION").to_string(),
            package: None,
        }
    }

    fn resolve_package(&self, doc: &EtlDocument) -> String {
        if let Some(ref p) = self.package {
            return p.clone();
        }
        sanitize_package(&doc.info.domain)
    }
}

impl CodeGenerator for PythonCodeGenerator {
    fn target_name(&self) -> &'static str {
        "python"
    }

    fn generate_all(
        &self,
        doc: &EtlDocument,
        fault_tree_probs: &BTreeMap<String, f64>,
        registry: &AsyncApiRegistry,
        stem: &str,
        _diagnostics: &mut Vec<Diagnostic>,
    ) -> Result<Vec<GeneratedFile>, String> {
        let package = self.resolve_package(doc);
        let _ = stem;

        let mut files = runtime_files();
        files.push(GeneratedFile::new(format!("{package}/__init__.py"), String::new()));

        let message_refs = collect_message_refs(doc);
        let mut messages_module = String::new();
        messages_module.push_str("\"\"\"AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY.\n\n");
        messages_module.push_str("Message dataclasses resolved from AsyncAPI schemas (both External and\n");
        messages_module.push_str("Internal References, via `AsyncApiRegistry.resolve_message` upstream in\n");
        messages_module.push_str("the compiler) referenced anywhere in this document.\n\"\"\"\n\n");
        messages_module.push_str("from __future__ import annotations\n\n");
        messages_module.push_str("from dataclasses import dataclass\n\n");
        for (type_name, msg_ref) in &message_refs {
            let value = registry
                .resolve_message(doc, msg_ref)
                .map_err(|e| format!("resolving message for '{type_name}': {e}"))?;
            messages_module.push_str(&generate_message_dataclass(type_name, &value));
        }
        if !message_refs.is_empty() {
            files.push(GeneratedFile::new(
                format!("{package}/messages.py"),
                messages_module,
            ));
        }

        for tree in doc.event_trees.values() {
            let tree_id = tree_id_of(doc, tree);
            let module_name = to_snake_case(&sanitize_ident(tree_id));
            let class_name = format!("{}Handlers", to_pascal_case(&sanitize_ident(tree_id)));
            let handler_methods = collect_handler_methods(tree);
            files.push(GeneratedFile::new(
                format!("{package}/{module_name}_handlers.py"),
                generate_handlers_module(&class_name, &handler_methods, !message_refs.is_empty()),
            ));
        }

        let workflow = generate_workflow_module(doc, fault_tree_probs, !message_refs.is_empty())?;
        files.push(GeneratedFile::new(format!("{package}/workflow.py"), workflow));

        Ok(files)
    }
}

fn tree_id_of<'a>(doc: &'a EtlDocument, tree: &EventTree) -> &'a str {
    doc.event_trees
        .iter()
        .find(|(_, t)| std::ptr::eq(*t, tree))
        .map(|(id, _)| id.as_str())
        .unwrap_or("Tree")
}

// ---------------------------------------------------------------------
// Fixed, document-independent runtime (`etdl/runtime/*.py`)
// ---------------------------------------------------------------------

fn runtime_files() -> Vec<GeneratedFile> {
    vec![
        GeneratedFile::new("etdl/__init__.py", String::new()),
        GeneratedFile::new("etdl/runtime/__init__.py", String::new()),
        GeneratedFile::new("etdl/runtime/errors.py", ERRORS_PY),
        GeneratedFile::new("etdl/runtime/native.py", NATIVE_PY),
        GeneratedFile::new("etdl/runtime/branch_monitor.py", BRANCH_MONITOR_PY),
        GeneratedFile::new("etdl/runtime/retry_policy.py", RETRY_POLICY_PY),
        GeneratedFile::new("etdl/runtime/condition.py", CONDITION_PY),
        GeneratedFile::new("etdl/runtime/publisher.py", PUBLISHER_PY),
    ]
}

const ERRORS_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY."""


class WorkflowError(Exception):
    """Raised for a retryable operation failure. Analogous to the Rust
    target's `etdl_core::WorkflowError` and the Java target's
    `etdl.runtime.WorkflowError`."""
"#;

const PUBLISHER_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY."""

from abc import ABC, abstractmethod
from typing import Any


class Publisher(ABC):
    """The stable API boundary generated orchestration code calls into for
    a `consequence: send` (core spec Section 5.5.3). A developer supplies a
    concrete subclass (wiring it to a real broker/client and handling
    serialization however fits their stack); generated code never depends
    on a specific transport or JSON library. Application integration, not
    ETDL runtime machinery — unlike `BranchMonitor`/`RetryPolicy`/
    `Condition`, there is no native binding here: there is no shared
    "publish" semantics in the Rust runtime to delegate to.
    """

    @abstractmethod
    def publish(self, channel: str, payload: Any) -> None:
        raise NotImplementedError
"#;

/// The native-binding layer: loads `libetdl_runtime_ffi` via `ctypes` and
/// declares every function from `etdl-runtime-ffi/src/lib.rs` with its
/// exact `argtypes`/`restype`. `BranchMonitor`/`RetryPolicy`/`Condition`
/// are thin facades over this; nothing here re-implements ETDL semantics.
const NATIVE_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY.

Binds to `libetdl_runtime_ffi` (built from the `etdl-runtime-ffi` Rust
crate — see `docs/architecture/targets.md`) via the standard-library
`ctypes` module. No third-party package is required to run generated
Python — only a built copy of the native library.

Library resolution order:
1. environment variable `ETDL_RUNTIME_LIBRARY` — an explicit absolute path
   to the shared library (`libetdl_runtime_ffi.so`/`.dylib`,
   `etdl_runtime_ffi.dll`).
2. `ctypes.CDLL(<platform default name>)` — the platform's standard
   shared-library search path (`LD_LIBRARY_PATH` on Linux, `DYLD_LIBRARY_PATH`
   on macOS, `PATH` on Windows).
"""

import ctypes
import os
import sys

RETRY_CALLBACK_TYPE = ctypes.CFUNCTYPE(ctypes.c_int32, ctypes.c_void_p, ctypes.c_uint32)

# Mirrors etdl-runtime-ffi's ETDL_OK / ETDL_ERR_* / ETDL_RETRY_* constants exactly.
OK = 0
ERR_NULL_HANDLE = -1
ERR_INVALID_ARG = -2
RETRY_OK = 0
RETRY_EXHAUSTED = 1
RETRY_FATAL = 2


def _default_library_name() -> str:
    if sys.platform == "darwin":
        return "libetdl_runtime_ffi.dylib"
    if sys.platform == "win32":
        return "etdl_runtime_ffi.dll"
    return "libetdl_runtime_ffi.so"


def _load() -> ctypes.CDLL:
    explicit = os.environ.get("ETDL_RUNTIME_LIBRARY")
    name = _default_library_name()
    if explicit:
        return ctypes.CDLL(explicit)
    try:
        return ctypes.CDLL(name)
    except OSError as e:
        raise OSError(
            f"could not load the ETDL Rust runtime ('{name}'). Build it with "
            "`cargo build -p etdl-runtime-ffi --release` and either put its output "
            "directory on your platform's shared-library search path, or set "
            f"ETDL_RUNTIME_LIBRARY=<absolute path to {name}> directly. "
            f"Original error: {e}"
        ) from e


def _configure(lib: ctypes.CDLL) -> None:
    lib.etdl_runtime_abi_version.restype = ctypes.c_uint32
    lib.etdl_runtime_abi_version.argtypes = []

    lib.etdl_last_error_message.restype = ctypes.c_void_p
    lib.etdl_last_error_message.argtypes = []

    lib.etdl_string_free.restype = None
    lib.etdl_string_free.argtypes = [ctypes.c_void_p]

    lib.etdl_branch_monitor_new.restype = ctypes.c_void_p
    lib.etdl_branch_monitor_new.argtypes = [ctypes.c_char_p]

    lib.etdl_branch_monitor_free.restype = None
    lib.etdl_branch_monitor_free.argtypes = [ctypes.c_void_p]

    lib.etdl_branch_monitor_record_branch.restype = ctypes.c_int32
    lib.etdl_branch_monitor_record_branch.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_double,
    ]

    lib.etdl_branch_monitor_record_success.restype = ctypes.c_int32
    lib.etdl_branch_monitor_record_success.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_double,
        ctypes.c_bool,
    ]

    lib.etdl_branch_monitor_record_failure.restype = ctypes.c_int32
    lib.etdl_branch_monitor_record_failure.argtypes = [
        ctypes.c_void_p,
        ctypes.c_char_p,
        ctypes.c_char_p,
        ctypes.c_double,
        ctypes.c_bool,
    ]

    lib.etdl_retry_policy_new.restype = ctypes.c_void_p
    lib.etdl_retry_policy_new.argtypes = [ctypes.c_uint32, ctypes.c_uint64, ctypes.c_int32]

    lib.etdl_retry_policy_free.restype = None
    lib.etdl_retry_policy_free.argtypes = [ctypes.c_void_p]

    lib.etdl_retry_policy_delay_ms.restype = ctypes.c_uint64
    lib.etdl_retry_policy_delay_ms.argtypes = [ctypes.c_void_p, ctypes.c_uint32]

    lib.etdl_retry_policy_execute.restype = ctypes.c_int32
    lib.etdl_retry_policy_execute.argtypes = [
        ctypes.c_void_p,
        RETRY_CALLBACK_TYPE,
        ctypes.c_void_p,
        ctypes.POINTER(ctypes.c_uint32),
    ]

    lib.etdl_condition_matches.restype = ctypes.c_int32
    lib.etdl_condition_matches.argtypes = [ctypes.c_char_p, ctypes.c_char_p]

    lib.etdl_condition_contains.restype = ctypes.c_int32
    lib.etdl_condition_contains.argtypes = [ctypes.c_char_p, ctypes.c_char_p]


lib = _load()
_configure(lib)


def last_error() -> str:
    """Reads and frees the calling thread's last native error message."""
    ptr = lib.etdl_last_error_message()
    if not ptr:
        return "(no further detail available from the native runtime)"
    message = ctypes.cast(ptr, ctypes.c_char_p).value.decode("utf-8", errors="replace")
    lib.etdl_string_free(ptr)
    return message
"#;

/// The Python analog of `etdl_core::BranchMonitor` — not a
/// reimplementation: every method is a direct call into the same
/// `BranchMonitor` the Rust target's generated code has always used, via
/// `etdl-runtime-ffi`.
const BRANCH_MONITOR_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY.

Thin facade over the native `etdl_core::BranchMonitor` (via
`etdl-runtime-ffi`). Owns one native handle; always use as a context
manager (generated orchestration code always does) so
`etdl_branch_monitor_free` runs deterministically.
"""

from __future__ import annotations

from . import native


class BranchMonitor:
    def __init__(self, node_id: str) -> None:
        self._handle = native.lib.etdl_branch_monitor_new(node_id.encode("utf-8"))
        if not self._handle:
            raise RuntimeError(f"etdl_branch_monitor_new failed: {native.last_error()}")
        self._closed = False

    def record_branch(self, outcome: str, probability: float) -> None:
        rc = native.lib.etdl_branch_monitor_record_branch(
            self._handle, outcome.encode("utf-8"), probability
        )
        self._check(rc, "etdl_branch_monitor_record_branch")

    def record_success(self, operation_id: str, probability: float | None) -> None:
        rc = native.lib.etdl_branch_monitor_record_success(
            self._handle,
            operation_id.encode("utf-8"),
            probability if probability is not None else 0.0,
            probability is not None,
        )
        self._check(rc, "etdl_branch_monitor_record_success")

    def record_failure(
        self, operation_id: str, error: BaseException, probability: float | None
    ) -> None:
        rc = native.lib.etdl_branch_monitor_record_failure(
            self._handle,
            operation_id.encode("utf-8"),
            str(error).encode("utf-8"),
            probability if probability is not None else 0.0,
            probability is not None,
        )
        self._check(rc, "etdl_branch_monitor_record_failure")

    def _check(self, rc: int, call: str) -> None:
        if rc != native.OK:
            raise RuntimeError(f"{call} failed (code {rc}): {native.last_error()}")

    def close(self) -> None:
        if not self._closed:
            self._closed = True
            native.lib.etdl_branch_monitor_free(self._handle)

    def __enter__(self) -> "BranchMonitor":
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.close()
"#;

/// The Python analog of `etdl_core::retry::RetryPolicy` — the
/// attempt-count/backoff *sequence* is computed once, in Rust
/// (`etdl_retry_policy_execute`'s native loop), via a `ctypes.CFUNCTYPE`
/// callback bound to `op`. Per-attempt timeout enforcement stays here (see
/// `docs/architecture/targets.md`'s "Ownership model": no FFI boundary can
/// safely cancel foreign code mid-flight) using a one-shot thread pool —
/// note Python threads cannot be forcibly killed, so a "cancelled" timed
/// out call's thread still runs to completion in the background; only the
/// *retry loop* moves on.
const RETRY_POLICY_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY.

Thin facade over the native `etdl_core::RetryPolicy` (via
`etdl-runtime-ffi`'s callback-driven `etdl_retry_policy_execute`). Owns one
native handle; always use as a context manager.
"""

from __future__ import annotations

import ctypes
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
from typing import Any, Callable

from . import native
from .errors import WorkflowError


class BackoffStrategy:
    FIXED = 0
    EXPONENTIAL = 1


class RetryPolicy:
    def __init__(self, max_attempts: int, backoff_ms: int, strategy: int) -> None:
        self._handle = native.lib.etdl_retry_policy_new(max_attempts, backoff_ms, strategy)
        if not self._handle:
            raise RuntimeError(f"etdl_retry_policy_new failed: {native.last_error()}")
        self._closed = False

    def execute(self, op: Callable[[], Any], timeout_s: float | None = None) -> Any:
        """`op` is a zero-argument callable. May raise `WorkflowError`
        (retryable) or any other exception (fatal, not retried) — either
        way, the exception is captured here and never crosses the native
        call (ctypes callbacks cannot safely propagate a Python exception
        into C); it is re-raised by this method afterward instead."""
        result_box: dict[str, Any] = {}

        def trampoline(_user_data: int, _attempt: int) -> int:
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(op)
                try:
                    result_box["value"] = future.result(timeout=timeout_s)
                    return 0
                except FutureTimeoutError:
                    result_box["checked_error"] = WorkflowError(
                        f"operation timed out after {timeout_s}s"
                    )
                    return 1
                except WorkflowError as e:
                    result_box["checked_error"] = e
                    return 1
                except BaseException as e:  # noqa: BLE001 - must never cross the FFI boundary
                    result_box["unchecked_error"] = e
                    return -1

        callback = native.RETRY_CALLBACK_TYPE(trampoline)
        attempts_used = ctypes.c_uint32(0)
        rc = native.lib.etdl_retry_policy_execute(
            self._handle, callback, None, ctypes.byref(attempts_used)
        )

        if rc == native.RETRY_OK:
            return result_box.get("value")
        if "unchecked_error" in result_box:
            raise result_box["unchecked_error"]
        if "checked_error" in result_box:
            raise result_box["checked_error"]
        kind = "retry failed fatally" if rc == native.RETRY_FATAL else "retry exhausted"
        raise WorkflowError(f"{kind}: {native.last_error()}")

    def close(self) -> None:
        if not self._closed:
            self._closed = True
            native.lib.etdl_retry_policy_free(self._handle)

    def __enter__(self) -> "RetryPolicy":
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.close()
"#;

/// The Python analog of `etdl_core::condition::{contains, matches}` — ECEL
/// `in`/`matches` semantics stay in Rust; this facade only marshals values
/// to/from the JSON text `etdl_condition_contains` expects, using Python's
/// standard-library `json` module (unlike the Java target, no
/// dependency-free hand-rolled encoder is needed here).
const CONDITION_PY: &str = r#""""Read-only ETDL runtime. DO NOT EDIT DIRECTLY.

ECEL (core spec Section 6) `in`/`matches` operators lower to these
helpers, delegating to the native `etdl_core::condition` module (via
`etdl-runtime-ffi`) — so a `matches` regex or an `in` membership check
behaves byte-for-byte identically no matter which target (Rust, Java,
Python, Go, .NET) evaluates it.
"""

from __future__ import annotations

import json
from typing import Any, Iterable

from . import native


def matches(value: str | None, pattern: str) -> bool:
    rc = native.lib.etdl_condition_matches(
        (value or "").encode("utf-8"), pattern.encode("utf-8")
    )
    if rc < 0:
        raise ValueError(f"etdl_condition_matches: {native.last_error()}")
    return rc == 1


def contains(haystack: Iterable[Any] | None, needle: Any) -> bool:
    needle_json = json.dumps(needle).encode("utf-8")
    haystack_json = json.dumps(list(haystack) if haystack is not None else []).encode("utf-8")
    rc = native.lib.etdl_condition_contains(needle_json, haystack_json)
    if rc < 0:
        raise ValueError(f"etdl_condition_contains: {native.last_error()}")
    return rc == 1
"#;

// ---------------------------------------------------------------------
// Message dataclasses
// ---------------------------------------------------------------------

fn collect_message_refs(doc: &EtlDocument) -> BTreeMap<String, MessageRef> {
    let mut refs: BTreeMap<String, MessageRef> = BTreeMap::new();
    let note = |m: &MessageRef, refs: &mut BTreeMap<String, MessageRef>| {
        let name = ref_to_type_name(m);
        refs.entry(name).or_insert_with(|| m.clone());
    };

    for tree in doc.event_trees.values() {
        note(&tree.initiating_event.message, &mut refs);
        for node in tree.nodes.values() {
            match node {
                Node::Operation(op) => {
                    if let Some(ref m) = op.emits {
                        note(m, &mut refs);
                    }
                }
                Node::Consequence(cons) => {
                    if let Some(ref m) = cons.message {
                        note(m, &mut refs);
                    }
                }
                _ => {}
            }
        }
    }

    if let Some(ref fault_trees) = doc.fault_trees {
        for ft in fault_trees.values() {
            if let Some(ref m) = ft.top_event.message {
                note(m, &mut refs);
            }
            for be in ft.basic_events.values() {
                if let Some(ref m) = be.message {
                    note(m, &mut refs);
                }
            }
        }
    }

    refs
}

fn ref_to_type_name(msg_ref: &MessageRef) -> String {
    let pointer = match msg_ref {
        MessageRef::External(r) => &r.pointer,
        MessageRef::Internal(r) => &r.pointer,
    };
    extract_last_segment(pointer)
}

fn extract_last_segment(pointer: &str) -> String {
    let last = pointer.rsplit('/').next().unwrap_or("Unknown");
    to_pascal_case(last)
}

fn generate_message_dataclass(type_name: &str, message: &serde_json::Value) -> String {
    let payload_type_name = format!("{type_name}Payload");
    let payload_schema = message.get("payload").cloned().unwrap_or(serde_json::Value::Null);

    let mut nested = String::new();
    let payload_py_type = schema_to_python_type(&payload_type_name, &payload_schema, &mut nested, false);

    let mut out = String::new();
    out.push_str(&nested);
    out.push_str("@dataclass\n");
    out.push_str(&format!("class {type_name}:\n"));
    out.push_str(&format!("    payload: {payload_py_type}\n"));
    out.push_str("    headers: dict[str, object] | None = None\n\n\n");
    out
}

/// Python analog of the Rust/Java target's struct/record generators, but
/// with one Python-specific constraint neither of those has: a
/// `@dataclass`'s fields without a default value must all precede fields
/// *with* a default, so (unlike the schema's own property order) required
/// fields are always emitted before optional ones.
fn generate_dataclass_from_schema(name: &str, schema: &serde_json::Value, output: &mut String) {
    let Some(properties) = schema.get("properties").and_then(|p| p.as_object()) else {
        return;
    };

    let required: std::collections::BTreeSet<&str> = schema
        .get("required")
        .and_then(|r| r.as_array())
        .map(|arr| arr.iter().filter_map(|v| v.as_str()).collect())
        .unwrap_or_default();

    let mut nested = String::new();
    let mut required_fields = Vec::new();
    let mut optional_fields = Vec::new();

    for (field_name, field_schema) in properties {
        let snake = to_snake_case(field_name);
        let nested_type_name = format!("{name}{}", to_pascal_case(field_name));
        let is_required = required.contains(field_name.as_str());
        let py_type = schema_to_python_type(&nested_type_name, field_schema, &mut nested, !is_required);
        if is_required {
            required_fields.push(format!("    {snake}: {py_type}\n"));
        } else {
            optional_fields.push(format!("    {snake}: {py_type} = None\n"));
        }
    }

    output.push_str(&nested);
    output.push_str("@dataclass\n");
    output.push_str(&format!("class {name}:\n"));
    if required_fields.is_empty() && optional_fields.is_empty() {
        output.push_str("    pass\n");
    } else {
        for f in &required_fields {
            output.push_str(f);
        }
        for f in &optional_fields {
            output.push_str(f);
        }
    }
    output.push_str("\n\n");
}

fn schema_to_python_type(
    candidate_name: &str,
    schema: &serde_json::Value,
    nested: &mut String,
    nullable: bool,
) -> String {
    let base = match schema.get("type").and_then(|t| t.as_str()) {
        Some("string") => "str".to_string(),
        Some("integer") => "int".to_string(),
        Some("number") => "float".to_string(),
        Some("boolean") => "bool".to_string(),
        Some("array") => {
            let item_type = match schema.get("items") {
                Some(items_schema) => schema_to_python_type(
                    &format!("{candidate_name}Item"),
                    items_schema,
                    nested,
                    true,
                ),
                None => "object".to_string(),
            };
            format!("list[{item_type}]")
        }
        Some("object") => {
            if schema.get("properties").is_some() {
                generate_dataclass_from_schema(candidate_name, schema, nested);
                candidate_name.to_string()
            } else {
                "dict[str, object]".to_string()
            }
        }
        _ => "object".to_string(),
    };
    if nullable {
        format!("{base} | None")
    } else {
        base
    }
}

// ---------------------------------------------------------------------
// Handler ABCs (one per event tree — the user-editable boundary)
// ---------------------------------------------------------------------

fn collect_handler_methods(tree: &EventTree) -> Vec<String> {
    let mut seen = std::collections::BTreeSet::new();
    let mut methods = Vec::new();
    for node in tree.nodes.values() {
        if let Node::Operation(op) = node {
            let name = to_snake_case(&op.handler);
            if seen.insert(name.clone()) {
                methods.push(name);
            }
        }
    }
    methods
}

fn generate_handlers_module(class_name: &str, methods: &[String], has_messages: bool) -> String {
    let mut out = String::new();
    out.push_str("\"\"\"AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY.\n\n");
    out.push_str("Implement this class in your own (never regenerated) module; the generated\n");
    out.push_str("orchestration in workflow.py calls through it and never the other way\n");
    out.push_str("around.\n\"\"\"\n\n");
    out.push_str("from __future__ import annotations\n\n");
    out.push_str("from abc import ABC, abstractmethod\n");
    out.push_str("from typing import Any\n");
    if has_messages {
        out.push_str("\nfrom .messages import *  # noqa: F401,F403 - message types used in signatures below\n");
    }
    out.push_str(&format!("\n\nclass {class_name}(ABC):\n"));
    if methods.is_empty() {
        out.push_str("    pass\n");
    }
    for method in methods {
        out.push_str("    @abstractmethod\n");
        out.push_str(&format!("    def {method}(self, message: Any) -> Any:\n"));
        out.push_str("        raise NotImplementedError\n\n");
    }
    out
}

// ---------------------------------------------------------------------
// Workflow orchestration module
// ---------------------------------------------------------------------

fn generate_workflow_module(
    doc: &EtlDocument,
    fault_tree_probs: &BTreeMap<String, f64>,
    has_messages: bool,
) -> Result<String, String> {
    let mut out = String::new();
    out.push_str("\"\"\"AUTOGENERATED BY ETDL COMPILER - DO NOT EDIT DIRECTLY.\"\"\"\n\n");
    out.push_str("from __future__ import annotations\n\n");
    out.push_str("from etdl.runtime.branch_monitor import BranchMonitor\n");
    out.push_str("from etdl.runtime.condition import contains, matches\n");
    out.push_str("from etdl.runtime.errors import WorkflowError\n");
    out.push_str("from etdl.runtime.publisher import Publisher\n");
    out.push_str("from etdl.runtime.retry_policy import BackoffStrategy, RetryPolicy\n");
    if has_messages {
        out.push_str("\nfrom .messages import *  # noqa: F401,F403\n");
    }
    for tree in doc.event_trees.values() {
        let tree_id = tree_id_of(doc, tree);
        let module_name = to_snake_case(&sanitize_ident(tree_id));
        let class_name = format!("{}Handlers", to_pascal_case(&sanitize_ident(tree_id)));
        out.push_str(&format!("from .{module_name}_handlers import {class_name}\n"));
    }
    out.push('\n');

    out.push_str(&generate_fault_tree_constants(doc, fault_tree_probs));

    for tree in doc.event_trees.values() {
        let method = generate_event_tree_function(doc, tree, fault_tree_probs)?;
        out.push_str(&method);
        out.push('\n');
    }

    Ok(out)
}

fn generate_fault_tree_constants(doc: &EtlDocument, fault_tree_probs: &BTreeMap<String, f64>) -> String {
    let mut output = String::new();
    for tree in doc.event_trees.values() {
        for (node_id, node) in &tree.nodes {
            if let Node::Operation(op) = node {
                if let Some(ref ps) = op.on_failure_probability_source {
                    if let Some((ft_id, prob)) = find_fault_tree_prob(ps, fault_tree_probs) {
                        output.push_str(&format!(
                            "# Computed from faultTrees.{ft_id}.topEvent at build time (Section 5.16)\n"
                        ));
                        let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
                        output.push_str(&format!("{const_name}: float = {prob:.6}\n\n"));
                    }
                }
            }
        }
    }
    output
}

fn find_fault_tree_prob(
    ps: &etdl_parser::ast::InternalRef,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Option<(String, f64)> {
    let ft_id = extract_ft_id(&ps.pointer);
    fault_tree_probs.get(&ft_id).map(|&v| (ft_id.clone(), v))
}

fn extract_ft_id(pointer: &str) -> String {
    pointer
        .trim_start_matches("#/faultTrees/")
        .trim_end_matches("/topEvent")
        .to_string()
}

fn generate_event_tree_function(
    doc: &EtlDocument,
    tree: &EventTree,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Result<String, String> {
    let mut out = String::new();

    let tree_id = tree_id_of(doc, tree);
    let class_name = format!("{}Handlers", to_pascal_case(&sanitize_ident(tree_id)));
    let function_name = format!("handle_{}", to_snake_case(&sanitize_ident(&tree.initiating_event.id)));

    out.push_str(&format!(
        "def {function_name}(message, publisher: Publisher, handlers: {class_name}) -> None:\n"
    ));

    let first_barrier = find_first_barrier(tree);
    let monitor_name = first_barrier.map(to_snake_case).unwrap_or_else(|| "monitor".to_string());
    out.push_str(&format!(
        "    with BranchMonitor(\"{}\") as {monitor_name}:\n",
        first_barrier.unwrap_or("root")
    ));

    let start_node_id = &tree.initiating_event.next;
    let body = generate_node_code(doc, tree, start_node_id, 2, fault_tree_probs, &monitor_name)?;
    out.push_str(&body);

    Ok(out)
}

fn find_first_barrier(tree: &EventTree) -> Option<&str> {
    let mut current = &tree.initiating_event.next;
    loop {
        match tree.nodes.get(current.as_str()) {
            Some(Node::Barrier(_)) => return Some(current.as_str()),
            Some(Node::Operation(op)) => current = &op.next,
            Some(Node::Consequence(_)) => return None,
            None => return None,
        }
    }
}

fn generate_node_code(
    doc: &EtlDocument,
    tree: &EventTree,
    node_id: &str,
    depth: usize,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let node = tree
        .nodes
        .get(node_id)
        .ok_or_else(|| format!("node '{node_id}' not found"))?;

    match node {
        Node::Barrier(barrier) => {
            generate_barrier_code(tree, node_id, barrier, depth, doc, fault_tree_probs, monitor_name)
        }
        Node::Operation(op) => {
            generate_operation_code(tree, node_id, op, depth, doc, fault_tree_probs, monitor_name)
        }
        Node::Consequence(cons) => generate_consequence_code(cons, depth),
    }
}

fn generate_barrier_code(
    tree: &EventTree,
    node_id: &str,
    barrier: &etdl_parser::ast::Barrier,
    depth: usize,
    doc: &EtlDocument,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    for (i, branch) in barrier.branches.iter().enumerate() {
        let is_default = branch.condition == Condition::Default;
        let keyword = if i == 0 {
            "if"
        } else if is_default {
            "else"
        } else {
            "elif"
        };

        if is_default && i > 0 {
            output.push_str(&format!("{indent}else:\n"));
        } else {
            let cond = condition_to_python_code(&branch.condition);
            output.push_str(&format!("{indent}{keyword} {cond}:\n"));
        }

        emit_branch_record(&format!("{indent}    "), monitor_name, branch, node_id, fault_tree_probs, &mut output);
        let body = generate_node_code(doc, tree, &branch.next, depth + 1, fault_tree_probs, monitor_name)?;
        output.push_str(&body);
    }

    Ok(output)
}

fn emit_branch_record(
    indent: &str,
    monitor_name: &str,
    branch: &etdl_parser::ast::Branch,
    node_id: &str,
    fault_tree_probs: &BTreeMap<String, f64>,
    output: &mut String,
) {
    if let Some(p) = get_branch_prob(branch, node_id, fault_tree_probs) {
        output.push_str(&format!(
            "{indent}{monitor_name}.record_branch(\"{}\", {p:.6})\n",
            branch.outcome
        ));
    } else {
        output.push_str(&format!("{indent}pass\n"));
    }
}

fn get_branch_prob(
    branch: &etdl_parser::ast::Branch,
    _node_id: &str,
    fault_tree_probs: &BTreeMap<String, f64>,
) -> Option<f64> {
    if let Some(ref ps) = branch.probability_source {
        let ft_id = extract_ft_id(&ps.pointer);
        return fault_tree_probs.get(&ft_id).copied();
    }
    branch.effective_probability()
}

fn generate_operation_code(
    tree: &EventTree,
    node_id: &str,
    op: &etdl_parser::ast::Operation,
    depth: usize,
    doc: &EtlDocument,
    fault_tree_probs: &BTreeMap<String, f64>,
    monitor_name: &str,
) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    let handler_name = to_snake_case(&op.handler);
    let timeout_s = op.timeout_ms.unwrap_or(5000) as f64 / 1000.0;

    output.push_str(&format!("{indent}try:\n"));
    if let Some(ref retry) = op.retry_policy {
        let strategy = match retry.backoff_strategy.as_ref().unwrap_or(&AstBackoffStrategy::Fixed) {
            AstBackoffStrategy::Exponential => "BackoffStrategy.EXPONENTIAL",
            AstBackoffStrategy::Fixed => "BackoffStrategy.FIXED",
        };
        output.push_str(&format!(
            "{indent}    with RetryPolicy({}, {}, {strategy}) as retry:\n",
            retry.max_attempts, retry.backoff_ms
        ));
        output.push_str(&format!(
            "{indent}        _result = retry.execute(lambda: handlers.{handler_name}(message), timeout_s={timeout_s})\n"
        ));
    } else {
        output.push_str(&format!(
            "{indent}    _result = handlers.{handler_name}(message)\n"
        ));
    }

    if op.on_failure.is_some() {
        if let Some(ref ps) = op.on_failure_probability_source {
            let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
            let prob_exists = find_fault_tree_prob(ps, fault_tree_probs).is_some();
            let indent2 = "    ".repeat(depth + 1 + usize::from(op.retry_policy.is_some()));
            if prob_exists {
                output.push_str(&format!(
                    "{indent2}{monitor_name}.record_success(\"{node_id}\", {const_name})\n"
                ));
            } else {
                output.push_str(&format!(
                    "{indent2}{monitor_name}.record_success(\"{node_id}\", None)\n"
                ));
            }
        }
    }

    let body_indent = depth + 1 + usize::from(op.retry_policy.is_some());
    let next_node = tree.nodes.get(&op.next);
    match next_node {
        Some(Node::Consequence(cons)) => {
            emit_send(cons, "_result", &"    ".repeat(body_indent), &mut output);
        }
        Some(_) => {
            let body = generate_node_code(doc, tree, &op.next, body_indent, fault_tree_probs, monitor_name)?;
            output.push_str(&body);
        }
        None => {
            output.push_str(&format!("{}pass\n", "    ".repeat(body_indent)));
        }
    }

    output.push_str(&format!("{indent}except WorkflowError as err:\n"));

    if let Some(ref ps) = op.on_failure_probability_source {
        let const_name = to_upper_snake(&format!("{node_id}_failure_probability"));
        let prob_exists = find_fault_tree_prob(ps, fault_tree_probs).is_some();
        if prob_exists {
            output.push_str(&format!(
                "{indent}    {monitor_name}.record_failure(\"{node_id}\", err, {const_name})\n"
            ));
        } else {
            output.push_str(&format!(
                "{indent}    {monitor_name}.record_failure(\"{node_id}\", err, None)\n"
            ));
        }
    } else {
        output.push_str(&format!(
            "{indent}    {monitor_name}.record_failure(\"{node_id}\", err, None)\n"
        ));
    }

    if let Some(ref on_failure_id) = op.on_failure {
        match tree.nodes.get(on_failure_id) {
            Some(Node::Consequence(cons)) => {
                emit_send(cons, "message", &format!("{indent}    "), &mut output);
            }
            _ => {
                let body = generate_node_code(doc, tree, on_failure_id, depth + 1, fault_tree_probs, monitor_name)?;
                output.push_str(&body);
            }
        }
    } else {
        output.push_str(&format!("{indent}    raise\n"));
    }

    Ok(output)
}

fn emit_send(cons: &etdl_parser::ast::Consequence, payload_expr: &str, indent: &str, output: &mut String) {
    if let etdl_parser::ast::ConsequenceOperation::Send = cons.consequence_operation {
        if let Some(ref channel_ref) = cons.channel {
            let channel_name = channel_ref_name(channel_ref);
            output.push_str(&format!(
                "{indent}publisher.publish(\"{channel_name}\", {payload_expr})\n"
            ));
            return;
        }
    }
    output.push_str(&format!("{indent}pass\n"));
}

fn generate_consequence_code(cons: &etdl_parser::ast::Consequence, depth: usize) -> Result<String, String> {
    let indent = "    ".repeat(depth);
    let mut output = String::new();

    if let etdl_parser::ast::ConsequenceOperation::Send = cons.consequence_operation {
        if let Some(ref channel_ref) = cons.channel {
            let channel_name = channel_ref_name(channel_ref);
            output.push_str(&format!("{indent}publisher.publish(\"{channel_name}\", message)\n"));
            return Ok(output);
        }
    }
    output.push_str(&format!("{indent}pass\n"));
    Ok(output)
}

fn channel_ref_name(channel_ref: &ChannelRef) -> String {
    match channel_ref {
        ChannelRef::External(ext_ref) => extract_last_segment(&ext_ref.pointer),
        ChannelRef::Bare(name) => name.clone(),
    }
}

// ---------------------------------------------------------------------
// ECEL condition -> Python
// ---------------------------------------------------------------------

fn condition_to_python_code(condition: &Condition) -> String {
    match condition {
        Condition::Default => "True".to_string(),
        Condition::Expr(expr) => render_bool_expr(expr),
    }
}

/// `boolean-expr` (spec §6.2): `and`/`or`/`not` are keywords in Python, not
/// symbolic operators like `&&`/`||`/`!` — everything else about combining
/// comparisons is unchanged.
fn render_bool_expr(expr: &ecel::BoolExpr) -> String {
    use ecel::BoolExpr as B;
    match expr {
        B::And(a, b) => format!("({}) and ({})", render_bool_expr(a), render_bool_expr(b)),
        B::Or(a, b) => format!("({}) or ({})", render_bool_expr(a), render_bool_expr(b)),
        B::Not(a) => format!("not ({})", render_bool_expr(a)),
        B::Comparison(cmp) => render_comparison(cmp),
        B::Quantifier(q) => render_quantifier(q),
        B::Defined(path) => render_defined(path),
    }
}

fn render_comparison(cmp: &ecel::Comparison) -> String {
    use ecel::Comparator as C;
    match cmp.op {
        C::In => {
            let left = path_or_literal(&cmp.left);
            let right = array_or_literal(&cmp.right);
            format!("contains({right}, {left})")
        }
        C::Matches => {
            let left = operand_to_path_expr_or_literal(&cmp.left);
            let right = literal_to_val_str(&literal_of(&cmp.right));
            format!("matches({left}, {right})")
        }
        _ => {
            let (left_path, has_wildcard) = match as_path(&cmp.left) {
                Some(p) => build_path_expression(p),
                None => (PathParts { path_prefix: String::new(), remaining_path: String::new() }, false),
            };
            let right = operand_to_val_str(&cmp.right);
            let op = comparator_str(&cmp.op);

            if has_wildcard {
                format!(
                    "all(item{} {} {} for item in {})",
                    left_path.remaining_path, op, right, left_path.path_prefix
                )
            } else {
                let l = if left_path.path_prefix.is_empty() && left_path.remaining_path.is_empty() {
                    operand_to_val_str(&cmp.left)
                } else {
                    left_path.path_prefix + &left_path.remaining_path
                };
                let r = if right.is_empty() { operand_to_path_expr(&cmp.right) } else { right };
                format!("{l} {op} {r}")
            }
        }
    }
}

/// `quantifier-expr` (spec §6.4): `any`/`all` over a wildcarded path's
/// matching elements — Python's own `any(... for ...)`/`all(... for ...)`
/// generator-expression builtins map directly onto the two
/// `QuantifierKind` values, reusing the same wildcard-split machinery a
/// bare comparison's implicit-`all` case already uses.
fn render_quantifier(q: &ecel::QuantifierExpr) -> String {
    if let Some(path_expr) = as_path(&q.comparison.left) {
        if path_has_wildcard(path_expr) && !is_headers_path(path_expr) {
            let (parts, _has_wildcard) = build_path_expression(path_expr);
            let right = operand_to_val_str(&q.comparison.right);
            let op = comparator_str(&q.comparison.op);
            let builtin = match q.kind {
                ecel::QuantifierKind::Any => "any",
                ecel::QuantifierKind::All => "all",
            };
            return format!(
                "{}(item{} {} {} for item in {})",
                builtin, parts.remaining_path, op, right, parts.path_prefix
            );
        }
    }
    render_comparison(&q.comparison)
}

/// `defined-expr` (spec §6.2): whether a path both exists and is non-null.
/// `headers` is a plain `dict`, so a missing key and an absent field are
/// both `... is not None` after a chain of `.get(...)` lookups; a payload
/// path is a `@dataclass` attribute chain, so only null-ness (`is not
/// None`) is checkable — a required, non-optional field is always
/// "defined" by construction, matching Rust/Java's own treatment.
fn render_defined(path: &ecel::PathExpr) -> String {
    if is_headers_path(path) {
        let mut expr = String::from("message.headers");
        for seg in path.segments.iter().skip(2) {
            let key = match seg {
                ecel::PathSegment::Field(name) | ecel::PathSegment::QuotedKey(name) => {
                    format!("\"{name}\"")
                }
                ecel::PathSegment::Index(idx) => idx.to_string(),
                ecel::PathSegment::Wildcard => continue,
            };
            expr = format!("({expr} or {{}}).get({key})");
        }
        format!("(({expr}) is not None)")
    } else {
        format!("(({}) is not None)", render_path_expr(path))
    }
}

fn path_or_literal(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(_) => operand_to_path_expr(operand),
        None => operand_to_val_str(operand),
    }
}

fn array_or_literal(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(_) => operand_to_path_expr(operand),
        None => operand_to_val_str(operand),
    }
}

fn operand_to_path_expr_or_literal(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(_) => operand_to_path_expr(operand),
        None => operand_to_val_str(operand),
    }
}

fn literal_of(operand: &ecel::Operand) -> ecel::Literal {
    match operand {
        ecel::Operand::Literal(lit) => lit.clone(),
        ecel::Operand::Value(_) => ecel::Literal::String(String::new()),
    }
}

/// `Some(path)` if `operand` is a bare path (spec §6.2's `value-expr ::=
/// path-expr`), independent of whether it's used at the top level of a
/// comparison or nested inside arithmetic/a function-call argument.
fn as_path(operand: &ecel::Operand) -> Option<&ecel::PathExpr> {
    match operand {
        ecel::Operand::Value(ecel::ValueExpr::Path(p)) => Some(p),
        _ => None,
    }
}

fn path_has_wildcard(path: &ecel::PathExpr) -> bool {
    path.segments.iter().any(|s| matches!(s, ecel::PathSegment::Wildcard))
}

/// Whether `path` is rooted at `message.headers` (spec §6.3's second root)
/// rather than `message.payload` — decided purely syntactically, matching
/// `RustCodeGenerator::path_root`'s own discipline.
fn is_headers_path(path: &ecel::PathExpr) -> bool {
    matches!(path.segments.get(1), Some(ecel::PathSegment::Field(name)) if name == "headers")
}

struct PathParts {
    path_prefix: String,
    remaining_path: String,
}

/// Python analog of the Rust/Java targets' path-expression builder.
/// Segments render as plain attribute access (`.field`) — unlike Java's
/// records, a Python `@dataclass`'s fields are ordinary attributes, not
/// accessor methods — and `Index`/`QuotedKey` segments render as `[...]`
/// subscripting, which works directly on both `list` (index) and `dict`
/// (key) without the method-call detour Java's `List`/`Map` fallback
/// types needed.
fn build_path_expression(path_expr: &ecel::PathExpr) -> (PathParts, bool) {
    let segments = &path_expr.segments;
    let mut pre_wildcard = Vec::new();
    let mut post_wildcard = Vec::new();
    let mut has_wildcard = false;

    for (i, seg) in segments.iter().enumerate() {
        if i == 0 {
            continue;
        }
        if has_wildcard {
            post_wildcard.push(seg.clone());
        } else if matches!(seg, ecel::PathSegment::Wildcard) {
            has_wildcard = true;
        } else {
            pre_wildcard.push(seg.clone());
        }
    }

    let mut prefix = String::from("message");
    for seg in &pre_wildcard {
        append_segment(&mut prefix, seg);
    }

    let mut suffix = String::new();
    for seg in &post_wildcard {
        append_segment(&mut suffix, seg);
    }

    (
        PathParts { path_prefix: prefix, remaining_path: suffix },
        has_wildcard,
    )
}

fn append_segment(out: &mut String, seg: &ecel::PathSegment) {
    match seg {
        ecel::PathSegment::Field(name) => {
            out.push('.');
            out.push_str(&to_snake_case(name));
        }
        ecel::PathSegment::Index(idx) => {
            out.push_str(&format!("[{idx}]"));
        }
        ecel::PathSegment::QuotedKey(name) => {
            out.push_str(&format!("[\"{name}\"]"));
        }
        _ => {}
    }
}

/// Core path-segment rendering, shared by the top-level
/// `operand_to_path_expr` (bare `Operand::Value(Path(_))`) and
/// `render_value_expr`'s `Path` arm (a path nested inside an arithmetic
/// expression or function-call argument, e.g.
/// `length(message.payload.name)`).
fn render_path_expr(path: &ecel::PathExpr) -> String {
    let mut out = String::from("message");
    for seg in path.segments.iter().skip(1) {
        if matches!(seg, ecel::PathSegment::Wildcard) {
            continue;
        }
        append_segment(&mut out, seg);
    }
    out
}

fn operand_to_path_expr(operand: &ecel::Operand) -> String {
    match as_path(operand) {
        Some(path) => render_path_expr(path),
        None => String::new(),
    }
}

fn operand_to_val_str(operand: &ecel::Operand) -> String {
    match operand {
        // A bare path renders empty here on purpose: `render_comparison`
        // tries `build_path_expression`/`operand_to_path_expr` first for a
        // path operand (to get wildcard-quantification right) and only
        // falls back to this function for the non-path cases below.
        ecel::Operand::Value(ecel::ValueExpr::Path(_)) => String::new(),
        ecel::Operand::Value(v) => render_value_expr(v),
        ecel::Operand::Literal(lit) => literal_to_val_str(lit),
    }
}

/// Renders a `value-expr` (spec §6.2) as a Python expression. Python's
/// `len()` builtin already handles both `str` and `list` uniformly at
/// runtime — unlike Java (a payload accessor has a concrete static type,
/// making a naive `isinstance` check unnecessary here in the first place;
/// there's no static-type system to conflict with), no dedicated runtime
/// helper is needed for `length()`.
fn render_value_expr(expr: &ecel::ValueExpr) -> String {
    use ecel::{FuncName as F, ValueExpr as V};
    match expr {
        V::Path(path) => render_path_expr(path),
        V::Number(n) => n.to_string(),
        V::Call(func, arg) => {
            let arg_code = render_value_expr(arg);
            match func {
                F::Length => format!("len({arg_code})"),
                F::Abs => format!("abs({arg_code})"),
                F::Lower => format!("({arg_code}).lower()"),
                F::Upper => format!("({arg_code}).upper()"),
            }
        }
        V::Add(a, b) => format!("(({}) + ({}))", render_value_expr(a), render_value_expr(b)),
        V::Sub(a, b) => format!("(({}) - ({}))", render_value_expr(a), render_value_expr(b)),
        V::Mul(a, b) => format!("(({}) * ({}))", render_value_expr(a), render_value_expr(b)),
        V::Div(a, b) => format!("(({}) / ({}))", render_value_expr(a), render_value_expr(b)),
    }
}

fn literal_to_val_str(lit: &ecel::Literal) -> String {
    match lit {
        ecel::Literal::Number(n) => n.to_string(),
        ecel::Literal::String(s) => format!("\"{s}\""),
        ecel::Literal::Bool(b) => if *b { "True".to_string() } else { "False".to_string() },
        ecel::Literal::Null => "None".to_string(),
        ecel::Literal::Array(items) => {
            let inner: Vec<String> = items.iter().map(literal_to_val_str).collect();
            format!("[{}]", inner.join(", "))
        }
    }
}

fn comparator_str(op: &ecel::Comparator) -> &str {
    match op {
        ecel::Comparator::Eq => "==",
        ecel::Comparator::Neq => "!=",
        ecel::Comparator::Gte => ">=",
        ecel::Comparator::Lte => "<=",
        ecel::Comparator::Gt => ">",
        ecel::Comparator::Lt => "<",
        ecel::Comparator::In => "in",
        ecel::Comparator::Matches => "matches",
    }
}

// ---------------------------------------------------------------------
// Naming helpers
// ---------------------------------------------------------------------

fn sanitize_package(domain: &str) -> String {
    let snake = to_snake_case(domain);
    if snake.is_empty() {
        "etdl_generated".to_string()
    } else {
        snake
    }
}

fn sanitize_ident(s: &str) -> String {
    let cleaned: String = s
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect();
    if cleaned.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(true) {
        format!("_{cleaned}")
    } else {
        cleaned
    }
}

fn to_snake_case(s: &str) -> String {
    let mut result = String::new();
    for (i, c) in s.chars().enumerate() {
        if c.is_uppercase() {
            if i > 0 {
                result.push('_');
            }
            result.push(c.to_lowercase().next().unwrap());
        } else if c == '-' || c == ' ' {
            result.push('_');
        } else {
            result.push(c);
        }
    }
    result
}

fn to_upper_snake(s: &str) -> String {
    to_snake_case(s).to_uppercase()
}

fn to_pascal_case(s: &str) -> String {
    let mut result = String::new();
    let mut capitalize = true;
    for c in s.chars() {
        if c == '_' || c == '-' || c == ' ' {
            capitalize = true;
        } else if capitalize {
            result.extend(c.to_uppercase());
            capitalize = false;
        } else {
            result.push(c);
        }
    }
    result
}

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

    #[test]
    fn package_snake_cases_domain() {
        assert_eq!(sanitize_package("FulfillmentContext"), "fulfillment_context");
        assert_eq!(sanitize_package(""), "etdl_generated");
    }

    #[test]
    fn snake_and_pascal_case() {
        assert_eq!(to_snake_case("orderId"), "order_id");
        assert_eq!(to_pascal_case("order_placed"), "OrderPlaced");
    }

    #[test]
    fn in_operator_lowers_to_native_contains_call() {
        let cond = ecel::parse_condition(
            "message.payload.status in [\"PAID\", \"AUTHORIZED\"]",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains("contains("), "got: {code}");
        assert!(code.contains("\"PAID\""));
    }

    #[test]
    fn matches_operator_lowers_to_native_matches_call() {
        let cond = ecel::parse_condition(
            "message.payload.reference matches \"^ORD-[0-9]{8}$\"",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains("matches("), "got: {code}");
    }

    #[test]
    fn comparison_uses_plain_attribute_access() {
        let cond = ecel::parse_condition("message.payload.amount >= 10000").unwrap();
        let code = condition_to_python_code(&cond);
        assert_eq!(code, "message.payload.amount >= 10000");
    }

    #[test]
    fn wildcard_condition_uses_all_comprehension() {
        let cond = ecel::parse_condition("message.payload.items[*].qty > 0").unwrap();
        let code = condition_to_python_code(&cond);
        assert_eq!(code, "all(item.qty > 0 for item in message.payload.items)");
    }

    #[test]
    fn boolean_and_or_not_combine_comparisons() {
        let cond = ecel::parse_condition(
            "message.payload.amount >= 10000 && !(message.payload.status == \"VOID\")",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains(" and "), "got: {code}");
        assert!(code.contains("not ("), "got: {code}");

        let cond = ecel::parse_condition(
            "message.payload.amount >= 10000 || message.payload.priority == true",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains(" or "), "got: {code}");
    }

    #[test]
    fn arithmetic_and_function_calls_render_as_python_expressions() {
        let cond = ecel::parse_condition("message.payload.amount + 5 > 100").unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains("message.payload.amount"), "got: {code}");
        assert!(code.contains('+'), "got: {code}");

        let cond = ecel::parse_condition("abs(message.payload.amount) > 100").unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains("abs("), "got: {code}");

        let cond = ecel::parse_condition("length(message.payload.reference) > 5").unwrap();
        let code = condition_to_python_code(&cond);
        assert_eq!(code, "len(message.payload.reference) > 5");
    }

    #[test]
    fn quantifier_lowers_to_any_or_all_generator_expression() {
        let cond = ecel::parse_condition(
            "any(message.payload.items, message.payload.items[*].qty > 0)",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.starts_with("any("), "got: {code}");

        let cond = ecel::parse_condition(
            "all(message.payload.items, message.payload.items[*].qty > 0)",
        )
        .unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.starts_with("all("), "got: {code}");
    }

    #[test]
    fn defined_checks_payload_and_headers_paths() {
        let cond = ecel::parse_condition("defined(message.payload.reference)").unwrap();
        let code = condition_to_python_code(&cond);
        assert_eq!(code, "((message.payload.reference) is not None)");

        let cond = ecel::parse_condition("defined(message.headers.traceId)").unwrap();
        let code = condition_to_python_code(&cond);
        assert!(code.contains("message.headers"), "got: {code}");
        assert!(code.contains("is not None"), "got: {code}");
    }
}