Skip to main content

cu29_runtime/
config.rs

1//! This module defines the configuration of the copper runtime.
2//! The configuration is a directed graph where nodes are tasks and edges are connections between tasks.
3//! The configuration is serialized in the RON format.
4//! The configuration is used to generate the runtime code at compile time.
5#[cfg(not(feature = "std"))]
6extern crate alloc;
7
8use ConfigGraphs::{Missions, Simple};
9#[cfg(feature = "std")]
10use core::fmt;
11#[cfg(feature = "std")]
12use core::fmt::Display;
13use cu29_traits::{CuError, CuResult};
14use hashbrown::HashMap;
15pub use petgraph::Direction::Incoming;
16pub use petgraph::Direction::Outgoing;
17use petgraph::stable_graph::{EdgeIndex, NodeIndex, StableDiGraph};
18#[cfg(feature = "std")]
19use petgraph::visit::IntoEdgeReferences;
20use petgraph::visit::{Bfs, EdgeRef};
21use ron::Options;
22use ron::extensions::Extensions;
23use serde::{Deserialize, Deserializer, Serialize, Serializer};
24
25#[cfg(not(feature = "std"))]
26use alloc::boxed::Box;
27#[cfg(not(feature = "std"))]
28use alloc::collections::BTreeMap;
29#[cfg(not(feature = "std"))]
30use alloc::vec;
31#[cfg(feature = "std")]
32use std::collections::BTreeMap;
33
34#[cfg(not(feature = "std"))]
35mod imp {
36    pub use alloc::borrow::ToOwned;
37    pub use alloc::format;
38    pub use alloc::string::String;
39    pub use alloc::string::ToString;
40    pub use alloc::vec::Vec;
41}
42
43#[cfg(feature = "std")]
44mod imp {
45    pub use html_escape::encode_text;
46    pub use std::fs::read_to_string;
47}
48
49use imp::*;
50
51mod value;
52
53pub use value::{
54    ComponentConfig, ConfigError, ConstantConfig, ConstantNumber, ConstantStorage, Value,
55};
56
57/// NodeId is the unique identifier of a node in the configuration graph for petgraph
58/// and the code generation.
59pub type NodeId = u32;
60pub const DEFAULT_MISSION_ID: &str = "default";
61/// Default number of preallocated CopperLists compiled into a runtime.
62#[doc(hidden)]
63pub const DEFAULT_COPPERLIST_COUNT: usize = 2;
64
65/// Logging policy for a `CuHandle`'s payload content.
66///
67/// Set by the source that produces the handle (typically via this enum's slot under
68/// `NodeLogging`) and propagated through clones. The unified-log encoder reads this to
69/// decide whether to write the payload bytes or just a metadata-only record for the
70/// frame. See `cu29_runtime::pool::CuHandle` for the runtime side.
71///
72/// Defined here because this policy is part of the serialized runtime configuration.
73#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
74#[repr(u8)]
75pub enum HandleContent {
76    /// Always log the full payload (current default).
77    #[serde(rename = "all", alias = "All")]
78    #[default]
79    All = 0,
80    /// Log the payload only if a downstream consumer called `CuHandle::mark_touched`.
81    #[serde(rename = "touched_only", alias = "TouchedOnly")]
82    TouchedOnly = 1,
83    /// Never log the payload; keep only the surrounding metadata (timestamps, status).
84    #[serde(rename = "none", alias = "None")]
85    None = 2,
86}
87
88impl HandleContent {
89    /// Reconstruct a [`HandleContent`] from its `AtomicU8` representation. Unknown
90    /// values fall back to `All` so corrupt state never silently drops payload bytes.
91    #[allow(dead_code)] // Only the lib's pool module calls this; the graph viewer doesn't.
92    pub fn from_u8(v: u8) -> Self {
93        match v {
94            1 => HandleContent::TouchedOnly,
95            2 => HandleContent::None,
96            _ => HandleContent::All,
97        }
98    }
99}
100
101/// Configuration for logging in the node.
102#[derive(Serialize, Deserialize, Debug, Clone)]
103pub struct NodeLogging {
104    #[serde(default = "default_as_true")]
105    enabled: bool,
106    #[serde(skip_serializing_if = "Option::is_none")]
107    codec: Option<String>,
108    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
109    #[serde(serialize_with = "serialize_ordered_map")]
110    codecs: HashMap<String, String>,
111    /// Logging policy applied to the source's pool-acquired `CuHandle`s. Surfaced
112    /// in user RON config as e.g. `logging: ( handle_content: "touched_only" )`.
113    #[serde(default, skip_serializing_if = "is_default_handle_content")]
114    handle_content: HandleContent,
115}
116
117fn is_default_handle_content(c: &HandleContent) -> bool {
118    *c == HandleContent::default()
119}
120
121impl NodeLogging {
122    #[allow(dead_code)]
123    pub fn enabled(&self) -> bool {
124        self.enabled
125    }
126
127    #[allow(dead_code)]
128    pub fn codec(&self) -> Option<&str> {
129        self.codec.as_deref()
130    }
131
132    #[allow(dead_code)]
133    pub fn codecs(&self) -> &HashMap<String, String> {
134        &self.codecs
135    }
136
137    #[allow(dead_code)]
138    pub fn codec_for_msg_type(&self, msg_type: &str) -> Option<&str> {
139        self.codecs
140            .get(msg_type)
141            .map(String::as_str)
142            .or(self.codec.as_deref())
143    }
144
145    /// Logging policy applied to handles minted by this node's pool. Defaults to
146    /// `HandleContent::All` — i.e. existing behavior.
147    pub fn handle_content(&self) -> HandleContent {
148        self.handle_content
149    }
150}
151
152impl Default for NodeLogging {
153    fn default() -> Self {
154        Self {
155            enabled: true,
156            codec: None,
157            codecs: HashMap::new(),
158            handle_content: HandleContent::default(),
159        }
160    }
161}
162
163/// Distinguishes regular tasks from bridge nodes so downstream stages can apply
164/// bridge-specific instantiation rules.
165#[derive(Default, Debug, Copy, Clone, PartialEq, Eq)]
166pub enum Flavor {
167    #[default]
168    Task,
169    Bridge,
170}
171
172/// Declares which Copper task trait a task node implements.
173///
174/// This lets config express the runtime role explicitly instead of forcing the
175/// proc-macro to guess from graph shape alone.
176#[derive(Serialize, Deserialize, Debug, Copy, Clone, PartialEq, Eq)]
177pub enum TaskKind {
178    #[serde(rename = "source", alias = "src")]
179    Source,
180    #[serde(rename = "task", alias = "regular", alias = "cutask")]
181    Regular,
182    /// A transform implementing `CuStatelessTask`. It has the regular graph
183    /// shape but immutable per-CopperList callbacks.
184    #[serde(rename = "stateless_task", alias = "stateless")]
185    Stateless,
186    #[serde(rename = "sink", alias = "snk")]
187    Sink,
188}
189
190impl TaskKind {
191    #[allow(dead_code)]
192    pub fn as_str(&self) -> &'static str {
193        match self {
194            TaskKind::Source => "source",
195            TaskKind::Regular => "task",
196            TaskKind::Stateless => "stateless_task",
197            TaskKind::Sink => "sink",
198        }
199    }
200}
201
202/// Default thread pool name used by `background: true` tasks.
203pub const DEFAULT_BACKGROUND_POOL: &str = "background";
204
205/// Reserved thread pool name driving the Pipeline execution engine. Applied to
206/// each lane worker at startup; never task-bound nor built as a rayon pool.
207#[allow(dead_code)] // consumed by cu29_derive; unused in some binary targets
208pub const RT_POOL: &str = "rt";
209
210/// How a task is backgrounded.
211///
212/// Either a simple on/off flag (`background: true`), which runs the task on the
213/// default [`DEFAULT_BACKGROUND_POOL`] pool, or an explicit pool selection
214/// (`background: (pool: "vision")`).
215#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
216#[serde(untagged)]
217pub enum BackgroundConfig {
218    /// `background: true` / `background: false`.
219    Flag(bool),
220    /// `background: (pool: "vision")`.
221    Pool { pool: String },
222}
223
224/// Refinement policy for an anytime node (`anytime:` on a task).
225///
226/// Every field is optional, but validation requires at least one - `time_budget_ms`,
227/// `max_age_ms` and `max_refines`; see [`CuConfig::validate_anytime_configs`].
228///
229/// Two orthogonal axes organize the fields:
230///
231/// - **Budget** — how much to *spend* per result: `time_budget_ms` and
232///   `max_refines` are hard bounds, `quality_target` and `max_stall` stop
233///   spending early when more is provably not worth it.
234/// - **Utility** — whether the result is *worth having* at all: `max_age_ms`
235///   (worthless because too old) and `quality_floor` (worthless because too
236///   crude).
237#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default)]
238pub struct AnytimeConfig {
239    /// Wall-clock window for one job in milliseconds, measured from the start of
240    /// the base computation and checked *between* refinement quanta.
241    /// In background placement it is measured on the worker thread and may exceed the
242    /// copperlist period.
243    #[serde(skip_serializing_if = "Option::is_none")]
244    pub time_budget_ms: Option<f64>,
245
246    /// Validity deadline in milliseconds, measured from the input's earliest time
247    /// of validity (Tov): past this data age a result is no longer worth starting
248    /// or waiting for.
249    #[serde(skip_serializing_if = "Option::is_none")]
250    pub max_age_ms: Option<f64>,
251
252    /// Stop refining early once the reported quality reaches this target, in
253    /// `(0.0, 1.0]` on the normalized quality scale. Only valid for tasks that
254    /// report a comparable quality.
255    #[serde(skip_serializing_if = "Option::is_none")]
256    pub quality_target: Option<f32>,
257
258    /// Publish only if the final reported quality is at least this floor, in
259    /// `(0.0, 1.0)` on the normalized quality scale; below it the payload is
260    /// cleared. Only valid for tasks that report a comparable quality.
261    #[serde(skip_serializing_if = "Option::is_none")]
262    pub quality_floor: Option<f32>,
263
264    /// Hard bound on refinement quanta per job.
265    #[serde(skip_serializing_if = "Option::is_none")]
266    pub max_refines: Option<u32>,
267
268    /// Stop after this many quanta without the published quality improving.
269    /// Only valid for tasks that report a comparable quality.
270    #[serde(skip_serializing_if = "Option::is_none")]
271    pub max_stall: Option<u32>,
272}
273
274impl AnytimeConfig {
275    /// Validates the node-local invariants of this policy.
276    ///
277    /// Ranges are written as positive containment checks so a NaN coming from
278    /// the RON fails the check and is rejected, and at least one hard bound is
279    /// mandatory: `quality_target`, `max_stall` and `quality_floor` alone leave
280    /// refinement unbounded.
281    fn validate(&self, task_id: &str) -> CuResult<()> {
282        if let Some(budget) = self.time_budget_ms {
283            let valid = budget.is_finite() && budget > 0.0;
284            if !valid {
285                return Err(CuError::from(format!(
286                    "Task '{task_id}': anytime.time_budget_ms must be a positive number of milliseconds (got {budget})."
287                )));
288            }
289        }
290        if let Some(age) = self.max_age_ms {
291            let valid = age.is_finite() && age > 0.0;
292            if !valid {
293                return Err(CuError::from(format!(
294                    "Task '{task_id}': anytime.max_age_ms must be a positive number of milliseconds (got {age})."
295                )));
296            }
297        }
298        if let Some(target) = self.quality_target {
299            let valid = target > 0.0 && target <= 1.0;
300            if !valid {
301                return Err(CuError::from(format!(
302                    "Task '{task_id}': anytime.quality_target must be within (0.0, 1.0] (got {target})."
303                )));
304            }
305        }
306        if let Some(floor) = self.quality_floor {
307            let valid = floor > 0.0 && floor < 1.0;
308            if !valid {
309                return Err(CuError::from(format!(
310                    "Task '{task_id}': anytime.quality_floor must be within (0.0, 1.0) (got {floor})."
311                )));
312            }
313        }
314        if let Some(refines) = self.max_refines
315            && refines == 0
316        {
317            return Err(CuError::from(format!(
318                "Task '{task_id}': anytime.max_refines must be at least 1."
319            )));
320        }
321        if let Some(stall) = self.max_stall
322            && stall == 0
323        {
324            return Err(CuError::from(format!(
325                "Task '{task_id}': anytime.max_stall must be at least 1."
326            )));
327        }
328        if let (Some(floor), Some(target)) = (self.quality_floor, self.quality_target)
329            && floor > target
330        {
331            return Err(CuError::from(format!(
332                "Task '{task_id}': anytime.quality_floor ({floor}) must not exceed anytime.quality_target ({target}): refinement could stop at the target and then always discard the result."
333            )));
334        }
335        if self.time_budget_ms.is_none() && self.max_age_ms.is_none() && self.max_refines.is_none()
336        {
337            return Err(CuError::from(format!(
338                "Task '{task_id}': anytime needs at least one hard bound: set time_budget_ms, max_age_ms or max_refines. quality_target, max_stall and quality_floor alone leave refinement unbounded."
339            )));
340        }
341        Ok(())
342    }
343}
344
345/// Whether a task output is transmitted or deterministically recomputed on the ground.
346#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default, PartialEq, Eq)]
347#[serde(rename_all = "snake_case")]
348pub enum StreamReplay {
349    #[default]
350    Capture,
351    Reconstruct,
352}
353
354/// Static streaming contract for an ordinary deterministic task.
355#[derive(Serialize, Deserialize, Debug, Clone, Copy, Default)]
356#[serde(deny_unknown_fields)]
357pub struct NodeStreaming {
358    #[serde(default)]
359    pub replay: StreamReplay,
360}
361
362/// A node in the configuration graph.
363/// A node represents a Task in the system Graph.
364#[derive(Serialize, Deserialize, Debug, Clone)]
365pub struct Node {
366    /// Unique node identifier.
367    id: String,
368
369    /// Task rust struct underlying type, e.g. "mymodule::Sensor", etc.
370    #[serde(rename = "type", skip_serializing_if = "Option::is_none")]
371    type_: Option<String>,
372
373    /// Declared Copper task role. When omitted, legacy configs still infer it
374    /// from graph shape when that is unambiguous.
375    #[serde(skip_serializing_if = "Option::is_none")]
376    kind: Option<TaskKind>,
377
378    /// Config passed to the task.
379    #[serde(skip_serializing_if = "Option::is_none")]
380    config: Option<ComponentConfig>,
381
382    /// Resources requested by the task.
383    #[serde(skip_serializing_if = "Option::is_none")]
384    #[serde(serialize_with = "serialize_optional_ordered_map")]
385    resources: Option<HashMap<String, String>>,
386
387    /// Missions for which this task is run.
388    missions: Option<Vec<String>>,
389
390    /// Run this task in the background:
391    /// ie. Will be set to run on a background thread and until it is finished `CuTask::process` will return None.
392    ///
393    /// Accepts either a simple flag (`background: true`, which uses the default
394    /// [`DEFAULT_BACKGROUND_POOL`] pool) or an explicit pool selection
395    /// (`background: (pool: "vision")`).
396    #[serde(skip_serializing_if = "Option::is_none")]
397    background: Option<BackgroundConfig>,
398
399    /// Dispatch empty inputs as background jobs. Completed results are collected on empty ticks.
400    /// Requires a regular background task; defaults to false (empty inputs are skipped).
401    #[serde(skip_serializing_if = "Option::is_none")]
402    background_process_empty: Option<bool>,
403
404    /// Anytime refinement policy for this task (base + bounded refinements).
405    ///
406    /// Only supported on regular tasks. Orthogonal to `background:`, which adds
407    /// the async placement layer on top of the refinement loop.
408    #[serde(skip_serializing_if = "Option::is_none")]
409    anytime: Option<AnytimeConfig>,
410
411    /// Option to include/exclude stubbing for simulation.
412    /// By default, sources and sinks are replaces (stubbed) by the runtime to avoid trying to compile hardware specific code for sensing or actuation.
413    /// In some cases, for example a sink or source used as a middleware bridge, you might want to run the real code even in simulation.
414    /// This option allows to control this behavior.
415    /// Note: Normal tasks will be run in sim and this parameter ignored.
416    #[serde(skip_serializing_if = "Option::is_none")]
417    run_in_sim: Option<bool>,
418
419    /// Config passed to the task.
420    #[serde(skip_serializing_if = "Option::is_none")]
421    logging: Option<NodeLogging>,
422
423    #[serde(skip_serializing_if = "Option::is_none")]
424    streaming: Option<NodeStreaming>,
425
426    /// Node role in the runtime graph (normal task or bridge endpoint).
427    #[serde(skip, default)]
428    flavor: Flavor,
429    /// Message types that are intentionally not connected (NC) in configuration.
430    #[serde(skip, default)]
431    nc_outputs: Vec<String>,
432    /// Original config connection order for each NC output message type.
433    #[serde(skip, default)]
434    nc_output_orders: Vec<usize>,
435}
436
437impl Node {
438    #[allow(dead_code)]
439    pub fn new(id: &str, ptype: &str) -> Self {
440        Node {
441            id: id.to_string(),
442            type_: Some(ptype.to_string()),
443            kind: None,
444            config: None,
445            resources: None,
446            missions: None,
447            background: None,
448            background_process_empty: None,
449            anytime: None,
450            run_in_sim: None,
451            logging: None,
452            streaming: None,
453            flavor: Flavor::Task,
454            nc_outputs: Vec::new(),
455            nc_output_orders: Vec::new(),
456        }
457    }
458
459    #[allow(dead_code)]
460    pub fn new_with_flavor(id: &str, ptype: &str, flavor: Flavor) -> Self {
461        let mut node = Self::new(id, ptype);
462        node.flavor = flavor;
463        node
464    }
465
466    #[allow(dead_code)]
467    pub fn get_id(&self) -> String {
468        self.id.clone()
469    }
470
471    #[allow(dead_code)]
472    pub fn get_type(&self) -> &str {
473        self.type_.as_ref().unwrap()
474    }
475
476    #[allow(dead_code)]
477    pub fn set_type(mut self, name: Option<String>) -> Self {
478        self.type_ = name;
479        self
480    }
481
482    #[allow(dead_code)]
483    pub fn get_declared_task_kind(&self) -> Option<TaskKind> {
484        self.kind
485    }
486
487    #[allow(dead_code)]
488    pub fn set_task_kind(&mut self, kind: Option<TaskKind>) {
489        self.kind = kind;
490    }
491
492    #[allow(dead_code)]
493    pub fn set_resources<I>(&mut self, resources: Option<I>)
494    where
495        I: IntoIterator<Item = (String, String)>,
496    {
497        self.resources = resources.map(|iter| iter.into_iter().collect());
498    }
499
500    #[allow(dead_code)]
501    pub fn is_background(&self) -> bool {
502        match &self.background {
503            Some(BackgroundConfig::Flag(flag)) => *flag,
504            Some(BackgroundConfig::Pool { .. }) => true,
505            None => false,
506        }
507    }
508
509    /// Compile-time empty-input policy consumed by runtime generation.
510    #[doc(hidden)]
511    #[allow(dead_code)]
512    pub fn background_process_empty(&self) -> bool {
513        self.background_process_empty.unwrap_or(false)
514    }
515
516    /// Name of the thread pool this task should run on when backgrounded.
517    /// Defaults to [`DEFAULT_BACKGROUND_POOL`] when no explicit pool is set.
518    #[allow(dead_code)]
519    pub fn background_pool(&self) -> &str {
520        match &self.background {
521            Some(BackgroundConfig::Pool { pool }) => pool.as_str(),
522            _ => DEFAULT_BACKGROUND_POOL,
523        }
524    }
525
526    #[allow(dead_code)]
527    pub fn is_anytime(&self) -> bool {
528        self.anytime.is_some()
529    }
530
531    /// Anytime refinement policy configured on this node, if any.
532    #[allow(dead_code)]
533    pub fn anytime(&self) -> Option<&AnytimeConfig> {
534        self.anytime.as_ref()
535    }
536
537    /// Sets the anytime refinement policy for this node.
538    #[allow(dead_code)]
539    pub fn set_anytime(&mut self, anytime: Option<AnytimeConfig>) {
540        self.anytime = anytime;
541    }
542
543    #[allow(dead_code)]
544    pub fn get_instance_config(&self) -> Option<&ComponentConfig> {
545        self.config.as_ref()
546    }
547
548    #[allow(dead_code)]
549    pub fn get_resources(&self) -> Option<&HashMap<String, String>> {
550        self.resources.as_ref()
551    }
552
553    /// By default, assume a source or a sink is not run in sim.
554    /// Normal tasks will be run in sim and this parameter ignored.
555    #[allow(dead_code)]
556    pub fn is_run_in_sim(&self) -> bool {
557        self.run_in_sim.unwrap_or(false)
558    }
559
560    #[allow(dead_code)]
561    pub fn is_logging_enabled(&self) -> bool {
562        if let Some(logging) = &self.logging {
563            logging.enabled()
564        } else {
565            true
566        }
567    }
568
569    /// Convenience wrapper around [`NodeLogging::handle_content`]: returns the per-handle
570    /// logging policy for this node, defaulting to [`HandleContent::All`] when no
571    /// `logging` block is configured.
572    #[allow(dead_code)]
573    pub fn handle_content_policy(&self) -> HandleContent {
574        self.logging
575            .as_ref()
576            .map(NodeLogging::handle_content)
577            .unwrap_or_default()
578    }
579
580    #[allow(dead_code)]
581    pub fn streaming(&self) -> NodeStreaming {
582        self.streaming.unwrap_or_default()
583    }
584
585    #[allow(dead_code)]
586    pub fn get_logging(&self) -> Option<&NodeLogging> {
587        self.logging.as_ref()
588    }
589
590    #[allow(dead_code)]
591    pub fn get_param<T>(&self, key: &str) -> Result<Option<T>, ConfigError>
592    where
593        T: for<'a> TryFrom<&'a Value, Error = ConfigError>,
594    {
595        let pc = match self.config.as_ref() {
596            Some(pc) => pc,
597            None => return Ok(None),
598        };
599        let ComponentConfig(pc) = pc;
600        match pc.get(key) {
601            Some(v) => T::try_from(v).map(Some),
602            None => Ok(None),
603        }
604    }
605
606    #[allow(dead_code)]
607    pub fn set_param<T: Into<Value>>(&mut self, key: &str, value: T) {
608        if self.config.is_none() {
609            self.config = Some(ComponentConfig(HashMap::new()));
610        }
611        let ComponentConfig(config) = self.config.as_mut().unwrap();
612        config.insert(key.to_string(), value.into());
613    }
614
615    /// Returns whether this node is treated as a normal task or as a bridge.
616    #[allow(dead_code)]
617    pub fn get_flavor(&self) -> Flavor {
618        self.flavor
619    }
620
621    /// Overrides the node flavor; primarily used when injecting bridge nodes.
622    #[allow(dead_code)]
623    pub fn set_flavor(&mut self, flavor: Flavor) {
624        self.flavor = flavor;
625    }
626
627    /// Registers an intentionally unconnected output message type for this node.
628    #[allow(dead_code)]
629    pub fn add_nc_output(&mut self, msg_type: &str, order: usize) {
630        if let Some(pos) = self
631            .nc_outputs
632            .iter()
633            .position(|existing| existing == msg_type)
634        {
635            if order < self.nc_output_orders[pos] {
636                self.nc_output_orders[pos] = order;
637            }
638            return;
639        }
640        self.nc_outputs.push(msg_type.to_string());
641        self.nc_output_orders.push(order);
642    }
643
644    /// Returns message types intentionally marked as not connected.
645    #[allow(dead_code)]
646    pub fn nc_outputs(&self) -> &[String] {
647        &self.nc_outputs
648    }
649
650    /// Returns NC outputs paired with original config order.
651    #[allow(dead_code)]
652    pub fn nc_outputs_with_order(&self) -> impl Iterator<Item = (&String, usize)> {
653        self.nc_outputs
654            .iter()
655            .zip(self.nc_output_orders.iter().copied())
656    }
657}
658
659/// Directional mapping for bridge channels.
660#[derive(Serialize, Deserialize, Debug, Clone)]
661pub enum BridgeChannelConfigRepresentation {
662    /// Channel that receives data from the bridge into the graph.
663    Rx {
664        id: String,
665        /// Optional transport/topic identifier specific to the bridge backend.
666        #[serde(skip_serializing_if = "Option::is_none")]
667        route: Option<String>,
668        /// Optional per-channel configuration forwarded to the bridge implementation.
669        #[serde(skip_serializing_if = "Option::is_none")]
670        config: Option<ComponentConfig>,
671    },
672    /// Channel that transmits data from the graph into the bridge.
673    Tx {
674        id: String,
675        /// Optional transport/topic identifier specific to the bridge backend.
676        #[serde(skip_serializing_if = "Option::is_none")]
677        route: Option<String>,
678        /// Optional per-channel configuration forwarded to the bridge implementation.
679        #[serde(skip_serializing_if = "Option::is_none")]
680        config: Option<ComponentConfig>,
681    },
682}
683
684impl BridgeChannelConfigRepresentation {
685    /// Stable logical identifier to reference this channel in connections.
686    #[allow(dead_code)]
687    pub fn id(&self) -> &str {
688        match self {
689            BridgeChannelConfigRepresentation::Rx { id, .. }
690            | BridgeChannelConfigRepresentation::Tx { id, .. } => id,
691        }
692    }
693
694    /// Bridge-specific transport path (topic, route, path...) describing this channel.
695    #[allow(dead_code)]
696    pub fn route(&self) -> Option<&str> {
697        match self {
698            BridgeChannelConfigRepresentation::Rx { route, .. }
699            | BridgeChannelConfigRepresentation::Tx { route, .. } => route.as_deref(),
700        }
701    }
702}
703
704enum EndpointRole {
705    Source,
706    Destination,
707}
708
709fn validate_bridge_channel(
710    bridge: &BridgeConfig,
711    channel_id: &str,
712    role: EndpointRole,
713) -> Result<(), String> {
714    let channel = bridge
715        .channels
716        .iter()
717        .find(|ch| ch.id() == channel_id)
718        .ok_or_else(|| {
719            format!(
720                "Bridge '{}' does not declare a channel named '{}'",
721                bridge.id, channel_id
722            )
723        })?;
724
725    match (role, channel) {
726        (EndpointRole::Source, BridgeChannelConfigRepresentation::Rx { .. }) => Ok(()),
727        (EndpointRole::Destination, BridgeChannelConfigRepresentation::Tx { .. }) => Ok(()),
728        (EndpointRole::Source, BridgeChannelConfigRepresentation::Tx { .. }) => Err(format!(
729            "Bridge '{}' channel '{}' is Tx and cannot act as a source",
730            bridge.id, channel_id
731        )),
732        (EndpointRole::Destination, BridgeChannelConfigRepresentation::Rx { .. }) => Err(format!(
733            "Bridge '{}' channel '{}' is Rx and cannot act as a destination",
734            bridge.id, channel_id
735        )),
736    }
737}
738
739/// Declarative definition of a resource bundle.
740#[derive(Serialize, Deserialize, Debug, Clone)]
741pub struct ResourceBundleConfig {
742    /// Resource inputs consumed by this provider at startup.
743    #[serde(default, skip_serializing_if = "Option::is_none")]
744    #[serde(serialize_with = "serialize_optional_ordered_map")]
745    pub resources: Option<HashMap<String, String>>,
746    pub id: String,
747    #[serde(rename = "provider")]
748    pub provider: String,
749    #[serde(skip_serializing_if = "Option::is_none")]
750    pub config: Option<ComponentConfig>,
751    #[serde(skip_serializing_if = "Option::is_none")]
752    pub missions: Option<Vec<String>>,
753}
754
755/// Static log-streaming policy compiled into a Copper application.
756#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
757#[serde(deny_unknown_fields)]
758pub struct LogStreamingConfig {
759    pub destinations: Vec<LogStreamDestinationConfig>,
760}
761
762/// One statically generated log-stream destination.
763#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
764#[serde(deny_unknown_fields)]
765pub struct LogStreamDestinationConfig {
766    #[serde(default, skip_serializing_if = "Option::is_none")]
767    pub feedback: Option<LogStreamFeedbackConfig>,
768    pub id: String,
769    pub transport: LogStreamTransportConfig,
770    pub link: LogStreamLinkConfig,
771    pub fec: LogStreamFecConfig,
772    /// Recovery interval in CopperLists; a nonzero multiple of logging.keyframe_interval.
773    pub recovery_interval: u32,
774    pub max_record_bytes: u64,
775}
776
777/// Explicit reverse resource and advisory feedback policy for one destination.
778#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
779#[serde(deny_unknown_fields)]
780pub struct LogStreamFeedbackConfig {
781    pub transport: LogStreamTransportConfig,
782    pub report_interval_ms: u32,
783    pub timeout_ms: u32,
784    #[serde(default, skip_serializing_if = "Option::is_none")]
785    pub adaptation: Option<LogStreamAdaptationConfig>,
786}
787
788/// Bounds on the number of future source symbols between continuous repairs.
789#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
790#[serde(deny_unknown_fields)]
791pub struct LogStreamAdaptationConfig {
792    pub min_repair_every_source_symbols: u16,
793    pub max_repair_every_source_symbols: u16,
794}
795
796/// Concrete Copper resource used as the destination's packet transmitter.
797#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
798#[serde(deny_unknown_fields)]
799pub struct LogStreamTransportConfig {
800    #[serde(rename = "type")]
801    pub type_: String,
802    pub resource: String,
803}
804
805/// Physical-link assumptions and sender bounds.
806#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
807#[serde(deny_unknown_fields)]
808pub struct LogStreamLinkConfig {
809    pub mtu_bytes: u16,
810    pub bitrate_bps: u64,
811    pub memory_budget_kib: u32,
812    pub max_latency_ms: u32,
813    pub burst_packets: u32,
814}
815
816/// Explicit FEC policy. Lane identity fixes the algorithms: continuous is RLC
817/// and objects is RaptorQ, so there is deliberately no configurable scheme.
818#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
819#[serde(deny_unknown_fields)]
820pub struct LogStreamFecConfig {
821    pub continuous: LogStreamContinuousFecConfig,
822    pub objects: LogStreamObjectFecConfig,
823}
824
825#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
826#[serde(deny_unknown_fields)]
827pub struct LogStreamContinuousFecConfig {
828    pub field: LogStreamRlcField,
829    pub window_symbols: u16,
830    pub repair_every_source_symbols: u16,
831    pub repair_density: LogStreamRepairDensity,
832}
833
834#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
835pub enum LogStreamRlcField {
836    Gf2,
837    Gf256,
838}
839
840#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
841pub enum LogStreamRepairDensity {
842    Full,
843    Threshold(u8),
844}
845
846impl LogStreamRepairDensity {
847    pub const fn threshold(self) -> u8 {
848        match self {
849            Self::Full => 15,
850            Self::Threshold(value) => value,
851        }
852    }
853}
854
855#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
856#[serde(deny_unknown_fields)]
857pub struct LogStreamObjectFecConfig {
858    pub max_object_bytes: u64,
859    pub repair_symbols_per_block: u32,
860}
861
862/// Declarative definition of a bridge component with a list of channels.
863#[derive(Serialize, Deserialize, Debug, Clone)]
864pub struct BridgeConfig {
865    pub id: String,
866    #[serde(rename = "type")]
867    pub type_: String,
868    #[serde(skip_serializing_if = "Option::is_none")]
869    pub config: Option<ComponentConfig>,
870    #[serde(skip_serializing_if = "Option::is_none")]
871    #[serde(serialize_with = "serialize_optional_ordered_map")]
872    pub resources: Option<HashMap<String, String>>,
873    #[serde(skip_serializing_if = "Option::is_none")]
874    pub missions: Option<Vec<String>>,
875    /// Whether this bridge should run as the real implementation in simulation mode.
876    ///
877    /// Default is `true` to preserve historical behavior where bridges were always
878    /// instantiated in sim mode.
879    #[serde(skip_serializing_if = "Option::is_none")]
880    pub run_in_sim: Option<bool>,
881    /// List of logical endpoints exposed by this bridge.
882    pub channels: Vec<BridgeChannelConfigRepresentation>,
883}
884
885impl BridgeConfig {
886    /// By default, bridges run as real implementations in sim mode for backward compatibility.
887    #[allow(dead_code)]
888    pub fn is_run_in_sim(&self) -> bool {
889        self.run_in_sim.unwrap_or(true)
890    }
891
892    fn to_node(&self) -> Node {
893        let mut node = Node::new_with_flavor(&self.id, &self.type_, Flavor::Bridge);
894        node.config = self.config.clone();
895        node.resources = self.resources.clone();
896        node.missions = self.missions.clone();
897        node
898    }
899}
900
901fn insert_bridge_node(graph: &mut CuGraph, bridge: &BridgeConfig) -> Result<(), String> {
902    if graph.get_node_id_by_name(bridge.id.as_str()).is_some() {
903        return Err(format!(
904            "Bridge '{}' reuses an existing node id. Bridge ids must be unique.",
905            bridge.id
906        ));
907    }
908    graph
909        .add_node(bridge.to_node())
910        .map(|_| ())
911        .map_err(|e| e.to_string())
912}
913
914/// Serialized representation of a connection used for the RON config.
915#[derive(Serialize, Deserialize, Debug, Clone)]
916struct SerializedCnx {
917    src: String,
918    dst: String,
919    msg: String,
920    missions: Option<Vec<String>>,
921}
922
923/// Special destination endpoint used to mark an output as intentionally not connected.
924pub const NC_ENDPOINT: &str = "__nc__";
925
926/// This represents a connection between 2 tasks (nodes) in the configuration graph.
927#[derive(Debug, Clone)]
928pub struct Cnx {
929    /// Source node id.
930    pub src: String,
931    /// Destination node id.
932    pub dst: String,
933    /// Message type exchanged between src and dst.
934    pub msg: String,
935    /// Restrict this connection for this list of missions.
936    pub missions: Option<Vec<String>>,
937    /// Optional channel id when the source endpoint is a bridge.
938    pub src_channel: Option<String>,
939    /// Optional channel id when the destination endpoint is a bridge.
940    pub dst_channel: Option<String>,
941    /// Original serialized connection index used to preserve output ordering.
942    pub order: usize,
943}
944
945impl From<&Cnx> for SerializedCnx {
946    fn from(cnx: &Cnx) -> Self {
947        SerializedCnx {
948            src: format_endpoint(&cnx.src, cnx.src_channel.as_deref()),
949            dst: format_endpoint(&cnx.dst, cnx.dst_channel.as_deref()),
950            msg: cnx.msg.clone(),
951            missions: cnx.missions.clone(),
952        }
953    }
954}
955
956fn format_endpoint(node: &str, channel: Option<&str>) -> String {
957    match channel {
958        Some(ch) => format!("{node}/{ch}"),
959        None => node.to_string(),
960    }
961}
962
963fn parse_endpoint(
964    endpoint: &str,
965    role: EndpointRole,
966    bridges: &HashMap<&str, &BridgeConfig>,
967) -> Result<(String, Option<String>), String> {
968    if let Some((node, channel)) = endpoint.split_once('/') {
969        if let Some(bridge) = bridges.get(node) {
970            validate_bridge_channel(bridge, channel, role)?;
971            return Ok((node.to_string(), Some(channel.to_string())));
972        } else {
973            return Err(format!(
974                "Endpoint '{endpoint}' references an unknown bridge '{node}'"
975            ));
976        }
977    }
978
979    if let Some(bridge) = bridges.get(endpoint) {
980        return Err(format!(
981            "Bridge '{}' connections must reference a channel using '{}/<channel>'",
982            bridge.id, bridge.id
983        ));
984    }
985
986    Ok((endpoint.to_string(), None))
987}
988
989fn build_bridge_lookup(bridges: Option<&Vec<BridgeConfig>>) -> HashMap<&str, &BridgeConfig> {
990    let mut map = HashMap::new();
991    if let Some(bridges) = bridges {
992        for bridge in bridges {
993            map.insert(bridge.id.as_str(), bridge);
994        }
995    }
996    map
997}
998
999fn mission_applies(missions: &Option<Vec<String>>, mission_id: &str) -> bool {
1000    missions
1001        .as_ref()
1002        .map(|mission_list| mission_list.iter().any(|m| m == mission_id))
1003        .unwrap_or(true)
1004}
1005
1006fn merge_connection_missions(existing: &mut Option<Vec<String>>, incoming: &Option<Vec<String>>) {
1007    if incoming.is_none() {
1008        *existing = None;
1009        return;
1010    }
1011    if existing.is_none() {
1012        return;
1013    }
1014
1015    if let (Some(existing_missions), Some(incoming_missions)) =
1016        (existing.as_mut(), incoming.as_ref())
1017    {
1018        for mission in incoming_missions {
1019            if !existing_missions
1020                .iter()
1021                .any(|existing_mission| existing_mission == mission)
1022            {
1023                existing_missions.push(mission.clone());
1024            }
1025        }
1026        existing_missions.sort();
1027        existing_missions.dedup();
1028    }
1029}
1030
1031fn register_nc_output<E>(
1032    graph: &mut CuGraph,
1033    src_endpoint: &str,
1034    msg_type: &str,
1035    order: usize,
1036    bridge_lookup: &HashMap<&str, &BridgeConfig>,
1037) -> Result<(), E>
1038where
1039    E: From<String>,
1040{
1041    let (src_name, src_channel) =
1042        parse_endpoint(src_endpoint, EndpointRole::Source, bridge_lookup).map_err(E::from)?;
1043    if src_channel.is_some() {
1044        return Err(E::from(format!(
1045            "NC destination '{}' does not support bridge channels in source endpoint '{}'",
1046            NC_ENDPOINT, src_endpoint
1047        )));
1048    }
1049
1050    let src = graph
1051        .get_node_id_by_name(src_name.as_str())
1052        .ok_or_else(|| E::from(format!("Source node not found: {src_endpoint}")))?;
1053    let src_node = graph
1054        .get_node_mut(src)
1055        .ok_or_else(|| E::from(format!("Source node id {src} not found for NC output")))?;
1056    if src_node.get_flavor() != Flavor::Task {
1057        return Err(E::from(format!(
1058            "NC destination '{}' is only supported for task outputs (source '{}')",
1059            NC_ENDPOINT, src_endpoint
1060        )));
1061    }
1062    src_node.add_nc_output(msg_type, order);
1063    Ok(())
1064}
1065
1066/// A simple wrapper enum for `petgraph::Direction`,
1067/// designed to be converted *into* it via the `From` trait.
1068#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1069pub enum CuDirection {
1070    Outgoing,
1071    Incoming,
1072}
1073
1074impl From<CuDirection> for petgraph::Direction {
1075    fn from(dir: CuDirection) -> Self {
1076        match dir {
1077            CuDirection::Outgoing => petgraph::Direction::Outgoing,
1078            CuDirection::Incoming => petgraph::Direction::Incoming,
1079        }
1080    }
1081}
1082
1083#[derive(Default, Debug, Clone)]
1084pub struct CuGraph(pub StableDiGraph<Node, Cnx, NodeId>);
1085
1086impl CuGraph {
1087    #[allow(dead_code)]
1088    pub fn get_all_nodes(&self) -> Vec<(NodeId, &Node)> {
1089        self.0
1090            .node_indices()
1091            .map(|index| (index.index() as u32, &self.0[index]))
1092            .collect()
1093    }
1094
1095    #[allow(dead_code)]
1096    pub fn get_neighbor_ids(&self, node_id: NodeId, dir: CuDirection) -> Vec<NodeId> {
1097        self.0
1098            .neighbors_directed(node_id.into(), dir.into())
1099            .map(|petgraph_index| petgraph_index.index() as NodeId)
1100            .collect()
1101    }
1102
1103    #[allow(dead_code)]
1104    pub fn node_ids(&self) -> Vec<NodeId> {
1105        self.0
1106            .node_indices()
1107            .map(|index| index.index() as NodeId)
1108            .collect()
1109    }
1110
1111    #[allow(dead_code)]
1112    pub fn edge_id_between(&self, source: NodeId, target: NodeId) -> Option<usize> {
1113        self.0
1114            .find_edge(source.into(), target.into())
1115            .map(|edge| edge.index())
1116    }
1117
1118    #[allow(dead_code)]
1119    pub fn edge(&self, edge_id: usize) -> Option<&Cnx> {
1120        self.0.edge_weight(EdgeIndex::new(edge_id))
1121    }
1122
1123    #[allow(dead_code)]
1124    pub fn edges(&self) -> impl Iterator<Item = &Cnx> {
1125        self.0
1126            .edge_indices()
1127            .filter_map(|edge| self.0.edge_weight(edge))
1128    }
1129
1130    #[allow(dead_code)]
1131    pub fn bfs_nodes(&self, start: NodeId) -> Vec<NodeId> {
1132        let mut visitor = Bfs::new(&self.0, start.into());
1133        let mut nodes = Vec::new();
1134        while let Some(node) = visitor.next(&self.0) {
1135            nodes.push(node.index() as NodeId);
1136        }
1137        nodes
1138    }
1139
1140    #[allow(dead_code)]
1141    pub fn incoming_neighbor_count(&self, node_id: NodeId) -> usize {
1142        self.0.neighbors_directed(node_id.into(), Incoming).count()
1143    }
1144
1145    #[allow(dead_code)]
1146    pub fn outgoing_neighbor_count(&self, node_id: NodeId) -> usize {
1147        self.0.neighbors_directed(node_id.into(), Outgoing).count()
1148    }
1149
1150    pub fn node_indices(&self) -> Vec<petgraph::stable_graph::NodeIndex> {
1151        self.0.node_indices().collect()
1152    }
1153
1154    pub fn add_node(&mut self, node: Node) -> CuResult<NodeId> {
1155        Ok(self.0.add_node(node).index() as NodeId)
1156    }
1157
1158    #[allow(dead_code)]
1159    pub fn connection_exists(&self, source: NodeId, target: NodeId) -> bool {
1160        self.0.find_edge(source.into(), target.into()).is_some()
1161    }
1162
1163    pub fn connect_ext(
1164        &mut self,
1165        source: NodeId,
1166        target: NodeId,
1167        msg_type: &str,
1168        missions: Option<Vec<String>>,
1169        src_channel: Option<String>,
1170        dst_channel: Option<String>,
1171    ) -> CuResult<()> {
1172        self.connect_ext_with_order(
1173            source,
1174            target,
1175            msg_type,
1176            missions,
1177            src_channel,
1178            dst_channel,
1179            usize::MAX,
1180        )
1181    }
1182
1183    #[allow(clippy::too_many_arguments)]
1184    pub fn connect_ext_with_order(
1185        &mut self,
1186        source: NodeId,
1187        target: NodeId,
1188        msg_type: &str,
1189        missions: Option<Vec<String>>,
1190        src_channel: Option<String>,
1191        dst_channel: Option<String>,
1192        order: usize,
1193    ) -> CuResult<()> {
1194        let (src_id, dst_id) = (
1195            self.0
1196                .node_weight(source.into())
1197                .ok_or("Source node not found")?
1198                .id
1199                .clone(),
1200            self.0
1201                .node_weight(target.into())
1202                .ok_or("Target node not found")?
1203                .id
1204                .clone(),
1205        );
1206
1207        let _ = self.0.add_edge(
1208            petgraph::stable_graph::NodeIndex::from(source),
1209            petgraph::stable_graph::NodeIndex::from(target),
1210            Cnx {
1211                src: src_id,
1212                dst: dst_id,
1213                msg: msg_type.to_string(),
1214                missions,
1215                src_channel,
1216                dst_channel,
1217                order,
1218            },
1219        );
1220        Ok(())
1221    }
1222    /// Get the node with the given id.
1223    /// If mission_id is provided, get the node from that mission's graph.
1224    /// Otherwise get the node from the simple graph.
1225    #[allow(dead_code)]
1226    pub fn get_node(&self, node_id: NodeId) -> Option<&Node> {
1227        self.0.node_weight(node_id.into())
1228    }
1229
1230    #[allow(dead_code)]
1231    pub fn get_node_weight(&self, index: NodeId) -> Option<&Node> {
1232        self.0.node_weight(index.into())
1233    }
1234
1235    #[allow(dead_code)]
1236    pub fn get_node_mut(&mut self, node_id: NodeId) -> Option<&mut Node> {
1237        self.0.node_weight_mut(node_id.into())
1238    }
1239
1240    pub fn get_node_id_by_name(&self, name: &str) -> Option<NodeId> {
1241        self.0
1242            .node_indices()
1243            .into_iter()
1244            .find(|idx| self.0[*idx].get_id() == name)
1245            .map(|i| i.index() as NodeId)
1246    }
1247
1248    #[allow(dead_code)]
1249    pub fn get_edge_weight(&self, index: usize) -> Option<Cnx> {
1250        self.0.edge_weight(EdgeIndex::new(index)).cloned()
1251    }
1252
1253    #[allow(dead_code)]
1254    pub fn get_node_output_msg_type(&self, node_id: &str) -> Option<String> {
1255        self.get_node_output_msg_types(node_id)
1256            .and_then(|mut msgs| msgs.drain(..1).next())
1257    }
1258
1259    #[allow(dead_code)]
1260    pub fn get_node_output_msg_types(&self, node_id: &str) -> Option<Vec<String>> {
1261        let node_id = self.get_node_id_by_name(node_id)?;
1262        let msgs = self.get_node_output_msg_types_by_id(node_id).ok()?;
1263        (!msgs.is_empty()).then_some(msgs)
1264    }
1265
1266    #[allow(dead_code)]
1267    pub fn get_node_output_msg_types_by_id(&self, node_id: NodeId) -> CuResult<Vec<String>> {
1268        let mut edge_ids = self.get_src_edges(node_id)?;
1269        edge_ids.sort();
1270
1271        let node = self
1272            .get_node(node_id)
1273            .ok_or_else(|| CuError::from(format!("Node id {node_id} not found")))?;
1274
1275        let mut msg_order: Vec<(usize, String)> = Vec::new();
1276        let mut record_msg = |msg: String, order: usize| {
1277            if let Some((existing_order, _)) = msg_order
1278                .iter_mut()
1279                .find(|(_, existing_msg)| *existing_msg == msg)
1280            {
1281                if order < *existing_order {
1282                    *existing_order = order;
1283                }
1284                return;
1285            }
1286            msg_order.push((order, msg));
1287        };
1288
1289        for edge_id in edge_ids {
1290            let Some(edge) = self.edge(edge_id) else {
1291                continue;
1292            };
1293            let order = if edge.order == usize::MAX {
1294                edge_id
1295            } else {
1296                edge.order
1297            };
1298            record_msg(edge.msg.clone(), order);
1299        }
1300
1301        for (msg, order) in node.nc_outputs_with_order() {
1302            record_msg(msg.clone(), order);
1303        }
1304
1305        msg_order.sort_by(|(order_a, msg_a), (order_b, msg_b)| {
1306            order_a.cmp(order_b).then_with(|| msg_a.cmp(msg_b))
1307        });
1308        Ok(msg_order.into_iter().map(|(_, msg)| msg).collect())
1309    }
1310
1311    /// Channel-aware variant of [`CuGraph::get_node_output_msg_types_by_id`].
1312    ///
1313    /// Returns each distinct output port as `(msg_type, src_channel)`, where
1314    /// `src_channel` is `None` for ordinary task ports and `Some(id)` for
1315    /// bridge-channel ports. Unlike the msg-only variant, two ports sharing the
1316    /// same message type but living on different bridge channels are kept as
1317    /// separate entries, preserving their relative ordering (see #791).
1318    #[allow(dead_code)]
1319    pub fn get_node_output_ports_by_id(
1320        &self,
1321        node_id: NodeId,
1322    ) -> CuResult<Vec<(String, Option<String>)>> {
1323        let mut edge_ids = self.get_src_edges(node_id)?;
1324        edge_ids.sort();
1325
1326        let node = self
1327            .get_node(node_id)
1328            .ok_or_else(|| CuError::from(format!("Node id {node_id} not found")))?;
1329
1330        let mut port_order: Vec<(usize, String, Option<String>)> = Vec::new();
1331        let mut record_port = |msg: String, channel: Option<String>, order: usize| {
1332            if let Some((existing_order, _, _)) = port_order
1333                .iter_mut()
1334                .find(|(_, m, c)| *m == msg && *c == channel)
1335            {
1336                if order < *existing_order {
1337                    *existing_order = order;
1338                }
1339                return;
1340            }
1341            port_order.push((order, msg, channel));
1342        };
1343
1344        for edge_id in edge_ids {
1345            let Some(edge) = self.edge(edge_id) else {
1346                continue;
1347            };
1348            let order = if edge.order == usize::MAX {
1349                edge_id
1350            } else {
1351                edge.order
1352            };
1353            record_port(edge.msg.clone(), edge.src_channel.clone(), order);
1354        }
1355
1356        for (msg, order) in node.nc_outputs_with_order() {
1357            record_port(msg.clone(), None, order);
1358        }
1359
1360        port_order.sort_by(|(order_a, msg_a, ch_a), (order_b, msg_b, ch_b)| {
1361            order_a
1362                .cmp(order_b)
1363                .then_with(|| msg_a.cmp(msg_b))
1364                .then_with(|| ch_a.cmp(ch_b))
1365        });
1366        Ok(port_order
1367            .into_iter()
1368            .map(|(_, msg, ch)| (msg, ch))
1369            .collect())
1370    }
1371
1372    #[allow(dead_code)]
1373    pub fn get_node_input_msg_type(&self, node_id: &str) -> Option<String> {
1374        self.get_node_input_msg_types(node_id)
1375            .and_then(|mut v| v.pop())
1376    }
1377
1378    pub fn get_node_input_msg_types(&self, node_id: &str) -> Option<Vec<String>> {
1379        self.0.node_indices().find_map(|node_index| {
1380            if let Some(node) = self.0.node_weight(node_index) {
1381                if node.id != node_id {
1382                    return None;
1383                }
1384                let edges: Vec<_> = self
1385                    .0
1386                    .edges_directed(node_index, Incoming)
1387                    .map(|edge| edge.id().index())
1388                    .collect();
1389                if edges.is_empty() {
1390                    return None;
1391                }
1392                let mut edges = edges;
1393                edges.sort();
1394                let msgs = edges
1395                    .into_iter()
1396                    .map(|edge_id| {
1397                        let cnx = self
1398                            .0
1399                            .edge_weight(EdgeIndex::new(edge_id))
1400                            .expect("Found an cnx id but could not retrieve it back");
1401                        cnx.msg.clone()
1402                    })
1403                    .collect();
1404                return Some(msgs);
1405            }
1406            None
1407        })
1408    }
1409
1410    #[allow(dead_code)]
1411    pub fn get_connection_msg_type(&self, source: NodeId, target: NodeId) -> Option<&str> {
1412        self.0
1413            .find_edge(source.into(), target.into())
1414            .map(|edge_index| self.0[edge_index].msg.as_str())
1415    }
1416
1417    /// Get the list of edges that are connected to the given node as a source.
1418    fn get_edges_by_direction(
1419        &self,
1420        node_id: NodeId,
1421        direction: petgraph::Direction,
1422    ) -> CuResult<Vec<usize>> {
1423        Ok(self
1424            .0
1425            .edges_directed(node_id.into(), direction)
1426            .map(|edge| edge.id().index())
1427            .collect())
1428    }
1429
1430    pub fn get_src_edges(&self, node_id: NodeId) -> CuResult<Vec<usize>> {
1431        self.get_edges_by_direction(node_id, Outgoing)
1432    }
1433
1434    /// Get the list of edges that are connected to the given node as a destination.
1435    pub fn get_dst_edges(&self, node_id: NodeId) -> CuResult<Vec<usize>> {
1436        self.get_edges_by_direction(node_id, Incoming)
1437    }
1438
1439    #[allow(dead_code)]
1440    pub fn node_count(&self) -> usize {
1441        self.0.node_count()
1442    }
1443
1444    #[allow(dead_code)]
1445    pub fn edge_count(&self) -> usize {
1446        self.0.edge_count()
1447    }
1448
1449    /// Adds an edge between two nodes/tasks in the configuration graph.
1450    /// msg_type is the type of message exchanged between the two nodes/tasks.
1451    #[allow(dead_code)]
1452    pub fn connect(&mut self, source: NodeId, target: NodeId, msg_type: &str) -> CuResult<()> {
1453        self.connect_ext(source, target, msg_type, None, None, None)
1454    }
1455}
1456
1457fn validate_task_kind(
1458    node_id: &str,
1459    kind: TaskKind,
1460    has_inputs: bool,
1461    has_outputs: bool,
1462) -> CuResult<()> {
1463    match kind {
1464        TaskKind::Source if has_inputs => Err(CuError::from(format!(
1465            "Task '{node_id}' is declared as kind 'source' but has incoming connections. Sources map to CuSrcTask and cannot consume inputs. Use kind: task instead."
1466        ))),
1467        TaskKind::Regular | TaskKind::Stateless if !has_inputs => Err(CuError::from(format!(
1468            "Task '{node_id}' is declared as kind '{}' but has no incoming connections. Transform tasks need at least one input connection. Use kind: source if it is input-free.",
1469            kind.as_str()
1470        ))),
1471        TaskKind::Sink if has_outputs => Err(CuError::from(format!(
1472            "Task '{node_id}' is declared as kind 'sink' but has outgoing or NC outputs. Sinks map to CuSinkTask and cannot produce outputs. Use kind: task instead."
1473        ))),
1474        TaskKind::Sink if !has_inputs => Err(CuError::from(format!(
1475            "Task '{node_id}' is declared as kind 'sink' but has no incoming connections. Sinks need at least one input connection so Copper can determine their input message type."
1476        ))),
1477        _ => Ok(()),
1478    }
1479}
1480
1481#[allow(dead_code)]
1482pub fn infer_task_kind_for_id(graph: &CuGraph, node_id: NodeId) -> Option<TaskKind> {
1483    let node = graph.get_node(node_id)?;
1484    if node.get_flavor() != Flavor::Task {
1485        return None;
1486    }
1487
1488    let has_inputs = !graph.get_dst_edges(node_id).ok()?.is_empty();
1489    let has_outputs = !graph
1490        .get_node_output_msg_types_by_id(node_id)
1491        .ok()?
1492        .is_empty();
1493
1494    match (has_inputs, has_outputs) {
1495        (false, true) => Some(TaskKind::Source),
1496        (true, true) => Some(TaskKind::Regular),
1497        (true, false) => Some(TaskKind::Sink),
1498        (false, false) => None,
1499    }
1500}
1501
1502#[allow(dead_code)]
1503pub fn resolve_task_kind_for_id(graph: &CuGraph, node_id: NodeId) -> CuResult<TaskKind> {
1504    let node = graph
1505        .get_node(node_id)
1506        .ok_or_else(|| CuError::from(format!("Task node id {node_id} not found")))?;
1507    if node.get_flavor() != Flavor::Task {
1508        return Err(CuError::from(format!(
1509            "Node '{}' is not a task and does not have a task kind.",
1510            node.id
1511        )));
1512    }
1513
1514    let has_inputs = !graph.get_dst_edges(node_id)?.is_empty();
1515    let has_outputs = !graph.get_node_output_msg_types_by_id(node_id)?.is_empty();
1516
1517    if let Some(kind) = node.get_declared_task_kind() {
1518        validate_task_kind(node.id.as_str(), kind, has_inputs, has_outputs)?;
1519        return Ok(kind);
1520    }
1521
1522    let inferred = match (has_inputs, has_outputs) {
1523        (false, true) => TaskKind::Source,
1524        (true, true) => TaskKind::Regular,
1525        (true, false) => TaskKind::Sink,
1526        (false, false) => {
1527            return Err(CuError::from(format!(
1528                "Task '{}' has no declared inputs or outputs, so Copper cannot infer whether it is a source, task, or sink. Add `kind: source|task|sink`; source/task nodes also need an output declaration via a connection or `dst: \"{NC_ENDPOINT}\"`.",
1529                node.id
1530            )));
1531        }
1532    };
1533
1534    validate_task_kind(node.id.as_str(), inferred, has_inputs, has_outputs)?;
1535    Ok(inferred)
1536}
1537
1538impl core::ops::Index<NodeIndex> for CuGraph {
1539    type Output = Node;
1540
1541    fn index(&self, index: NodeIndex) -> &Self::Output {
1542        &self.0[index]
1543    }
1544}
1545
1546#[derive(Debug, Clone)]
1547pub enum ConfigGraphs {
1548    Simple(CuGraph),
1549    Missions(HashMap<String, CuGraph>),
1550}
1551
1552impl ConfigGraphs {
1553    /// Returns a consistent hashmap of mission names to Graphs whatever the shape of the config is.
1554    /// Note: if there is only one anonymous mission it will be called "default"
1555    #[allow(dead_code)]
1556    pub fn get_all_missions_graphs(&self) -> HashMap<String, CuGraph> {
1557        match self {
1558            Simple(graph) => HashMap::from([(DEFAULT_MISSION_ID.to_string(), graph.clone())]),
1559            Missions(graphs) => graphs.clone(),
1560        }
1561    }
1562
1563    #[allow(dead_code)]
1564    pub fn get_default_mission_graph(&self) -> CuResult<&CuGraph> {
1565        match self {
1566            Simple(graph) => Ok(graph),
1567            Missions(graphs) => {
1568                if graphs.len() == 1 {
1569                    Ok(graphs.values().next().unwrap())
1570                } else {
1571                    Err("Cannot get default mission graph from mission config".into())
1572                }
1573            }
1574        }
1575    }
1576
1577    #[allow(dead_code)]
1578    pub fn get_graph(&self, mission_id: Option<&str>) -> CuResult<&CuGraph> {
1579        match self {
1580            Simple(graph) => match mission_id {
1581                None | Some(DEFAULT_MISSION_ID) => Ok(graph),
1582                Some(_) => Err("Cannot get mission graph from simple config".into()),
1583            },
1584            Missions(graphs) => {
1585                let id = mission_id
1586                    .ok_or_else(|| "Mission ID required for mission configs".to_string())?;
1587                graphs
1588                    .get(id)
1589                    .ok_or_else(|| format!("Mission {id} not found").into())
1590            }
1591        }
1592    }
1593
1594    #[allow(dead_code)]
1595    pub fn get_graph_mut(&mut self, mission_id: Option<&str>) -> CuResult<&mut CuGraph> {
1596        match self {
1597            Simple(graph) => match mission_id {
1598                None => Ok(graph),
1599                Some(_) => Err("Cannot get mission graph from simple config".into()),
1600            },
1601            Missions(graphs) => {
1602                let id = mission_id
1603                    .ok_or_else(|| "Mission ID required for mission configs".to_string())?;
1604                graphs
1605                    .get_mut(id)
1606                    .ok_or_else(|| format!("Mission {id} not found").into())
1607            }
1608        }
1609    }
1610
1611    pub fn add_mission(&mut self, mission_id: &str) -> CuResult<&mut CuGraph> {
1612        match self {
1613            Simple(_) => Err("Cannot add mission to simple config".into()),
1614            Missions(graphs) => match graphs.entry(mission_id.to_string()) {
1615                hashbrown::hash_map::Entry::Occupied(_) => {
1616                    Err(format!("Mission {mission_id} already exists").into())
1617                }
1618                hashbrown::hash_map::Entry::Vacant(entry) => Ok(entry.insert(CuGraph::default())),
1619            },
1620        }
1621    }
1622}
1623
1624/// CuConfig is the programmatic representation of the configuration graph.
1625/// It is a directed graph where nodes are tasks and edges are connections between tasks.
1626///
1627/// The core of CuConfig is its `graphs` field which can be either a simple graph
1628/// or a collection of mission-specific graphs. The graph structure is based on petgraph.
1629#[derive(Debug, Clone)]
1630pub struct CuConfig {
1631    /// Values baked into the application by `#[copper_runtime]`.
1632    #[doc(hidden)]
1633    pub constants: Vec<ConstantConfig>,
1634    /// Monitoring configuration list.
1635    pub monitors: Vec<MonitorConfig>,
1636    /// Optional logging configuration
1637    pub logging: Option<LoggingConfig>,
1638    /// Optional runtime configuration
1639    pub runtime: Option<RuntimeConfig>,
1640    /// Declarative resource bundle definitions
1641    pub resources: Vec<ResourceBundleConfig>,
1642    /// Optional statically generated semantic log-stream destinations.
1643    pub log_streaming: Option<LogStreamingConfig>,
1644    /// Declarative bridge definitions that are yet to be expanded into the graph
1645    pub bridges: Vec<BridgeConfig>,
1646    /// Graph structure - either a single graph or multiple mission-specific graphs
1647    pub graphs: ConfigGraphs,
1648}
1649
1650/// Every reconstructed node needs recorded or reconstructed inputs. Checking
1651/// direct edges at every reconstructed node also covers chains and fan-in.
1652fn validate_reconstruction_inputs(graph: &CuGraph) -> CuResult<()> {
1653    for index in graph.0.node_indices() {
1654        let node = &graph.0[index];
1655        if node.streaming().replay != StreamReplay::Reconstruct {
1656            continue;
1657        }
1658        for edge in graph.0.edges_directed(index, Incoming) {
1659            let source = &graph.0[edge.source()];
1660            if !source.is_logging_enabled() {
1661                return Err(CuError::from(format!(
1662                    "Task '{}' uses streaming.replay: reconstruct but input '{}' from '{}' has logging.enabled: false. Enable logging on '{}' or use streaming.replay: capture on '{}'.",
1663                    node.id,
1664                    edge.weight().msg,
1665                    source.id,
1666                    source.id,
1667                    node.id
1668                )));
1669            }
1670        }
1671    }
1672    Ok(())
1673}
1674
1675impl CuConfig {
1676    /// Validates reconstruction inputs, static log-stream topology and bounds before code generation.
1677    pub fn validate_log_streaming_config(&self) -> CuResult<()> {
1678        match &self.graphs {
1679            Simple(graph) => validate_reconstruction_inputs(graph)?,
1680            Missions(graphs) => {
1681                for (mission, graph) in graphs {
1682                    validate_reconstruction_inputs(graph)
1683                        .map_err(|error| CuError::from(format!("Mission '{mission}': {error}")))?;
1684                }
1685            }
1686        }
1687
1688        let Some(streaming) = &self.log_streaming else {
1689            return Ok(());
1690        };
1691        if streaming.destinations.is_empty() {
1692            return Err(CuError::from(
1693                "log_streaming.destinations must contain at least one destination",
1694            ));
1695        }
1696
1697        let keyframe_interval = self
1698            .logging
1699            .as_ref()
1700            .and_then(|logging| logging.keyframe_interval)
1701            .unwrap_or(DEFAULT_KEYFRAME_INTERVAL);
1702        for (index, destination) in streaming.destinations.iter().enumerate() {
1703            if destination.id.trim().is_empty() {
1704                return Err(CuError::from(format!(
1705                    "log_streaming destination at index {index} has an empty id"
1706                )));
1707            }
1708            if streaming.destinations[..index]
1709                .iter()
1710                .any(|other| other.id == destination.id)
1711            {
1712                return Err(CuError::from(format!(
1713                    "Duplicate log_streaming destination id '{}'",
1714                    destination.id
1715                )));
1716            }
1717            if destination.transport.type_.trim().is_empty() {
1718                return Err(CuError::from(format!(
1719                    "log_streaming destination '{}' has an empty transport type",
1720                    destination.id
1721                )));
1722            }
1723            let Some((bundle_id, resource_name)) = destination.transport.resource.split_once('.')
1724            else {
1725                return Err(CuError::from(format!(
1726                    "log_streaming destination '{}' resource '{}' must use 'bundle.resource' syntax",
1727                    destination.id, destination.transport.resource
1728                )));
1729            };
1730            if bundle_id.is_empty()
1731                || resource_name.is_empty()
1732                || resource_name.contains('.')
1733                || !self.resources.iter().any(|bundle| bundle.id == bundle_id)
1734            {
1735                return Err(CuError::from(format!(
1736                    "log_streaming destination '{}' references invalid resource '{}'",
1737                    destination.id, destination.transport.resource
1738                )));
1739            }
1740            if streaming.destinations[..index]
1741                .iter()
1742                .any(|other| other.transport.resource == destination.transport.resource)
1743            {
1744                return Err(CuError::from(format!(
1745                    "log_streaming resource '{}' is bound by more than one destination",
1746                    destination.transport.resource
1747                )));
1748            }
1749            if let Some(feedback) = &destination.feedback {
1750                let baseline = destination.fec.continuous.repair_every_source_symbols;
1751                if feedback.report_interval_ms == 0
1752                    || feedback.timeout_ms <= feedback.report_interval_ms
1753                    || feedback.adaptation.is_some_and(|bounds| {
1754                        bounds.min_repair_every_source_symbols == 0
1755                            || bounds.min_repair_every_source_symbols > baseline
1756                            || baseline > bounds.max_repair_every_source_symbols
1757                    })
1758                {
1759                    return Err(CuError::from(
1760                        "Invalid log_streaming feedback cadence or FEC bounds",
1761                    ));
1762                }
1763                let resource = &feedback.transport.resource;
1764                let valid_resource = resource.split_once('.').is_some_and(|(bundle, slot)| {
1765                    !slot.is_empty()
1766                        && !slot.contains('.')
1767                        && self.resources.iter().any(|r| r.id == bundle)
1768                });
1769                if feedback.transport.type_.trim().is_empty() || !valid_resource {
1770                    return Err(CuError::from("Invalid log_streaming feedback resource"));
1771                }
1772            }
1773            // All logical receivers have one owner, including shared-carrier handles.
1774            let resources = core::iter::once(&destination.transport.resource)
1775                .chain(destination.feedback.iter().map(|f| &f.transport.resource));
1776            for resource in resources {
1777                let uses = streaming
1778                    .destinations
1779                    .iter()
1780                    .map(|d| {
1781                        usize::from(&d.transport.resource == resource)
1782                            + usize::from(
1783                                d.feedback
1784                                    .as_ref()
1785                                    .is_some_and(|f| &f.transport.resource == resource),
1786                            )
1787                    })
1788                    .sum::<usize>();
1789                if uses > 1 {
1790                    return Err(CuError::from(format!(
1791                        "Log-stream resource '{resource}' has multiple owners"
1792                    )));
1793                }
1794            }
1795            if destination.link.mtu_bytes <= 72 {
1796                return Err(CuError::from(format!(
1797                    "log_streaming destination '{}' mtu_bytes must exceed the 72-byte packet header",
1798                    destination.id
1799                )));
1800            }
1801            if destination.link.bitrate_bps == 0
1802                || destination.link.memory_budget_kib == 0
1803                || destination.link.max_latency_ms == 0
1804                || destination.link.burst_packets == 0
1805            {
1806                return Err(CuError::from(format!(
1807                    "log_streaming destination '{}' link values must be nonzero",
1808                    destination.id
1809                )));
1810            }
1811            if destination.fec.continuous.window_symbols == 0
1812                || destination.fec.continuous.repair_every_source_symbols == 0
1813            {
1814                return Err(CuError::from(format!(
1815                    "log_streaming destination '{}' continuous FEC values must be nonzero",
1816                    destination.id
1817                )));
1818            }
1819            if destination.fec.continuous.repair_density.threshold() > 15 {
1820                return Err(CuError::from(format!(
1821                    "log_streaming destination '{}' repair density threshold must be in 0..=15",
1822                    destination.id
1823                )));
1824            }
1825            if destination.fec.objects.max_object_bytes == 0
1826                || destination.fec.objects.repair_symbols_per_block == 0
1827                || destination.max_record_bytes == 0
1828            {
1829                return Err(CuError::from(format!(
1830                    "log_streaming destination '{}' object and record bounds must be nonzero",
1831                    destination.id
1832                )));
1833            }
1834            if destination.recovery_interval == 0
1835                || !destination
1836                    .recovery_interval
1837                    .is_multiple_of(keyframe_interval)
1838            {
1839                return Err(CuError::from(format!(
1840                    "log_streaming destination '{}' recovery_interval must be a nonzero multiple of logging.keyframe_interval ({keyframe_interval})",
1841                    destination.id
1842                )));
1843            }
1844        }
1845        Ok(())
1846    }
1847
1848    /// Guarantees that a default `"background"` thread pool entry exists in
1849    /// `runtime.thread_pools` whenever the graph has any `background: true`
1850    /// task that didn't explicitly select a pool. Thread pools are otherwise
1851    /// constructed straight from `runtime.thread_pools` by the runtime — they
1852    /// are not stored in `ResourceManager`.
1853    #[cfg(feature = "std")]
1854    fn ensure_default_background_pool(&mut self) {
1855        if !self.has_background_tasks() {
1856            return;
1857        }
1858
1859        const DEFAULT_BACKGROUND_THREADS: usize = 2;
1860
1861        let runtime = self.runtime.get_or_insert_with(RuntimeConfig::default);
1862        if !runtime
1863            .thread_pools
1864            .iter()
1865            .any(|pool| pool.id == DEFAULT_BACKGROUND_POOL)
1866        {
1867            runtime.thread_pools.push(ThreadPoolConfig {
1868                id: DEFAULT_BACKGROUND_POOL.to_string(),
1869                threads: DEFAULT_BACKGROUND_THREADS,
1870                affinity: None,
1871                policy: SchedulingPolicy::Fair,
1872                on_error: OnError::Warn,
1873            });
1874        }
1875    }
1876
1877    /// The configured planner selection, if any (absent means serial execution).
1878    pub fn planner_config(&self) -> Option<&PlannerConfig> {
1879        self.runtime.as_ref()?.planner.as_ref()
1880    }
1881
1882    /// Execution strategy selected by the configuration.
1883    pub fn planner_kind(&self) -> PlannerKind {
1884        self.planner_config()
1885            .map_or(PlannerKind::Serial, PlannerConfig::kind)
1886    }
1887
1888    #[cfg(feature = "std")]
1889    fn has_background_tasks(&self) -> bool {
1890        match &self.graphs {
1891            ConfigGraphs::Simple(graph) => graph
1892                .get_all_nodes()
1893                .iter()
1894                .any(|(_, node)| node.is_background()),
1895            ConfigGraphs::Missions(graphs) => graphs.values().any(|graph| {
1896                graph
1897                    .get_all_nodes()
1898                    .iter()
1899                    .any(|(_, node)| node.is_background())
1900            }),
1901        }
1902    }
1903}
1904
1905#[derive(Serialize, Deserialize, Default, Debug, Clone)]
1906pub struct MonitorConfig {
1907    #[serde(rename = "type")]
1908    type_: String,
1909    #[serde(skip_serializing_if = "Option::is_none")]
1910    config: Option<ComponentConfig>,
1911}
1912
1913impl MonitorConfig {
1914    #[allow(dead_code)]
1915    pub fn get_type(&self) -> &str {
1916        &self.type_
1917    }
1918
1919    #[allow(dead_code)]
1920    pub fn get_config(&self) -> Option<&ComponentConfig> {
1921        self.config.as_ref()
1922    }
1923}
1924
1925fn default_as_true() -> bool {
1926    true
1927}
1928
1929pub const DEFAULT_KEYFRAME_INTERVAL: u32 = 100;
1930
1931fn default_keyframe_interval() -> Option<u32> {
1932    Some(DEFAULT_KEYFRAME_INTERVAL)
1933}
1934
1935#[derive(Serialize, Deserialize, Debug, Clone)]
1936pub struct LoggingConfig {
1937    /// Enable task logging to the log file.
1938    #[serde(default = "default_as_true", skip_serializing_if = "Clone::clone")]
1939    pub enable_task_logging: bool,
1940
1941    /// Generate and record task-state keyframes.
1942    ///
1943    /// Without another generated keyframe consumer, this is a compile-time application
1944    /// property and `#[copper_runtime]` emits no capture calls when it is `false`.
1945    /// A log-stream destination independently requests keyframe capture for recovery points.
1946    #[serde(default = "default_as_true", skip_serializing_if = "Clone::clone")]
1947    pub enable_keyframe_logging: bool,
1948
1949    /// Number of preallocated CopperLists available to the runtime.
1950    ///
1951    /// This is consumed by proc-macro codegen and must match the value compiled into the
1952    /// application binary.
1953    #[serde(skip_serializing_if = "Option::is_none")]
1954    pub copperlist_count: Option<usize>,
1955
1956    /// Size of each slab in the log file. (it is the size of the memory mapped file at a time)
1957    #[serde(skip_serializing_if = "Option::is_none")]
1958    pub slab_size_mib: Option<u64>,
1959
1960    /// Pre-allocated size for each section in the log file.
1961    #[serde(skip_serializing_if = "Option::is_none")]
1962    pub section_size_mib: Option<u64>,
1963
1964    /// Interval in copperlists between two "keyframes" in the log file i.e. freezing tasks.
1965    #[serde(
1966        default = "default_keyframe_interval",
1967        skip_serializing_if = "Option::is_none"
1968    )]
1969    pub keyframe_interval: Option<u32>,
1970
1971    /// Named log codec specs reusable across task output bindings.
1972    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1973    pub codecs: Vec<LoggingCodecSpec>,
1974}
1975
1976impl Default for LoggingConfig {
1977    fn default() -> Self {
1978        Self {
1979            enable_task_logging: true,
1980            enable_keyframe_logging: true,
1981            copperlist_count: None,
1982            slab_size_mib: None,
1983            section_size_mib: None,
1984            keyframe_interval: default_keyframe_interval(),
1985            codecs: Vec::new(),
1986        }
1987    }
1988}
1989
1990#[derive(Serialize, Deserialize, Debug, Clone)]
1991pub struct LoggingCodecSpec {
1992    pub id: String,
1993    #[serde(rename = "type")]
1994    pub type_: String,
1995    #[serde(skip_serializing_if = "Option::is_none")]
1996    pub config: Option<ComponentConfig>,
1997}
1998
1999#[derive(Serialize, Deserialize, Default, Debug, Clone)]
2000#[serde(try_from = "RuntimeConfigInput")]
2001pub struct RuntimeConfig {
2002    /// Reference-clock configuration; requires `clock-sync`.
2003    #[serde(default, skip_serializing_if = "Option::is_none")]
2004    #[cfg(feature = "clock-sync")]
2005    pub clock: Option<ClockSyncConfig>,
2006    /// Set a CopperList execution rate target in Hz
2007    /// It will act as a rate limiter: if the execution is slower than this rate,
2008    /// it will continue to execute at "best effort".
2009    ///
2010    /// The main usecase is to not waste cycles when the system doesn't need an unbounded execution rate.
2011    #[serde(skip_serializing_if = "Option::is_none")]
2012    pub rate_target_hz: Option<u64>,
2013
2014    /// Declarative thread pool definitions used by the background-task pools and
2015    /// the Pipeline execution engine. Each pool carries an optional CPU
2016    /// affinity and a scheduling policy/priority.
2017    ///
2018    /// This is a `std`-only concept; on `no_std`/embedded targets there are no
2019    /// threads and this section is ignored.
2020    #[serde(default, skip_serializing_if = "Vec::is_empty")]
2021    pub thread_pools: Vec<ThreadPoolConfig>,
2022
2023    /// Execution planner selection, one for the whole config.
2024    ///
2025    /// This is a codegen input: `#[copper_runtime]` bakes the resulting plan
2026    /// into the binary. Editing it in a deployed app's RON at startup does not
2027    /// change the compiled plan (same class as `logging.copperlist_count`); the
2028    /// RON must match the binary that wrote the log.
2029    #[serde(default, skip_serializing_if = "Option::is_none")]
2030    pub planner: Option<PlannerConfig>,
2031}
2032
2033// Preserve legacy unknown-field handling while rejecting a declared clock when
2034// this binary cannot discipline it. Keep feature-disabled RuntimeConfig's fields
2035// unchanged for applications constructing it directly.
2036#[derive(Deserialize)]
2037struct RuntimeConfigInput {
2038    clock: Option<ClockSyncConfig>,
2039    rate_target_hz: Option<u64>,
2040    #[serde(default)]
2041    thread_pools: Vec<ThreadPoolConfig>,
2042    planner: Option<PlannerConfig>,
2043}
2044impl TryFrom<RuntimeConfigInput> for RuntimeConfig {
2045    type Error = &'static str;
2046    fn try_from(input: RuntimeConfigInput) -> Result<Self, Self::Error> {
2047        #[cfg(not(feature = "clock-sync"))]
2048        if input.clock.is_some() {
2049            return Err("runtime.clock requires the clock-sync feature");
2050        }
2051        Ok(Self {
2052            #[cfg(feature = "clock-sync")]
2053            clock: input.clock,
2054            rate_target_hz: input.rate_target_hz,
2055            thread_pools: input.thread_pools,
2056            planner: input.planner,
2057        })
2058    }
2059}
2060
2061/// Reference-clock wiring and acquisition/error bounds.
2062#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2063#[serde(deny_unknown_fields)]
2064pub struct ClockSyncConfig {
2065    /// Owned reference resource, resolved by the runtime generator.
2066    pub parent: String,
2067    /// Required maximum timestamp uncertainty in nanoseconds.
2068    pub max_error_ns: u64,
2069    /// Raw-time acquisition timeout; defaults to ten seconds.
2070    #[serde(default = "clock_acquisition_timeout")]
2071    pub acquisition_timeout_ns: u64,
2072    /// Raw-time sample interval; defaults to one second.
2073    #[serde(default = "clock_sample_interval")]
2074    pub sample_interval_ns: u64,
2075    /// Maximum raw observation age; defaults to five seconds.
2076    #[serde(default = "clock_max_age")]
2077    pub max_age_ns: u64,
2078    /// Relative holdover oscillator error; defaults to 100 ppm.
2079    #[serde(default = "clock_drift_bound")]
2080    pub drift_bound_ppb: u32,
2081    /// Additional phase correction bound; defaults to 500 ppm.
2082    #[serde(default = "clock_slew_bound")]
2083    pub max_slew_ppb: u32,
2084}
2085fn clock_acquisition_timeout() -> u64 {
2086    10_000_000_000
2087}
2088fn clock_sample_interval() -> u64 {
2089    1_000_000_000
2090}
2091fn clock_max_age() -> u64 {
2092    5_000_000_000
2093}
2094fn clock_drift_bound() -> u32 {
2095    100_000
2096}
2097fn clock_slew_bound() -> u32 {
2098    500_000
2099}
2100
2101impl ClockSyncConfig {
2102    /// Validates timing, resource selection and positive clock-rate bounds.
2103    pub fn validate(&self) -> CuResult<()> {
2104        if self
2105            .parent
2106            .split_once('.')
2107            .is_none_or(|(bundle, slot)| bundle.is_empty() || slot.is_empty() || slot.contains('.'))
2108        {
2109            return Err(CuError::from(
2110                "runtime.clock.parent must be a bundle.resource path",
2111            ));
2112        }
2113        if self.max_error_ns == 0
2114            || self.max_age_ns == 0
2115            || self.acquisition_timeout_ns == 0
2116            || self.sample_interval_ns == 0
2117            || self.sample_interval_ns >= self.acquisition_timeout_ns
2118            || u64::from(self.drift_bound_ppb) + u64::from(self.max_slew_ppb) >= 1_000_000_000
2119        {
2120            return Err(CuError::from(
2121                "runtime.clock has invalid acquisition, error, age or rate bounds",
2122            ));
2123        }
2124        Ok(())
2125    }
2126
2127    /// Converts the declarative bounds into maintenance settings.
2128    #[cfg(feature = "clock-sync")]
2129    pub fn maintenance(&self) -> crate::clock_sync::MaintenanceConfig {
2130        crate::clock_sync::MaintenanceConfig {
2131            acquisition_timeout: core::time::Duration::from_nanos(self.acquisition_timeout_ns),
2132            sample_interval: core::time::Duration::from_nanos(self.sample_interval_ns),
2133            sync: cu29_clock::sync::SyncConfig {
2134                max_error: cu29_clock::CuDuration(self.max_error_ns),
2135                max_age: cu29_clock::CuDuration(self.max_age_ns),
2136                drift_bound_ppb: self.drift_bound_ppb,
2137                max_slew_ppb: self.max_slew_ppb,
2138            },
2139        }
2140    }
2141}
2142
2143/// Selects the compile-time execution strategy for every mission graph.
2144///
2145/// Pipeline execution is selected explicitly with:
2146///
2147/// ```text
2148/// runtime: (
2149///     planner: (
2150///         kind: Pipeline,
2151///         config: { "max_in_flight": 2 },
2152///     ),
2153/// )
2154/// ```
2155///
2156/// `Pipeline` requires the `parallel-rt` Cargo feature. Omitting `planner`
2157/// selects [`PlannerKind::Serial`], even when that feature is enabled.
2158#[derive(Serialize, Deserialize, Debug, Clone)]
2159pub struct PlannerConfig {
2160    pub(crate) kind: PlannerKind,
2161    #[serde(skip_serializing_if = "Option::is_none")]
2162    pub(crate) config: Option<ComponentConfig>,
2163}
2164
2165impl PlannerConfig {
2166    #[allow(dead_code)]
2167    pub fn kind(&self) -> PlannerKind {
2168        self.kind
2169    }
2170
2171    #[allow(dead_code)]
2172    pub fn get_config(&self) -> Option<&ComponentConfig> {
2173        self.config.as_ref()
2174    }
2175
2176    /// Pipeline capacity, defaulting to the number of preallocated CopperLists.
2177    pub fn max_in_flight(&self, copperlist_count: usize) -> CuResult<usize> {
2178        if self.kind != PlannerKind::Pipeline {
2179            return Ok(1);
2180        }
2181        let configured = match self.config.as_ref() {
2182            Some(config) => config
2183                .get_value::<usize>("max_in_flight")?
2184                .unwrap_or(copperlist_count),
2185            None => copperlist_count,
2186        };
2187        if configured == 0 {
2188            return Err(CuError::from(
2189                "Pipeline planner max_in_flight cannot be zero.",
2190            ));
2191        }
2192        if configured > copperlist_count {
2193            return Err(CuError::from(format!(
2194                "Pipeline planner max_in_flight ({configured}) exceeds logging.copperlist_count ({copperlist_count})."
2195            )));
2196        }
2197        Ok(configured)
2198    }
2199}
2200
2201/// Built-in compile-time execution strategies.
2202#[derive(Serialize, Deserialize, Default, Debug, Clone, Copy, PartialEq, Eq)]
2203pub enum PlannerKind {
2204    /// Deterministic single-threaded execution in graph order.
2205    #[default]
2206    Serial,
2207    /// Deterministic single-threaded execution in the configured task order.
2208    TaskOrder,
2209    /// An exact portable schedule validated and compiled with the application.
2210    ExplicitSchedule,
2211    /// One ordered worker lane per process stage.
2212    Pipeline,
2213}
2214
2215/// Smallest valid real-time priority for [`SchedulingPolicy::Fifo`]/[`SchedulingPolicy::RoundRobin`].
2216pub const MIN_RT_PRIORITY: u8 = 1;
2217/// Largest valid real-time priority for [`SchedulingPolicy::Fifo`]/[`SchedulingPolicy::RoundRobin`].
2218pub const MAX_RT_PRIORITY: u8 = 99;
2219/// Lowest valid niceness for [`SchedulingPolicy::Nice`] (most favorable).
2220pub const MIN_NICE: i8 = -20;
2221/// Highest valid niceness for [`SchedulingPolicy::Nice`] (least favorable).
2222pub const MAX_NICE: i8 = 19;
2223
2224/// Scheduling policy applied to every worker thread of a [`ThreadPoolConfig`].
2225///
2226/// On Linux these map directly onto the POSIX scheduling policies. On other
2227/// platforms they are applied best-effort (see the per-pool
2228/// [`ThreadPoolConfig::on_error`] behavior).
2229#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
2230pub enum SchedulingPolicy {
2231    /// Normal fair time-sharing scheduler (`SCHED_OTHER`/CFS on Linux) with default
2232    /// niceness. The OS shares the CPU fairly across threads and no thread starves.
2233    ///
2234    /// Use for everything that isn't latency-critical. This is the default.
2235    #[default]
2236    Fair,
2237    /// Fair scheduler with an explicit niceness (`-20..=19`, lower is more favorable).
2238    ///
2239    /// A soft priority hint, not a guarantee: a higher (nicer) value yields the CPU
2240    /// more readily. Use to bias a pool below or above normal work without leaving
2241    /// the fair scheduler — e.g. `Nice(10)` for heavy background work that should
2242    /// step aside for the control loop.
2243    Nice(i8),
2244    /// `SCHED_FIFO` real-time policy, priority `1..=99` (higher wins).
2245    ///
2246    /// Hard real-time: a FIFO thread runs ahead of every fair thread and is not
2247    /// time-sliced — it runs until it blocks or a higher-priority RT thread preempts
2248    /// it. Use for the latency-critical pipeline, and pin it with `affinity` so a
2249    /// busy worker cannot starve other work on the same core. Linux-only; typically
2250    /// needs `CAP_SYS_NICE`.
2251    Fifo { priority: u8 },
2252    /// `SCHED_RR` real-time policy, priority `1..=99` (higher wins).
2253    ///
2254    /// Same real-time semantics as [`Fifo`](Self::Fifo), except threads at the same
2255    /// priority are round-robin time-sliced rather than run-to-block. Use when
2256    /// several RT workers share a priority and should interleave fairly. Linux-only;
2257    /// typically needs `CAP_SYS_NICE`.
2258    RoundRobin { priority: u8 },
2259}
2260
2261/// What to do when a pool's affinity or scheduling request cannot be applied
2262/// (for example, setting a real-time priority without `CAP_SYS_NICE`).
2263#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
2264pub enum OnError {
2265    /// Log a warning and fall back to default scheduling. This keeps unprivileged
2266    /// dev/laptop runs working out of the box.
2267    #[default]
2268    Warn,
2269    /// Hard-fail at startup if the requested affinity/scheduler cannot be applied.
2270    /// Use this for deployed real-time robots that must fail loudly.
2271    Strict,
2272}
2273
2274/// Declarative definition of a single thread pool.
2275#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2276pub struct ThreadPoolConfig {
2277    /// Unique pool id. Reserved ids: [`RT_POOL`] (the Pipeline execution
2278    /// engine) and [`DEFAULT_BACKGROUND_POOL`] (the default background pool).
2279    pub id: String,
2280    /// Number of worker threads in the pool.
2281    pub threads: usize,
2282    /// Optional set of logical CPU cores the pool may use. When set, worker `i`
2283    /// is pinned to `affinity[i % affinity.len()]` (Spread): `threads ==
2284    /// affinity.len()` yields one worker pinned per dedicated core.
2285    #[serde(default, skip_serializing_if = "Option::is_none")]
2286    pub affinity: Option<Vec<usize>>,
2287    /// Scheduling policy/priority applied to each worker thread.
2288    #[serde(default)]
2289    pub policy: SchedulingPolicy,
2290    /// What to do if affinity/scheduling cannot be applied.
2291    #[serde(default)]
2292    pub on_error: OnError,
2293}
2294
2295/// Validates the declarative thread pool definitions of a runtime config.
2296///
2297/// Checks ids are non-empty and unique, thread counts are non-zero, real-time
2298/// priorities and niceness values are in range, and affinity lists are non-empty
2299/// when present. This is purely a config-level check; pools are built later.
2300fn validate_thread_pools<E>(runtime: &Option<RuntimeConfig>) -> Result<(), E>
2301where
2302    E: From<String>,
2303{
2304    let Some(runtime) = runtime else {
2305        return Ok(());
2306    };
2307
2308    let mut seen: Vec<&str> = Vec::new();
2309    for pool in &runtime.thread_pools {
2310        if pool.id.is_empty() {
2311            return Err(E::from("Thread pool id cannot be empty".to_string()));
2312        }
2313        if seen.contains(&pool.id.as_str()) {
2314            return Err(E::from(format!("Duplicate thread pool id '{}'", pool.id)));
2315        }
2316        seen.push(pool.id.as_str());
2317
2318        if pool.threads == 0 {
2319            return Err(E::from(format!(
2320                "Thread pool '{}' must have at least 1 thread",
2321                pool.id
2322            )));
2323        }
2324
2325        match pool.policy {
2326            SchedulingPolicy::Fifo { priority } | SchedulingPolicy::RoundRobin { priority } => {
2327                if !(MIN_RT_PRIORITY..=MAX_RT_PRIORITY).contains(&priority) {
2328                    return Err(E::from(format!(
2329                        "Thread pool '{}' real-time priority {priority} is out of range ({MIN_RT_PRIORITY}..={MAX_RT_PRIORITY})",
2330                        pool.id
2331                    )));
2332                }
2333            }
2334            SchedulingPolicy::Nice(nice) => {
2335                if !(MIN_NICE..=MAX_NICE).contains(&nice) {
2336                    return Err(E::from(format!(
2337                        "Thread pool '{}' niceness {nice} is out of range ({MIN_NICE}..={MAX_NICE})",
2338                        pool.id
2339                    )));
2340                }
2341            }
2342            SchedulingPolicy::Fair => {}
2343        }
2344
2345        if let Some(affinity) = &pool.affinity
2346            && affinity.is_empty()
2347        {
2348            return Err(E::from(format!(
2349                "Thread pool '{}' has an empty affinity list; omit `affinity` for no pinning",
2350                pool.id
2351            )));
2352        }
2353    }
2354
2355    Ok(())
2356}
2357
2358/// Maximum representable Copper runtime rate target in whole Hertz.
2359///
2360/// Copper stores runtime periods in integer nanoseconds, so anything above 1 GHz
2361/// would round down to a zero-duration period.
2362pub const MAX_RATE_TARGET_HZ: u64 = 1_000_000_000;
2363
2364/// Missions are used to generate alternative DAGs within the same configuration.
2365#[derive(Serialize, Deserialize, Debug, Clone)]
2366pub struct MissionsConfig {
2367    pub id: String,
2368}
2369
2370/// A compile-time predicate controlling whether a configuration fragment is included.
2371#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2372pub enum ConfigPredicate {
2373    Feature(String),
2374    Not(Box<ConfigPredicate>),
2375    All(Vec<ConfigPredicate>),
2376    Any(Vec<ConfigPredicate>),
2377}
2378
2379#[cfg(feature = "std")]
2380impl ConfigPredicate {
2381    fn evaluate(&self, active_features: &[&str]) -> bool {
2382        match self {
2383            Self::Feature(feature) => active_features.contains(&feature.as_str()),
2384            Self::Not(predicate) => !predicate.evaluate(active_features),
2385            Self::All(predicates) => predicates
2386                .iter()
2387                .all(|predicate| predicate.evaluate(active_features)),
2388            Self::Any(predicates) => predicates
2389                .iter()
2390                .any(|predicate| predicate.evaluate(active_features)),
2391        }
2392    }
2393}
2394
2395/// Includes are used to include other configuration files.
2396#[derive(Serialize, Deserialize, Debug, Clone)]
2397pub struct IncludesConfig {
2398    pub path: String,
2399    #[serde(default)]
2400    #[serde(serialize_with = "serialize_ordered_map")]
2401    pub params: HashMap<String, Value>,
2402    #[serde(default)]
2403    pub missions: Option<Vec<String>>,
2404    #[serde(default)]
2405    pub when: Option<ConfigPredicate>,
2406}
2407
2408/// One subsystem participating in a multi-Copper deployment.
2409#[cfg(feature = "std")]
2410#[allow(dead_code)]
2411#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2412pub struct MultiCopperSubsystemConfig {
2413    pub id: String,
2414    pub config: String,
2415}
2416
2417/// One explicit interconnect between two subsystem bridge channels.
2418#[cfg(feature = "std")]
2419#[allow(dead_code)]
2420#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2421pub struct MultiCopperInterconnectConfig {
2422    pub from: String,
2423    pub to: String,
2424    pub msg: String,
2425    #[serde(default)]
2426    pub when: Option<ConfigPredicate>,
2427}
2428
2429/// One path-based config overlay applied to a parsed local Copper config.
2430#[cfg(feature = "std")]
2431#[allow(dead_code)]
2432#[derive(Serialize, Deserialize, Debug, Clone)]
2433pub struct InstanceConfigSetOperation {
2434    pub path: String,
2435    pub value: ComponentConfig,
2436}
2437
2438/// Typed endpoint reference used by validated multi-Copper interconnects.
2439#[cfg(feature = "std")]
2440#[allow(dead_code)]
2441#[derive(Debug, Clone, PartialEq, Eq, Hash)]
2442pub struct MultiCopperEndpoint {
2443    pub subsystem_id: String,
2444    pub bridge_id: String,
2445    pub channel_id: String,
2446}
2447
2448#[cfg(feature = "std")]
2449impl Display for MultiCopperEndpoint {
2450    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2451        write!(
2452            f,
2453            "{}/{}/{}",
2454            self.subsystem_id, self.bridge_id, self.channel_id
2455        )
2456    }
2457}
2458
2459/// Validated subsystem entry with its compiler-assigned numeric subsystem code and parsed local Copper config.
2460#[cfg(feature = "std")]
2461#[allow(dead_code)]
2462#[derive(Debug, Clone)]
2463pub struct MultiCopperSubsystem {
2464    pub id: String,
2465    pub subsystem_code: u16,
2466    pub config_path: String,
2467    pub config: CuConfig,
2468}
2469
2470/// Validated explicit interconnect between two subsystem endpoints.
2471#[cfg(feature = "std")]
2472#[allow(dead_code)]
2473#[derive(Debug, Clone, PartialEq, Eq)]
2474pub struct MultiCopperInterconnect {
2475    pub from: MultiCopperEndpoint,
2476    pub to: MultiCopperEndpoint,
2477    pub msg: String,
2478    pub bridge_type: String,
2479}
2480
2481/// Strict umbrella configuration describing multiple Copper subsystems and their explicit links.
2482#[cfg(feature = "std")]
2483#[allow(dead_code)]
2484#[derive(Debug, Clone)]
2485pub struct MultiCopperConfig {
2486    pub subsystems: Vec<MultiCopperSubsystem>,
2487    pub interconnects: Vec<MultiCopperInterconnect>,
2488    pub instance_overrides_root: Option<String>,
2489}
2490
2491#[cfg(feature = "std")]
2492impl MultiCopperConfig {
2493    #[allow(dead_code)]
2494    pub fn subsystem(&self, id: &str) -> Option<&MultiCopperSubsystem> {
2495        self.subsystems.iter().find(|subsystem| subsystem.id == id)
2496    }
2497
2498    #[allow(dead_code)]
2499    pub fn resolve_subsystem_config_for_instance(
2500        &self,
2501        subsystem_id: &str,
2502        instance_id: u32,
2503    ) -> CuResult<CuConfig> {
2504        let subsystem = self.subsystem(subsystem_id).ok_or_else(|| {
2505            CuError::from(format!(
2506                "Multi-Copper config does not define subsystem '{}'.",
2507                subsystem_id
2508            ))
2509        })?;
2510        let mut config = subsystem.config.clone();
2511
2512        let Some(root) = &self.instance_overrides_root else {
2513            return Ok(config);
2514        };
2515
2516        let override_path = std::path::Path::new(root)
2517            .join(instance_id.to_string())
2518            .join(format!("{subsystem_id}.ron"));
2519        if !override_path.exists() {
2520            return Ok(config);
2521        }
2522
2523        apply_instance_overrides_from_file(&mut config, &override_path)?;
2524        Ok(config)
2525    }
2526}
2527
2528#[cfg(feature = "std")]
2529#[allow(dead_code)]
2530#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
2531struct MultiCopperConfigRepresentation {
2532    subsystems: Vec<MultiCopperSubsystemConfig>,
2533    interconnects: Vec<MultiCopperInterconnectConfig>,
2534    instance_overrides_root: Option<String>,
2535}
2536
2537#[cfg(feature = "std")]
2538#[derive(Serialize, Deserialize, Debug, Clone, Default)]
2539struct InstanceConfigOverridesRepresentation {
2540    #[serde(default)]
2541    set: Vec<InstanceConfigSetOperation>,
2542}
2543
2544#[cfg(feature = "std")]
2545#[allow(dead_code)]
2546#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2547enum MultiCopperChannelDirection {
2548    Rx,
2549    Tx,
2550}
2551
2552#[cfg(feature = "std")]
2553#[allow(dead_code)]
2554#[derive(Debug, Clone)]
2555struct MultiCopperChannelContract {
2556    bridge_type: String,
2557    direction: MultiCopperChannelDirection,
2558    msg: Option<String>,
2559}
2560
2561#[cfg(feature = "std")]
2562#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2563enum InstanceConfigTargetKind {
2564    Task,
2565    Resource,
2566    Bridge,
2567}
2568
2569/// This is the main Copper configuration representation.
2570#[derive(Serialize, Deserialize, Default)]
2571struct CuConfigRepresentation {
2572    constants: Option<Vec<ConstantConfig>>,
2573    tasks: Option<Vec<Node>>,
2574    resources: Option<Vec<ResourceBundleConfig>>,
2575    log_streaming: Option<LogStreamingConfig>,
2576    bridges: Option<Vec<BridgeConfig>>,
2577    cnx: Option<Vec<SerializedCnx>>,
2578    #[serde(
2579        default,
2580        alias = "monitor",
2581        deserialize_with = "deserialize_monitor_configs"
2582    )]
2583    monitors: Option<Vec<MonitorConfig>>,
2584    logging: Option<LoggingConfig>,
2585    runtime: Option<RuntimeConfig>,
2586    missions: Option<Vec<MissionsConfig>>,
2587    includes: Option<Vec<IncludesConfig>>,
2588}
2589
2590#[derive(Deserialize)]
2591#[serde(untagged)]
2592enum OneOrManyMonitorConfig {
2593    One(MonitorConfig),
2594    Many(Vec<MonitorConfig>),
2595}
2596
2597fn deserialize_monitor_configs<'de, D>(
2598    deserializer: D,
2599) -> Result<Option<Vec<MonitorConfig>>, D::Error>
2600where
2601    D: Deserializer<'de>,
2602{
2603    let parsed = Option::<OneOrManyMonitorConfig>::deserialize(deserializer)?;
2604    Ok(parsed.map(|value| match value {
2605        OneOrManyMonitorConfig::One(single) => vec![single],
2606        OneOrManyMonitorConfig::Many(many) => many,
2607    }))
2608}
2609
2610/// Shared implementation for deserializing a CuConfigRepresentation into a CuConfig
2611fn deserialize_config_representation<E>(
2612    representation: &CuConfigRepresentation,
2613) -> Result<CuConfig, E>
2614where
2615    E: From<String>,
2616{
2617    let mut cuconfig = CuConfig::default();
2618    let bridge_lookup = build_bridge_lookup(representation.bridges.as_ref());
2619
2620    if let Some(mission_configs) = &representation.missions {
2621        // This is the multi-mission case
2622        let mut missions = Missions(HashMap::new());
2623
2624        for mission_config in mission_configs {
2625            let mission_id = mission_config.id.as_str();
2626            let graph = missions
2627                .add_mission(mission_id)
2628                .map_err(|e| E::from(e.to_string()))?;
2629
2630            if let Some(tasks) = &representation.tasks {
2631                for task in tasks {
2632                    if let Some(task_missions) = &task.missions {
2633                        // if there is a filter by mission on the task, only add the task to the mission if it matches the filter.
2634                        if task_missions.contains(&mission_id.to_owned()) {
2635                            graph
2636                                .add_node(task.clone())
2637                                .map_err(|e| E::from(e.to_string()))?;
2638                        }
2639                    } else {
2640                        // if there is no filter by mission on the task, add the task to the mission.
2641                        graph
2642                            .add_node(task.clone())
2643                            .map_err(|e| E::from(e.to_string()))?;
2644                    }
2645                }
2646            }
2647
2648            if let Some(bridges) = &representation.bridges {
2649                for bridge in bridges {
2650                    if mission_applies(&bridge.missions, mission_id) {
2651                        insert_bridge_node(graph, bridge).map_err(E::from)?;
2652                    }
2653                }
2654            }
2655
2656            if let Some(cnx) = &representation.cnx {
2657                for (connection_order, c) in cnx.iter().enumerate() {
2658                    if let Some(cnx_missions) = &c.missions {
2659                        // if there is a filter by mission on the connection, only add the connection to the mission if it matches the filter.
2660                        if cnx_missions.contains(&mission_id.to_owned()) {
2661                            if c.dst == NC_ENDPOINT {
2662                                register_nc_output::<E>(
2663                                    graph,
2664                                    &c.src,
2665                                    &c.msg,
2666                                    connection_order,
2667                                    &bridge_lookup,
2668                                )?;
2669                                continue;
2670                            }
2671                            let (src_name, src_channel) =
2672                                parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2673                                    .map_err(E::from)?;
2674                            let (dst_name, dst_channel) =
2675                                parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2676                                    .map_err(E::from)?;
2677                            let src =
2678                                graph
2679                                    .get_node_id_by_name(src_name.as_str())
2680                                    .ok_or_else(|| {
2681                                        E::from(format!("Source node not found: {}", c.src))
2682                                    })?;
2683                            let dst =
2684                                graph
2685                                    .get_node_id_by_name(dst_name.as_str())
2686                                    .ok_or_else(|| {
2687                                        E::from(format!("Destination node not found: {}", c.dst))
2688                                    })?;
2689                            graph
2690                                .connect_ext_with_order(
2691                                    src,
2692                                    dst,
2693                                    &c.msg,
2694                                    Some(cnx_missions.clone()),
2695                                    src_channel,
2696                                    dst_channel,
2697                                    connection_order,
2698                                )
2699                                .map_err(|e| E::from(e.to_string()))?;
2700                        }
2701                    } else {
2702                        // if there is no filter by mission on the connection, add the connection to the mission.
2703                        if c.dst == NC_ENDPOINT {
2704                            register_nc_output::<E>(
2705                                graph,
2706                                &c.src,
2707                                &c.msg,
2708                                connection_order,
2709                                &bridge_lookup,
2710                            )?;
2711                            continue;
2712                        }
2713                        let (src_name, src_channel) =
2714                            parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2715                                .map_err(E::from)?;
2716                        let (dst_name, dst_channel) =
2717                            parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2718                                .map_err(E::from)?;
2719                        let src = graph
2720                            .get_node_id_by_name(src_name.as_str())
2721                            .ok_or_else(|| E::from(format!("Source node not found: {}", c.src)))?;
2722                        let dst =
2723                            graph
2724                                .get_node_id_by_name(dst_name.as_str())
2725                                .ok_or_else(|| {
2726                                    E::from(format!("Destination node not found: {}", c.dst))
2727                                })?;
2728                        graph
2729                            .connect_ext_with_order(
2730                                src,
2731                                dst,
2732                                &c.msg,
2733                                None,
2734                                src_channel,
2735                                dst_channel,
2736                                connection_order,
2737                            )
2738                            .map_err(|e| E::from(e.to_string()))?;
2739                    }
2740                }
2741            }
2742        }
2743        cuconfig.graphs = missions;
2744    } else {
2745        // this is the simple case
2746        let mut graph = CuGraph::default();
2747
2748        if let Some(tasks) = &representation.tasks {
2749            for task in tasks {
2750                graph
2751                    .add_node(task.clone())
2752                    .map_err(|e| E::from(e.to_string()))?;
2753            }
2754        }
2755
2756        if let Some(bridges) = &representation.bridges {
2757            for bridge in bridges {
2758                insert_bridge_node(&mut graph, bridge).map_err(E::from)?;
2759            }
2760        }
2761
2762        if let Some(cnx) = &representation.cnx {
2763            for (connection_order, c) in cnx.iter().enumerate() {
2764                if c.dst == NC_ENDPOINT {
2765                    register_nc_output::<E>(
2766                        &mut graph,
2767                        &c.src,
2768                        &c.msg,
2769                        connection_order,
2770                        &bridge_lookup,
2771                    )?;
2772                    continue;
2773                }
2774                let (src_name, src_channel) =
2775                    parse_endpoint(&c.src, EndpointRole::Source, &bridge_lookup)
2776                        .map_err(E::from)?;
2777                let (dst_name, dst_channel) =
2778                    parse_endpoint(&c.dst, EndpointRole::Destination, &bridge_lookup)
2779                        .map_err(E::from)?;
2780                let src = graph
2781                    .get_node_id_by_name(src_name.as_str())
2782                    .ok_or_else(|| E::from(format!("Source node not found: {}", c.src)))?;
2783                let dst = graph
2784                    .get_node_id_by_name(dst_name.as_str())
2785                    .ok_or_else(|| E::from(format!("Destination node not found: {}", c.dst)))?;
2786                graph
2787                    .connect_ext_with_order(
2788                        src,
2789                        dst,
2790                        &c.msg,
2791                        None,
2792                        src_channel,
2793                        dst_channel,
2794                        connection_order,
2795                    )
2796                    .map_err(|e| E::from(e.to_string()))?;
2797            }
2798        }
2799        cuconfig.graphs = Simple(graph);
2800    }
2801
2802    cuconfig.monitors = representation.monitors.clone().unwrap_or_default();
2803    cuconfig.constants = representation.constants.clone().unwrap_or_default();
2804    cuconfig.logging = representation.logging.clone();
2805    cuconfig.runtime = representation.runtime.clone();
2806    cuconfig.resources = representation.resources.clone().unwrap_or_default();
2807    cuconfig.log_streaming = representation.log_streaming.clone();
2808    cuconfig.bridges = representation.bridges.clone().unwrap_or_default();
2809
2810    validate_thread_pools::<E>(&cuconfig.runtime)?;
2811
2812    Ok(cuconfig)
2813}
2814
2815impl<'de> Deserialize<'de> for CuConfig {
2816    /// This is a custom serialization to make this implementation independent of petgraph.
2817    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
2818    where
2819        D: Deserializer<'de>,
2820    {
2821        let representation =
2822            CuConfigRepresentation::deserialize(deserializer).map_err(serde::de::Error::custom)?;
2823
2824        // Convert String errors to D::Error using serde::de::Error::custom
2825        match deserialize_config_representation::<String>(&representation) {
2826            Ok(config) => Ok(config),
2827            Err(e) => Err(serde::de::Error::custom(e)),
2828        }
2829    }
2830}
2831
2832fn serialize_ordered_map<T: Serialize, S: Serializer>(
2833    map: &HashMap<String, T>,
2834    serializer: S,
2835) -> Result<S::Ok, S::Error> {
2836    map.iter().collect::<BTreeMap<_, _>>().serialize(serializer)
2837}
2838fn serialize_optional_ordered_map<T: Serialize, S: Serializer>(
2839    map: &Option<HashMap<String, T>>,
2840    serializer: S,
2841) -> Result<S::Ok, S::Error> {
2842    map.as_ref()
2843        .map(|map| map.iter().collect::<BTreeMap<_, _>>())
2844        .serialize(serializer)
2845}
2846
2847impl Serialize for CuConfig {
2848    /// This is a custom serialization to make this implementation independent of petgraph.
2849    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
2850    where
2851        S: Serializer,
2852    {
2853        let bridges = if self.bridges.is_empty() {
2854            None
2855        } else {
2856            Some(self.bridges.clone())
2857        };
2858        let resources = if self.resources.is_empty() {
2859            None
2860        } else {
2861            Some(self.resources.clone())
2862        };
2863        let monitors = (!self.monitors.is_empty()).then_some(self.monitors.clone());
2864        match &self.graphs {
2865            Simple(graph) => {
2866                let tasks: Vec<Node> = graph
2867                    .0
2868                    .node_indices()
2869                    .map(|idx| graph.0[idx].clone())
2870                    .filter(|node| node.get_flavor() == Flavor::Task)
2871                    .collect();
2872
2873                let mut ordered_cnx: Vec<(usize, SerializedCnx)> = graph
2874                    .0
2875                    .edge_indices()
2876                    .map(|edge_idx| {
2877                        let edge = &graph.0[edge_idx];
2878                        let order = if edge.order == usize::MAX {
2879                            edge_idx.index()
2880                        } else {
2881                            edge.order
2882                        };
2883                        (order, SerializedCnx::from(edge))
2884                    })
2885                    .collect();
2886                for node_idx in graph.0.node_indices() {
2887                    let node = &graph.0[node_idx];
2888                    if node.get_flavor() != Flavor::Task {
2889                        continue;
2890                    }
2891                    for (msg, order) in node.nc_outputs_with_order() {
2892                        ordered_cnx.push((
2893                            order,
2894                            SerializedCnx {
2895                                src: node.get_id(),
2896                                dst: NC_ENDPOINT.to_string(),
2897                                msg: msg.clone(),
2898                                missions: None,
2899                            },
2900                        ));
2901                    }
2902                }
2903                ordered_cnx.sort_by(|(order_a, cnx_a), (order_b, cnx_b)| {
2904                    order_a
2905                        .cmp(order_b)
2906                        .then_with(|| cnx_a.src.cmp(&cnx_b.src))
2907                        .then_with(|| cnx_a.dst.cmp(&cnx_b.dst))
2908                        .then_with(|| cnx_a.msg.cmp(&cnx_b.msg))
2909                });
2910                let cnx: Vec<SerializedCnx> = ordered_cnx
2911                    .into_iter()
2912                    .map(|(_, serialized)| serialized)
2913                    .collect();
2914
2915                CuConfigRepresentation {
2916                    constants: (!self.constants.is_empty()).then_some(self.constants.clone()),
2917                    tasks: Some(tasks),
2918                    bridges: bridges.clone(),
2919                    cnx: Some(cnx),
2920                    monitors: monitors.clone(),
2921                    logging: self.logging.clone(),
2922                    runtime: self.runtime.clone(),
2923                    resources: resources.clone(),
2924                    log_streaming: self.log_streaming.clone(),
2925                    missions: None,
2926                    includes: None,
2927                }
2928                .serialize(serializer)
2929            }
2930            Missions(graphs) => {
2931                let graphs = graphs.iter().collect::<BTreeMap<_, _>>();
2932                let missions = graphs
2933                    .keys()
2934                    .map(|id| MissionsConfig { id: (*id).clone() })
2935                    .collect();
2936
2937                // Collect all unique tasks across missions
2938                let mut tasks = Vec::new();
2939                let mut ordered_cnx: Vec<(usize, SerializedCnx)> = Vec::new();
2940
2941                for (mission_id, graph) in graphs {
2942                    // Add all nodes from this mission
2943                    for node_idx in graph.node_indices() {
2944                        let node = &graph[node_idx];
2945                        if node.get_flavor() == Flavor::Task
2946                            && !tasks.iter().any(|n: &Node| n.id == node.id)
2947                        {
2948                            tasks.push(node.clone());
2949                        }
2950                    }
2951
2952                    // Add all edges from this mission
2953                    for edge_idx in graph.0.edge_indices() {
2954                        let edge = &graph.0[edge_idx];
2955                        let order = if edge.order == usize::MAX {
2956                            edge_idx.index()
2957                        } else {
2958                            edge.order
2959                        };
2960                        let serialized = SerializedCnx::from(edge);
2961                        if let Some((existing_order, existing_serialized)) =
2962                            ordered_cnx.iter_mut().find(|(_, c)| {
2963                                c.src == serialized.src
2964                                    && c.dst == serialized.dst
2965                                    && c.msg == serialized.msg
2966                            })
2967                        {
2968                            if order < *existing_order {
2969                                *existing_order = order;
2970                            }
2971                            merge_connection_missions(
2972                                &mut existing_serialized.missions,
2973                                &serialized.missions,
2974                            );
2975                        } else {
2976                            ordered_cnx.push((order, serialized));
2977                        }
2978                    }
2979                    for node_idx in graph.0.node_indices() {
2980                        let node = &graph.0[node_idx];
2981                        if node.get_flavor() != Flavor::Task {
2982                            continue;
2983                        }
2984                        for (msg, order) in node.nc_outputs_with_order() {
2985                            let serialized = SerializedCnx {
2986                                src: node.get_id(),
2987                                dst: NC_ENDPOINT.to_string(),
2988                                msg: msg.clone(),
2989                                missions: Some(vec![mission_id.clone()]),
2990                            };
2991                            if let Some((existing_order, existing_serialized)) =
2992                                ordered_cnx.iter_mut().find(|(_, c)| {
2993                                    c.src == serialized.src
2994                                        && c.dst == serialized.dst
2995                                        && c.msg == serialized.msg
2996                                })
2997                            {
2998                                if order < *existing_order {
2999                                    *existing_order = order;
3000                                }
3001                                merge_connection_missions(
3002                                    &mut existing_serialized.missions,
3003                                    &serialized.missions,
3004                                );
3005                            } else {
3006                                ordered_cnx.push((order, serialized));
3007                            }
3008                        }
3009                    }
3010                }
3011                ordered_cnx.sort_by(|(order_a, cnx_a), (order_b, cnx_b)| {
3012                    order_a
3013                        .cmp(order_b)
3014                        .then_with(|| cnx_a.src.cmp(&cnx_b.src))
3015                        .then_with(|| cnx_a.dst.cmp(&cnx_b.dst))
3016                        .then_with(|| cnx_a.msg.cmp(&cnx_b.msg))
3017                });
3018                let cnx: Vec<SerializedCnx> = ordered_cnx
3019                    .into_iter()
3020                    .map(|(_, serialized)| serialized)
3021                    .collect();
3022
3023                CuConfigRepresentation {
3024                    constants: (!self.constants.is_empty()).then_some(self.constants.clone()),
3025                    tasks: Some(tasks),
3026                    resources: resources.clone(),
3027                    log_streaming: self.log_streaming.clone(),
3028                    bridges,
3029                    cnx: Some(cnx),
3030                    monitors,
3031                    logging: self.logging.clone(),
3032                    runtime: self.runtime.clone(),
3033                    missions: Some(missions),
3034                    includes: None,
3035                }
3036                .serialize(serializer)
3037            }
3038        }
3039    }
3040}
3041
3042impl Default for CuConfig {
3043    fn default() -> Self {
3044        CuConfig {
3045            constants: Vec::new(),
3046            graphs: Simple(CuGraph(StableDiGraph::new())),
3047            monitors: Vec::new(),
3048            logging: None,
3049            runtime: None,
3050            resources: Vec::new(),
3051            log_streaming: None,
3052            bridges: Vec::new(),
3053        }
3054    }
3055}
3056
3057/// The implementation has a lot of convenience methods to manipulate
3058/// the configuration to give some flexibility into programmatically creating the configuration.
3059impl CuConfig {
3060    #[allow(dead_code)]
3061    pub fn new_simple_type() -> Self {
3062        Self::default()
3063    }
3064
3065    #[allow(dead_code)]
3066    pub fn new_mission_type() -> Self {
3067        CuConfig {
3068            constants: Vec::new(),
3069            graphs: Missions(HashMap::new()),
3070            monitors: Vec::new(),
3071            logging: None,
3072            runtime: None,
3073            resources: Vec::new(),
3074            log_streaming: None,
3075            bridges: Vec::new(),
3076        }
3077    }
3078
3079    pub(crate) fn get_options() -> Options {
3080        Options::default()
3081            .with_default_extension(Extensions::IMPLICIT_SOME)
3082            .with_default_extension(Extensions::UNWRAP_NEWTYPES)
3083            .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
3084    }
3085
3086    #[allow(dead_code)]
3087    pub fn serialize_ron(&self) -> CuResult<String> {
3088        let ron = Self::get_options();
3089        let pretty = ron::ser::PrettyConfig::default();
3090        ron.to_string_pretty(&self, pretty)
3091            .map_err(|e| CuError::from(format!("Error serializing configuration: {e}")))
3092    }
3093
3094    #[allow(dead_code)]
3095    pub fn deserialize_ron(ron: &str) -> CuResult<Self> {
3096        let representation = Self::get_options().from_str(ron).map_err(|e| {
3097            CuError::from(format!(
3098                "Syntax Error in config: {} at position {}",
3099                e.code, e.span
3100            ))
3101        })?;
3102        Self::deserialize_impl(representation)
3103            .map_err(|e| CuError::from(format!("Error deserializing configuration: {e}")))
3104    }
3105
3106    fn deserialize_impl(representation: CuConfigRepresentation) -> Result<Self, String> {
3107        deserialize_config_representation(&representation)
3108    }
3109
3110    /// Render the configuration graph in the dot format.
3111    #[cfg(feature = "std")]
3112    #[allow(dead_code)]
3113    pub fn render(
3114        &self,
3115        output: &mut dyn std::io::Write,
3116        mission_id: Option<&str>,
3117    ) -> CuResult<()> {
3118        writeln!(output, "digraph G {{")
3119            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3120        writeln!(output, "    graph [rankdir=LR, nodesep=0.8, ranksep=1.2];")
3121            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3122        writeln!(output, "    node [shape=plain, fontname=\"Noto Sans\"];")
3123            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3124        writeln!(output, "    edge [fontname=\"Noto Sans\"];")
3125            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3126
3127        let sections = match (&self.graphs, mission_id) {
3128            (Simple(graph), _) => vec![RenderSection { label: None, graph }],
3129            (Missions(graphs), Some(id)) => {
3130                let graph = graphs
3131                    .get(id)
3132                    .ok_or_else(|| CuError::from(format!("Mission {id} not found")))?;
3133                vec![RenderSection {
3134                    label: Some(id.to_string()),
3135                    graph,
3136                }]
3137            }
3138            (Missions(graphs), None) => {
3139                let mut missions: Vec<_> = graphs.iter().collect();
3140                missions.sort_by(|a, b| a.0.cmp(b.0));
3141                missions
3142                    .into_iter()
3143                    .map(|(label, graph)| RenderSection {
3144                        label: Some(label.clone()),
3145                        graph,
3146                    })
3147                    .collect()
3148            }
3149        };
3150
3151        for section in sections {
3152            self.render_section(output, section.graph, section.label.as_deref())?;
3153        }
3154
3155        writeln!(output, "}}")
3156            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3157        Ok(())
3158    }
3159
3160    #[allow(dead_code)]
3161    pub fn get_all_instances_configs(
3162        &self,
3163        mission_id: Option<&str>,
3164    ) -> Vec<Option<&ComponentConfig>> {
3165        let graph = self.graphs.get_graph(mission_id).unwrap();
3166        graph
3167            .get_all_nodes()
3168            .iter()
3169            .map(|(_, node)| node.get_instance_config())
3170            .collect()
3171    }
3172
3173    #[allow(dead_code)]
3174    pub fn get_graph(&self, mission_id: Option<&str>) -> CuResult<&CuGraph> {
3175        self.graphs.get_graph(mission_id)
3176    }
3177
3178    #[allow(dead_code)]
3179    pub fn get_graph_mut(&mut self, mission_id: Option<&str>) -> CuResult<&mut CuGraph> {
3180        self.graphs.get_graph_mut(mission_id)
3181    }
3182
3183    #[allow(dead_code)]
3184    pub fn get_monitor_config(&self) -> Option<&MonitorConfig> {
3185        self.monitors.first()
3186    }
3187
3188    #[allow(dead_code)]
3189    pub fn get_monitor_configs(&self) -> &[MonitorConfig] {
3190        &self.monitors
3191    }
3192
3193    #[allow(dead_code)]
3194    pub fn get_runtime_config(&self) -> Option<&RuntimeConfig> {
3195        self.runtime.as_ref()
3196    }
3197
3198    #[allow(dead_code)]
3199    pub fn find_task_node(&self, mission_id: Option<&str>, task_id: &str) -> Option<&Node> {
3200        self.get_graph(mission_id)
3201            .ok()?
3202            .get_all_nodes()
3203            .into_iter()
3204            .find_map(|(_, node)| {
3205                (node.get_flavor() == Flavor::Task && node.id == task_id).then_some(node)
3206            })
3207    }
3208
3209    #[allow(dead_code)]
3210    pub fn find_logging_codec_spec(&self, codec_id: &str) -> Option<&LoggingCodecSpec> {
3211        self.logging
3212            .as_ref()?
3213            .codecs
3214            .iter()
3215            .find(|spec| spec.id == codec_id)
3216    }
3217
3218    /// Validate compile-time constant names, shapes, scalar ranges, and unit compatibility.
3219    pub fn validate_constants(&self) -> CuResult<()> {
3220        fn validate_integer(
3221            id: &str,
3222            storage: ConstantStorage,
3223            number: ConstantNumber,
3224        ) -> CuResult<()> {
3225            let valid = match (storage, number) {
3226                (ConstantStorage::I8, ConstantNumber::Signed(value)) => i8::try_from(value).is_ok(),
3227                (ConstantStorage::I16, ConstantNumber::Signed(value)) => {
3228                    i16::try_from(value).is_ok()
3229                }
3230                (ConstantStorage::I32, ConstantNumber::Signed(value)) => {
3231                    i32::try_from(value).is_ok()
3232                }
3233                (ConstantStorage::I64, ConstantNumber::Signed(_)) => true,
3234                (ConstantStorage::Isize, ConstantNumber::Signed(value)) => {
3235                    isize::try_from(value).is_ok()
3236                }
3237                (ConstantStorage::U8, ConstantNumber::Unsigned(value)) => {
3238                    u8::try_from(value).is_ok()
3239                }
3240                (ConstantStorage::U16, ConstantNumber::Unsigned(value)) => {
3241                    u16::try_from(value).is_ok()
3242                }
3243                (ConstantStorage::U32, ConstantNumber::Unsigned(value)) => {
3244                    u32::try_from(value).is_ok()
3245                }
3246                (ConstantStorage::U64, ConstantNumber::Unsigned(_)) => true,
3247                (ConstantStorage::Usize, ConstantNumber::Unsigned(value)) => {
3248                    usize::try_from(value).is_ok()
3249                }
3250                _ => false,
3251            };
3252            if valid {
3253                Ok(())
3254            } else {
3255                Err(CuError::from(format!(
3256                    "Constant '{id}' value {number:?} cannot be represented as {}",
3257                    storage.rust_type()
3258                )))
3259            }
3260        }
3261
3262        let mut ids = HashMap::new();
3263        for constant in &self.constants {
3264            if constant.id().is_empty() {
3265                return Err(CuError::from("Constant ids cannot be empty"));
3266            }
3267            if ids
3268                .insert((constant.module_path(), constant.id()), ())
3269                .is_some()
3270            {
3271                return Err(CuError::from(format!(
3272                    "Duplicate constant '{}'. Constant ids must be unique within a module.",
3273                    constant.qualified_id()
3274                )));
3275            }
3276
3277            match (
3278                constant.value.is_some(),
3279                constant.rust_type.as_deref(),
3280                constant.expression.as_deref(),
3281            ) {
3282                (true, None, None) => {}
3283                (true, _, _) => {
3284                    return Err(CuError::from(format!(
3285                        "Constant '{}' cannot combine numeric 'value' with 'type' or 'expression'",
3286                        constant.id()
3287                    )));
3288                }
3289                (false, Some(rust_type), Some(expression)) => {
3290                    if constant.storage.is_some()
3291                        || constant.quantity.is_some()
3292                        || constant.unit.is_some()
3293                    {
3294                        return Err(CuError::from(format!(
3295                            "Constant '{}' cannot combine 'type' and 'expression' with numeric 'storage', 'quantity', or 'unit'",
3296                            constant.id()
3297                        )));
3298                    }
3299                    if rust_type.trim().is_empty() {
3300                        return Err(CuError::from(format!(
3301                            "Constant '{}' type cannot be empty",
3302                            constant.id()
3303                        )));
3304                    }
3305                    if expression.trim().is_empty() {
3306                        return Err(CuError::from(format!(
3307                            "Constant '{}' expression cannot be empty",
3308                            constant.id()
3309                        )));
3310                    }
3311                    continue;
3312                }
3313                (false, Some(_), None) => {
3314                    return Err(CuError::from(format!(
3315                        "Constant '{}' declares 'type' without 'expression'",
3316                        constant.id()
3317                    )));
3318                }
3319                (false, None, Some(_)) => {
3320                    return Err(CuError::from(format!(
3321                        "Constant '{}' declares 'expression' without 'type'",
3322                        constant.id()
3323                    )));
3324                }
3325                (false, None, None) => {
3326                    return Err(CuError::from(format!(
3327                        "Constant '{}' must declare either numeric 'value' or both 'type' and 'expression'",
3328                        constant.id()
3329                    )));
3330                }
3331            }
3332
3333            if constant.quantity().is_none() && constant.explicit_unit().is_some() {
3334                return Err(CuError::from(format!(
3335                    "Constant '{}' declares a unit without a quantity",
3336                    constant.id()
3337                )));
3338            }
3339
3340            if constant.quantity().is_some() {
3341                if !constant.storage().supports_quantity() {
3342                    return Err(CuError::from(format!(
3343                        "Constant '{}' quantity '{}' requires storage f32 or f64, not {}",
3344                        constant.id(),
3345                        constant.quantity().map_or("", |quantity| quantity.name()),
3346                        constant.storage().rust_type()
3347                    )));
3348                }
3349                let normalized = match constant.storage() {
3350                    ConstantStorage::F32 => constant.normalized_f32().map(|_| ()),
3351                    ConstantStorage::F64 => constant.normalized_f64().map(|_| ()),
3352                    _ => unreachable!("quantity storage was checked above"),
3353                };
3354                normalized.map_err(CuError::from)?;
3355                continue;
3356            }
3357
3358            let (_, numbers) = constant.numbers().map_err(CuError::from)?;
3359            for number in numbers {
3360                match constant.storage() {
3361                    ConstantStorage::F32 => {
3362                        if !(number.as_f64() as f32).is_finite() {
3363                            return Err(CuError::from(format!(
3364                                "Constant '{}' values must be finite",
3365                                constant.id()
3366                            )));
3367                        }
3368                    }
3369                    ConstantStorage::F64 => {
3370                        if !number.as_f64().is_finite() {
3371                            return Err(CuError::from(format!(
3372                                "Constant '{}' values must be finite",
3373                                constant.id()
3374                            )));
3375                        }
3376                    }
3377                    storage => validate_integer(constant.id(), storage, number)?,
3378                }
3379            }
3380        }
3381        Ok(())
3382    }
3383
3384    /// Validate the logging configuration to ensure section pre-allocation sizes do not exceed slab sizes.
3385    /// This method is wrapper around [LoggingConfig::validate]
3386    pub fn validate_logging_config(&self) -> CuResult<()> {
3387        if let Some(logging) = &self.logging {
3388            return logging.validate();
3389        }
3390        Ok(())
3391    }
3392
3393    /// Validate the runtime configuration.
3394    pub fn validate_runtime_config(&self) -> CuResult<()> {
3395        if let Some(runtime) = &self.runtime {
3396            runtime.validate()?;
3397            if let Some(planner) = &runtime.planner {
3398                let copperlist_count = self
3399                    .logging
3400                    .as_ref()
3401                    .and_then(|logging| logging.copperlist_count)
3402                    .unwrap_or(DEFAULT_COPPERLIST_COUNT);
3403                planner.max_in_flight(copperlist_count)?;
3404            }
3405        }
3406        Ok(())
3407    }
3408
3409    fn validate_background_input_configs(&self) -> CuResult<()> {
3410        let validate = |graph: &CuGraph| -> CuResult<()> {
3411            for (node_id, node) in graph.get_all_nodes() {
3412                if node.background_process_empty()
3413                    && (!node.is_background()
3414                        || resolve_task_kind_for_id(graph, node_id)? != TaskKind::Regular)
3415                {
3416                    return Err(CuError::from(format!(
3417                        "Task '{}': background_process_empty requires a regular background task",
3418                        node.id
3419                    )));
3420                }
3421            }
3422            Ok(())
3423        };
3424        match &self.graphs {
3425            Simple(graph) => validate(graph),
3426            Missions(graphs) => {
3427                for graph in graphs.values() {
3428                    validate(graph)?;
3429                }
3430                Ok(())
3431            }
3432        }
3433    }
3434
3435    fn validate_stateless_configs(&self) -> CuResult<()> {
3436        match &self.graphs {
3437            Simple(graph) => validate_stateless_graph(graph),
3438            Missions(graphs) => {
3439                for graph in graphs.values() {
3440                    validate_stateless_graph(graph)?;
3441                }
3442                Ok(())
3443            }
3444        }
3445    }
3446
3447    /// Validates every `anytime:` policy in the resolved graphs.
3448    ///
3449    /// Runs at configuration-resolution time, the first point where both the
3450    /// resolved graphs and `runtime.rate_target_hz` are known:
3451    ///
3452    /// 1. node-local bounds and ranges (see [`AnytimeConfig`]);
3453    /// 2. `anytime:` is only supported on regular tasks — refinement needs both
3454    ///    an input and an output;
3455    /// 3. an anytime task has exactly one input connection (the runner anchors
3456    ///    the job on the input's Tov) and at most one output message type
3457    ///    (`base()` and every `refine()` write the same output slot);
3458    /// 4. a *foreground* anytime task needs `max_refines`: the execution plan
3459    ///    is static (the node compiles to a base step plus `max_refines` refine
3460    ///    steps, see `curuntime::expand_anytime_steps`), so the refine count
3461    ///    must be known at compile time;
3462    /// 5. fit the period: a *foreground* anytime task in a rate-limited config
3463    ///    must set a time bound (`time_budget_ms` or `max_age_ms`), and the
3464    ///    worst-case window — `min` of the ones set — must be smaller than the
3465    ///    loop period. Background nodes and configs without a rate target skip
3466    ///    this check.
3467    pub fn validate_anytime_configs(&self) -> CuResult<()> {
3468        let rate_target_hz = self.runtime.as_ref().and_then(|r| r.rate_target_hz);
3469        match &self.graphs {
3470            Simple(graph) => validate_anytime_graph(graph, rate_target_hz),
3471            Missions(graphs) => {
3472                for graph in graphs.values() {
3473                    validate_anytime_graph(graph, rate_target_hz)?;
3474                }
3475                Ok(())
3476            }
3477        }
3478    }
3479}
3480
3481/// Checks every `anytime:` node of one graph: local bounds, regular-task kind,
3482/// single-input/single-output arity, and the foreground fit-the-period rule
3483/// (see [`CuConfig::validate_anytime_configs`]).
3484fn validate_anytime_graph(graph: &CuGraph, rate_target_hz: Option<u64>) -> CuResult<()> {
3485    for (node_id, node) in graph.get_all_nodes() {
3486        let Some(anytime) = node.anytime() else {
3487            continue;
3488        };
3489        anytime.validate(&node.id)?;
3490
3491        let kind = resolve_task_kind_for_id(graph, node_id)?;
3492        if kind != TaskKind::Regular {
3493            return Err(CuError::from(format!(
3494                "Task '{}' is declared with an anytime: policy but resolves to kind '{}'. Anytime refinement needs both an input and an output, so it is only supported on regular tasks.",
3495                node.id,
3496                kind.as_str()
3497            )));
3498        }
3499
3500        // Foreground and background alike: the runner reads the job anchor
3501        // from the single input's Tov, and base()/refine() write one stable
3502        // output slot. Zero declared outputs is fine when the kind is
3503        // declared — the macro synthesizes exactly one nc output.
3504        let input_count = graph.get_dst_edges(node_id)?.len();
3505        if input_count != 1 {
3506            return Err(CuError::from(format!(
3507                "Task '{}' is an anytime task and must have exactly one input connection (found {input_count}): the runner anchors the job on the input's Tov.",
3508                node.id
3509            )));
3510        }
3511        let output_count = graph.get_node_output_msg_types_by_id(node_id)?.len();
3512        if output_count > 1 {
3513            return Err(CuError::from(format!(
3514                "Task '{}' is an anytime task and must have exactly one output message type (found {output_count}): base() and every refine() write the same output slot.",
3515                node.id
3516            )));
3517        }
3518
3519        // Background placement: the refinement window runs on a worker thread
3520        // and may exceed the copperlist period — that is the point of it.
3521        if node.is_background() {
3522            continue;
3523        }
3524
3525        // Foreground placement compiles to a static plan: the node's step is
3526        // followed by exactly max_refines refine steps, so the count must be
3527        // known here — a time-only hard bound cannot produce a static plan.
3528        if anytime.max_refines.is_none() {
3529            return Err(CuError::from(format!(
3530                "Task '{}' is a foreground anytime task and needs anytime.max_refines: the execution plan is static, so the refine step count must be known at compile time. time_budget_ms/max_age_ms remain early-stop conditions within those quanta.",
3531                node.id
3532            )));
3533        }
3534
3535        let Some(rate_target_hz) = rate_target_hz else {
3536            continue;
3537        };
3538        let window_ms = match (anytime.time_budget_ms, anytime.max_age_ms) {
3539            (Some(budget), Some(age)) => budget.min(age),
3540            (Some(budget), None) => budget,
3541            (None, Some(age)) => age,
3542            (None, None) => {
3543                return Err(CuError::from(format!(
3544                    "Task '{}' is a foreground anytime task in a rate-limited config and needs a time bound: set anytime.time_budget_ms or anytime.max_age_ms. max_refines alone gives the runtime no time quantity to check against the {rate_target_hz} Hz loop period, and one slow quantum would silently overrun it.",
3545                    node.id
3546                )));
3547            }
3548        };
3549        let period_ms = 1_000.0 / rate_target_hz as f64;
3550        if window_ms >= period_ms {
3551            return Err(CuError::from(format!(
3552                "Task '{}': the worst-case anytime window ({window_ms} ms) does not fit within the {rate_target_hz} Hz loop period ({period_ms} ms) with headroom for the rest of the copperlist. Tighten the time bound, lower runtime.rate_target_hz, or run the task with background: true.",
3553                node.id
3554            )));
3555        }
3556    }
3557    Ok(())
3558}
3559
3560fn validate_stateless_graph(graph: &CuGraph) -> CuResult<()> {
3561    for (node_id, node) in graph.get_all_nodes() {
3562        if node.get_declared_task_kind() != Some(TaskKind::Stateless) {
3563            continue;
3564        }
3565
3566        resolve_task_kind_for_id(graph, node_id)?;
3567        if node.is_background() {
3568            return Err(CuError::from(format!(
3569                "Task '{}' is declared as kind 'stateless_task' and cannot be backgrounded.",
3570                node.id
3571            )));
3572        }
3573        if node.is_anytime() {
3574            return Err(CuError::from(format!(
3575                "Task '{}' is declared as kind 'stateless_task' and cannot use an anytime policy.",
3576                node.id
3577            )));
3578        }
3579    }
3580    Ok(())
3581}
3582
3583#[cfg(feature = "std")]
3584#[derive(Default)]
3585#[doc(hidden)]
3586pub struct PortLookup {
3587    pub inputs: HashMap<String, String>,
3588    pub outputs: HashMap<String, String>,
3589    pub default_input: Option<String>,
3590    pub default_output: Option<String>,
3591}
3592
3593#[cfg(feature = "std")]
3594#[derive(Clone)]
3595#[doc(hidden)]
3596pub struct RenderNode {
3597    pub id: String,
3598    pub type_name: String,
3599    pub flavor: Flavor,
3600    pub inputs: Vec<String>,
3601    pub outputs: Vec<String>,
3602}
3603
3604#[cfg(feature = "std")]
3605#[derive(Clone)]
3606#[doc(hidden)]
3607pub struct RenderConnection {
3608    pub src: String,
3609    pub src_port: Option<String>,
3610    #[allow(dead_code)]
3611    pub src_channel: Option<String>,
3612    pub dst: String,
3613    pub dst_port: Option<String>,
3614    #[allow(dead_code)]
3615    pub dst_channel: Option<String>,
3616    pub msg: String,
3617}
3618
3619#[cfg(feature = "std")]
3620#[doc(hidden)]
3621pub struct RenderTopology {
3622    pub nodes: Vec<RenderNode>,
3623    pub connections: Vec<RenderConnection>,
3624}
3625
3626#[cfg(feature = "std")]
3627impl RenderTopology {
3628    pub fn sort_connections(&mut self) {
3629        self.connections.sort_by(|a, b| {
3630            a.src
3631                .cmp(&b.src)
3632                .then(a.dst.cmp(&b.dst))
3633                .then(a.msg.cmp(&b.msg))
3634        });
3635    }
3636}
3637
3638#[cfg(feature = "std")]
3639#[allow(dead_code)]
3640struct RenderSection<'a> {
3641    label: Option<String>,
3642    graph: &'a CuGraph,
3643}
3644
3645#[cfg(feature = "std")]
3646impl CuConfig {
3647    #[allow(dead_code)]
3648    fn render_section(
3649        &self,
3650        output: &mut dyn std::io::Write,
3651        graph: &CuGraph,
3652        label: Option<&str>,
3653    ) -> CuResult<()> {
3654        use std::fmt::Write as FmtWrite;
3655
3656        let mut topology = build_render_topology(graph, &self.bridges);
3657        topology.nodes.sort_by(|a, b| a.id.cmp(&b.id));
3658        topology.sort_connections();
3659
3660        let cluster_id = label.map(|lbl| format!("cluster_{}", sanitize_identifier(lbl)));
3661        if let Some(ref cluster_id) = cluster_id {
3662            writeln!(output, "    subgraph \"{cluster_id}\" {{")
3663                .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3664            writeln!(
3665                output,
3666                "        label=<<B>Mission: {}</B>>;",
3667                encode_text(label.unwrap())
3668            )
3669            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3670            writeln!(
3671                output,
3672                "        labelloc=t; labeljust=l; color=\"#bbbbbb\"; style=\"rounded\"; margin=20;"
3673            )
3674            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3675        }
3676        let indent = if cluster_id.is_some() {
3677            "        "
3678        } else {
3679            "    "
3680        };
3681        let node_prefix = label
3682            .map(|lbl| format!("{}__", sanitize_identifier(lbl)))
3683            .unwrap_or_default();
3684
3685        let mut port_lookup: HashMap<String, PortLookup> = HashMap::new();
3686        let mut id_lookup: HashMap<String, String> = HashMap::new();
3687
3688        for node in &topology.nodes {
3689            let node_idx = graph
3690                .get_node_id_by_name(node.id.as_str())
3691                .ok_or_else(|| CuError::from(format!("Node '{}' missing from graph", node.id)))?;
3692            let node_weight = graph
3693                .get_node(node_idx)
3694                .ok_or_else(|| CuError::from(format!("Node '{}' missing weight", node.id)))?;
3695
3696            let fillcolor = match node.flavor {
3697                Flavor::Bridge => "#faedcd",
3698                Flavor::Task => match resolve_task_kind_for_id(graph, node_idx)? {
3699                    TaskKind::Source => "#ddefc7",
3700                    TaskKind::Sink => "#cce0ff",
3701                    TaskKind::Regular => "#f2f2f2",
3702                    TaskKind::Stateless => "#cba6f7",
3703                },
3704            };
3705
3706            let port_base = format!("{}{}", node_prefix, sanitize_identifier(&node.id));
3707            let (inputs_table, input_map, default_input) =
3708                build_port_table("Inputs", &node.inputs, &port_base, "in");
3709            let (outputs_table, output_map, default_output) =
3710                build_port_table("Outputs", &node.outputs, &port_base, "out");
3711            let config_html = node_weight.config.as_ref().and_then(build_config_table);
3712
3713            let mut label_html = String::new();
3714            write!(
3715                label_html,
3716                "<TABLE BORDER=\"0\" CELLBORDER=\"1\" CELLSPACING=\"0\" CELLPADDING=\"6\" COLOR=\"gray\" BGCOLOR=\"white\">"
3717            )
3718            .unwrap();
3719            write!(
3720                label_html,
3721                "<TR><TD COLSPAN=\"2\" ALIGN=\"LEFT\" BGCOLOR=\"{fillcolor}\"><FONT POINT-SIZE=\"12\"><B>{}</B></FONT><BR/><FONT COLOR=\"dimgray\">[{}]</FONT></TD></TR>",
3722                encode_text(&node.id),
3723                encode_text(&node.type_name)
3724            )
3725            .unwrap();
3726            write!(
3727                label_html,
3728                "<TR><TD ALIGN=\"LEFT\" VALIGN=\"TOP\">{inputs_table}</TD><TD ALIGN=\"LEFT\" VALIGN=\"TOP\">{outputs_table}</TD></TR>"
3729            )
3730            .unwrap();
3731
3732            if let Some(config_html) = config_html {
3733                write!(
3734                    label_html,
3735                    "<TR><TD COLSPAN=\"2\" ALIGN=\"LEFT\">{config_html}</TD></TR>"
3736                )
3737                .unwrap();
3738            }
3739
3740            label_html.push_str("</TABLE>");
3741
3742            let identifier_raw = if node_prefix.is_empty() {
3743                node.id.clone()
3744            } else {
3745                format!("{node_prefix}{}", node.id)
3746            };
3747            let identifier = escape_dot_id(&identifier_raw);
3748            writeln!(output, "{indent}\"{identifier}\" [label=<{label_html}>];")
3749                .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3750
3751            id_lookup.insert(node.id.clone(), identifier);
3752            port_lookup.insert(
3753                node.id.clone(),
3754                PortLookup {
3755                    inputs: input_map,
3756                    outputs: output_map,
3757                    default_input,
3758                    default_output,
3759                },
3760            );
3761        }
3762
3763        for cnx in &topology.connections {
3764            let src_id = id_lookup
3765                .get(&cnx.src)
3766                .ok_or_else(|| CuError::from(format!("Unknown node '{}'", cnx.src)))?;
3767            let dst_id = id_lookup
3768                .get(&cnx.dst)
3769                .ok_or_else(|| CuError::from(format!("Unknown node '{}'", cnx.dst)))?;
3770            let src_suffix = port_lookup
3771                .get(&cnx.src)
3772                .and_then(|lookup| lookup.resolve_output(cnx.src_port.as_deref()))
3773                .map(|port| format!(":\"{port}\":e"))
3774                .unwrap_or_default();
3775            let dst_suffix = port_lookup
3776                .get(&cnx.dst)
3777                .and_then(|lookup| lookup.resolve_input(cnx.dst_port.as_deref()))
3778                .map(|port| format!(":\"{port}\":w"))
3779                .unwrap_or_default();
3780            let msg = encode_text(&cnx.msg);
3781            writeln!(
3782                output,
3783                "{indent}\"{src_id}\"{src_suffix} -> \"{dst_id}\"{dst_suffix} [label=< <B><FONT COLOR=\"gray\">{msg}</FONT></B> >];"
3784            )
3785            .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3786        }
3787
3788        if cluster_id.is_some() {
3789            writeln!(output, "    }}")
3790                .map_err(|e| CuError::new_with_cause("Failed to write render output", e))?;
3791        }
3792
3793        Ok(())
3794    }
3795}
3796
3797#[cfg(feature = "std")]
3798#[doc(hidden)]
3799pub fn build_render_topology(graph: &CuGraph, bridges: &[BridgeConfig]) -> RenderTopology {
3800    let mut bridge_lookup = HashMap::new();
3801    for bridge in bridges {
3802        bridge_lookup.insert(bridge.id.as_str(), bridge);
3803    }
3804
3805    let mut nodes: Vec<RenderNode> = Vec::new();
3806    let mut node_lookup: HashMap<String, usize> = HashMap::new();
3807    for (node_idx, node) in graph.get_all_nodes() {
3808        let node_id = node.get_id();
3809        let mut inputs = Vec::new();
3810        let mut outputs = Vec::new();
3811        if node.get_flavor() == Flavor::Bridge
3812            && let Some(bridge) = bridge_lookup.get(node_id.as_str())
3813        {
3814            for channel in &bridge.channels {
3815                match channel {
3816                    // Rx brings data from the bridge into the graph, so treat it as an output.
3817                    BridgeChannelConfigRepresentation::Rx { id, .. } => outputs.push(id.clone()),
3818                    // Tx consumes data from the graph heading into the bridge, so show it on the input side.
3819                    BridgeChannelConfigRepresentation::Tx { id, .. } => inputs.push(id.clone()),
3820                }
3821            }
3822        } else if node.get_flavor() == Flavor::Task {
3823            for (idx, msg) in graph
3824                .get_node_output_msg_types_by_id(node_idx)
3825                .unwrap_or_default()
3826                .into_iter()
3827                .enumerate()
3828            {
3829                outputs.push(format!("out{idx}: {msg}"));
3830            }
3831        }
3832
3833        node_lookup.insert(node_id.clone(), nodes.len());
3834        nodes.push(RenderNode {
3835            id: node_id,
3836            type_name: node.get_type().to_string(),
3837            flavor: node.get_flavor(),
3838            inputs,
3839            outputs,
3840        });
3841    }
3842
3843    let mut output_port_lookup: Vec<HashMap<String, String>> = vec![HashMap::new(); nodes.len()];
3844    for (node_idx, node) in graph.get_all_nodes() {
3845        let Some(&idx) = node_lookup.get(&node.get_id()) else {
3846            continue;
3847        };
3848        if node.get_flavor() != Flavor::Task {
3849            continue;
3850        }
3851        for (port_idx, msg) in graph
3852            .get_node_output_msg_types_by_id(node_idx)
3853            .unwrap_or_default()
3854            .into_iter()
3855            .enumerate()
3856        {
3857            output_port_lookup[idx].insert(msg.clone(), format!("out{port_idx}: {msg}"));
3858        }
3859    }
3860
3861    let mut auto_input_counts = vec![0usize; nodes.len()];
3862    for edge in graph.0.edge_references() {
3863        let cnx = edge.weight();
3864        if let Some(&idx) = node_lookup.get(&cnx.dst)
3865            && nodes[idx].flavor == Flavor::Task
3866            && cnx.dst_channel.is_none()
3867        {
3868            auto_input_counts[idx] += 1;
3869        }
3870    }
3871
3872    let mut next_auto_input = vec![0usize; nodes.len()];
3873    let mut connections = Vec::new();
3874    for edge in graph.0.edge_references() {
3875        let cnx = edge.weight();
3876        let mut src_port = cnx.src_channel.clone();
3877        let mut dst_port = cnx.dst_channel.clone();
3878
3879        if let Some(&idx) = node_lookup.get(&cnx.src) {
3880            let node = &mut nodes[idx];
3881            if node.flavor == Flavor::Task && src_port.is_none() {
3882                src_port = output_port_lookup[idx].get(&cnx.msg).cloned();
3883            }
3884        }
3885        if let Some(&idx) = node_lookup.get(&cnx.dst) {
3886            let node = &mut nodes[idx];
3887            if node.flavor == Flavor::Task && dst_port.is_none() {
3888                let count = auto_input_counts[idx];
3889                let next = if count <= 1 {
3890                    "in".to_string()
3891                } else {
3892                    let next = format!("in.{}", next_auto_input[idx]);
3893                    next_auto_input[idx] += 1;
3894                    next
3895                };
3896                node.inputs.push(next.clone());
3897                dst_port = Some(next);
3898            }
3899        }
3900
3901        connections.push(RenderConnection {
3902            src: cnx.src.clone(),
3903            src_port,
3904            src_channel: cnx.src_channel.clone(),
3905            dst: cnx.dst.clone(),
3906            dst_port,
3907            dst_channel: cnx.dst_channel.clone(),
3908            msg: cnx.msg.clone(),
3909        });
3910    }
3911
3912    RenderTopology { nodes, connections }
3913}
3914
3915#[cfg(feature = "std")]
3916impl PortLookup {
3917    pub fn resolve_input(&self, name: Option<&str>) -> Option<&str> {
3918        if let Some(name) = name
3919            && let Some(port) = self.inputs.get(name)
3920        {
3921            return Some(port.as_str());
3922        }
3923        self.default_input.as_deref()
3924    }
3925
3926    pub fn resolve_output(&self, name: Option<&str>) -> Option<&str> {
3927        if let Some(name) = name
3928            && let Some(port) = self.outputs.get(name)
3929        {
3930            return Some(port.as_str());
3931        }
3932        self.default_output.as_deref()
3933    }
3934}
3935
3936#[cfg(feature = "std")]
3937#[allow(dead_code)]
3938fn build_port_table(
3939    title: &str,
3940    names: &[String],
3941    base_id: &str,
3942    prefix: &str,
3943) -> (String, HashMap<String, String>, Option<String>) {
3944    use std::fmt::Write as FmtWrite;
3945
3946    let mut html = String::new();
3947    write!(
3948        html,
3949        "<TABLE BORDER=\"0\" CELLBORDER=\"0\" CELLSPACING=\"0\" CELLPADDING=\"1\">"
3950    )
3951    .unwrap();
3952    write!(
3953        html,
3954        "<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"dimgray\">{}</FONT></TD></TR>",
3955        encode_text(title)
3956    )
3957    .unwrap();
3958
3959    let mut lookup = HashMap::new();
3960    let mut default_port = None;
3961
3962    if names.is_empty() {
3963        html.push_str("<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"lightgray\">&mdash;</FONT></TD></TR>");
3964    } else {
3965        for (idx, name) in names.iter().enumerate() {
3966            let port_id = format!("{base_id}_{prefix}_{idx}");
3967            write!(
3968                html,
3969                "<TR><TD PORT=\"{port_id}\" ALIGN=\"LEFT\">{}</TD></TR>",
3970                encode_text(name)
3971            )
3972            .unwrap();
3973            lookup.insert(name.clone(), port_id.clone());
3974            if idx == 0 {
3975                default_port = Some(port_id);
3976            }
3977        }
3978    }
3979
3980    html.push_str("</TABLE>");
3981    (html, lookup, default_port)
3982}
3983
3984#[cfg(feature = "std")]
3985#[allow(dead_code)]
3986fn build_config_table(config: &ComponentConfig) -> Option<String> {
3987    use std::fmt::Write as FmtWrite;
3988
3989    if config.0.is_empty() {
3990        return None;
3991    }
3992
3993    let mut entries: Vec<_> = config.0.iter().collect();
3994    entries.sort_by(|a, b| a.0.cmp(b.0));
3995
3996    let mut html = String::new();
3997    html.push_str("<TABLE BORDER=\"0\" CELLBORDER=\"0\" CELLSPACING=\"0\" CELLPADDING=\"1\">");
3998    for (key, value) in entries {
3999        let value_txt = format!("{value}");
4000        write!(
4001            html,
4002            "<TR><TD ALIGN=\"LEFT\"><FONT COLOR=\"dimgray\">{}</FONT> = {}</TD></TR>",
4003            encode_text(key),
4004            encode_text(&value_txt)
4005        )
4006        .unwrap();
4007    }
4008    html.push_str("</TABLE>");
4009    Some(html)
4010}
4011
4012#[cfg(feature = "std")]
4013#[allow(dead_code)]
4014fn sanitize_identifier(value: &str) -> String {
4015    value
4016        .chars()
4017        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
4018        .collect()
4019}
4020
4021#[cfg(feature = "std")]
4022#[allow(dead_code)]
4023fn escape_dot_id(value: &str) -> String {
4024    let mut escaped = String::with_capacity(value.len());
4025    for ch in value.chars() {
4026        match ch {
4027            '"' => escaped.push_str("\\\""),
4028            '\\' => escaped.push_str("\\\\"),
4029            _ => escaped.push(ch),
4030        }
4031    }
4032    escaped
4033}
4034
4035impl LoggingConfig {
4036    /// Validate the logging configuration to ensure section pre-allocation sizes do not exceed slab sizes.
4037    pub fn validate(&self) -> CuResult<()> {
4038        if let Some(copperlist_count) = self.copperlist_count
4039            && copperlist_count == 0
4040        {
4041            return Err(CuError::from(
4042                "CopperList count cannot be zero. Set logging.copperlist_count to at least 1.",
4043            ));
4044        }
4045
4046        if let Some(section_size_mib) = self.section_size_mib
4047            && let Some(slab_size_mib) = self.slab_size_mib
4048            && section_size_mib > slab_size_mib
4049        {
4050            return Err(CuError::from(format!(
4051                "Section size ({section_size_mib} MiB) cannot be larger than slab size ({slab_size_mib} MiB). Adjust the parameters accordingly."
4052            )));
4053        }
4054
4055        let mut codec_ids = HashMap::new();
4056        for codec in &self.codecs {
4057            if codec_ids.insert(codec.id.as_str(), ()).is_some() {
4058                return Err(CuError::from(format!(
4059                    "Duplicate logging codec id '{}'. Codec ids must be unique.",
4060                    codec.id
4061                )));
4062            }
4063        }
4064
4065        Ok(())
4066    }
4067}
4068
4069impl RuntimeConfig {
4070    /// Reference configuration when clock discipline is enabled.
4071    pub fn clock_sync_config(&self) -> Option<&ClockSyncConfig> {
4072        #[cfg(feature = "clock-sync")]
4073        {
4074            self.clock.as_ref()
4075        }
4076        #[cfg(not(feature = "clock-sync"))]
4077        {
4078            None
4079        }
4080    }
4081
4082    /// Validate runtime loop-rate and reference-clock settings.
4083    pub fn validate(&self) -> CuResult<()> {
4084        if let Some(clock) = self.clock_sync_config() {
4085            clock.validate()?;
4086        }
4087        if let Some(rate_target_hz) = self.rate_target_hz {
4088            if rate_target_hz == 0 {
4089                return Err(CuError::from(
4090                    "Runtime rate target cannot be zero. Set runtime.rate_target_hz to at least 1.",
4091                ));
4092            }
4093
4094            if rate_target_hz > MAX_RATE_TARGET_HZ {
4095                return Err(CuError::from(format!(
4096                    "Runtime rate target ({rate_target_hz} Hz) exceeds the supported maximum of {MAX_RATE_TARGET_HZ} Hz."
4097                )));
4098            }
4099        }
4100
4101        Ok(())
4102    }
4103}
4104
4105#[allow(dead_code)] // dead in no-std
4106fn substitute_parameters(content: &str, params: &HashMap<String, Value>) -> String {
4107    let mut result = content.to_string();
4108
4109    for (key, value) in params {
4110        let pattern = format!("{{{{{key}}}}}");
4111        result = result.replace(&pattern, &value.to_string());
4112    }
4113
4114    result
4115}
4116
4117/// Returns a merged CuConfigRepresentation.
4118#[cfg(feature = "std")]
4119fn process_includes(
4120    file_path: &str,
4121    base_representation: CuConfigRepresentation,
4122    processed_files: &mut Vec<String>,
4123    active_features: &[&str],
4124) -> CuResult<CuConfigRepresentation> {
4125    // Note: Circular dependency detection removed
4126    processed_files.push(file_path.to_string());
4127
4128    let mut result = base_representation;
4129
4130    if let Some(includes) = result.includes.take() {
4131        for include in includes {
4132            if include
4133                .when
4134                .as_ref()
4135                .is_some_and(|predicate| !predicate.evaluate(active_features))
4136            {
4137                continue;
4138            }
4139
4140            let include_path = if include.path.starts_with('/') {
4141                include.path.clone()
4142            } else {
4143                let current_dir = std::path::Path::new(file_path).parent();
4144
4145                match current_dir.map(|path| path.to_string_lossy().to_string()) {
4146                    Some(current_dir) if !current_dir.is_empty() => {
4147                        format!("{}/{}", current_dir, include.path)
4148                    }
4149                    _ => include.path,
4150                }
4151            };
4152
4153            let include_content = read_to_string(&include_path).map_err(|e| {
4154                CuError::from(format!("Failed to read include file: {include_path}"))
4155                    .add_cause(e.to_string().as_str())
4156            })?;
4157
4158            let processed_content = substitute_parameters(&include_content, &include.params);
4159
4160            let mut included_representation: CuConfigRepresentation = match Options::default()
4161                .with_default_extension(Extensions::IMPLICIT_SOME)
4162                .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4163                .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4164                .from_str(&processed_content)
4165            {
4166                Ok(rep) => rep,
4167                Err(e) => {
4168                    return Err(CuError::from(format!(
4169                        "Failed to parse include file: {} - Error: {} at position {}",
4170                        include_path, e.code, e.span
4171                    )));
4172                }
4173            };
4174
4175            included_representation = process_includes(
4176                &include_path,
4177                included_representation,
4178                processed_files,
4179                active_features,
4180            )?;
4181
4182            if let Some(included_constants) = included_representation.constants {
4183                if result.constants.is_none() {
4184                    result.constants = Some(included_constants);
4185                } else {
4186                    let mut constants = result.constants.take().unwrap();
4187                    for included_constant in included_constants {
4188                        if !constants.iter().any(|constant| {
4189                            constant.id == included_constant.id
4190                                && constant.module_path() == included_constant.module_path()
4191                        }) {
4192                            constants.push(included_constant);
4193                        }
4194                    }
4195                    result.constants = Some(constants);
4196                }
4197            }
4198
4199            if let Some(included_tasks) = included_representation.tasks {
4200                if result.tasks.is_none() {
4201                    result.tasks = Some(included_tasks);
4202                } else {
4203                    let mut tasks = result.tasks.take().unwrap();
4204                    for included_task in included_tasks {
4205                        if !tasks.iter().any(|t| t.id == included_task.id) {
4206                            tasks.push(included_task);
4207                        }
4208                    }
4209                    result.tasks = Some(tasks);
4210                }
4211            }
4212
4213            if let Some(included_bridges) = included_representation.bridges {
4214                if result.bridges.is_none() {
4215                    result.bridges = Some(included_bridges);
4216                } else {
4217                    let mut bridges = result.bridges.take().unwrap();
4218                    for included_bridge in included_bridges {
4219                        if !bridges.iter().any(|b| b.id == included_bridge.id) {
4220                            bridges.push(included_bridge);
4221                        }
4222                    }
4223                    result.bridges = Some(bridges);
4224                }
4225            }
4226
4227            if let Some(included_resources) = included_representation.resources {
4228                if result.resources.is_none() {
4229                    result.resources = Some(included_resources);
4230                } else {
4231                    let mut resources = result.resources.take().unwrap();
4232                    for included_resource in included_resources {
4233                        if !resources.iter().any(|r| r.id == included_resource.id) {
4234                            resources.push(included_resource);
4235                        }
4236                    }
4237                    result.resources = Some(resources);
4238                }
4239            }
4240
4241            if let Some(included_cnx) = included_representation.cnx {
4242                if result.cnx.is_none() {
4243                    result.cnx = Some(included_cnx);
4244                } else {
4245                    let mut cnx = result.cnx.take().unwrap();
4246                    for included_c in included_cnx {
4247                        if let Some(existing_cnx) = cnx.iter_mut().find(|c| {
4248                            c.src == included_c.src
4249                                && c.dst == included_c.dst
4250                                && c.msg == included_c.msg
4251                        }) {
4252                            merge_connection_missions(
4253                                &mut existing_cnx.missions,
4254                                &included_c.missions,
4255                            );
4256                        } else {
4257                            cnx.push(included_c);
4258                        }
4259                    }
4260                    result.cnx = Some(cnx);
4261                }
4262            }
4263
4264            if let Some(included_monitors) = included_representation.monitors {
4265                if result.monitors.is_none() {
4266                    result.monitors = Some(included_monitors);
4267                } else {
4268                    let mut monitors = result.monitors.take().unwrap();
4269                    for included_monitor in included_monitors {
4270                        if !monitors.iter().any(|m| m.type_ == included_monitor.type_) {
4271                            monitors.push(included_monitor);
4272                        }
4273                    }
4274                    result.monitors = Some(monitors);
4275                }
4276            }
4277
4278            if result.logging.is_none() {
4279                result.logging = included_representation.logging;
4280            }
4281
4282            if result.runtime.is_none() {
4283                result.runtime = included_representation.runtime;
4284            }
4285
4286            if result.log_streaming.is_none() {
4287                result.log_streaming = included_representation.log_streaming;
4288            }
4289
4290            if let Some(included_missions) = included_representation.missions {
4291                if result.missions.is_none() {
4292                    result.missions = Some(included_missions);
4293                } else {
4294                    let mut missions = result.missions.take().unwrap();
4295                    for included_mission in included_missions {
4296                        if !missions.iter().any(|m| m.id == included_mission.id) {
4297                            missions.push(included_mission);
4298                        }
4299                    }
4300                    result.missions = Some(missions);
4301                }
4302            }
4303        }
4304    }
4305
4306    Ok(result)
4307}
4308
4309#[cfg(feature = "std")]
4310fn parse_instance_config_overrides_string(
4311    content: &str,
4312) -> CuResult<InstanceConfigOverridesRepresentation> {
4313    Options::default()
4314        .with_default_extension(Extensions::IMPLICIT_SOME)
4315        .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4316        .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4317        .from_str(content)
4318        .map_err(|e| {
4319            CuError::from(format!(
4320                "Failed to parse instance override file: Error: {} at position {}",
4321                e.code, e.span
4322            ))
4323        })
4324}
4325
4326#[cfg(feature = "std")]
4327fn merge_component_config(target: &mut Option<ComponentConfig>, value: &ComponentConfig) {
4328    if let Some(existing) = target {
4329        existing.merge_from(value);
4330    } else {
4331        *target = Some(value.clone());
4332    }
4333}
4334
4335#[cfg(feature = "std")]
4336fn apply_task_config_override_to_graph(
4337    graph: &mut CuGraph,
4338    task_id: &str,
4339    value: &ComponentConfig,
4340) -> usize {
4341    let mut matches = 0usize;
4342    let node_indices: Vec<_> = graph.0.node_indices().collect();
4343    for node_index in node_indices {
4344        let node = &mut graph.0[node_index];
4345        if node.get_flavor() == Flavor::Task && node.id == task_id {
4346            merge_component_config(&mut node.config, value);
4347            matches += 1;
4348        }
4349    }
4350    matches
4351}
4352
4353#[cfg(feature = "std")]
4354fn apply_bridge_node_config_override_to_graph(
4355    graph: &mut CuGraph,
4356    bridge_id: &str,
4357    value: &ComponentConfig,
4358) {
4359    let node_indices: Vec<_> = graph.0.node_indices().collect();
4360    for node_index in node_indices {
4361        let node = &mut graph.0[node_index];
4362        if node.get_flavor() == Flavor::Bridge && node.id == bridge_id {
4363            merge_component_config(&mut node.config, value);
4364        }
4365    }
4366}
4367
4368#[cfg(feature = "std")]
4369fn parse_instance_override_target(path: &str) -> CuResult<(InstanceConfigTargetKind, String)> {
4370    let mut parts = path.split('/');
4371    let scope = parts.next().unwrap_or_default();
4372    let id = parts.next().unwrap_or_default();
4373    let leaf = parts.next().unwrap_or_default();
4374
4375    if scope.is_empty() || id.is_empty() || leaf.is_empty() || parts.next().is_some() {
4376        return Err(CuError::from(format!(
4377            "Invalid instance override path '{}'. Expected 'tasks/<id>/config', 'resources/<id>/config', or 'bridges/<id>/config'.",
4378            path
4379        )));
4380    }
4381
4382    if leaf != "config" {
4383        return Err(CuError::from(format!(
4384            "Invalid instance override path '{}'. Only the '/config' leaf is supported.",
4385            path
4386        )));
4387    }
4388
4389    let kind = match scope {
4390        "tasks" => InstanceConfigTargetKind::Task,
4391        "resources" => InstanceConfigTargetKind::Resource,
4392        "bridges" => InstanceConfigTargetKind::Bridge,
4393        _ => {
4394            return Err(CuError::from(format!(
4395                "Invalid instance override path '{}'. Supported roots are 'tasks', 'resources', and 'bridges'.",
4396                path
4397            )));
4398        }
4399    };
4400
4401    Ok((kind, id.to_string()))
4402}
4403
4404#[cfg(feature = "std")]
4405fn apply_instance_config_set_operation(
4406    config: &mut CuConfig,
4407    operation: &InstanceConfigSetOperation,
4408) -> CuResult<()> {
4409    let (target_kind, target_id) = parse_instance_override_target(&operation.path)?;
4410
4411    match target_kind {
4412        InstanceConfigTargetKind::Task => {
4413            let matches = match &mut config.graphs {
4414                ConfigGraphs::Simple(graph) => {
4415                    apply_task_config_override_to_graph(graph, &target_id, &operation.value)
4416                }
4417                ConfigGraphs::Missions(graphs) => graphs
4418                    .values_mut()
4419                    .map(|graph| {
4420                        apply_task_config_override_to_graph(graph, &target_id, &operation.value)
4421                    })
4422                    .sum(),
4423            };
4424
4425            if matches == 0 {
4426                return Err(CuError::from(format!(
4427                    "Instance override path '{}' targets unknown task '{}'.",
4428                    operation.path, target_id
4429                )));
4430            }
4431        }
4432        InstanceConfigTargetKind::Resource => {
4433            let mut matches = 0usize;
4434            for resource in &mut config.resources {
4435                if resource.id == target_id {
4436                    merge_component_config(&mut resource.config, &operation.value);
4437                    matches += 1;
4438                }
4439            }
4440            if matches == 0 {
4441                return Err(CuError::from(format!(
4442                    "Instance override path '{}' targets unknown resource '{}'.",
4443                    operation.path, target_id
4444                )));
4445            }
4446        }
4447        InstanceConfigTargetKind::Bridge => {
4448            let mut matches = 0usize;
4449            for bridge in &mut config.bridges {
4450                if bridge.id == target_id {
4451                    merge_component_config(&mut bridge.config, &operation.value);
4452                    matches += 1;
4453                }
4454            }
4455            if matches == 0 {
4456                return Err(CuError::from(format!(
4457                    "Instance override path '{}' targets unknown bridge '{}'.",
4458                    operation.path, target_id
4459                )));
4460            }
4461
4462            match &mut config.graphs {
4463                ConfigGraphs::Simple(graph) => {
4464                    apply_bridge_node_config_override_to_graph(graph, &target_id, &operation.value);
4465                }
4466                ConfigGraphs::Missions(graphs) => {
4467                    for graph in graphs.values_mut() {
4468                        apply_bridge_node_config_override_to_graph(
4469                            graph,
4470                            &target_id,
4471                            &operation.value,
4472                        );
4473                    }
4474                }
4475            }
4476        }
4477    }
4478
4479    Ok(())
4480}
4481
4482#[cfg(feature = "std")]
4483fn apply_instance_overrides(
4484    config: &mut CuConfig,
4485    overrides: &InstanceConfigOverridesRepresentation,
4486) -> CuResult<()> {
4487    for operation in &overrides.set {
4488        apply_instance_config_set_operation(config, operation)?;
4489    }
4490    Ok(())
4491}
4492
4493#[cfg(feature = "std")]
4494fn apply_instance_overrides_from_file(
4495    config: &mut CuConfig,
4496    override_path: &std::path::Path,
4497) -> CuResult<()> {
4498    let override_content = read_to_string(override_path).map_err(|e| {
4499        CuError::from(format!(
4500            "Failed to read instance override file '{}'",
4501            override_path.display()
4502        ))
4503        .add_cause(e.to_string().as_str())
4504    })?;
4505    let overrides = parse_instance_config_overrides_string(&override_content).map_err(|e| {
4506        CuError::from(format!(
4507            "Failed to parse instance override file '{}': {e}",
4508            override_path.display()
4509        ))
4510    })?;
4511    apply_instance_overrides(config, &overrides)
4512}
4513
4514#[cfg(feature = "std")]
4515#[allow(dead_code)]
4516fn parse_multi_config_string(content: &str) -> CuResult<MultiCopperConfigRepresentation> {
4517    Options::default()
4518        .with_default_extension(Extensions::IMPLICIT_SOME)
4519        .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4520        .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4521        .from_str(content)
4522        .map_err(|e| {
4523            CuError::from(format!(
4524                "Failed to parse multi-Copper configuration: Error: {} at position {}",
4525                e.code, e.span
4526            ))
4527        })
4528}
4529
4530#[cfg(feature = "std")]
4531#[allow(dead_code)]
4532fn resolve_relative_config_path(base_path: Option<&str>, referenced_path: &str) -> String {
4533    if referenced_path.starts_with('/') || base_path.is_none() {
4534        return referenced_path.to_string();
4535    }
4536
4537    let current_dir = std::path::Path::new(base_path.expect("checked above"))
4538        .parent()
4539        .unwrap_or_else(|| std::path::Path::new(""))
4540        .to_path_buf();
4541    current_dir
4542        .join(referenced_path)
4543        .to_string_lossy()
4544        .to_string()
4545}
4546
4547#[cfg(feature = "std")]
4548#[allow(dead_code)]
4549fn parse_multi_endpoint(endpoint: &str) -> CuResult<MultiCopperEndpoint> {
4550    let mut parts = endpoint.split('/');
4551    let subsystem_id = parts.next().unwrap_or_default();
4552    let bridge_id = parts.next().unwrap_or_default();
4553    let channel_id = parts.next().unwrap_or_default();
4554
4555    if subsystem_id.is_empty()
4556        || bridge_id.is_empty()
4557        || channel_id.is_empty()
4558        || parts.next().is_some()
4559    {
4560        return Err(CuError::from(format!(
4561            "Invalid multi-Copper endpoint '{endpoint}'. Expected 'subsystem/bridge/channel'."
4562        )));
4563    }
4564
4565    Ok(MultiCopperEndpoint {
4566        subsystem_id: subsystem_id.to_string(),
4567        bridge_id: bridge_id.to_string(),
4568        channel_id: channel_id.to_string(),
4569    })
4570}
4571
4572#[cfg(feature = "std")]
4573#[allow(dead_code)]
4574fn multi_channel_key(bridge_id: &str, channel_id: &str) -> String {
4575    format!("{bridge_id}/{channel_id}")
4576}
4577
4578#[cfg(feature = "std")]
4579#[allow(dead_code)]
4580fn register_multi_channel_msg(
4581    contracts: &mut HashMap<String, MultiCopperChannelContract>,
4582    bridge_id: &str,
4583    channel_id: &str,
4584    expected_direction: MultiCopperChannelDirection,
4585    msg: &str,
4586) -> CuResult<()> {
4587    let key = multi_channel_key(bridge_id, channel_id);
4588    let contract = contracts.get_mut(&key).ok_or_else(|| {
4589        CuError::from(format!(
4590            "Bridge channel '{bridge_id}/{channel_id}' is referenced by the graph but not declared in the bridge config."
4591        ))
4592    })?;
4593
4594    if contract.direction != expected_direction {
4595        let expected = match expected_direction {
4596            MultiCopperChannelDirection::Rx => "Rx",
4597            MultiCopperChannelDirection::Tx => "Tx",
4598        };
4599        return Err(CuError::from(format!(
4600            "Bridge channel '{bridge_id}/{channel_id}' is used as {expected} in the graph but declared with the opposite direction."
4601        )));
4602    }
4603
4604    match &contract.msg {
4605        Some(existing) if existing != msg => Err(CuError::from(format!(
4606            "Bridge channel '{bridge_id}/{channel_id}' carries inconsistent message types '{existing}' and '{msg}'."
4607        ))),
4608        Some(_) => Ok(()),
4609        None => {
4610            contract.msg = Some(msg.to_string());
4611            Ok(())
4612        }
4613    }
4614}
4615
4616#[cfg(feature = "std")]
4617#[allow(dead_code)]
4618fn build_multi_bridge_channel_contracts(
4619    config: &CuConfig,
4620) -> CuResult<HashMap<String, MultiCopperChannelContract>> {
4621    let graph = config
4622        .graphs
4623        .get_graph(Some(DEFAULT_MISSION_ID))
4624        .map_err(|e| {
4625            CuError::from(format!(
4626                "Multi-Copper subsystem configs with missions must define a '{DEFAULT_MISSION_ID}' mission: {e}"
4627            ))
4628        })?;
4629
4630    let mut contracts = HashMap::new();
4631    for bridge in &config.bridges {
4632        for channel in &bridge.channels {
4633            let (channel_id, direction) = match channel {
4634                BridgeChannelConfigRepresentation::Rx { id, .. } => {
4635                    (id.as_str(), MultiCopperChannelDirection::Rx)
4636                }
4637                BridgeChannelConfigRepresentation::Tx { id, .. } => {
4638                    (id.as_str(), MultiCopperChannelDirection::Tx)
4639                }
4640            };
4641
4642            let key = multi_channel_key(&bridge.id, channel_id);
4643            if contracts.contains_key(&key) {
4644                return Err(CuError::from(format!(
4645                    "Duplicate bridge channel declaration for '{key}'."
4646                )));
4647            }
4648
4649            contracts.insert(
4650                key,
4651                MultiCopperChannelContract {
4652                    bridge_type: bridge.type_.clone(),
4653                    direction,
4654                    msg: None,
4655                },
4656            );
4657        }
4658    }
4659
4660    for edge in graph.edges() {
4661        if let Some(channel_id) = &edge.src_channel {
4662            register_multi_channel_msg(
4663                &mut contracts,
4664                &edge.src,
4665                channel_id,
4666                MultiCopperChannelDirection::Rx,
4667                &edge.msg,
4668            )?;
4669        }
4670        if let Some(channel_id) = &edge.dst_channel {
4671            register_multi_channel_msg(
4672                &mut contracts,
4673                &edge.dst,
4674                channel_id,
4675                MultiCopperChannelDirection::Tx,
4676                &edge.msg,
4677            )?;
4678        }
4679    }
4680
4681    Ok(contracts)
4682}
4683
4684#[cfg(feature = "std")]
4685#[allow(dead_code)]
4686fn validate_multi_config_representation(
4687    representation: MultiCopperConfigRepresentation,
4688    file_path: Option<&str>,
4689    active_features: &[&str],
4690) -> CuResult<MultiCopperConfig> {
4691    if representation
4692        .instance_overrides_root
4693        .as_ref()
4694        .is_some_and(|root| root.trim().is_empty())
4695    {
4696        return Err(CuError::from(
4697            "Multi-Copper instance_overrides_root must not be empty.",
4698        ));
4699    }
4700
4701    if representation.subsystems.is_empty() {
4702        return Err(CuError::from(
4703            "Multi-Copper config must declare at least one subsystem.",
4704        ));
4705    }
4706    if representation.subsystems.len() > usize::from(u16::MAX) + 1 {
4707        return Err(CuError::from(
4708            "Multi-Copper config supports at most 65536 distinct subsystem ids.",
4709        ));
4710    }
4711
4712    let mut seen_subsystems = std::collections::HashSet::new();
4713    for subsystem in &representation.subsystems {
4714        if subsystem.id.trim().is_empty() {
4715            return Err(CuError::from(
4716                "Multi-Copper subsystem ids must not be empty.",
4717            ));
4718        }
4719        if !seen_subsystems.insert(subsystem.id.clone()) {
4720            return Err(CuError::from(format!(
4721                "Duplicate multi-Copper subsystem id '{}'.",
4722                subsystem.id
4723            )));
4724        }
4725    }
4726
4727    let mut sorted_ids: Vec<_> = representation
4728        .subsystems
4729        .iter()
4730        .map(|subsystem| subsystem.id.clone())
4731        .collect();
4732    sorted_ids.sort();
4733    let subsystem_code_map: HashMap<_, _> = sorted_ids
4734        .into_iter()
4735        .enumerate()
4736        .map(|(idx, id)| {
4737            (
4738                id,
4739                u16::try_from(idx).expect("subsystem count was validated against u16 range"),
4740            )
4741        })
4742        .collect();
4743
4744    let mut subsystem_contracts: HashMap<String, HashMap<String, MultiCopperChannelContract>> =
4745        HashMap::new();
4746    let mut subsystems = Vec::with_capacity(representation.subsystems.len());
4747
4748    for subsystem in representation.subsystems {
4749        let resolved_config_path = resolve_relative_config_path(file_path, &subsystem.config);
4750        let config = read_configuration_with_features(&resolved_config_path, active_features)
4751            .map_err(|e| {
4752                CuError::from(format!(
4753                    "Failed to read subsystem '{}' from '{}': {e}",
4754                    subsystem.id, resolved_config_path
4755                ))
4756            })?;
4757        let contracts = build_multi_bridge_channel_contracts(&config).map_err(|e| {
4758            CuError::from(format!(
4759                "Invalid subsystem '{}' for multi-Copper validation: {e}",
4760                subsystem.id
4761            ))
4762        })?;
4763        subsystem_contracts.insert(subsystem.id.clone(), contracts);
4764        subsystems.push(MultiCopperSubsystem {
4765            subsystem_code: *subsystem_code_map
4766                .get(&subsystem.id)
4767                .expect("subsystem code map must contain every subsystem"),
4768            id: subsystem.id,
4769            config_path: resolved_config_path,
4770            config,
4771        });
4772    }
4773
4774    let mut interconnects = Vec::with_capacity(representation.interconnects.len());
4775    for interconnect in representation.interconnects {
4776        if interconnect
4777            .when
4778            .as_ref()
4779            .is_some_and(|predicate| !predicate.evaluate(active_features))
4780        {
4781            continue;
4782        }
4783
4784        let from = parse_multi_endpoint(&interconnect.from).map_err(|e| {
4785            CuError::from(format!(
4786                "Invalid multi-Copper interconnect source '{}': {e}",
4787                interconnect.from
4788            ))
4789        })?;
4790        let to = parse_multi_endpoint(&interconnect.to).map_err(|e| {
4791            CuError::from(format!(
4792                "Invalid multi-Copper interconnect destination '{}': {e}",
4793                interconnect.to
4794            ))
4795        })?;
4796
4797        let from_contracts = subsystem_contracts.get(&from.subsystem_id).ok_or_else(|| {
4798            CuError::from(format!(
4799                "Interconnect source '{}' references unknown subsystem '{}'.",
4800                from, from.subsystem_id
4801            ))
4802        })?;
4803        let to_contracts = subsystem_contracts.get(&to.subsystem_id).ok_or_else(|| {
4804            CuError::from(format!(
4805                "Interconnect destination '{}' references unknown subsystem '{}'.",
4806                to, to.subsystem_id
4807            ))
4808        })?;
4809
4810        let from_contract = from_contracts
4811            .get(&multi_channel_key(&from.bridge_id, &from.channel_id))
4812            .ok_or_else(|| {
4813                CuError::from(format!(
4814                    "Interconnect source '{}' references unknown bridge channel.",
4815                    from
4816                ))
4817            })?;
4818        let to_contract = to_contracts
4819            .get(&multi_channel_key(&to.bridge_id, &to.channel_id))
4820            .ok_or_else(|| {
4821                CuError::from(format!(
4822                    "Interconnect destination '{}' references unknown bridge channel.",
4823                    to
4824                ))
4825            })?;
4826
4827        if from_contract.direction != MultiCopperChannelDirection::Tx {
4828            return Err(CuError::from(format!(
4829                "Interconnect source '{}' must reference a Tx bridge channel.",
4830                from
4831            )));
4832        }
4833        if to_contract.direction != MultiCopperChannelDirection::Rx {
4834            return Err(CuError::from(format!(
4835                "Interconnect destination '{}' must reference an Rx bridge channel.",
4836                to
4837            )));
4838        }
4839
4840        if from_contract.bridge_type != to_contract.bridge_type {
4841            return Err(CuError::from(format!(
4842                "Interconnect '{}' -> '{}' mixes incompatible bridge types '{}' and '{}'.",
4843                from, to, from_contract.bridge_type, to_contract.bridge_type
4844            )));
4845        }
4846
4847        let from_msg = from_contract.msg.as_ref().ok_or_else(|| {
4848            CuError::from(format!(
4849                "Interconnect source '{}' is not wired inside subsystem '{}', so its message type cannot be inferred.",
4850                from, from.subsystem_id
4851            ))
4852        })?;
4853        let to_msg = to_contract.msg.as_ref().ok_or_else(|| {
4854            CuError::from(format!(
4855                "Interconnect destination '{}' is not wired inside subsystem '{}', so its message type cannot be inferred.",
4856                to, to.subsystem_id
4857            ))
4858        })?;
4859
4860        if from_msg != to_msg {
4861            return Err(CuError::from(format!(
4862                "Interconnect '{}' -> '{}' connects incompatible message types '{}' and '{}'.",
4863                from, to, from_msg, to_msg
4864            )));
4865        }
4866        if interconnect.msg != *from_msg {
4867            return Err(CuError::from(format!(
4868                "Interconnect '{}' -> '{}' declares message type '{}' but subsystem graphs require '{}'.",
4869                from, to, interconnect.msg, from_msg
4870            )));
4871        }
4872
4873        interconnects.push(MultiCopperInterconnect {
4874            from,
4875            to,
4876            msg: interconnect.msg,
4877            bridge_type: from_contract.bridge_type.clone(),
4878        });
4879    }
4880
4881    let instance_overrides_root = representation
4882        .instance_overrides_root
4883        .as_ref()
4884        .map(|root| resolve_relative_config_path(file_path, root));
4885
4886    Ok(MultiCopperConfig {
4887        subsystems,
4888        interconnects,
4889        instance_overrides_root,
4890    })
4891}
4892
4893/// Read a copper configuration from a file.
4894#[cfg(feature = "std")]
4895pub fn read_configuration(config_filename: &str) -> CuResult<CuConfig> {
4896    read_configuration_with_features(config_filename, &[])
4897}
4898
4899/// Read a Copper configuration using the supplied compile-time Cargo features.
4900#[cfg(feature = "std")]
4901pub fn read_configuration_with_features(
4902    config_filename: &str,
4903    active_features: &[&str],
4904) -> CuResult<CuConfig> {
4905    let config_content = read_configuration_content(config_filename)?;
4906    read_configuration_str_with_features(config_content, Some(config_filename), active_features)
4907}
4908
4909#[cfg(feature = "std")]
4910fn read_configuration_content(config_filename: &str) -> CuResult<String> {
4911    read_to_string(config_filename).map_err(|e| {
4912        CuError::from(format!(
4913            "Failed to read configuration file: {:?}",
4914            config_filename
4915        ))
4916        .add_cause(e.to_string().as_str())
4917    })
4918}
4919
4920/// Read a copper configuration from a String.
4921/// Parse a RON string into a CuConfigRepresentation, using the standard options.
4922/// Returns an error if the parsing fails.
4923fn parse_config_string(content: &str) -> CuResult<CuConfigRepresentation> {
4924    Options::default()
4925        .with_default_extension(Extensions::IMPLICIT_SOME)
4926        .with_default_extension(Extensions::UNWRAP_NEWTYPES)
4927        .with_default_extension(Extensions::UNWRAP_VARIANT_NEWTYPES)
4928        .from_str(content)
4929        .map_err(|e| {
4930            CuError::from(format!(
4931                "Failed to parse configuration: Error: {} at position {}",
4932                e.code, e.span
4933            ))
4934        })
4935}
4936
4937/// Convert a CuConfigRepresentation to a CuConfig.
4938/// Uses the deserialize_impl method and validates the logging configuration.
4939fn config_representation_to_config(representation: CuConfigRepresentation) -> CuResult<CuConfig> {
4940    #[allow(unused_mut)]
4941    let mut cuconfig = CuConfig::deserialize_impl(representation)
4942        .map_err(|e| CuError::from(format!("Error deserializing configuration: {e}")))?;
4943
4944    #[cfg(feature = "std")]
4945    cuconfig.ensure_default_background_pool();
4946
4947    cuconfig.validate_logging_config()?;
4948    cuconfig.validate_runtime_config()?;
4949    cuconfig.validate_stateless_configs()?;
4950    cuconfig.validate_background_input_configs()?;
4951    cuconfig.validate_anytime_configs()?;
4952    cuconfig.validate_constants()?;
4953    cuconfig.validate_log_streaming_config()?;
4954
4955    Ok(cuconfig)
4956}
4957
4958#[allow(unused_variables)]
4959fn resolve_configuration_representation(
4960    config_content: &str,
4961    file_path: Option<&str>,
4962    active_features: &[&str],
4963) -> CuResult<CuConfigRepresentation> {
4964    // Parse the configuration string
4965    let representation = parse_config_string(config_content)?;
4966
4967    // Process includes and generate a merged configuration if a file path is provided
4968    // includes are only available with std.
4969    #[cfg(feature = "std")]
4970    let representation = if let Some(path) = file_path {
4971        process_includes(path, representation, &mut Vec::new(), active_features)?
4972    } else {
4973        representation
4974    };
4975
4976    Ok(representation)
4977}
4978
4979/// Read a Copper configuration and return the include-expanded RON used by proc-macro bundling.
4980///
4981/// The RON is serialized from the ordered source representation before it is lowered into
4982/// mission graph hash maps. This keeps task ordering aligned with generated runtime code.
4983#[cfg(feature = "std")]
4984#[doc(hidden)]
4985#[allow(dead_code)]
4986pub fn read_configuration_with_resolved_ron(config_filename: &str) -> CuResult<(CuConfig, String)> {
4987    read_configuration_with_resolved_ron_and_features(config_filename, &[])
4988}
4989
4990/// Read and expand a Copper configuration using the supplied compile-time Cargo features.
4991#[cfg(feature = "std")]
4992#[doc(hidden)]
4993pub fn read_configuration_with_resolved_ron_and_features(
4994    config_filename: &str,
4995    active_features: &[&str],
4996) -> CuResult<(CuConfig, String)> {
4997    let config_content = read_configuration_content(config_filename)?;
4998    let representation = resolve_configuration_representation(
4999        &config_content,
5000        Some(config_filename),
5001        active_features,
5002    )?;
5003    let resolved_ron = CuConfig::get_options()
5004        .to_string_pretty(&representation, ron::ser::PrettyConfig::default())
5005        .map_err(|e| CuError::from(format!("Error serializing configuration: {e}")))?;
5006    let config = config_representation_to_config(representation)?;
5007    Ok((config, resolved_ron))
5008}
5009
5010#[allow(dead_code)]
5011pub fn read_configuration_str(
5012    config_content: String,
5013    file_path: Option<&str>,
5014) -> CuResult<CuConfig> {
5015    read_configuration_str_with_features(config_content, file_path, &[])
5016}
5017
5018/// Read a Copper configuration string using the supplied compile-time Cargo features.
5019pub fn read_configuration_str_with_features(
5020    config_content: String,
5021    file_path: Option<&str>,
5022    active_features: &[&str],
5023) -> CuResult<CuConfig> {
5024    let representation =
5025        resolve_configuration_representation(&config_content, file_path, active_features)?;
5026
5027    // Convert the representation to a CuConfig and validate
5028    config_representation_to_config(representation)
5029}
5030
5031/// Read a strict multi-Copper umbrella configuration from a file.
5032#[cfg(feature = "std")]
5033#[allow(dead_code)]
5034pub fn read_multi_configuration(config_filename: &str) -> CuResult<MultiCopperConfig> {
5035    read_multi_configuration_with_features(config_filename, &[])
5036}
5037
5038/// Read a multi-Copper configuration using the supplied compile-time Cargo features.
5039#[cfg(feature = "std")]
5040#[allow(dead_code)]
5041pub fn read_multi_configuration_with_features(
5042    config_filename: &str,
5043    active_features: &[&str],
5044) -> CuResult<MultiCopperConfig> {
5045    let config_content = read_to_string(config_filename).map_err(|e| {
5046        CuError::from(format!(
5047            "Failed to read multi-Copper configuration file: {:?}",
5048            config_filename
5049        ))
5050        .add_cause(e.to_string().as_str())
5051    })?;
5052    read_multi_configuration_str_with_features(
5053        config_content,
5054        Some(config_filename),
5055        active_features,
5056    )
5057}
5058
5059/// Read a strict multi-Copper umbrella configuration from a string.
5060#[cfg(feature = "std")]
5061#[allow(dead_code)]
5062pub fn read_multi_configuration_str(
5063    config_content: String,
5064    file_path: Option<&str>,
5065) -> CuResult<MultiCopperConfig> {
5066    read_multi_configuration_str_with_features(config_content, file_path, &[])
5067}
5068
5069/// Read a multi-Copper configuration string using the supplied compile-time Cargo features.
5070#[cfg(feature = "std")]
5071#[allow(dead_code)]
5072pub fn read_multi_configuration_str_with_features(
5073    config_content: String,
5074    file_path: Option<&str>,
5075    active_features: &[&str],
5076) -> CuResult<MultiCopperConfig> {
5077    let representation = parse_multi_config_string(&config_content)?;
5078    validate_multi_config_representation(representation, file_path, active_features)
5079}
5080
5081#[cfg(test)]
5082mod tests;