hyperi-rustlib 2.8.6

There's plenty of sage advice out there about how to run Rust services in production at scale — config cascades, structured logging, masking secrets, multi-backend secrets management, Prometheus, OpenTelemetry, Kafka transports, tiered disk-spillover sinks, adaptive worker pools, graceful shutdown — but almost none of it as code you can just install and use. This is that code. Opinionated, drop-in, working out of the box. The patterns from blog posts, watercooler chats and beers with your Google mates as actual library — not a framework you assemble from twenty crates and 8 weeks of munging.
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
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
// Project:   hyperi-rustlib
// File:      src/transport/codec.rs
// Purpose:   Parse-on-demand WorkBatch codec (native JSON + MsgPack, no bridge)
// Language:  Rust
//
// License:   BUSL-1.1
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! # Parse-on-demand codec (Task 0.3a)
//!
//! The data-plane spine frames bytes into a [`WorkBatch`](super::WorkBatch)
//! WITHOUT parsing (Task 0.2). A transform / router that needs to read a field
//! parses on demand -- and it should not care whether the record arrived as
//! JSON or MsgPack. This module is that parse step.
//!
//! ## Native, no JSON bridge
//!
//! There is no `rmp_serde -> serde_json::Value -> serde_json::to_vec ->
//! sonic_rs` bridge anywhere on the parse path -- that double-parse-and-
//! re-serialise was killed in Phase 0.7c (the engine `parse.rs` MsgPack path
//! now walks `rmpv` straight into a `sonic_rs::Value`). Both this codec and
//! the engine decode natively:
//!
//! - **JSON** is parsed once with [`sonic_rs`] (SIMD, AVX2/NEON).
//! - **MsgPack** is parsed once with [`rmpv`] -- the schema-less `Value` decoder
//!   from the same `3Hren/msgpack-rust` workspace as `rmp-serde`. No
//!   intermediate `serde_json::Value`, no JSON re-serialise.
//!
//! ## Unified routing-field accessor
//!
//! A router keys off ONE field (`_table`, `org_id`, ...) and must not branch on
//! wire format. [`ParsedPayload`] exposes a format-agnostic accessor:
//!
//! - [`ParsedPayload::field_str`] -- the common case: a top-level string field.
//! - [`ParsedPayload::field`] -- a [`FieldRef`] covering the scalar routing
//!   cases (string / int / float / bool / null), with everything else
//!   ([`FieldRef::Other`]) deliberately collapsed because routers do not key
//!   off nested containers.
//!
//! **Scope:** top-level object-key lookup only. No deep JSON-path -- routing
//! keys live at the top level, and a deep-path query is a separate concern that
//! YAGNI keeps out of the hot routing path.
//!
//! See `docs/MIGRATIONS.md` (codec consolidation: native rmpv, JSON bridge
//! removed) and `docs/SELF-REGULATION.md` for where this codec sits in the
//! `WorkBatch` data-plane spine. The block contract is in
//! [`WorkBatch`](crate::transport::WorkBatch).

use super::types::PayloadFormat;
use bytes::Bytes;
use sonic_rs::JsonValueTrait as _;
use thiserror::Error;

/// A parse failure, tagged by the format that failed.
#[derive(Debug, Error)]
pub enum CodecError {
    /// JSON parse failed (sonic_rs SIMD parser).
    #[error("json parse error: {0}")]
    Json(#[from] sonic_rs::Error),

    /// MsgPack parse failed (native rmpv decoder).
    #[error("msgpack parse error: {0}")]
    MsgPack(#[from] rmpv::decode::Error),

    /// MsgPack serialise failed (native rmpv encoder).
    ///
    /// `rmpv::encode::Error` is `rmp::encode::ValueWriteError` -- an I/O write
    /// failure from the underlying writer. Serialising into an in-memory `Vec`
    /// effectively never fails, but the encoder is fallible so we surface it
    /// rather than panic. JSON serialise reuses [`CodecError::Json`]
    /// (`sonic_rs::Error` already covers both parse and serialise).
    #[error("msgpack encode error: {0}")]
    Encode(#[from] rmpv::encode::Error),

    /// Trailing bytes remain after a complete MsgPack value was decoded.
    ///
    /// A single-record payload must encode exactly ONE value with no leftover
    /// bytes. Trailing bytes indicate corruption, a framing error, or two
    /// values concatenated (MsgPack stream framing is a separate, deferred
    /// feature -- it is NOT supported here).
    ///
    /// The `usize` is the number of bytes that remained unconsumed.
    #[error("msgpack trailing bytes: {0} byte(s) remain after value")]
    TrailingBytes(usize),
}

/// A parsed payload, retaining its native value representation.
///
/// JSON stays a [`sonic_rs::Value`] (so the SIMD parse is not thrown away) and
/// MsgPack stays an [`rmpv::Value`] (native, no JSON bridge). A consumer that
/// only needs a routing field should reach for [`ParsedPayload::field_str`] /
/// [`ParsedPayload::field`] rather than matching the variant -- that is the
/// whole point of the unified accessor.
#[derive(Debug, Clone)]
pub enum ParsedPayload {
    /// JSON value parsed by sonic_rs (SIMD).
    Json(sonic_rs::Value),
    /// MsgPack value parsed natively by rmpv (no serde_json bridge).
    MsgPack(rmpv::Value),
}

/// A borrowed view of one routing field, format-agnostic.
///
/// This is the shared currency the unified accessor returns so a router need
/// not know whether the record was JSON or MsgPack. It covers the scalar cases
/// a router actually keys off; nested objects / arrays / binary / ext collapse
/// to [`FieldRef::Other`] because routing never branches on a container.
///
/// `Str` borrows from the parsed value (zero-copy); the numeric / bool variants
/// are `Copy` scalars.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum FieldRef<'a> {
    /// A string field (borrowed from the parsed value).
    Str(&'a str),
    /// An integer field (MsgPack ints and JSON integers fold to `i64`).
    Int(i64),
    /// A floating-point field.
    Float(f64),
    /// A boolean field.
    Bool(bool),
    /// An explicit null / nil field.
    Null,
    /// Present but not a routing scalar (object, array, binary, ext, ...).
    Other,
}

/// Parse a framed payload into a native [`ParsedPayload`].
///
/// - [`PayloadFormat::Json`] -> sonic_rs (SIMD).
/// - [`PayloadFormat::MsgPack`] -> rmpv (native, no JSON bridge).
/// - [`PayloadFormat::Auto`] -> [`PayloadFormat::detect`] then dispatch. An empty
///   blob detects as JSON (matching `detect`'s contract) and surfaces a
///   [`CodecError::Json`] -- empty input is not valid JSON.
///
/// # Errors
///
/// Returns [`CodecError::Json`] or [`CodecError::MsgPack`] when the bytes are
/// malformed for the (detected or declared) format.
pub fn parse(payload: &Bytes, format: PayloadFormat) -> Result<ParsedPayload, CodecError> {
    let effective = match format {
        PayloadFormat::Auto => PayloadFormat::detect(payload),
        other => other,
    };

    match effective {
        // detect() never yields Auto, but treat a residual Auto as JSON.
        PayloadFormat::Json | PayloadFormat::Auto => {
            let value: sonic_rs::Value = sonic_rs::from_slice(payload)?;
            Ok(ParsedPayload::Json(value))
        }
        PayloadFormat::MsgPack => {
            // rmpv::decode::read_value reads from any `io::Read`; a byte slice
            // is one. `&mut &[u8]` advances the cursor as it decodes. This is a
            // SINGLE native decode -- no rmp_serde, no serde_json, no re-encode.
            let mut cursor: &[u8] = payload.as_ref();
            let value = rmpv::decode::read_value(&mut cursor)?;
            // A single-record payload encodes exactly ONE value. Any bytes still
            // in `cursor` after the value was decoded indicate corruption,
            // framing misalignment, or concatenated values. Reject them.
            // MsgPack-stream framing (multiple values per payload) is a separate
            // deferred feature -- do NOT silently accept trailing bytes here.
            let remaining = cursor.len();
            if remaining > 0 {
                return Err(CodecError::TrailingBytes(remaining));
            }
            Ok(ParsedPayload::MsgPack(value))
        }
    }
}

/// Serialise a JSON value to bytes via [`sonic_rs`] (SIMD), no bridge.
///
/// The inverse of the JSON arm of [`parse`]. Reuses sonic_rs end-to-end so a
/// transform that mutates a parsed JSON value re-emits it without ever touching
/// `serde_json`.
///
/// # Errors
///
/// Returns [`CodecError::Json`] if sonic_rs fails to serialise the value.
pub fn to_json_bytes(value: &sonic_rs::Value) -> Result<Bytes, CodecError> {
    let buf = sonic_rs::to_vec(value)?;
    Ok(Bytes::from(buf))
}

/// Serialise a MsgPack value to bytes via NATIVE [`rmpv::encode::write_value`].
///
/// The inverse of the MsgPack arm of [`parse`]. This is the native rmpv encoder
/// -- NOT `rmp_serde`, NOT a JSON bridge. A transform that mutates a parsed
/// `rmpv::Value` re-emits MsgPack wire bytes with a single native encode, no
/// intermediate `serde_json::Value`, no re-parse.
///
/// # Errors
///
/// Returns [`CodecError::Encode`] if the encoder fails to write the value. For
/// an in-memory `Vec` writer this is effectively unreachable, but the encoder
/// is fallible so the error is surfaced rather than unwrapped.
pub fn to_msgpack_bytes(value: &rmpv::Value) -> Result<Bytes, CodecError> {
    // write_value writes into any `io::Write`; a Vec<u8> is one and never
    // returns a short write, so the only failure path is the encoder's own.
    let mut buf: Vec<u8> = Vec::new();
    rmpv::encode::write_value(&mut buf, value)?;
    Ok(Bytes::from(buf))
}

impl ParsedPayload {
    /// Whether the payload was decoded from JSON.
    #[must_use]
    pub fn is_json(&self) -> bool {
        matches!(self, Self::Json(_))
    }

    /// Whether the payload was decoded from MsgPack.
    #[must_use]
    pub fn is_msgpack(&self) -> bool {
        matches!(self, Self::MsgPack(_))
    }

    /// Read a top-level string field, format-agnostic.
    ///
    /// The common routing case: a router keys off one string field and does not
    /// care about wire format. Returns `None` if the value is not a top-level
    /// object, the key is absent, or the field is not a string. Borrows from the
    /// parsed value (zero-copy).
    ///
    /// Top-level lookup only -- see the module docs.
    #[must_use]
    pub fn field_str(&self, name: &str) -> Option<&str> {
        match self {
            Self::Json(v) => v.get(name).and_then(|f| f.as_str()),
            Self::MsgPack(v) => msgpack_field(v, name).and_then(rmpv::Value::as_str),
        }
    }

    /// Read a top-level field as a format-agnostic [`FieldRef`].
    ///
    /// Returns `None` only when the value is not a top-level object or the key
    /// is absent. A present-but-non-scalar field yields [`FieldRef::Other`]
    /// (routers never key off containers). Borrows from the parsed value.
    ///
    /// Top-level lookup only -- see the module docs.
    #[must_use]
    pub fn field(&self, name: &str) -> Option<FieldRef<'_>> {
        match self {
            Self::Json(v) => v.get(name).map(json_field_ref),
            Self::MsgPack(v) => msgpack_field(v, name).map(msgpack_field_ref),
        }
    }

    /// Serialise back to the payload's OWN wire format (Task 0.3b).
    ///
    /// `Json` -> JSON bytes (via [`to_json_bytes`]), `MsgPack` -> MsgPack bytes
    /// (via [`to_msgpack_bytes`]). Same format in, same format out -- no
    /// cross-format conversion, no bridge.
    ///
    /// ## Pass-through contract -- DO NOT round-trip untouched records
    ///
    /// This is the egress face of a *parse-on-demand* spine. The governing
    /// principle is "serde is the enemy / zero re-representation": a record that
    /// a transform did NOT change must re-use its original `Record.payload`
    /// (the `Bytes` it arrived as) directly on egress. `to_bytes` is ONLY for a
    /// record a transform actually mutated.
    ///
    /// Calling `to_bytes` on an unmodified record is a correctness *and*
    /// performance bug: it pays a full parse + re-serialise for nothing AND can
    /// alter the wire bytes (key order, number formatting, whitespace) even
    /// though the logical value is identical. Reuse the original `Bytes`; only
    /// reach for `to_bytes` once the value has been edited.
    ///
    /// There is deliberately NO `to_bytes_as` cross-format egress. JSON and
    /// MsgPack have distinct value models (`sonic_rs::Value` vs `rmpv::Value`)
    /// with no native conversion between them; bridging would mean either a
    /// hand-rolled recursive value walker or a `serde_json` hop -- the exact
    /// double-representation this spine exists to avoid. Cross-format egress, if
    /// a consumer ever needs it, is a separate, explicit concern (YAGNI).
    ///
    /// # Errors
    ///
    /// Returns [`CodecError::Json`] (JSON serialise) or [`CodecError::Encode`]
    /// (MsgPack serialise) on encoder failure.
    pub fn to_bytes(&self) -> Result<Bytes, CodecError> {
        match self {
            Self::Json(v) => to_json_bytes(v),
            Self::MsgPack(v) => to_msgpack_bytes(v),
        }
    }
}

/// Classify a sonic_rs JSON value into a [`FieldRef`] (borrows from `v`).
///
/// Order matters: probe the scalar accessors in turn. `as_i64` is tried before
/// `as_f64` so integers stay [`FieldRef::Int`]; a JSON number with a fractional
/// part falls through to [`FieldRef::Float`].
fn json_field_ref(v: &sonic_rs::Value) -> FieldRef<'_> {
    if let Some(s) = v.as_str() {
        FieldRef::Str(s)
    } else if v.is_null() {
        FieldRef::Null
    } else if let Some(b) = v.as_bool() {
        FieldRef::Bool(b)
    } else if let Some(i) = v.as_i64() {
        FieldRef::Int(i)
    } else if let Some(f) = v.as_f64() {
        FieldRef::Float(f)
    } else {
        FieldRef::Other
    }
}

/// Find a top-level value for `name` in an rmpv MsgPack value.
///
/// Only a [`rmpv::Value::Map`] has named fields. The map is a `Vec<(Value,
/// Value)>`, so this is a linear scan -- routing maps are small (a handful of
/// keys), so a linear scan beats building an index. Only string keys match.
fn msgpack_field<'a>(v: &'a rmpv::Value, name: &str) -> Option<&'a rmpv::Value> {
    match v {
        rmpv::Value::Map(pairs) => pairs
            .iter()
            .find(|(k, _)| k.as_str() == Some(name))
            .map(|(_, val)| val),
        _ => None,
    }
}

/// Classify an rmpv MsgPack value into a [`FieldRef`].
///
/// MsgPack integers split into signed/unsigned at the wire level; both fold to
/// `i64` here when they fit. An unsigned value above `i64::MAX` cannot fit `i64`
/// and is surfaced as [`FieldRef::Float`] via `as_f64` (lossy but it keeps a
/// numeric field numeric for routing) rather than dropped to `Other`.
fn msgpack_field_ref(v: &rmpv::Value) -> FieldRef<'_> {
    match v {
        rmpv::Value::String(s) => s.as_str().map_or(FieldRef::Other, FieldRef::Str),
        rmpv::Value::Nil => FieldRef::Null,
        rmpv::Value::Boolean(b) => FieldRef::Bool(*b),
        rmpv::Value::Integer(_) => v
            .as_i64()
            .map(FieldRef::Int)
            .or_else(|| v.as_f64().map(FieldRef::Float))
            .unwrap_or(FieldRef::Other),
        rmpv::Value::F32(f) => FieldRef::Float(f64::from(*f)),
        rmpv::Value::F64(f) => FieldRef::Float(*f),
        // Map / Array / Binary / Ext: routers do not key off containers.
        _ => FieldRef::Other,
    }
}

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

    // ---- Helpers: build real MsgPack blobs by hand (no serde encode) -------
    //
    // We hand-roll the MsgPack bytes so the test exercises the NATIVE rmpv
    // decoder against the real wire format, not a serde round-trip.

    /// fixstr: 0xa0 | len, then the UTF-8 bytes (len < 32).
    fn fixstr(s: &str) -> Vec<u8> {
        let bytes = s.as_bytes();
        assert!(bytes.len() < 32, "fixstr helper only handles len < 32");
        let len = u8::try_from(bytes.len()).expect("len < 32 fits u8");
        let mut out = vec![0xa0 | len];
        out.extend_from_slice(bytes);
        out
    }

    /// fixmap header: 0x80 | n (n < 16 entries).
    fn fixmap_header(n: u8) -> u8 {
        assert!(n < 16, "fixmap helper only handles < 16 entries");
        0x80 | n
    }

    /// Build a logical record `{"_table": "events", "org_id": 42, "live":
    /// true, "ratio": <f64>, "missing": nil}` as a MsgPack fixmap.
    fn sample_msgpack() -> Bytes {
        let mut buf = vec![fixmap_header(5)];
        // "_table": "events"
        buf.extend(fixstr("_table"));
        buf.extend(fixstr("events"));
        // "org_id": 42  (positive fixint -- the byte is its own value)
        buf.extend(fixstr("org_id"));
        buf.push(42);
        // "live": true (0xc3)
        buf.extend(fixstr("live"));
        buf.push(0xc3);
        // "ratio": 1.5 (float64 0xcb + 8 bytes big-endian)
        buf.extend(fixstr("ratio"));
        buf.push(0xcb);
        buf.extend_from_slice(&1.5f64.to_be_bytes());
        // "missing": nil (0xc0)
        buf.extend(fixstr("missing"));
        buf.push(0xc0);
        Bytes::from(buf)
    }

    /// The same logical record as JSON.
    fn sample_json() -> Bytes {
        Bytes::from_static(
            br#"{"_table":"events","org_id":42,"live":true,"ratio":1.5,"missing":null}"#,
        )
    }

    // ---- parse(): JSON -----------------------------------------------------

    #[test]
    fn parse_json_object() {
        let parsed = parse(&sample_json(), PayloadFormat::Json).unwrap();
        assert!(parsed.is_json());
        assert!(!parsed.is_msgpack());
        assert_eq!(parsed.field_str("_table"), Some("events"));
    }

    #[test]
    fn parse_json_array_is_ok() {
        // A top-level array is valid JSON; it simply has no named fields.
        let parsed = parse(&Bytes::from_static(b"[1,2,3]"), PayloadFormat::Json).unwrap();
        assert!(parsed.is_json());
        assert_eq!(parsed.field_str("anything"), None);
    }

    // ---- parse(): MsgPack (native rmpv, hand-rolled bytes) -----------------

    #[test]
    fn parse_msgpack_map() {
        let parsed = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert!(parsed.is_msgpack());
        assert!(!parsed.is_json());
        assert_eq!(parsed.field_str("_table"), Some("events"));
    }

    #[test]
    fn parse_minimal_fixmap() {
        // {"k": "v"} -- the smallest interesting map.
        let mut buf = vec![fixmap_header(1)];
        buf.extend(fixstr("k"));
        buf.extend(fixstr("v"));
        let parsed = parse(&Bytes::from(buf), PayloadFormat::MsgPack).unwrap();
        assert_eq!(parsed.field_str("k"), Some("v"));
    }

    // ---- Auto detection dispatch ------------------------------------------

    #[test]
    fn parse_auto_dispatches_to_json() {
        let parsed = parse(&sample_json(), PayloadFormat::Auto).unwrap();
        assert!(parsed.is_json(), "object byte '{{' must detect as JSON");
        assert_eq!(parsed.field_str("_table"), Some("events"));
    }

    #[test]
    fn parse_auto_dispatches_to_msgpack() {
        let parsed = parse(&sample_msgpack(), PayloadFormat::Auto).unwrap();
        assert!(
            parsed.is_msgpack(),
            "fixmap byte 0x85 must detect as MsgPack"
        );
        assert_eq!(parsed.field_str("_table"), Some("events"));
    }

    // ---- Unified accessor: SAME field value from BOTH formats --------------

    #[test]
    fn field_str_identical_across_formats() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        // The whole point: same logical record, same routing field, regardless
        // of wire format.
        assert_eq!(j.field_str("_table"), m.field_str("_table"));
        assert_eq!(j.field_str("_table"), Some("events"));
    }

    #[test]
    fn field_str_returns_none_for_non_string() {
        // org_id is an int -- field_str only returns strings.
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field_str("org_id"), None);
        assert_eq!(m.field_str("org_id"), None);
    }

    #[test]
    fn field_str_returns_none_for_missing_key() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field_str("nope"), None);
        assert_eq!(m.field_str("nope"), None);
    }

    #[test]
    fn field_str_value_is_present_via_field_too() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        assert_eq!(j.field("_table"), Some(FieldRef::Str("events")));
    }

    #[test]
    fn field_int_identical_across_formats() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field("org_id"), Some(FieldRef::Int(42)));
        assert_eq!(m.field("org_id"), Some(FieldRef::Int(42)));
    }

    #[test]
    fn field_bool_identical_across_formats() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field("live"), Some(FieldRef::Bool(true)));
        assert_eq!(m.field("live"), Some(FieldRef::Bool(true)));
    }

    #[test]
    fn field_float_identical_across_formats() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field("ratio"), Some(FieldRef::Float(1.5)));
        assert_eq!(m.field("ratio"), Some(FieldRef::Float(1.5)));
    }

    #[test]
    fn field_null_identical_across_formats() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field("missing"), Some(FieldRef::Null));
        assert_eq!(m.field("missing"), Some(FieldRef::Null));
    }

    #[test]
    fn field_missing_key_is_none_for_both() {
        let j = parse(&sample_json(), PayloadFormat::Json).unwrap();
        let m = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert_eq!(j.field("nope"), None);
        assert_eq!(m.field("nope"), None);
    }

    #[test]
    fn field_nested_object_is_other() {
        // A field whose value is a container collapses to Other, not None.
        let j = parse(
            &Bytes::from_static(br#"{"k":{"nested":1}}"#),
            PayloadFormat::Json,
        )
        .unwrap();
        assert_eq!(j.field("k"), Some(FieldRef::Other));
        // ...but it is not a routing scalar, so field_str is None.
        assert_eq!(j.field_str("k"), None);

        // MsgPack: {"k": [1]} -- an array value also collapses to Other.
        // fixmap(1) "k" -> fixarray(1) [positive fixint 1]
        let mut buf = vec![fixmap_header(1)];
        buf.extend(fixstr("k"));
        buf.push(0x91); // fixarray with 1 element
        buf.push(0x01); // positive fixint 1
        let m = parse(&Bytes::from(buf), PayloadFormat::MsgPack).unwrap();
        assert_eq!(m.field("k"), Some(FieldRef::Other));
    }

    #[test]
    fn field_on_non_object_top_level_is_none() {
        // A top-level JSON array has no named fields.
        let j = parse(&Bytes::from_static(b"[1,2,3]"), PayloadFormat::Json).unwrap();
        assert_eq!(j.field("0"), None);

        // A top-level MsgPack array (fixarray) likewise.
        // fixarray(2) [1, 2]
        let m = parse(&Bytes::from(vec![0x92, 0x01, 0x02]), PayloadFormat::MsgPack).unwrap();
        assert_eq!(m.field("0"), None);
    }

    // ---- Error cases -------------------------------------------------------

    #[test]
    fn malformed_json_errors() {
        let err = parse(&Bytes::from_static(b"{not valid json"), PayloadFormat::Json).unwrap_err();
        assert!(matches!(err, CodecError::Json(_)), "got {err:?}");
        assert!(!err.to_string().is_empty());
    }

    #[test]
    fn empty_blob_auto_errors_as_json() {
        // detect() maps empty -> Json; empty is not valid JSON.
        let err = parse(&Bytes::new(), PayloadFormat::Auto).unwrap_err();
        assert!(matches!(err, CodecError::Json(_)), "got {err:?}");
    }

    #[test]
    fn malformed_msgpack_errors() {
        // 0x81 declares a fixmap with one entry but supplies no key/value.
        let err = parse(&Bytes::from_static(&[0x81]), PayloadFormat::MsgPack).unwrap_err();
        assert!(matches!(err, CodecError::MsgPack(_)), "got {err:?}");
        assert!(!err.to_string().is_empty());
    }

    #[test]
    fn msgpack_truncated_float_errors() {
        // 0xcb declares a float64 but supplies only 3 of the 8 payload bytes.
        let mut buf = vec![fixmap_header(1)];
        buf.extend(fixstr("ratio"));
        buf.push(0xcb);
        buf.extend_from_slice(&[0x00, 0x01, 0x02]); // short
        let err = parse(&Bytes::from(buf), PayloadFormat::MsgPack).unwrap_err();
        assert!(matches!(err, CodecError::MsgPack(_)), "got {err:?}");
    }

    // ---- Task 0.3b: serialise-out (round-trips through real bytes) ---------
    //
    // The contract is "parse -> (mutate) -> serialise -> parse again preserves
    // the logical value". We assert on re-parsed VALUES (via the unified
    // accessor), NOT on raw bytes: re-serialise may reorder keys or reformat
    // numbers, so a byte-for-byte equality assertion would be wrong.

    /// Compare every routing field of the canonical sample across two payloads.
    fn assert_sample_fields_eq(a: &ParsedPayload, b: &ParsedPayload) {
        assert_eq!(a.field("_table"), b.field("_table"));
        assert_eq!(a.field("org_id"), b.field("org_id"));
        assert_eq!(a.field("live"), b.field("live"));
        assert_eq!(a.field("ratio"), b.field("ratio"));
        assert_eq!(a.field("missing"), b.field("missing"));
    }

    #[test]
    fn json_to_bytes_round_trips() {
        // parse JSON -> to_bytes -> parse again -> values equal.
        let original = parse(&sample_json(), PayloadFormat::Json).unwrap();
        assert!(original.is_json());

        let bytes = original.to_bytes().unwrap();
        assert!(!bytes.is_empty());

        let reparsed = parse(&bytes, PayloadFormat::Json).unwrap();
        assert!(reparsed.is_json(), "JSON must round-trip as JSON");
        assert_sample_fields_eq(&original, &reparsed);
    }

    #[test]
    fn msgpack_to_bytes_round_trips_via_native_bytes() {
        // parse MsgPack (hand-rolled bytes) -> to_bytes (native rmpv encode)
        // -> parse again -> values equal. No serde, no JSON bridge anywhere.
        let original = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        assert!(original.is_msgpack());

        let bytes = original.to_bytes().unwrap();
        assert!(!bytes.is_empty());
        // First byte must be a MsgPack map marker (fixmap 0x80..=0x8f for 5
        // entries -> 0x85), proving native MsgPack came out, not JSON.
        assert_eq!(bytes[0], fixmap_header(5), "expected fixmap(5) wire marker");

        let reparsed = parse(&bytes, PayloadFormat::MsgPack).unwrap();
        assert!(reparsed.is_msgpack(), "MsgPack must round-trip as MsgPack");
        assert_sample_fields_eq(&original, &reparsed);
    }

    #[test]
    fn to_json_bytes_reparses_to_same_value() {
        // Free function: serialise a sonic_rs::Value, re-parse, compare.
        let ParsedPayload::Json(value) = parse(&sample_json(), PayloadFormat::Json).unwrap() else {
            panic!("expected JSON");
        };
        let bytes = to_json_bytes(&value).unwrap();
        let reparsed = parse(&bytes, PayloadFormat::Json).unwrap();
        assert_eq!(reparsed.field("_table"), Some(FieldRef::Str("events")));
        assert_eq!(reparsed.field("org_id"), Some(FieldRef::Int(42)));
        assert_eq!(reparsed.field("ratio"), Some(FieldRef::Float(1.5)));
    }

    #[test]
    fn to_msgpack_bytes_reparses_to_same_value() {
        // Free function: serialise an rmpv::Value via native write_value,
        // re-parse, compare.
        let ParsedPayload::MsgPack(value) =
            parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap()
        else {
            panic!("expected MsgPack");
        };
        let bytes = to_msgpack_bytes(&value).unwrap();
        let reparsed = parse(&bytes, PayloadFormat::MsgPack).unwrap();
        assert_eq!(reparsed.field("_table"), Some(FieldRef::Str("events")));
        assert_eq!(reparsed.field("org_id"), Some(FieldRef::Int(42)));
        assert_eq!(reparsed.field("live"), Some(FieldRef::Bool(true)));
        assert_eq!(reparsed.field("ratio"), Some(FieldRef::Float(1.5)));
        assert_eq!(reparsed.field("missing"), Some(FieldRef::Null));
    }

    #[test]
    fn to_bytes_preserves_a_mutated_json_field() {
        // The realistic case: a transform CHANGED a field, then re-serialises.
        // Only then is to_bytes the right tool (unmodified records pass through
        // the original Bytes -- see the to_bytes doc).
        let ParsedPayload::Json(mut value) = parse(&sample_json(), PayloadFormat::Json).unwrap()
        else {
            panic!("expected JSON");
        };
        // Mutate _table in place via sonic_rs's object insert (overwrites the
        // existing key) -- the value model is what gets re-serialised.
        value.insert("_table", sonic_rs::Value::from("audit"));
        let bytes = to_json_bytes(&value).unwrap();
        let reparsed = parse(&bytes, PayloadFormat::Json).unwrap();
        assert_eq!(reparsed.field_str("_table"), Some("audit"));
        // Untouched siblings survive the round-trip.
        assert_eq!(reparsed.field("org_id"), Some(FieldRef::Int(42)));
    }

    #[test]
    fn json_to_bytes_handles_top_level_array() {
        // Egress is not object-only: a top-level array must round-trip too.
        let parsed = parse(&Bytes::from_static(b"[1,2,3]"), PayloadFormat::Json).unwrap();
        let bytes = parsed.to_bytes().unwrap();
        let reparsed = parse(&bytes, PayloadFormat::Json).unwrap();
        assert!(reparsed.is_json());
        // No named fields either way; the value re-parses without error.
        assert_eq!(reparsed.field_str("anything"), None);
    }

    #[test]
    fn msgpack_to_bytes_handles_top_level_scalar() {
        // A bare MsgPack integer (positive fixint 42) round-trips as itself,
        // not wrapped in a map.
        let parsed = parse(&Bytes::from(vec![42u8]), PayloadFormat::MsgPack).unwrap();
        let bytes = parsed.to_bytes().unwrap();
        assert_eq!(
            bytes.as_ref(),
            &[42u8],
            "fixint must re-emit byte-identical"
        );
        let reparsed = parse(&bytes, PayloadFormat::MsgPack).unwrap();
        assert!(reparsed.is_msgpack());
    }

    #[test]
    fn double_round_trip_is_stable() {
        // parse -> to_bytes -> parse -> to_bytes: the SECOND serialise must
        // equal the first (the value model is the fixed point, even if it
        // differs from the original hand-rolled bytes).
        let first = parse(&sample_msgpack(), PayloadFormat::MsgPack).unwrap();
        let b1 = first.to_bytes().unwrap();
        let second = parse(&b1, PayloadFormat::MsgPack).unwrap();
        let b2 = second.to_bytes().unwrap();
        assert_eq!(b1, b2, "re-serialising a re-parsed value must be stable");
    }

    // ---- Phase 5: trailing-bytes hardening ------------------------------------

    #[test]
    fn msgpack_rejects_trailing_bytes() {
        // A valid fixmap {"k": "v"} followed by a stray nil byte (0xc0).
        // Before the fix this returns Ok (silently ignoring 0xc0).
        // After the fix it must return CodecError::TrailingBytes(1).
        let mut buf = vec![fixmap_header(1)];
        buf.extend(fixstr("k"));
        buf.extend(fixstr("v"));
        buf.push(0xc0); // stray nil -- trailing garbage
        let err = parse(&Bytes::from(buf), PayloadFormat::MsgPack)
            .expect_err("trailing byte must be rejected");
        match err {
            CodecError::TrailingBytes(n) => assert_eq!(n, 1, "expected 1 trailing byte"),
            other => panic!("expected TrailingBytes(1), got {other:?}"),
        }
    }

    #[test]
    fn msgpack_rejects_concatenated_values() {
        // Two valid MsgPack values back-to-back: fixint 1 then fixint 2.
        // parse() decodes ONE value; the second byte is trailing and must error.
        let buf = vec![0x01u8, 0x02u8]; // positive fixint 1, positive fixint 2
        let err = parse(&Bytes::from(buf), PayloadFormat::MsgPack)
            .expect_err("concatenated values must be rejected");
        match err {
            CodecError::TrailingBytes(n) => assert_eq!(n, 1, "expected 1 trailing byte"),
            other => panic!("expected TrailingBytes(1), got {other:?}"),
        }
    }

    #[test]
    fn msgpack_clean_single_value_still_parses_ok() {
        // A single valid fixmap with no trailing bytes -- must still parse Ok.
        // Regression guard: the fix must not break the happy path.
        let mut buf = vec![fixmap_header(1)];
        buf.extend(fixstr("k"));
        buf.extend(fixstr("v"));
        let parsed = parse(&Bytes::from(buf), PayloadFormat::MsgPack).unwrap();
        assert_eq!(parsed.field_str("k"), Some("v"));
    }

    #[test]
    fn json_rejects_trailing_garbage() {
        // The MsgPack path rejects trailing bytes (CodecError::TrailingBytes).
        // The JSON path delegates to sonic_rs, which rejects trailing
        // NON-whitespace content after a complete value (serde_json semantics).
        // A valid object followed by stray garbage must error.
        let mut buf = br#"{"_table":"events"}"#.to_vec();
        buf.extend_from_slice(b"garbage");
        let err = parse(&Bytes::from(buf), PayloadFormat::Json)
            .expect_err("trailing non-whitespace garbage must be rejected");
        assert!(
            matches!(err, CodecError::Json(_)),
            "expected CodecError::Json for trailing garbage, got {err:?}"
        );
    }

    #[test]
    fn json_accepts_trailing_whitespace() {
        // sonic_rs (like serde_json) tolerates trailing whitespace after a
        // complete value -- a pretty-printer's trailing newline must not be a
        // parse error. Asserted alongside the trailing-garbage rejection so the
        // JSON trailing-byte contract is pinned both ways.
        let mut buf = br#"{"_table":"events"}"#.to_vec();
        buf.extend_from_slice(b" \t\r\n");
        let parsed = parse(&Bytes::from(buf), PayloadFormat::Json)
            .expect("trailing whitespace must be accepted");
        assert_eq!(parsed.field_str("_table"), Some("events"));
    }

    #[test]
    fn json_parsed_as_msgpack_errors() {
        // Force the wrong decoder: JSON bytes through the MsgPack path. '{' is
        // 0x7b, which rmpv reads as a positive fixint -- a single-byte value
        // -- leaving the remaining 69 bytes of the JSON payload as trailing
        // bytes. After Phase 5 hardening this MUST error with TrailingBytes.
        let err = parse(&sample_json(), PayloadFormat::MsgPack)
            .expect_err("JSON fed to MsgPack path must error after trailing-bytes hardening");
        assert!(
            matches!(err, CodecError::TrailingBytes(_)),
            "expected TrailingBytes, got {err:?}"
        );
    }
}