dora_message/

descriptor.rs

#![warn(missing_docs)]

use crate::{
    config::{ByteSize, Input, NodeRunConfig},
    id::{DataId, NodeId, OperatorId},
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_with_expand_env::with_expand_envs;
use std::{
    collections::{BTreeMap, BTreeSet},
    fmt,
    path::PathBuf,
};

/// Wire framing mode for an output.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum OutputFraming {
    /// Raw Arrow buffer layout (default, current behavior).
    #[default]
    Raw,
    /// Arrow IPC stream format — self-describing, schema + record batches.
    ArrowIpc,
}

/// Source identifier for shell-based nodes.
pub const SHELL_SOURCE: &str = "shell";
/// Set the [`Node::path`] field to this value to treat the node as a
/// [_dynamic node_](https://docs.rs/dora-node-api/latest/dora_node_api/).
pub const DYNAMIC_SOURCE: &str = "dynamic";

/// # Dataflow Specification
///
/// The main configuration structure for defining a Dora dataflow. Dataflows are
/// specified through YAML files that describe the nodes, their connections, and
/// execution parameters.
///
/// ## Structure
///
/// A dataflow consists of:
/// - **Nodes**: The computational units that process data
/// - **Deployment**: Optional deployment configuration (unstable)
/// - **Debug options**: Optional development and debugging settings (unstable)
///
/// ## Example
///
/// ```
/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
/// use dora_message::descriptor::Descriptor;
/// let yaml = r#"
/// nodes:
///   - id: webcam
///     operator:
///       python: webcam.py
///       inputs:
///         tick: dora/timer/millis/100
///       outputs:
///         - image
///   - id: plot
///     operator:
///       python: plot.py
///       inputs:
///         image: webcam/image
/// "#;
/// let descriptor: Descriptor = serde_yaml::from_str(yaml)?;
/// assert_eq!(descriptor.nodes.len(), 2);
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
#[schemars(title = "dora-rs specification")]
// Same rationale as `Node`: keeps a new *dataflow-level* key a minor release.
// This is where `exit_when_nodes_finish`, `health_check_interval`, `type_rules`
// and `strict_types` landed, so it grows at least as often as `Node` does.
// Construct with `Descriptor::new`; the fields remain `pub`.
#[non_exhaustive]
pub struct Descriptor {
    /// List of nodes in the dataflow
    ///
    /// This is the most important field of the dataflow specification.
    /// Each node must be identified by a unique `id`:
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: foo
    ///     path: path/to/the/executable
    ///     # ... (see below)
    ///   - id: bar
    ///     path: path/to/another/executable
    ///     # ... (see below)
    /// ```
    ///
    /// For each node, you need to specify the `path` of the executable or script that Dora should run when starting the node.
    /// Most of the other node fields are optional, but you typically want to specify at least some `inputs` and/or `outputs`.
    pub nodes: Vec<Node>,

    /// Deployment configuration (optional).
    #[schemars(skip)]
    pub deploy: Option<Deploy>,

    /// Debug options (optional).
    #[schemars(skip)]
    #[serde(default)]
    pub debug: Debug,

    /// How often the daemon checks node health (in seconds).
    ///
    /// Defaults to 5.0 seconds if not specified. Lower values detect hung nodes
    /// faster but add more overhead.
    #[serde(default)]
    pub health_check_interval: Option<f64>,

    /// Enable strict type checking: type warnings become errors during build.
    ///
    /// Can also be enabled via `--strict-types` CLI flag on `dora build`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub strict_types: Option<bool>,

    /// Finish the dataflow once every node has, treating
    /// `dora/timer/...` inputs as a clock rather than as work.
    ///
    /// A timer input has no upstream node, so it never closes. By default
    /// a node consuming one is therefore never told its inputs are done
    /// and the graph cannot end on its own, even after every node doing
    /// real work has exited (dora-rs/dora#2920).
    ///
    /// Off by default: for a long-lived dataflow the timer is precisely
    /// what keeps it alive. Nodes with no data inputs at all (timer-only
    /// sources, or no inputs) are unaffected either way -- they have no
    /// dependency that could finish, so they are treated as sources.
    ///
    /// Set by `dora run --exit-when-nodes-finish` and `dora start
    /// --exit-when-nodes-finish`, and settable directly in YAML. It lives
    /// on the descriptor rather than on the wire so that it survives the
    /// events a dataflow outlives: auto-recovery re-spawn, coordinator
    /// restart with state reconstruction, and `dora restart`.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// exit_when_nodes_finish: true
    /// nodes:
    ///   - id: worker
    ///     path: ./worker
    ///     inputs:
    ///       tick: dora/timer/millis/100
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exit_when_nodes_finish: Option<bool>,

    /// Custom type compatibility rules.
    ///
    /// Each rule declares that a source type can be implicitly converted to
    /// a target type. These supplement the built-in widening rules.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// type_rules:
    ///   - from: myproject/SensorV1
    ///     to: myproject/SensorV2
    /// ```
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub type_rules: Vec<TypeRuleDef>,

    /// Global environment variables inherited by every node.
    ///
    /// Each node's own `env` map takes precedence on key conflicts, so nodes
    /// can override a global default without repeating shared values like
    /// `RUST_LOG`, `OTEL_EXPORTER_OTLP_ENDPOINT`, or `CUDA_VISIBLE_DEVICES`.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// env:
    ///   RUST_LOG: info
    ///   OTEL_EXPORTER_OTLP_ENDPOINT: http://collector:4317
    /// nodes:
    ///   - id: verbose-node
    ///     path: path/to/node
    ///     env:
    ///       RUST_LOG: debug  # overrides the global RUST_LOG for this node
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub env: Option<BTreeMap<String, EnvValue>>,
}

impl Descriptor {
    /// A dataflow of `nodes` with every dataflow-level option left at its
    /// default (the state a YAML file with only a `nodes:` key deserializes
    /// to).
    ///
    /// `Descriptor` is `#[non_exhaustive]`, so other crates cannot build one
    /// with a struct literal. Start here and assign the options you need — the
    /// fields are all still `pub`.
    pub fn new(nodes: Vec<Node>) -> Self {
        Self {
            nodes,
            deploy: None,
            debug: Default::default(),
            health_check_interval: None,
            strict_types: None,
            exit_when_nodes_finish: None,
            type_rules: Default::default(),
            env: None,
        }
    }
}

/// A type compatibility rule declared in the dataflow YAML.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
// See the note on `Node`: `type_rules:` is a dataflow-level YAML surface, so a
// per-rule option added later (a direction flag, a coercion mode) must stay a
// minor release. Construct with `TypeRuleDef::new`; the fields remain `pub`.
#[non_exhaustive]
pub struct TypeRuleDef {
    /// Source type URN
    pub from: String,
    /// Target type URN
    pub to: String,
}

impl TypeRuleDef {
    /// A rule declaring that `from` is compatible with `to`.
    ///
    /// `TypeRuleDef` is `#[non_exhaustive]`, so other crates cannot build one
    /// with a struct literal. Start here and assign any further options — the
    /// fields are all still `pub`.
    pub fn new(from: String, to: String) -> Self {
        Self { from, to }
    }
}

/// Specifies when a node should be restarted.
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
// The descriptor *enums* — this one, `OutputFraming`, `DistributeStrategy`,
// `NodeSource`, `GitRepoRev`, `EnvValue`, `OperatorSource`, `PythonSourceDef`,
// the four `Ros2*` enums and `config::QueuePolicy` — are deliberately NOT
// `#[non_exhaustive]`, unlike the descriptor structs.
//
// The cost of that is real and known: adding a variant (`restart-policy:
// unless-stopped`, a third `output_framing`, an rsync `distribute` strategy) is
// `enum_variant_added` — a semver-major break — so it cannot land until 2.0.
//
// It is deliberate because the alternative is worse here. `#[non_exhaustive]`
// forces a `_ =>` arm in every downstream match. Marking all of them stops the
// build with 11 such matches in `dora-core` and the ROS2 bridge alone, before
// it even reaches the ones in `dora-daemon` and `dora-cli` — restart decisions,
// git-ref resolution, lockfile cache keys, operator-runtime dispatch, ROS2
// transport selection. Those crates ship in lockstep with this one, so today a
// new variant is a compile error naming every site that must handle it; behind
// a catch-all it becomes a silent wrong answer (a new `GitRepoRev` colliding in
// the build lockfile key, a new `RestartPolicy` reading as "never restart").
// Exhaustive matching is the thing actually preventing those bugs, and no
// external consumer gets a comparable guarantee back.
//
// The structs have no such tension: a new field breaks only struct literals,
// which the `::new` constructors already replace.
pub enum RestartPolicy {
    /// Never restart the node (default)
    #[default]
    Never,
    /// Restart the node if it exits with a non-zero exit code.
    OnFailure,
    /// Always restart the node when it exits, regardless of exit code.
    ///
    /// The node will not be restarted on the following conditions:
    ///
    /// - The node was stopped by the user (e.g., via `dora stop`).
    /// - All inputs to the node have been closed and the node finished with a non-zero exit code.
    Always,
}

/// Deployment configuration for distributing nodes across machines.
#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
// Same rationale as `Node`: keeps a new deployment key a minor release.
// Every field has a meaningful default, so `Deploy::default()` is the
// construction entry point rather than a bespoke `new`; the fields remain
// `pub`.
#[non_exhaustive]
pub struct Deploy {
    /// Target machine for deployment
    pub machine: Option<String>,
    /// Working directory for the deployment
    pub working_dir: Option<PathBuf>,
    /// Labels for label-based scheduling (e.g. `gpu: "true"`, `arch: arm64`).
    /// The coordinator matches these against daemon labels reported at registration.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub labels: BTreeMap<String, String>,
    /// How built binaries are distributed to remote daemons.
    #[serde(default)]
    pub distribute: DistributeStrategy,
}

/// Strategy for distributing built binaries to daemons.
#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum DistributeStrategy {
    /// Each daemon builds from source (current/default behavior).
    #[default]
    Local,
    /// CLI pushes built binary via SSH/SCP before spawn.
    Scp,
    /// Daemon pulls binary from coordinator HTTP artifact store before spawn.
    Http,
}

/// Debug options for dataflow development and troubleshooting.
#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
// See the note on `Node`: `debug:` is a dataflow-level YAML surface and a
// second debug option must stay a minor release. Every field has a meaningful
// default, so `Debug::default()` is the construction entry point; the fields
// remain `pub`.
#[non_exhaustive]
pub struct Debug {
    /// When true, daemons mirror every node output to the coordinator WebSocket
    /// so that `dora topic echo`, `dora topic hz`, and `dora topic info` can
    /// inspect runtime messages.
    #[serde(default)]
    pub enable_debug_inspection: bool,
}

/// # Dora Node Configuration
///
/// A node represents a computational unit in a Dora dataflow. Each node runs as a
/// separate process and can communicate with other nodes through inputs and outputs.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
// Adding a descriptor key must stay a *minor* release. Without this, every new
// per-node field is `constructible_struct_adds_field` — a semver-major break
// for `dora-message` — which turns routine feature work into a "land it before
// the next major or wait" scramble. `Node::new` is the construction entry point
// for other crates; the fields stay `pub`, so they are still freely readable
// and assignable. The *wire-protocol* enums in this crate (`daemon_to_node.rs`,
// `node_to_daemon.rs`, `daemon_to_coordinator.rs`, `daemon_to_daemon.rs`) were
// marked for the same reason in #3151, and `RunDataflowOptions` in the daemon
// is the existing struct-shaped precedent. The descriptor *enums* are a
// separate axis and are not covered — see the note on `RestartPolicy`.
//
// The construction advice lives here and on `Node::new`, not in the `///`
// doc: that doc is the description schemars writes into `dora-schema.json`,
// which YAML editors show to dataflow authors, and rustdoc already flags
// `#[non_exhaustive]` types on its own.
#[non_exhaustive]
pub struct Node {
    /// Unique node identifier. Must not contain `/` characters.
    ///
    /// Node IDs can be arbitrary strings with the following limitations:
    ///
    /// - They must not contain any `/` characters (slashes).
    /// - We do not recommend using whitespace characters (e.g. spaces) in IDs
    ///
    /// Each node must have an ID field.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: camera_node
    ///   - id: some_other_node
    /// ```
    pub id: NodeId,

    /// Human-readable node name for documentation.
    ///
    /// This optional field can be used to define a more descriptive name in addition to a short
    /// [`id`](Self::id).
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: camera_node
    ///     name: "Camera Input Handler"
    pub name: Option<String>,

    /// Detailed description of the node's functionality.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: camera_node
    ///     description: "Captures video frames from webcam"
    /// ```
    pub description: Option<String>,

    /// Path to executable or script that should be run.
    ///
    /// Specifies the path of the executable or script that Dora should run when starting the
    /// dataflow.
    /// This can point to a normal executable (e.g. when using a compiled language such as Rust) or
    /// a Python script.
    ///
    /// Dora will automatically append a `.exe` extension on Windows systems when the specified
    /// file name has no extension.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: rust-example
    ///     path: target/release/rust-node
    ///   - id: python-example
    ///     path: ./receive_data.py
    /// ```
    ///
    /// ## URL as Path
    ///
    /// The `path` field can also point to a URL instead of a local path.
    /// In this case, Dora will download the given file when starting the dataflow.
    ///
    /// Note that this is quite an old feature and using this functionality is **not recommended**
    /// anymore. Instead, we recommend using a [`git`][Self::git] and/or [`build`](Self::build)
    /// key.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub path: Option<String>,

    /// SHA-256 checksum the `path` download must match, verified after fetch
    /// and on cache reuse (spec §8.2/§8.4). Set internally when a `hub:`
    /// reference resolves to a prebuilt binary artifact; rarely set by hand.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub path_sha256: Option<String>,

    /// Command-line arguments passed to the executable.
    ///
    /// The command-line arguments that should be passed to the executable/script specified in `path`.
    /// The arguments should be separated by space.
    /// This field is optional and defaults to an empty argument list.
    ///
    /// ## Example
    /// ```yaml
    /// nodes:
    ///   - id: example
    ///     path: example-node
    ///     args: -v --some-flag foo
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub args: Option<String>,

    /// Environment variables for node builds and execution.
    ///
    /// Key-value map of environment variables that should be set for both the
    /// [`build`](Self::build) operation and the node execution (i.e. when the node is spawned
    /// through [`path`](Self::path)).
    ///
    /// Supports strings, numbers, and booleans.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: example-node
    ///     path: path/to/node
    ///     env:
    ///       DEBUG: true
    ///       PORT: 8080
    ///       API_KEY: "secret-key"
    /// ```
    pub env: Option<BTreeMap<String, EnvValue>>,

    /// Multiple operators running in a shared runtime process.
    ///
    /// Operators are an experimental, lightweight alternative to nodes.
    /// Instead of running as a separate process, operators are linked into a runtime process.
    /// This allows running multiple operators to share a single address space (not supported for
    /// Python currently).
    ///
    /// Operators are defined as part of the node list, as children of a runtime node.
    /// A runtime node is a special node that specifies no [`path`](Self::path) field, but contains
    /// an `operators` field instead.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: runtime-node
    ///     operators:
    ///       - id: processor
    ///         python: process.py
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operators: Option<RuntimeNode>,

    /// Single operator configuration.
    ///
    /// This is a convenience field for defining runtime nodes that contain only a single operator.
    /// This field is an alternative to the [`operators`](Self::operators) field, which can be used
    /// if there is only a single operator defined for the runtime node.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: runtime-node
    ///     operator:
    ///       id: processor
    ///       python: script.py
    ///       outputs: [data]
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub operator: Option<SingleOperatorDefinition>,

    /// ROS2 bridge configuration (unstable).
    ///
    /// Declares this node as a ROS2 bridge that automatically subscribes to or
    /// publishes on ROS2 topics. No custom code is needed -- the framework spawns
    /// a bridge binary that converts between ROS2 DDS messages and Dora's Arrow
    /// format.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: camera_bridge
    ///     ros2:
    ///       topic: /camera/image_raw
    ///       message_type: sensor_msgs/Image
    ///       direction: subscribe
    ///     outputs:
    ///       - image
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ros2: Option<Ros2BridgeConfig>,

    /// Output data identifiers produced by this node.
    ///
    /// List of output identifiers that the node sends.
    /// Must contain all `output_id` values that the node uses when sending output, e.g. through the
    /// [`send_output`](https://docs.rs/dora-node-api/latest/dora_node_api/struct.DoraNode.html#method.send_output)
    /// function.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: example-node
    ///     outputs:
    ///       - processed_image
    ///       - metadata
    /// ```
    #[serde(default)]
    pub outputs: BTreeSet<DataId>,

    /// Optional type annotations for outputs.
    ///
    /// Maps output identifiers to type URNs (e.g. `std/media/v1/Image`).
    /// Only annotated outputs are type-checked; unannotated outputs remain dynamic.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_types: BTreeMap<DataId, String>,

    /// Per-output framing overrides (default: Raw for all).
    ///
    /// Maps output identifiers to their wire framing mode.
    /// Outputs not listed here use the default `Raw` framing.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_framing: BTreeMap<DataId, OutputFraming>,

    /// Input data connections from other nodes.
    ///
    /// Defines the inputs that this node is subscribing to.
    ///
    /// The `inputs` field should be a key-value map of the following format:
    ///
    /// `input_id: source_node_id/source_node_output_id`
    ///
    /// The components are defined as follows:
    ///
    ///   - `input_id` is the local identifier that should be used for this input.
    ///
    ///     This will map to the `id` field of
    ///     [`Event::Input`](https://docs.rs/dora-node-api/latest/dora_node_api/enum.Event.html#variant.Input)
    ///     events sent to the node event loop.
    ///   - `source_node_id` should be the `id` field of the node that sends the output that we want
    ///     to subscribe to
    ///   - `source_node_output_id` should be the identifier of the output that that we want
    ///     to subscribe to
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: example-node
    ///     outputs:
    ///       - one
    ///       - two
    ///   - id: receiver
    ///     inputs:
    ///         my_input: example-node/two
    /// ```
    #[serde(default)]
    pub inputs: BTreeMap<DataId, Input>,

    /// Optional type annotations for inputs.
    ///
    /// Maps input identifiers to expected type URNs. Used by `dora validate`
    /// to check that upstream output types match expectations.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub input_types: BTreeMap<DataId, String>,

    /// Required metadata keys per output.
    ///
    /// Maps output identifiers to lists of required metadata key names.
    /// These are checked at build/validate time.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// output_metadata:
    ///   response: [request_id]
    /// ```
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_metadata: BTreeMap<DataId, Vec<String>>,

    /// Communication pattern shorthand (e.g. `service-server`).
    ///
    /// Automatically implies required metadata keys on all outputs.
    /// See `pattern_metadata_keys()` for supported patterns.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pattern: Option<String>,

    /// Redirect stdout/stderr to a data output.
    ///
    /// This field can be used to send all stdout and stderr output of the node as a Dora output.
    /// Each output line is sent as a separate message.
    ///
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: example
    ///     send_stdout_as: stdout_output
    ///   - id: logger
    ///     inputs:
    ///         example_output: example/stdout_output
    /// ```
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_stdout_as: Option<String>,

    /// Redirect structured log entries to a data output as JSON strings.
    ///
    /// Unlike `send_stdout_as` which sends raw stdout lines, this sends only
    /// parsed structured log entries (with level, timestamp, message, fields).
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: sensor
    ///     path: ./sensor
    ///     send_logs_as: logs
    ///     outputs:
    ///       - data
    ///       - logs
    /// ```
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_logs_as: Option<String>,

    /// Minimum log level for this node (error, warn, info, debug, trace, stdout).
    ///
    /// Logs below this level are suppressed from file output, coordinator
    /// forwarding, and `send_logs_as` routing.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: noisy_sensor
    ///     path: ./sensor
    ///     min_log_level: info
    /// ```
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_log_level: Option<String>,

    /// Maximum log file size before rotation (e.g. "50MB", "1GB").
    ///
    /// When the JSONL log file exceeds this size, it is rotated. Old files
    /// are renamed with numeric suffixes (`.1.jsonl`, `.2.jsonl`, etc.) and
    /// the oldest are deleted once 5 rotated files exist.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: sensor
    ///     path: ./sensor
    ///     max_log_size: "100MB"
    /// ```
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_log_size: Option<String>,
    /// Maximum number of rotated log files to keep (default: 5, range: 0-100)
    ///
    /// `0` keeps the active log only, rotating the previous one away.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[schemars(range(max = 100))]
    pub max_rotated_files: Option<u32>,

    /// Build commands executed during `dora build`. Each line runs separately.
    ///
    /// The `build` key specifies the command that should be invoked for building the node.
    /// The key expects a single- or multi-line string.
    ///
    /// Each line is run as a separate command.
    /// Spaces are used to separate arguments.
    ///
    /// Note that all the environment variables specified in the [`env`](Self::env) field are also
    /// applied to the build commands.
    ///
    /// ## Special treatment of `pip`
    ///
    /// Build lines that start with `pip` or `pip3` are treated in a special way:
    /// If the `--uv` argument is passed to the `dora build` command, all `pip`/`pip3` commands are
    /// run through the [`uv` package manager](https://docs.astral.sh/uv/).
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    /// - id: build-example
    ///   build: cargo build -p receive_data --release
    ///   path: target/release/receive_data
    /// - id: multi-line-example
    ///   build: |
    ///       pip install requirements.txt
    ///       pip install -e some/local/package
    ///   path: package
    /// ```
    ///
    /// In the above example, the `pip` commands will be replaced by `uv pip` when run through
    /// `dora build --uv`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build: Option<String>,

    /// Git repository URL for downloading nodes.
    ///
    /// The `git` key allows downloading nodes (i.e. their source code) from git repositories.
    /// This can be especially useful for distributed dataflows.
    ///
    /// When a `git` key is specified, `dora build` automatically clones the specified repository
    /// (or reuse an existing clone).
    /// Then it checks out the specified [`branch`](Self::branch), [`tag`](Self::tag), or
    /// [`rev`](Self::rev), or the default branch if none of them are specified.
    /// Afterwards it runs the [`build`](Self::build) command if specified.
    ///
    /// Note that the git clone directory is set as working directory for both the
    /// [`build`](Self::build) command and the specified [`path`](Self::path).
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: rust-node
    ///     git: https://github.com/dora-rs/dora.git
    ///     build: cargo build -p rust-dataflow-example-node
    ///     path: target/debug/rust-dataflow-example-node
    /// ```
    ///
    /// In the above example, `dora build` will first clone the specified `git` repository and then
    /// run the specified `build` inside the local clone directory.
    /// When `dora run` or `dora start` is invoked, the working directory will be the git clone
    /// directory too. So a relative `path` will start from the clone directory.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub git: Option<String>,

    /// Hub package reference.
    ///
    /// **Outside the 1.0 stability guarantee.** This field, the way it is
    /// resolved, and the `HubProvenance` recorded in the lockfile may change
    /// or be removed in a minor release. `dora build` and `dora validate`
    /// print a warning whenever a dataflow uses it.
    ///
    /// The reason is readiness rather than scope: stabilizing `hub:` would
    /// promise a typed-contract guarantee that no package in the catalog
    /// currently delivers. The path to stabilization is the node-typing
    /// workstream, not more code here — see `docs/plan-node-hub.md` §14 (P3.5).
    ///
    /// References a node published in the Dora Hub index:
    /// `[<namespace>/]<name>@<semver-requirement>`. A bare name is shorthand
    /// for the official `dora-rs/` namespace.
    ///
    /// `dora build` resolves the reference against the index to a pinned
    /// commit and the node is fetched/built through the same machinery as a
    /// [`git`](Self::git) node; the package manifest supplies the
    /// entrypoint, build command, and typed contracts. Mutually exclusive
    /// with `path`, `git`, and `build`.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: detector
    ///     hub: dora-yolo@^0.5
    ///     inputs:
    ///       image: camera/image
    ///     outputs:
    ///       - bbox
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub hub: Option<String>,

    /// Git branch to checkout after cloning.
    ///
    /// The `branch` field is only allowed in combination with the [`git`](#git) field.
    /// It specifies the branch that should be checked out after cloning.
    /// Only one of `branch`, `tag`, or `rev` can be specified.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: rust-node
    ///     git: https://github.com/dora-rs/dora.git
    ///     branch: some-branch-name
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub branch: Option<String>,

    /// Git tag to checkout after cloning.
    ///
    /// The `tag` field is only allowed in combination with the [`git`](#git) field.
    /// It specifies the git tag that should be checked out after cloning.
    /// Only one of `branch`, `tag`, or `rev` can be specified.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: rust-node
    ///     git: https://github.com/dora-rs/dora.git
    ///     tag: v0.1.0
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tag: Option<String>,

    /// Git revision (e.g. commit hash) to checkout after cloning.
    ///
    /// The `rev` field is only allowed in combination with the [`git`](#git) field.
    /// It specifies the git revision (e.g. a commit hash) that should be checked out after cloning.
    /// Only one of `branch`, `tag`, or `rev` can be specified.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: rust-node
    ///     git: https://github.com/dora-rs/dora.git
    ///     rev: 64ab0d7c
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub rev: Option<String>,

    /// Whether this node should be restarted on exit or error.
    ///
    /// Defaults to `RestartPolicy::Never`.
    #[serde(default)]
    pub restart_policy: RestartPolicy,

    /// Size of the zenoh shared memory pool for zero-copy output publishing.
    ///
    /// Accepts an integer (raw bytes) or a string with a unit suffix
    /// (`KB`, `MB`, `GB`, case-insensitive). If unset, the
    /// `DORA_NODE_SHM_POOL_SIZE` env var is used, falling back to a
    /// built-in default.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: camera-node
    ///     shared_memory_pool_size: 128MB
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub shared_memory_pool_size: Option<ByteSize>,

    /// Maximum number of restart attempts. 0 means unlimited.
    ///
    /// When combined with `restart_window`, this limits restarts within the window period.
    /// For example, `max_restarts: 5` with `restart_window: 300` means "5 restarts per 5 minutes".
    #[serde(default)]
    pub max_restarts: u32,

    /// Initial delay in seconds before restarting. Doubles each attempt (exponential backoff).
    ///
    /// For example, with `restart_delay: 1.0`, delays will be 1s, 2s, 4s, 8s, ...
    /// Use `max_restart_delay` to cap the backoff.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_delay: Option<f64>,

    /// Maximum delay in seconds for exponential backoff.
    ///
    /// Caps the exponentially growing `restart_delay`. For example, with
    /// `restart_delay: 1.0` and `max_restart_delay: 30.0`, delays grow as
    /// 1s, 2s, 4s, 8s, 16s, 30s, 30s, ...
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_restart_delay: Option<f64>,

    /// Time window in seconds for counting restarts.
    ///
    /// When set, the restart counter resets after this period of time elapses since the
    /// first restart in the current window. This enables "N restarts within M seconds" semantics.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_window: Option<f64>,

    /// Health check timeout in seconds.
    ///
    /// When set, the daemon monitors this node for activity **once it has
    /// connected** (i.e. subscribed to events during `Node::init`). If the
    /// connected node then does not communicate with the daemon within this
    /// timeout, it is killed and the restart policy is evaluated.
    ///
    /// This bounds post-connection liveness only, not startup time: a node
    /// still in a slow cold start has not connected yet and is never killed by
    /// this watchdog. A node that hangs before it ever subscribes is therefore
    /// not reaped here either.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub health_check_timeout: Option<f64>,

    /// Per-node finish-drain grace period in seconds.
    ///
    /// Overrides the global `DORA_FINISH_DRAIN_GRACE_SECS` for this node only.
    /// When all other nodes in a dataflow have finished, the daemon waits this
    /// long after the node's last input closes before force-stopping it.
    ///
    /// Set to a large value (e.g. `3600.0`) for nodes that need significant
    /// post-input compute time (ML training, large-batch inference, checkpoint
    /// writes) to prevent premature SIGKILL while the computation is in progress.
    ///
    /// When unset, the global grace period applies (default 120s, controlled
    /// by `DORA_FINISH_DRAIN_GRACE_SECS`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finish_grace_secs: Option<f64>,

    /// Path to a module definition file (e.g. `nav_module.yml`).
    ///
    /// A module is a reusable sub-dataflow: a group of nodes with declared
    /// inputs and outputs. At build time the module is expanded inline —
    /// internal node IDs are prefixed with `{module_id}.` and all wiring is
    /// rewritten so the runtime sees only flat nodes.
    ///
    /// A module node has no source or per-node runtime configuration of its own,
    /// so only `module`, `inputs`, `params`, `env`, `build`, and `deploy` are
    /// meaningful on it. Every other node field is rejected at expansion time
    /// rather than silently discarded -- both the source/kind fields (`path`,
    /// `args`, `path_sha256`, `git`, `hub`, `branch`, `tag`, `rev`, `operators`,
    /// `operator`, `ros2`) and per-node runtime fields (`outputs`,
    /// `output_types`, `cpu_affinity`, `restart_policy`, ...). The same rule
    /// applies at every nesting level.
    ///
    /// `env`, `build`, `deploy`, and `params` *are* accepted: they propagate
    /// into the module's inner nodes.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: nav_stack
    ///     module: modules/navigation_module.yml
    ///     inputs:
    ///       goal_pose: localization/goal
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub module: Option<String>,

    /// Parameters passed to a module for compile-time substitution.
    ///
    /// Only meaningful when `module` is set. Values are substituted into
    /// inner node `args` fields (using `${_param.name}` syntax) and can be
    /// injected into inner node `env` maps.
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: nav_stack
    ///     module: modules/navigation_module.yml
    ///     params:
    ///       speed: "2.0"
    ///       mode: turbo
    /// ```
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub params: BTreeMap<String, String>,

    /// CPU cores to pin this node's process to (Linux only, ignored on other platforms).
    ///
    /// ## Example
    ///
    /// ```yaml
    /// nodes:
    ///   - id: fast_node
    ///     path: ./fast_node
    ///     cpu_affinity: [0, 1]
    /// ```
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cpu_affinity: Option<Vec<usize>>,

    /// Machine deployment configuration.
    #[schemars(skip)]
    pub deploy: Option<Deploy>,

    /// Startup connection deadline in seconds.
    ///
    /// If the node process fails to connect (subscribe to events) within this
    /// many seconds after process spawn, the daemon kills the process and
    /// evaluates the `restart_policy`.
    ///
    /// Evaluated on each `health_check_interval` tick (default 5s).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub startup_timeout: Option<f64>,
}

impl Node {
    /// A node with the given ID and every other field left at its descriptor
    /// default (the state a YAML node with only an `id:` key deserializes to).
    ///
    /// `Node` is `#[non_exhaustive]`, so other crates cannot build one with a
    /// struct literal. Start here and assign the fields you need — they are all
    /// still `pub`:
    ///
    /// ```
    /// use dora_message::descriptor::Node;
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut node = Node::new("camera".parse()?);
    /// node.path = Some("./camera".to_owned());
    /// # Ok(())
    /// # }
    /// ```
    pub fn new(id: NodeId) -> Self {
        Self {
            id,
            name: None,
            description: None,
            path: None,
            path_sha256: None,
            args: None,
            env: None,
            operators: None,
            operator: None,
            ros2: None,
            outputs: Default::default(),
            output_types: Default::default(),
            output_framing: Default::default(),
            inputs: Default::default(),
            input_types: Default::default(),
            shared_memory_pool_size: None,
            output_metadata: Default::default(),
            pattern: None,
            send_stdout_as: None,
            send_logs_as: None,
            min_log_level: None,
            max_log_size: None,
            max_rotated_files: None,
            build: None,
            git: None,
            hub: None,
            branch: None,
            tag: None,
            rev: None,
            restart_policy: Default::default(),
            max_restarts: 0,
            restart_delay: None,
            max_restart_delay: None,
            restart_window: None,
            health_check_timeout: None,
            finish_grace_secs: None,
            module: None,
            params: Default::default(),
            cpu_affinity: None,
            deploy: None,
            startup_timeout: None,
        }
    }
}

/// A [`Node`] after alias resolution and defaulting, as the daemon runs it.
#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize)]
// Same rationale as `Node`: keeps a new per-node key a minor release. This is
// where keys that are not custom-node-specific land — `cpu_affinity` and
// `deploy` are threaded `Node` -> `ResolvedNode` without passing through
// `CustomNode`. Construct with `ResolvedNode::new`; the fields remain `pub`.
// Resolution itself goes through `ResolvedNode::from_node`, the in-crate
// literal that keeps a new field a compile error.
#[non_exhaustive]
pub struct ResolvedNode {
    pub id: NodeId,
    pub name: Option<String>,
    pub description: Option<String>,
    pub env: Option<BTreeMap<String, EnvValue>>,

    #[serde(default)]
    pub cpu_affinity: Option<Vec<usize>>,

    #[serde(default)]
    pub deploy: Option<Deploy>,

    #[serde(flatten)]
    pub kind: CoreNodeKind,
}

#[allow(missing_docs)]
impl ResolvedNode {
    /// A resolved node with the given ID and kind, and every other field left
    /// at its default.
    ///
    /// `ResolvedNode` is `#[non_exhaustive]`, so other crates cannot build one
    /// with a struct literal. Start here and assign the fields you need — they
    /// are all still `pub`.
    pub fn new(id: NodeId, kind: CoreNodeKind) -> Self {
        Self {
            id,
            name: None,
            description: None,
            env: None,
            cpu_affinity: None,
            deploy: None,
            kind,
        }
    }

    /// The resolved node for `node`'s node-level keys — `id`, `name`,
    /// `description`, `env`, `cpu_affinity`, `deploy` — around an already
    /// resolved `kind`. `env` is carried as declared; merging the
    /// dataflow-level `env` into it is the caller's job.
    ///
    /// The kind-level keys are dropped: for a custom node
    /// [`CustomNode::from_node`] has already moved them out, and a runtime
    /// node's live in its `operators`.
    ///
    /// This is a struct literal on purpose. `ResolvedNode` is
    /// `#[non_exhaustive]`, so a literal only compiles here, inside the
    /// defining crate — exactly where the compiler still insists that every
    /// field is accounted for. A field added to `ResolvedNode` is a build
    /// error on this line, not a key that resolution silently leaves at its
    /// default.
    pub fn from_node(node: Node, kind: CoreNodeKind) -> Self {
        Self {
            id: node.id,
            name: node.name,
            description: node.description,
            env: node.env,
            cpu_affinity: node.cpu_affinity,
            deploy: node.deploy,
            kind,
        }
    }

    pub fn has_git_source(&self) -> bool {
        self.kind
            .as_custom()
            .map(|n| n.source.is_git())
            .unwrap_or_default()
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[allow(clippy::large_enum_variant)]
pub enum CoreNodeKind {
    /// Dora runtime node
    #[serde(rename = "operators")]
    Runtime(RuntimeNode),
    Custom(CustomNode),
}

#[allow(missing_docs)]
impl CoreNodeKind {
    pub fn as_custom(&self) -> Option<&CustomNode> {
        match self {
            CoreNodeKind::Runtime(_) => None,
            CoreNodeKind::Custom(custom_node) => Some(custom_node),
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(transparent)]
pub struct RuntimeNode {
    /// List of operators running in this runtime
    pub operators: Vec<OperatorDefinition>,
}

#[allow(missing_docs)]
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)]
pub struct OperatorDefinition {
    /// Unique operator identifier within the runtime
    pub id: OperatorId,
    #[serde(flatten)]
    pub config: OperatorConfig,
}

#[allow(missing_docs)]
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)]
// Same rationale as `Node`: keeps a new operator-level descriptor key a minor
// release. No constructor yet because nothing constructs one outside this crate
// — every instance comes from deserialization. Adding a constructor later is
// itself a minor change, so only the `#[non_exhaustive]` half is time-critical;
// open an issue if you need to build one programmatically.
#[non_exhaustive]
pub struct SingleOperatorDefinition {
    /// Operator identifier (optional for single operators)
    pub id: Option<OperatorId>,
    #[serde(flatten)]
    pub config: OperatorConfig,
}

#[allow(missing_docs)]
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)]
// Same rationale as `Node`: keeps a new operator-level descriptor key a minor
// release. No constructor yet because nothing constructs one outside this crate
// — every instance comes from deserialization. Adding a constructor later is
// itself a minor change, so only the `#[non_exhaustive]` half is time-critical;
// open an issue if you need to build one programmatically.
//
// Kept as a `//` comment deliberately: a `///` doc here is `#[serde(flatten)]`ed
// by schemars onto `OperatorDefinition`'s entry in `dora-schema.json`, which
// YAML editors show to dataflow authors.
#[non_exhaustive]
pub struct OperatorConfig {
    /// Human-readable operator name
    pub name: Option<String>,
    /// Detailed description of the operator
    pub description: Option<String>,

    /// Input data connections
    #[serde(default)]
    pub inputs: BTreeMap<DataId, Input>,
    /// Output data identifiers
    #[serde(default)]
    pub outputs: BTreeSet<DataId>,
    /// Optional type annotations for outputs
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_types: BTreeMap<DataId, String>,

    /// Per-output framing overrides (default: Raw for all).
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_framing: BTreeMap<DataId, OutputFraming>,

    /// Optional type annotations for inputs
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub input_types: BTreeMap<DataId, String>,

    /// Required metadata keys per output
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub output_metadata: BTreeMap<DataId, Vec<String>>,

    /// Communication pattern shorthand (e.g. `service-server`)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pattern: Option<String>,

    /// Operator source configuration (Python, shared library, etc.)
    #[serde(flatten)]
    pub source: OperatorSource,

    /// Build commands for this operator
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build: Option<String>,
    /// Redirect stdout to data output
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_stdout_as: Option<String>,
    /// Redirect structured log entries to a data output as JSON strings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_logs_as: Option<String>,
    /// Minimum log level for this operator
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_log_level: Option<String>,
    /// Maximum log file size before rotation (e.g. "50MB", "1GB")
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_log_size: Option<String>,
    /// Maximum number of rotated log files to keep (default: 5, range: 0-100)
    ///
    /// `0` keeps the active log only, rotating the previous one away.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[schemars(range(max = 100))]
    pub max_rotated_files: Option<u32>,
}

#[allow(missing_docs)]
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone)]
#[serde(rename_all = "kebab-case")]
pub enum OperatorSource {
    SharedLibrary(String),
    Python(PythonSource),
    #[schemars(skip)]
    Wasm(String),
}
#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(from = "PythonSourceDef", into = "PythonSourceDef")]
pub struct PythonSource {
    pub source: String,
    pub conda_env: Option<String>,
}

#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(untagged)]
pub enum PythonSourceDef {
    SourceOnly(String),
    WithOptions {
        source: String,
        conda_env: Option<String>,
    },
}

impl From<PythonSource> for PythonSourceDef {
    fn from(input: PythonSource) -> Self {
        match input {
            PythonSource {
                source,
                conda_env: None,
            } => Self::SourceOnly(source),
            PythonSource { source, conda_env } => Self::WithOptions { source, conda_env },
        }
    }
}

impl From<PythonSourceDef> for PythonSource {
    fn from(value: PythonSourceDef) -> Self {
        match value {
            PythonSourceDef::SourceOnly(source) => Self {
                source,
                conda_env: None,
            },
            PythonSourceDef::WithOptions { source, conda_env } => Self { source, conda_env },
        }
    }
}

/// Built-in runtime name for shared-library operators.
pub const RUNTIME_SHARED_LIBRARY: &str = "shared-library";
/// Built-in runtime name for Python operators.
pub const RUNTIME_PYTHON: &str = "python";
/// Built-in runtime name for WebAssembly operators.
pub const RUNTIME_WASM: &str = "wasm";

impl OperatorSource {
    /// The name of the runtime that hosts operators declared with this source.
    ///
    /// This mapping is the single source of truth for "which runtime hosts this
    /// operator": the daemon's spawn logic and the CLI's build hashing key on
    /// the name rather than matching each variant.
    pub fn runtime_name(&self) -> &'static str {
        match self {
            OperatorSource::SharedLibrary(_) => RUNTIME_SHARED_LIBRARY,
            OperatorSource::Python(_) => RUNTIME_PYTHON,
            OperatorSource::Wasm(_) => RUNTIME_WASM,
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
// See the note on `Node`: keeps a new resolved-node field a minor release.
// Construct with `CustomNode::new`; the fields remain `pub`. Resolution goes
// through `CustomNode::from_node`, the in-crate literal that keeps a new field
// a compile error rather than a silently dropped descriptor key.
#[non_exhaustive]
pub struct CustomNode {
    /// Path of the source code
    ///
    /// If you want to use a specific `conda` environment.
    /// Provide the python path within the source.
    ///
    /// source: /home/peter/miniconda3/bin/python
    ///
    /// args: some_node.py
    ///
    /// Source can match any executable in PATH.
    pub path: String,
    pub source: NodeSource,
    /// SHA-256 the `path` download must match (set for hub binary artifacts,
    /// spec §8.2). When present the daemon fetches `path` as a verified URL
    /// download regardless of confinement — the checksum is the trust anchor.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub path_sha256: Option<String>,
    /// Args for the executable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub args: Option<String>,
    /// Environment variables injected during resolution.
    ///
    /// Not user-writable: [`Node::env`] is the YAML surface. Resolution folds
    /// it into this field, and the ROS2 bridge desugaring uses it to pass
    /// `DORA_ROS2_BRIDGE_CONFIG` to the spawned bridge binary.
    pub envs: Option<BTreeMap<String, EnvValue>>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub build: Option<String>,
    /// Send stdout and stderr to another node
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_stdout_as: Option<String>,
    /// Redirect structured log entries to a data output as JSON strings
    #[serde(skip_serializing_if = "Option::is_none")]
    pub send_logs_as: Option<String>,
    /// Minimum log level for this node
    #[serde(skip_serializing_if = "Option::is_none")]
    pub min_log_level: Option<String>,
    /// Maximum log file size before rotation (e.g. "50MB", "1GB")
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_log_size: Option<String>,
    /// Maximum number of rotated log files to keep (default: 5, range: 0-100)
    ///
    /// `0` keeps the active log only, rotating the previous one away.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[schemars(range(max = 100))]
    pub max_rotated_files: Option<u32>,

    #[serde(default)]
    pub restart_policy: RestartPolicy,

    /// Maximum number of restart attempts. 0 means unlimited.
    #[serde(default)]
    pub max_restarts: u32,

    /// Initial delay in seconds before restarting (exponential backoff).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_delay: Option<f64>,

    /// Maximum delay in seconds for exponential backoff.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_restart_delay: Option<f64>,

    /// Time window in seconds for counting restarts.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub restart_window: Option<f64>,

    /// Health check timeout in seconds.
    ///
    /// When set, the daemon monitors this node for activity **once it has
    /// connected** (i.e. subscribed to events during `Node::init`). If the
    /// connected node then does not communicate with the daemon within this
    /// timeout, it is killed and the restart policy is evaluated.
    ///
    /// This bounds post-connection liveness only, not startup time: a node
    /// still in a slow cold start has not connected yet and is never killed by
    /// this watchdog. A node that hangs before it ever subscribes is therefore
    /// not reaped here either.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub health_check_timeout: Option<f64>,

    /// Per-node finish-drain grace period in seconds.
    ///
    /// Overrides the global `DORA_FINISH_DRAIN_GRACE_SECS` for this node only.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finish_grace_secs: Option<f64>,

    #[serde(flatten)]
    pub run_config: NodeRunConfig,

    /// Startup connection deadline in seconds.
    ///
    /// If the node process fails to connect (subscribe to events) within this
    /// many seconds after process spawn, the daemon kills the process and
    /// evaluates the `restart_policy`.
    ///
    /// Evaluated on each `health_check_interval` tick (default 5s).
    // Appended last so existing fields keep their wire-format order.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub startup_timeout: Option<f64>,
}

impl CustomNode {
    /// A local-source node at `path` with every other field left at its
    /// default.
    ///
    /// `CustomNode` is `#[non_exhaustive]`, so other crates cannot build one
    /// with a struct literal. Start here and assign the fields you need — they
    /// are all still `pub`.
    pub fn new(path: String) -> Self {
        Self {
            path,
            source: NodeSource::Local,
            path_sha256: None,
            args: None,
            envs: None,
            build: None,
            send_stdout_as: None,
            send_logs_as: None,
            min_log_level: None,
            max_log_size: None,
            max_rotated_files: None,
            restart_policy: Default::default(),
            max_restarts: 0,
            restart_delay: None,
            max_restart_delay: None,
            restart_window: None,
            health_check_timeout: None,
            finish_grace_secs: None,
            run_config: NodeRunConfig::default(),
            startup_timeout: None,
        }
    }

    /// Move the custom-node keys out of `node` into a `CustomNode` at `path`.
    ///
    /// Every key that all custom-node kinds resolve identically is taken from
    /// `node`, leaving its `Option`s empty and its collections cleared. What
    /// stays behind are the node-level keys — `id`, `name`, `description`,
    /// `env`, `cpu_affinity`, `deploy` — for [`ResolvedNode::from_node`] to
    /// consume next, plus the kind-selection keys (`git`, `hub`, `operators`,
    /// `ros2`, …) that classification has already read.
    ///
    /// `source` and `envs` stay at their defaults: they are the two keys whose
    /// value depends on the node kind, so the caller sets them — `source` from
    /// classification for a `path:` node, `envs` for the ROS2 bridge, whose
    /// `path` is its fixed binary.
    ///
    /// This is a struct literal on purpose. `CustomNode` and `NodeRunConfig`
    /// are `#[non_exhaustive]`, so a literal only compiles here, inside the
    /// defining crate — exactly where the compiler still insists that every
    /// field is accounted for. A key added to either struct is a build error
    /// on this line, not a descriptor key that parses and is then silently
    /// dropped. `dora-core`'s `every_custom_node_field_is_carried_through`
    /// checks the values on top.
    pub fn from_node(node: &mut Node, path: String) -> Self {
        Self {
            path,
            source: NodeSource::Local,
            path_sha256: node.path_sha256.take(),
            args: node.args.take(),
            envs: None,
            build: node.build.take(),
            send_stdout_as: node.send_stdout_as.take(),
            send_logs_as: node.send_logs_as.take(),
            min_log_level: node.min_log_level.take(),
            max_log_size: node.max_log_size.take(),
            max_rotated_files: node.max_rotated_files.take(),
            restart_policy: node.restart_policy,
            max_restarts: node.max_restarts,
            restart_delay: node.restart_delay.take(),
            max_restart_delay: node.max_restart_delay.take(),
            restart_window: node.restart_window.take(),
            health_check_timeout: node.health_check_timeout.take(),
            finish_grace_secs: node.finish_grace_secs.take(),
            run_config: NodeRunConfig {
                inputs: std::mem::take(&mut node.inputs),
                outputs: std::mem::take(&mut node.outputs),
                output_types: std::mem::take(&mut node.output_types),
                output_framing: std::mem::take(&mut node.output_framing),
                input_types: std::mem::take(&mut node.input_types),
                shared_memory_pool_size: node.shared_memory_pool_size.take(),
            },
            startup_timeout: node.startup_timeout.take(),
        }
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub enum NodeSource {
    Local,
    GitBranch {
        repo: String,
        rev: Option<GitRepoRev>,
    },
}

#[allow(missing_docs)]
impl NodeSource {
    pub fn is_git(&self) -> bool {
        matches!(self, Self::GitBranch { .. })
    }
}

#[allow(missing_docs)]
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
pub enum GitRepoRev {
    Branch(String),
    Tag(String),
    Rev(String),
}

#[allow(missing_docs)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(untagged)]
pub enum EnvValue {
    #[serde(deserialize_with = "with_expand_envs")]
    Bool(bool),
    #[serde(deserialize_with = "with_expand_envs")]
    Integer(i64),
    #[serde(deserialize_with = "with_expand_envs")]
    Float(f64),
    #[serde(deserialize_with = "with_expand_envs")]
    String(String),
}

impl fmt::Display for EnvValue {
    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
        match self {
            EnvValue::Bool(bool) => fmt.write_str(&bool.to_string()),
            EnvValue::Integer(i64) => fmt.write_str(&i64.to_string()),
            EnvValue::Float(f64) => fmt.write_str(&f64.to_string()),
            EnvValue::String(str) => fmt.write_str(str),
        }
    }
}

/// ROS2 bridge configuration for declarative ROS2 bridging.
///
/// This allows nodes to interact with ROS2 topics, services, and actions
/// without writing any custom code. The framework spawns a bridge binary that
/// handles the ROS2 DDS communication and Arrow data conversion.
///
/// Exactly one of `topic`, `topics`, `service`, or `action` must be set.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct Ros2BridgeConfig {
    /// Native transport used to communicate with the ROS2 graph.
    ///
    /// Defaults to the existing DDS implementation.
    #[serde(default)]
    pub transport: Ros2TransportConfig,

    /// ROS2 topic name (e.g. "/camera/image_raw").
    /// Mutually exclusive with `topics`, `service`, `action`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topic: Option<String>,

    /// ROS2 message type (e.g. "sensor_msgs/Image").
    /// Required when `topic` is set.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub message_type: Option<String>,

    /// Direction: subscribe (ROS2 -> Dora) or publish (Dora -> ROS2).
    /// Defaults to subscribe. Only used with `topic`/`topics`.
    #[serde(default)]
    pub direction: Ros2Direction,

    /// Multiple topics on a single ROS2 node context.
    /// Mutually exclusive with `topic`, `service`, `action`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topics: Option<Vec<Ros2TopicConfig>>,

    /// ROS2 service name (e.g. "/add_two_ints").
    /// Mutually exclusive with `topic`, `topics`, `action`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub service: Option<String>,

    /// ROS2 service type (e.g. "example_interfaces/AddTwoInts").
    /// Required when `service` is set.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub service_type: Option<String>,

    /// ROS2 action name (e.g. "/navigate").
    /// Mutually exclusive with `topic`, `topics`, `service`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub action: Option<String>,

    /// ROS2 action type (e.g. "nav2_msgs/NavigateToPose").
    /// Required when `action` is set.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub action_type: Option<String>,

    /// Role: client or server. Required for `service` and `action`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub role: Option<Ros2Role>,

    /// QoS policies applied to all topics (can be overridden per-topic).
    #[serde(default)]
    pub qos: Ros2QosConfig,

    /// ROS2 namespace (default: "/").
    #[serde(default = "default_ros2_namespace")]
    pub namespace: String,

    /// ROS2 node name. Defaults to the dora node id.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_name: Option<String>,
}

impl Default for Ros2BridgeConfig {
    fn default() -> Self {
        Self {
            transport: Ros2TransportConfig::default(),
            topic: None,
            message_type: None,
            direction: Ros2Direction::default(),
            topics: None,
            service: None,
            service_type: None,
            action: None,
            action_type: None,
            role: None,
            qos: Ros2QosConfig::default(),
            namespace: default_ros2_namespace(),
            node_name: None,
        }
    }
}

/// Native transport used by a ROS2 bridge context.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum Ros2TransportConfig {
    /// The existing `ros2-client` and RustDDS transport.
    #[default]
    Dds,
    /// Direct interoperability with `rmw_zenoh_cpp` peers.
    Zenoh {
        /// Wire-compatibility profile used by the target ROS2 distribution.
        compatibility: RmwZenohCompatibility,
        /// Optional Zenoh session configuration path.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        config_uri: Option<PathBuf>,
    },
}

/// Wire-compatibility profile for the `rmw_zenoh_cpp` protocol.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum RmwZenohCompatibility {
    /// ROS2 Humble, whose endpoint identity uses `TypeHashNotSupported`.
    Humble,
    /// ROS2 distributions whose endpoint identity uses REP-2016 type hashes.
    Rep2016,
}

/// Role of a ROS2 service or action bridge node.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum Ros2Role {
    /// Client: sends requests/goals, receives responses/results.
    Client,
    /// Server: receives requests, sends responses.
    Server,
}

fn default_ros2_namespace() -> String {
    "/".to_string()
}

/// Configuration for a single ROS2 topic in multi-topic mode.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct Ros2TopicConfig {
    /// ROS2 topic name.
    pub topic: String,

    /// ROS2 message type (e.g. "geometry_msgs/Twist").
    pub message_type: String,

    /// Direction: subscribe or publish.
    #[serde(default)]
    pub direction: Ros2Direction,

    /// Maps to an dora output id (for subscribe direction).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub output: Option<String>,

    /// Maps to an dora input id (for publish direction).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub input: Option<String>,

    /// Per-topic QoS override.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub qos: Option<Ros2QosConfig>,
}

/// Strip the leading `/` of a ROS2 topic name, then map any remaining `/` to
/// `_` (e.g. `/robot/scan` -> `robot_scan`).
///
/// ```
/// use dora_message::descriptor::derive_port_id;
///
/// assert_eq!(derive_port_id("/robot/scan"), "robot_scan");
/// assert_eq!(derive_port_id("scan"), "scan");
/// ```
pub fn derive_port_id(topic: &str) -> String {
    topic.trim_start_matches('/').replace('/', "_")
}

impl Ros2TopicConfig {
    /// The dora output a `subscribe` topic feeds: the explicit `output:`
    /// mapping, or the topic-derived id when that is unset.
    ///
    /// Both the descriptor validator (`dora-core`) and the ros2 bridge node
    /// depend on producing the *same* id — the validator rejects configs whose
    /// port is not declared precisely because the bridge would otherwise
    /// silently drop the data — so the rule lives here in the shared message
    /// crate rather than being duplicated at each call site.
    pub fn output_port_id(&self) -> String {
        self.output
            .clone()
            .unwrap_or_else(|| derive_port_id(&self.topic))
    }

    /// The dora input a `publish` topic consumes; the counterpart of
    /// [`Self::output_port_id`].
    pub fn input_port_id(&self) -> String {
        self.input
            .clone()
            .unwrap_or_else(|| derive_port_id(&self.topic))
    }
}

/// The dora port a **single-topic** (`topic:`) ros2 bridge binds to, or `None`
/// when `declared_ports` leaves the choice ambiguous.
///
/// Single-topic mode has no explicit `output:`/`input:` field, so the port comes
/// from the node's declaration — `docs/ros2-bridge.md`: "In single-topic mode,
/// the node's declared `outputs` or `inputs` are used directly". Preferring the
/// topic-derived id when it *is* declared keeps a node that declares an extra
/// port (a `send_stdout_as` output, say) resolvable instead of ambiguous.
///
/// Both the descriptor resolver, which bakes the resolved id into the bridge
/// config, and the validator, which rejects the ambiguous case, call this — a
/// wrong choice here binds the bridge to a port nothing is wired to and drops
/// every message silently.
///
/// ```
/// use dora_message::descriptor::single_topic_port_id;
///
/// // The topic-derived id is preferred when the node declares it...
/// assert_eq!(
///     single_topic_port_id("/robot/scan", &["robot_scan", "status"]),
///     Some("robot_scan".to_string()),
/// );
/// // ...otherwise a node that declares a single port resolves to that port...
/// assert_eq!(
///     single_topic_port_id("/robot/scan", &["lidar"]),
///     Some("lidar".to_string()),
/// );
/// // ...but several non-matching ports leave the choice ambiguous.
/// assert_eq!(single_topic_port_id("/robot/scan", &["a", "b"]), None);
/// ```
pub fn single_topic_port_id(topic: &str, declared_ports: &[&str]) -> Option<String> {
    let derived = derive_port_id(topic);
    if declared_ports.contains(&derived.as_str()) {
        return Some(derived);
    }
    match declared_ports {
        [sole_port] => Some((*sole_port).to_owned()),
        _ => None,
    }
}

/// Direction of ROS2 bridge communication.
#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum Ros2Direction {
    /// Subscribe: receive from ROS2, forward to dora outputs.
    #[default]
    Subscribe,
    /// Publish: receive from dora inputs, publish to ROS2.
    Publish,
}

/// ROS2 Quality of Service configuration.
#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct Ros2QosConfig {
    /// Use reliable transport (default: false = best effort).
    #[serde(default)]
    pub reliable: bool,

    /// Durability: "volatile" (default), "transient_local".
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub durability: Option<String>,

    /// Liveliness: "automatic" (default), "manual_by_participant", "manual_by_topic".
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub liveliness: Option<String>,

    /// Lease duration in seconds (default: infinity).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub lease_duration: Option<f64>,

    /// Max blocking time in seconds for reliable transport.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_blocking_time: Option<f64>,

    /// History depth for KeepLast policy (default: 1).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub keep_last: Option<i32>,

    /// Use KeepAll history policy instead of KeepLast.
    #[serde(default)]
    pub keep_all: bool,
}

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

    /// Assert that `constructed` serializes to exactly what `yaml`
    /// deserializes to — that a hand-written constructor agrees with the
    /// per-field `#[serde(default)]`s.
    ///
    /// The two are independent sources of the same defaults, and both are
    /// used in production: the daemon builds dynamically registered nodes
    /// through `Node::new` (`Daemon::handle_add_node`) and declared nodes
    /// through serde, so a divergence — say a field that later grows a
    /// `#[serde(default = "…")]` custom default — would silently give the two
    /// kinds different configuration.
    fn assert_matches_yaml_defaults<T: Serialize + serde::de::DeserializeOwned>(
        yaml: &str,
        constructed: T,
        what: &str,
    ) {
        let from_yaml: T = serde_yaml::from_str(yaml).unwrap();
        assert_eq!(
            serde_yaml::to_value(&from_yaml).unwrap(),
            serde_yaml::to_value(&constructed).unwrap(),
            "`{what}` drifted from the defaults serde applies"
        );
    }

    /// `Node::new` must agree with what serde produces for a YAML node that
    /// sets nothing but `id`.
    #[test]
    fn node_new_matches_yaml_defaults() {
        assert_matches_yaml_defaults(
            "id: some-node\n",
            Node::new("some-node".to_owned().into()),
            "Node::new",
        );
    }

    /// The same contract for the other constructors this crate hands out.
    ///
    /// `Descriptor::new` is what the coordinator's `AddNode` resolution and the
    /// daemon's bench support build from; `Deploy::default()` and
    /// `Debug::default()` are the documented entry points for `deploy:` and
    /// `debug:`; `NodeRunConfig::default()` is the runtime node's I/O config in
    /// the daemon while custom nodes deserialize theirs.
    ///
    /// `CustomNode::new` is not covered here: `source` and `envs` have no serde
    /// default, so there is no "nothing set" YAML for it. Its guard is
    /// `CustomNode::from_node` — a struct literal in this crate, so a new field
    /// is a compile error — plus
    /// `dora_core::descriptor::tests::every_custom_node_field_is_carried_through`
    /// for the values.
    #[test]
    fn constructors_match_yaml_defaults() {
        assert_matches_yaml_defaults(
            "nodes: []\n",
            Descriptor::new(Vec::new()),
            "Descriptor::new",
        );
        assert_matches_yaml_defaults("{}\n", Deploy::default(), "Deploy::default");
        assert_matches_yaml_defaults("{}\n", Debug::default(), "Debug::default");
        assert_matches_yaml_defaults("{}\n", NodeRunConfig::default(), "NodeRunConfig::default");
        assert_matches_yaml_defaults(
            "from: a\nto: b\n",
            TypeRuleDef::new("a".to_owned(), "b".to_owned()),
            "TypeRuleDef::new",
        );
    }

    #[test]
    fn ros2_transport_defaults_to_dds() {
        let config: Ros2BridgeConfig =
            serde_yaml::from_str("topic: /chatter\nmessage_type: std_msgs/String\n").unwrap();
        assert!(matches!(config.transport, Ros2TransportConfig::Dds));
    }

    #[test]
    fn ros2_transport_parses_humble_zenoh() {
        let config: Ros2BridgeConfig = serde_yaml::from_str(
            "transport:\n  kind: zenoh\n  compatibility: humble\n  config_uri: /tmp/rmw.json5\n\
             topic: /chatter\nmessage_type: std_msgs/String\n",
        )
        .unwrap();
        assert_eq!(
            config.transport,
            Ros2TransportConfig::Zenoh {
                compatibility: RmwZenohCompatibility::Humble,
                config_uri: Some("/tmp/rmw.json5".into()),
            }
        );
    }

    #[test]
    fn ros2_transport_rejects_unknown_zenoh_compatibility() {
        let error = serde_yaml::from_str::<Ros2BridgeConfig>(
            "transport:\n  kind: zenoh\n  compatibility: automatic\n\
             topic: /chatter\nmessage_type: std_msgs/String\n",
        )
        .unwrap_err();
        assert!(error.to_string().contains("unknown variant `automatic`"));
    }

    #[test]
    fn single_topic_port_uses_the_sole_declared_port() {
        // `docs/ros2-bridge.md`: single-topic mode uses the declared port
        // directly, so `/turtle1/pose` binds to `pose` and not to the
        // topic-derived `turtle1_pose`.
        assert_eq!(
            single_topic_port_id("/turtle1/pose", &["pose"]),
            Some("pose".to_owned())
        );
    }

    #[test]
    fn single_topic_port_prefers_a_declared_topic_derived_id() {
        // A second declared port (here a `send_stdout_as` output) would make the
        // choice ambiguous, but the topic-derived id is itself declared.
        assert_eq!(
            single_topic_port_id("/turtle1/pose", &["log", "turtle1_pose"]),
            Some("turtle1_pose".to_owned())
        );
    }

    #[test]
    fn single_topic_port_is_ambiguous_with_several_unrelated_ports() {
        assert_eq!(
            single_topic_port_id("/turtle1/pose", &["pose", "log"]),
            None
        );
    }

    #[test]
    fn single_topic_port_needs_a_declared_port() {
        assert_eq!(single_topic_port_id("/turtle1/pose", &[]), None);
    }

    #[test]
    fn output_framing_defaults_to_raw() {
        let yaml = r#"
nodes:
  - id: test
    path: test.py
    outputs:
      - data
"#;
        let desc: Descriptor = serde_yaml::from_str(yaml).unwrap();
        assert!(desc.nodes[0].output_framing.is_empty());
    }

    #[test]
    fn output_framing_parses_arrow_ipc() {
        let yaml = r#"
nodes:
  - id: test
    path: test.py
    outputs:
      - data
    output_framing:
      data: arrow-ipc
"#;
        let desc: Descriptor = serde_yaml::from_str(yaml).unwrap();
        assert_eq!(
            desc.nodes[0].output_framing.get::<DataId>(&"data".into()),
            Some(&OutputFraming::ArrowIpc)
        );
    }

    #[test]
    fn cpu_affinity_parses() {
        let yaml = r#"
nodes:
  - id: test
    path: test.py
    cpu_affinity: [0, 2, 4]
"#;
        let desc: Descriptor = serde_yaml::from_str(yaml).unwrap();
        assert_eq!(desc.nodes[0].cpu_affinity, Some(vec![0, 2, 4]));
    }

    #[test]
    fn cpu_affinity_defaults_to_none() {
        let yaml = r#"
nodes:
  - id: test
    path: test.py
"#;
        let desc: Descriptor = serde_yaml::from_str(yaml).unwrap();
        assert_eq!(desc.nodes[0].cpu_affinity, None);
    }

    #[test]
    fn debug_flag_accepts_new_name() {
        let yaml = r#"
nodes:
  - id: test
    path: test.py
debug:
  enable_debug_inspection: true
"#;
        let desc: Descriptor = serde_yaml::from_str(yaml).unwrap();
        assert!(desc.debug.enable_debug_inspection);
    }

    #[test]
    fn removed_unstable_key_prefix_is_rejected_not_ignored() {
        // `_unstable_deploy` / `_unstable_debug` lost their prefix for 1.0.
        // Both must *error*, never deserialize to the default: a dataflow that
        // still says `_unstable_deploy` would otherwise run every node on the
        // local daemon while looking like it pinned them to machines, and a
        // stale `_unstable_debug` would leave `dora topic echo` silently
        // empty. `deny_unknown_fields` on `Descriptor` is what makes these
        // diagnosable, so this test guards that attribute as much as the
        // rename.
        for (key, block) in [
            ("_unstable_deploy", "_unstable_deploy:\n  machine: m1\n"),
            (
                "_unstable_debug",
                "_unstable_debug:\n  enable_debug_inspection: true\n",
            ),
        ] {
            let yaml = format!("nodes:\n  - id: test\n    path: test.py\n{block}");
            let err = serde_yaml::from_str::<Descriptor>(&yaml)
                .expect_err("the pre-1.0 `_unstable_` key must be rejected");
            assert!(
                err.to_string().contains(key),
                "error should name `{key}`, got: {err}"
            );
        }
    }

    #[test]
    fn debug_flag_rejects_the_removed_legacy_alias() {
        // The `publish_all_messages_to_zenoh` alias was removed for 1.0. It
        // must *error* rather than deserialize to the default: silently
        // ignoring it would leave debug inspection off while the dataflow
        // looks like it enabled it, and `dora topic echo` would return
        // nothing with no explanation. `deny_unknown_fields` on `Debug` is
        // what turns that into a diagnosable failure.
        let yaml = r#"
nodes:
  - id: test
    path: test.py
debug:
  publish_all_messages_to_zenoh: true
"#;
        let err = serde_yaml::from_str::<Descriptor>(yaml)
            .expect_err("removed alias must be rejected, not silently ignored");
        assert!(
            err.to_string().contains("publish_all_messages_to_zenoh"),
            "error should name the offending field, got: {err}"
        );
    }

    #[test]
    fn operator_source_shared_library_names_its_runtime() {
        let cfg: OperatorConfig = serde_yaml::from_str("shared-library: build/op").unwrap();
        assert!(matches!(&cfg.source, OperatorSource::SharedLibrary(s) if s == "build/op"));
        assert_eq!(cfg.source.runtime_name(), RUNTIME_SHARED_LIBRARY);
        assert_eq!(cfg.source.runtime_name(), "shared-library");
    }

    #[test]
    fn operator_source_python_source_only_names_its_runtime() {
        let cfg: OperatorConfig = serde_yaml::from_str("python: op.py").unwrap();
        assert!(matches!(&cfg.source, OperatorSource::Python(py) if py.source == "op.py"));
        assert_eq!(cfg.source.runtime_name(), RUNTIME_PYTHON);
    }

    #[test]
    fn operator_source_python_with_conda_env_names_its_runtime() {
        let cfg: OperatorConfig =
            serde_yaml::from_str("python:\n  source: op.py\n  conda_env: my-env").unwrap();
        match &cfg.source {
            OperatorSource::Python(py) => {
                assert_eq!(py.source, "op.py");
                assert_eq!(py.conda_env.as_deref(), Some("my-env"));
            }
            other => panic!("expected python source, got {other:?}"),
        }
        assert_eq!(cfg.source.runtime_name(), RUNTIME_PYTHON);
    }

    #[test]
    fn operator_source_wasm_names_its_runtime() {
        let cfg: OperatorConfig = serde_yaml::from_str("wasm: op.wasm").unwrap();
        assert!(matches!(&cfg.source, OperatorSource::Wasm(s) if s == "op.wasm"));
        assert_eq!(cfg.source.runtime_name(), RUNTIME_WASM);
    }

    /// The runtime a node is spawned with must survive a descriptor round-trip:
    /// the daemon re-parses the serialized descriptor before spawning.
    #[test]
    fn operator_source_runtime_survives_a_serde_roundtrip() {
        for yaml in ["shared-library: build/op", "python: op.py", "wasm: op.wasm"] {
            let cfg: OperatorConfig = serde_yaml::from_str(yaml).unwrap();
            let serialized = serde_yaml::to_string(&cfg).unwrap();
            let reparsed: OperatorConfig = serde_yaml::from_str(&serialized).unwrap();
            assert_eq!(cfg.source.runtime_name(), reparsed.source.runtime_name());
        }
    }
}