dora-message 1.0.0

`dora` goal is to be a low latency, composable, and distributed data flow.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
use std::collections::BTreeMap;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Additional data that is sent as part of output messages.
///
/// Includes a timestamp and additional user-provided parameters. The payload is
/// a self-describing Arrow IPC stream, so the message carries no separate type
/// descriptor.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Metadata {
    metadata_version: u16,
    timestamp: uhlc::Timestamp,
    pub parameters: MetadataParameters,
}

impl Metadata {
    /// Current metadata wire-format version, stamped on every outgoing message.
    ///
    /// Bumped from 0 to 1 when the `ArrowTypeInfo` sidecar was dropped and the
    /// wire format became Arrow-IPC-only, and from 1 to 2 when the binary
    /// encoding moved from bincode to postcard (varint integers and length
    /// prefixes, so the byte layout differs even though the field order does
    /// not). A receiver can compare
    /// [`metadata_version`](Self::metadata_version) against this to detect a peer
    /// speaking an incompatible format and report it clearly instead of failing
    /// with a cryptic positional-deserialization error.
    pub const CURRENT_VERSION: u16 = 2;

    /// Create metadata with the given timestamp and no user parameters.
    pub fn new(timestamp: uhlc::Timestamp) -> Self {
        Self::from_parameters(timestamp, Default::default())
    }

    /// Metadata for a startup route-probe marker (see [`STARTUP_MARKER_PARAM`]).
    pub fn startup_marker(timestamp: uhlc::Timestamp) -> Self {
        Self::from_parameters(
            timestamp,
            BTreeMap::from([(STARTUP_MARKER_PARAM.to_owned(), Parameter::Bool(true))]),
        )
    }

    /// Whether this message is a startup route-probe marker rather than node
    /// data. Markers are consumed by the receiving node's startup barrier and
    /// must never be decoded or surfaced to user code — see
    /// [`STARTUP_MARKER_PARAM`].
    pub fn is_startup_marker(&self) -> bool {
        get_bool_param(&self.parameters, STARTUP_MARKER_PARAM).unwrap_or(false)
    }

    /// Metadata for a startup route-probe **ack**: the consumer-side reply to a
    /// startup marker, identifying which consumer input received it (see
    /// [`STARTUP_ACK_PARAM`]).
    pub fn startup_ack(timestamp: uhlc::Timestamp, consumer_node: &str, input_id: &str) -> Self {
        Self::from_parameters(
            timestamp,
            BTreeMap::from([
                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true)),
                (
                    STARTUP_ACK_CONSUMER_PARAM.to_owned(),
                    Parameter::String(consumer_node.to_owned()),
                ),
                (
                    STARTUP_ACK_INPUT_PARAM.to_owned(),
                    Parameter::String(input_id.to_owned()),
                ),
            ]),
        )
    }

    /// `Some((consumer_node, input_id))` iff this message is a well-formed
    /// startup route-probe ack — see [`STARTUP_ACK_PARAM`]. Malformed acks
    /// (missing or wrongly-typed identity parameters) return `None` and are
    /// ignored by producers, which keeps the affected output on the reliable
    /// daemon path instead of switching on bad evidence.
    pub fn startup_ack_identity(&self) -> Option<(&str, &str)> {
        if !get_bool_param(&self.parameters, STARTUP_ACK_PARAM).unwrap_or(false) {
            return None;
        }
        let consumer = get_string_param(&self.parameters, STARTUP_ACK_CONSUMER_PARAM)?;
        let input = get_string_param(&self.parameters, STARTUP_ACK_INPUT_PARAM)?;
        Some((consumer, input))
    }

    /// Create metadata with the given timestamp and user parameters, stamping
    /// the current wire-format version ([`CURRENT_VERSION`](Self::CURRENT_VERSION)).
    pub fn from_parameters(timestamp: uhlc::Timestamp, parameters: MetadataParameters) -> Self {
        Self {
            metadata_version: Self::CURRENT_VERSION,
            timestamp,
            parameters,
        }
    }

    /// The wire-format version stamped on this metadata. Compare against
    /// [`CURRENT_VERSION`](Self::CURRENT_VERSION) on receive to reject peers
    /// using an incompatible format.
    pub fn metadata_version(&self) -> u16 {
        self.metadata_version
    }

    /// The hybrid-logical-clock timestamp assigned when this message was sent.
    pub fn timestamp(&self) -> uhlc::Timestamp {
        self.timestamp
    }

    /// The serialized OpenTelemetry propagation context carried in the
    /// `open_telemetry_context` parameter, or an empty string if absent.
    pub fn open_telemetry_context(&self) -> String {
        get_string_param(&self.parameters, OPEN_TELEMETRY_CONTEXT)
            .unwrap_or("")
            .to_string()
    }
}

/// Reserved [`MetadataParameters`] key marking a message as a **startup
/// route-probe marker** rather than node data.
///
/// The zenoh data plane is direct node-to-node pub/sub, so a producer that
/// publishes before a consumer's subscription has propagated would drop those
/// early samples. Rather than infer route-readiness from zenoh declarations,
/// each producer publishes markers on its real output topic while an output is
/// still on the reliable daemon path, and each consumer answers every received
/// marker with an ack (see [`STARTUP_ACK_PARAM`]): an ack arriving back at the
/// producer is end-to-end proof that the route pair carries data. The producer
/// stops marking an output — and switches it to the direct zenoh path — once
/// all its required consumers have acked, but only within a bounded startup
/// window: an output still un-acked when the window closes is pinned to the
/// daemon path for the rest of the run, so a topic's messages never straddle a
/// path switch (dora-rs/dora#2891). A route that never proves itself just keeps
/// the output on the daemon path.
///
/// The `__dora_` prefix is reserved; user parameters must not use it. Receivers
/// filter markers before decoding the payload, so they never reach user code.
pub const STARTUP_MARKER_PARAM: &str = "__dora_startup_marker";

/// Reserved [`MetadataParameters`] key marking a message as a **startup
/// route-probe ack**: the consumer-side half of the startup handshake.
///
/// When a consumer's data subscriber receives a startup marker (see
/// [`STARTUP_MARKER_PARAM`]) for an input, it replies with an ack on the
/// output's dedicated `@ack` topic. An ack arriving back at the producer is
/// end-to-end proof that the route works in *both* directions; once every
/// required consumer of an output has acked, the producer switches that output
/// from the reliable daemon path to the direct node-to-node zenoh path. Ack
/// timing is load-bearing: the switch can only happen inside the producer's
/// bounded startup window (see [`STARTUP_MARKER_PARAM`]), so a consumer that
/// acks late costs that output the fast path for the whole run. A missing or
/// late ack never fails anything — the output simply stays on the daemon path.
///
/// Acks travel as an empty-payload message whose attachment carries this flag
/// plus the acking consumer's identity under [`STARTUP_ACK_CONSUMER_PARAM`] and
/// [`STARTUP_ACK_INPUT_PARAM`] — the identity rides in the attachment rather
/// than the zenoh key so consumer/input ids never need key escaping.
pub const STARTUP_ACK_PARAM: &str = "__dora_startup_ack";

/// Reserved key carrying the acking consumer's node id as a
/// [`Parameter::String`] — see [`STARTUP_ACK_PARAM`].
pub const STARTUP_ACK_CONSUMER_PARAM: &str = "__dora_startup_ack_consumer";

/// Reserved key carrying the acking consumer's input id as a
/// [`Parameter::String`] — see [`STARTUP_ACK_PARAM`].
pub const STARTUP_ACK_INPUT_PARAM: &str = "__dora_startup_ack_input";

/// Additional metadata that can be sent as part of output messages.
pub type MetadataParameters = BTreeMap<String, Parameter>;

/// A typed metadata parameter sent as part of output messages.
///
/// Parameters are stored by key in [`MetadataParameters`]. The `get_*_param`
/// helpers ([`get_string_param`], [`get_integer_param`], [`get_bool_param`])
/// are type-checked: they return the value only when the stored variant matches
/// the requested type, and `None` both for a missing key **and** for a key whose
/// stored value has a different type. Callers therefore never need to match on
/// the variant themselves for the common scalar cases.
///
/// ```
/// use dora_message::metadata::{
///     get_integer_param, get_string_param, MetadataParameters, Parameter,
/// };
///
/// let mut params = MetadataParameters::new();
/// params.insert("frame".to_string(), Parameter::Integer(7));
///
/// // Matching type -> the value.
/// assert_eq!(get_integer_param(&params, "frame"), Some(7));
/// // Wrong requested type -> None (not a panic, not a coercion).
/// assert_eq!(get_string_param(&params, "frame"), None);
/// // Missing key -> None.
/// assert_eq!(get_integer_param(&params, "absent"), None);
/// ```
#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
pub enum Parameter {
    /// A boolean value.
    Bool(bool),
    /// A signed 64-bit integer value.
    Integer(i64),
    /// A UTF-8 string value.
    String(String),
    /// A list of signed 64-bit integers.
    ListInt(Vec<i64>),
    /// A 64-bit floating-point value.
    Float(f64),
    /// A list of 64-bit floating-point values.
    ListFloat(Vec<f64>),
    /// A list of UTF-8 strings.
    ListString(Vec<String>),
    /// A UTC timestamp.
    Timestamp(DateTime<Utc>),
}

/// Extract a string parameter from metadata, returning `None` if missing or
/// not a `Parameter::String`.
pub fn get_string_param<'a>(params: &'a MetadataParameters, key: &str) -> Option<&'a str> {
    params.get(key).and_then(|p| match p {
        Parameter::String(s) => Some(s.as_str()),
        _ => None,
    })
}

/// Extract an integer parameter from metadata, returning `None` if missing or
/// not a `Parameter::Integer`.
pub fn get_integer_param(params: &MetadataParameters, key: &str) -> Option<i64> {
    params.get(key).and_then(|p| match p {
        Parameter::Integer(n) => Some(*n),
        _ => None,
    })
}

/// Extract a bool parameter from metadata, returning `None` if missing or
/// not a `Parameter::Bool`.
pub fn get_bool_param(params: &MetadataParameters, key: &str) -> Option<bool> {
    params.get(key).and_then(|p| match p {
        Parameter::Bool(b) => Some(*b),
        _ => None,
    })
}

// ---------------------------------------------------------------------------
// Well-known metadata parameter keys for service and action patterns
// ---------------------------------------------------------------------------

/// Metadata key for correlating a service request with its response.
pub const REQUEST_ID: &str = "request_id";

/// Metadata key for identifying an action goal across feedback/result messages.
pub const GOAL_ID: &str = "goal_id";

/// Metadata key for the completion status of an action goal.
pub const GOAL_STATUS: &str = "goal_status";

/// Goal completed successfully.
pub const GOAL_STATUS_SUCCEEDED: &str = "succeeded";

/// Goal was aborted by the server.
pub const GOAL_STATUS_ABORTED: &str = "aborted";

/// Goal was canceled by the client.
pub const GOAL_STATUS_CANCELED: &str = "canceled";

// ---------------------------------------------------------------------------
// Well-known metadata parameter key for distributed tracing
// ---------------------------------------------------------------------------

/// Metadata key carrying the serialized OpenTelemetry propagation context, so a
/// trace can be continued across a dora message hop. Read via
/// [`Metadata::open_telemetry_context`]; stamped by the node and runtime send
/// paths when tracing is enabled. Keep this the single source of truth so the
/// write and read sides can never drift.
pub const OPEN_TELEMETRY_CONTEXT: &str = "open_telemetry_context";

// ---------------------------------------------------------------------------
// Well-known metadata parameter keys for the streaming pattern
// ---------------------------------------------------------------------------

/// Metadata key identifying the conversation/session.
pub const SESSION_ID: &str = "session_id";

/// Metadata key for the logical segment within a session (e.g. one utterance).
pub const SEGMENT_ID: &str = "segment_id";

/// Metadata key for chunk sequence number within a segment.
pub const SEQ: &str = "seq";

/// Metadata key marking the last chunk of a segment (`true` on final chunk).
pub const FIN: &str = "fin";

/// Metadata key to discard older queued messages on this input (`true` to flush).
pub const FLUSH: &str = "flush";

/// Metadata key indicating the wire framing of the data payload.
/// When set to `"arrow-ipc"`, the payload is an Arrow IPC stream.
pub const FRAMING: &str = "_framing";

/// Value for [`FRAMING`] indicating Arrow IPC stream framing.
pub const FRAMING_ARROW_IPC: &str = "arrow-ipc";

/// Metadata key carrying the FNV-1a hash (as an `i64`) of the Arrow IPC schema
/// for a zenoh data message. Present on schema-once messages so a receiver can
/// tell which primed decoder a schema-less batch belongs to and detect schema
/// changes. Absent on messages a receiver should decode as a standalone full
/// stream (large/SHM and daemon-path payloads).
pub const SCHEMA_HASH: &str = "_schema_hash";

/// Metadata key carrying the true on-wire byte size (as an `i64`) of a
/// `dora topic` debug frame's data sample.
///
/// The daemon rebuilds a self-describing Arrow IPC stream for inspection by
/// prepending the retained schema block to each schema-once batch, so the
/// rebuilt stream the CLI receives is larger than what actually travelled on
/// the wire (the schema-less batch — the schema is published only once on the
/// `@schema` subtopic). To keep `dora topic info`'s bandwidth accounting
/// accurate, the daemon stamps the original data-sample length here; the CLI
/// measures this instead of the rebuilt stream length (dora-rs/dora#2584).
///
/// Debug/inspection path only — never set on real node→node outputs.
pub const WIRE_SIZE: &str = "_wire_size";

/// Byte size to charge for a `dora topic` debug frame when accounting bandwidth.
///
/// Prefers the daemon-stamped [`WIRE_SIZE`] (the real on-wire data-sample
/// length): the `data` the CLI receives is a rebuilt self-describing stream
/// whose schema was re-prepended for inspection, so for a schema-once output
/// `data.len()` over-reports what actually travelled. Falls back to the buffer
/// length when the key is absent — an older daemon, or a non-debug frame that
/// never carried it (dora-rs/dora#2584). A present-but-out-of-range stamp (a
/// negative `i64` that can't be a byte count) also falls back rather than
/// silently counting zero. Keep this the single reader of [`WIRE_SIZE`] so the
/// daemon stamp and the CLI accounting can never disagree on the fallback rule.
pub fn debug_frame_wire_size(params: &MetadataParameters, data: Option<&[u8]>) -> usize {
    get_integer_param(params, WIRE_SIZE)
        .and_then(|n| usize::try_from(n).ok())
        .or_else(|| data.map(|d| d.len()))
        .unwrap_or(0)
}

/// Returns `true` if the given parameters carry any pattern-correlation key
/// ([`REQUEST_ID`], [`GOAL_ID`], or [`GOAL_STATUS`]).
///
/// Messages marked with these keys belong to a service or action pattern where
/// multiple Arrow schemas can legitimately flow through a single output/input,
/// distinguished by metadata rather than a fixed Arrow type. Runtime type
/// checks skip such messages (dora-rs/adora#150), and the schema-once zenoh
/// optimization excludes them — on the send side (they always travel as full
/// self-describing streams), on the node receive side (their schemas must not
/// churn the per-input decoder), and on the daemon's `dora topic` debug path
/// (same, for its schema cache). Keep this the single definition so those
/// layers can never disagree on what "pattern-correlated" means.
pub fn carries_pattern_correlation(params: &MetadataParameters) -> bool {
    params.contains_key(REQUEST_ID)
        || params.contains_key(GOAL_ID)
        || params.contains_key(GOAL_STATUS)
}

/// Remove internal wire-protocol keys ([`SCHEMA_HASH`], [`FRAMING`]) from a
/// parameter map. Call this at every wire→user boundary: the keys are
/// meaningless after decode, and a stale [`SCHEMA_HASH`] forwarded from an
/// input's metadata into `send_output` parameters (a standard pattern, e.g.
/// replay) would ride onto outputs that don't overwrite it, making receivers
/// hash-mismatch and silently drop them (dora-rs/dora#2366 review).
pub fn strip_internal_parameters(params: &mut MetadataParameters) {
    params.remove(SCHEMA_HASH);
    params.remove(FRAMING);
}

/// FNV-1a-64 hash with a fixed seed (cross-process deterministic). Used to
/// fingerprint an Arrow IPC schema block so a schema-less batch can be matched
/// to the schema it was encoded against (see [`SCHEMA_HASH`]). The producer
/// node, the consumer node, and the daemon's `dora topic` debug path all hash
/// the same schema-block bytes with this function, so the value must stay
/// identical across crates — keep this the single source of truth.
pub fn fnv1a(bytes: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf29ce484222325;
    for b in bytes {
        hash ^= *b as u64;
        hash = hash.wrapping_mul(0x100000001b3);
    }
    hash
}

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

    #[test]
    fn fnv1a_matches_standard_vectors() {
        // Canonical FNV-1a-64 vectors — pin the algorithm so the producer and
        // consumers (across crates/processes) never disagree on a schema hash.
        assert_eq!(fnv1a(b""), 0xcbf29ce484222325);
        assert_eq!(fnv1a(b"a"), 0xaf63dc4c8601ec8c);
    }

    fn test_timestamp() -> uhlc::Timestamp {
        uhlc::HLC::default().new_timestamp()
    }

    #[test]
    fn startup_marker_is_detected_and_survives_the_wire() {
        // A marker is recognized only via the reserved parameter, and the flag
        // must survive postcard round-tripping — it travels as the zenoh
        // attachment, and a receiver that failed to recognize it would decode
        // the marker as node data and surface it to user code.
        let marker = Metadata::startup_marker(test_timestamp());
        assert!(marker.is_startup_marker());

        let bytes = crate::encode(&marker).expect("serialize");
        let decoded: Metadata = crate::decode(&bytes).expect("deserialize");
        assert!(decoded.is_startup_marker());
    }

    #[test]
    fn ordinary_metadata_is_not_a_startup_marker() {
        // Guard the other direction: real data must never be mistaken for a
        // marker (it would be silently dropped instead of delivered).
        assert!(!Metadata::new(test_timestamp()).is_startup_marker());

        // Wrong type under the reserved key must not count as a marker.
        let wrong_type = Metadata::from_parameters(
            test_timestamp(),
            BTreeMap::from([(
                STARTUP_MARKER_PARAM.to_owned(),
                Parameter::String("true".into()),
            )]),
        );
        assert!(!wrong_type.is_startup_marker());

        // Explicit `false` is not a marker either.
        let explicit_false = Metadata::from_parameters(
            test_timestamp(),
            BTreeMap::from([(STARTUP_MARKER_PARAM.to_owned(), Parameter::Bool(false))]),
        );
        assert!(!explicit_false.is_startup_marker());
    }

    #[test]
    fn startup_marker_key_is_reserved_and_distinct() {
        // The `__dora_` prefix keeps it out of the user parameter namespace, and
        // it must not collide with any well-known protocol key.
        assert_eq!(STARTUP_MARKER_PARAM, "__dora_startup_marker");
        assert!(STARTUP_MARKER_PARAM.starts_with("__dora_"));
        for key in [
            REQUEST_ID,
            GOAL_ID,
            GOAL_STATUS,
            SESSION_ID,
            SEGMENT_ID,
            SEQ,
            FIN,
            FLUSH,
        ] {
            assert_ne!(STARTUP_MARKER_PARAM, key);
        }
    }

    #[test]
    fn startup_ack_round_trips_and_extracts_identity() {
        // The ack travels as a postcard attachment on the `@ack` topic; the
        // producer must recover exactly the (consumer, input) identity it needs
        // to tick off a required acker.
        let ack = Metadata::startup_ack(test_timestamp(), "camera-consumer", "image/depth");
        assert_eq!(
            ack.startup_ack_identity(),
            Some(("camera-consumer", "image/depth"))
        );
        // An ack is not a marker (and vice versa, checked below): the two
        // travel on different topics but share the filtering code path.
        assert!(!ack.is_startup_marker());

        let bytes = crate::encode(&ack).expect("serialize");
        let decoded: Metadata = crate::decode(&bytes).expect("deserialize");
        assert_eq!(
            decoded.startup_ack_identity(),
            Some(("camera-consumer", "image/depth"))
        );
    }

    #[test]
    fn malformed_startup_acks_are_rejected() {
        // Ordinary metadata and markers are not acks.
        assert_eq!(Metadata::new(test_timestamp()).startup_ack_identity(), None);
        assert_eq!(
            Metadata::startup_marker(test_timestamp()).startup_ack_identity(),
            None
        );

        // Flag present but identity missing → not a valid ack: a producer must
        // never count an acker it cannot identify.
        let flag_only = Metadata::from_parameters(
            test_timestamp(),
            BTreeMap::from([(STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true))]),
        );
        assert_eq!(flag_only.startup_ack_identity(), None);

        // Wrongly-typed identity parameters are rejected too.
        let wrong_types = Metadata::from_parameters(
            test_timestamp(),
            BTreeMap::from([
                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(true)),
                (STARTUP_ACK_CONSUMER_PARAM.to_owned(), Parameter::Integer(1)),
                (STARTUP_ACK_INPUT_PARAM.to_owned(), Parameter::Integer(2)),
            ]),
        );
        assert_eq!(wrong_types.startup_ack_identity(), None);

        // `Bool(false)` under the flag key is not an ack.
        let explicit_false = Metadata::from_parameters(
            test_timestamp(),
            BTreeMap::from([
                (STARTUP_ACK_PARAM.to_owned(), Parameter::Bool(false)),
                (
                    STARTUP_ACK_CONSUMER_PARAM.to_owned(),
                    Parameter::String("c".into()),
                ),
                (
                    STARTUP_ACK_INPUT_PARAM.to_owned(),
                    Parameter::String("i".into()),
                ),
            ]),
        );
        assert_eq!(explicit_false.startup_ack_identity(), None);
    }

    #[test]
    fn startup_ack_keys_are_reserved_and_distinct() {
        let ack_keys = [
            STARTUP_ACK_PARAM,
            STARTUP_ACK_CONSUMER_PARAM,
            STARTUP_ACK_INPUT_PARAM,
        ];
        for key in ack_keys {
            assert!(key.starts_with("__dora_"));
            assert_ne!(key, STARTUP_MARKER_PARAM);
        }
        for (i, a) in ack_keys.iter().enumerate() {
            for b in &ack_keys[i + 1..] {
                assert_ne!(a, b);
            }
        }
    }

    #[test]
    fn well_known_keys_have_stable_values() {
        // These string values are part of the cross-language protocol
        // (Python nodes use the same literal strings). Do not change.
        assert_eq!(REQUEST_ID, "request_id");
        assert_eq!(GOAL_ID, "goal_id");
        assert_eq!(GOAL_STATUS, "goal_status");
        assert_eq!(GOAL_STATUS_SUCCEEDED, "succeeded");
        assert_eq!(GOAL_STATUS_ABORTED, "aborted");
        assert_eq!(GOAL_STATUS_CANCELED, "canceled");
        assert_eq!(SESSION_ID, "session_id");
        assert_eq!(SEGMENT_ID, "segment_id");
        assert_eq!(SEQ, "seq");
        assert_eq!(FIN, "fin");
        assert_eq!(FLUSH, "flush");
    }

    #[test]
    fn well_known_keys_are_distinct() {
        let keys = [
            REQUEST_ID,
            GOAL_ID,
            GOAL_STATUS,
            SESSION_ID,
            SEGMENT_ID,
            SEQ,
            FIN,
            FLUSH,
        ];
        for (i, a) in keys.iter().enumerate() {
            for b in &keys[i + 1..] {
                assert_ne!(a, b);
            }
        }
    }

    #[test]
    fn goal_status_values_are_distinct() {
        let vals = [
            GOAL_STATUS_SUCCEEDED,
            GOAL_STATUS_ABORTED,
            GOAL_STATUS_CANCELED,
        ];
        for (i, a) in vals.iter().enumerate() {
            for b in &vals[i + 1..] {
                assert_ne!(a, b);
            }
        }
    }

    #[test]
    fn outgoing_metadata_is_stamped_with_current_version() {
        // The wire format is positional (postcard) and carries no separate type
        // descriptor, so `metadata_version` is the only in-band signal of an
        // incompatible layout. Every constructor must stamp `CURRENT_VERSION`.
        assert_eq!(Metadata::CURRENT_VERSION, 2);
        let ts = uhlc::HLC::default().new_timestamp();
        assert_eq!(
            Metadata::new(ts).metadata_version(),
            Metadata::CURRENT_VERSION
        );
        assert_eq!(
            Metadata::from_parameters(ts, Default::default()).metadata_version(),
            Metadata::CURRENT_VERSION
        );
    }

    #[test]
    fn get_string_param_extracts_string() {
        let mut params = MetadataParameters::default();
        params.insert("key".to_string(), Parameter::String("value".to_string()));
        assert_eq!(get_string_param(&params, "key"), Some("value"));
        assert_eq!(get_string_param(&params, "missing"), None);
    }

    #[test]
    fn get_string_param_returns_none_for_non_string() {
        let mut params = MetadataParameters::default();
        params.insert("num".to_string(), Parameter::Integer(42));
        assert_eq!(get_string_param(&params, "num"), None);
    }

    #[test]
    fn get_integer_param_extracts_integer() {
        let mut params = MetadataParameters::default();
        params.insert("key".to_string(), Parameter::Integer(42));
        assert_eq!(get_integer_param(&params, "key"), Some(42));
        assert_eq!(get_integer_param(&params, "missing"), None);
    }

    #[test]
    fn get_integer_param_returns_none_for_non_integer() {
        let mut params = MetadataParameters::default();
        params.insert("s".to_string(), Parameter::String("hello".to_string()));
        assert_eq!(get_integer_param(&params, "s"), None);
    }

    #[test]
    fn get_bool_param_extracts_bool() {
        let mut params = MetadataParameters::default();
        params.insert("key".to_string(), Parameter::Bool(true));
        assert_eq!(get_bool_param(&params, "key"), Some(true));
        assert_eq!(get_bool_param(&params, "missing"), None);
    }

    #[test]
    fn get_bool_param_returns_none_for_non_bool() {
        let mut params = MetadataParameters::default();
        params.insert("n".to_string(), Parameter::Integer(1));
        assert_eq!(get_bool_param(&params, "n"), None);
    }

    #[test]
    fn debug_frame_wire_size_prefers_stamped_value() {
        // The stamped size wins over the (larger, schema-inflated) buffer: a
        // schema-once frame's rebuilt `data` is bigger than what travelled.
        let mut params = MetadataParameters::default();
        params.insert(WIRE_SIZE.to_string(), Parameter::Integer(17));
        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 17);
    }

    #[test]
    fn debug_frame_wire_size_falls_back_to_buffer_len() {
        // No stamp (older daemon / non-debug frame) ⇒ use the buffer length.
        let params = MetadataParameters::default();
        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 42);
    }

    #[test]
    fn debug_frame_wire_size_falls_back_on_out_of_range_stamp() {
        // A negative i64 can't be a byte count; fall back rather than count 0.
        let mut params = MetadataParameters::default();
        params.insert(WIRE_SIZE.to_string(), Parameter::Integer(-1));
        assert_eq!(debug_frame_wire_size(&params, Some(&[0u8; 42])), 42);
    }

    #[test]
    fn debug_frame_wire_size_zero_without_stamp_or_buffer() {
        assert_eq!(
            debug_frame_wire_size(&MetadataParameters::default(), None),
            0
        );
    }
}