clawft-kernel 0.8.3

WeftOS kernel: boot sequence, process table, service registry, and health subsystem
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
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
//! Noise Protocol encryption for mesh transport (K6.1).
//!
//! Wraps raw [`MeshStream`] with Noise XX or IK handshake encryption.
//! The actual `snow` integration is planned for a future milestone;
//! this module provides the trait contract and a passthrough
//! implementation for testing.

use async_trait::async_trait;
use serde::{Deserialize, Serialize};

use crate::mesh::{MeshError, MeshStream};

/// Encrypted channel wrapping a raw mesh stream.
///
/// After a Noise handshake completes, all traffic flows through
/// this interface. Each `send_encrypted` / `recv_encrypted` call
/// handles framing, encryption, and authentication transparently.
///
/// # Cancel-safety (ADR-010 / WEFT-18)
///
/// `recv_encrypted` is raced in `tokio::select!` (mesh peer loop). It
/// must be cancel-safe: implementors delegate to [`MeshStream::recv`],
/// which itself must retain partial frame progress across cancellation.
#[async_trait]
pub trait EncryptedChannel: Send + Sync + 'static {
    /// Send encrypted data (plaintext in, ciphertext on the wire).
    async fn send_encrypted(&mut self, plaintext: &[u8]) -> Result<(), MeshError>;

    /// Receive and decrypt data (ciphertext from wire, plaintext out).
    ///
    /// Cancel-safe: may be dropped while pending inside `tokio::select!`.
    async fn recv_encrypted(&mut self) -> Result<Vec<u8>, MeshError>;

    /// Get the remote peer's static public key (after handshake).
    fn remote_static_key(&self) -> Option<&[u8]>;

    /// The Noise handshake hash of this session, when the channel has
    /// one. Both sides of a session compute the same value; admission
    /// signs it to bind a node identity to this exact session (P3-K1).
    /// `None` for channels without a handshake (plaintext passthrough).
    fn handshake_hash(&self) -> Option<&[u8]> {
        None
    }

    /// Close the encrypted channel gracefully.
    async fn close(&mut self) -> Result<(), MeshError>;
}

/// Noise handshake pattern.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum NoisePattern {
    /// XX pattern: mutual authentication, first contact.
    XX,
    /// IK pattern: known responder key, 1-RTT.
    IK,
}

/// Established encrypted peer session metadata (WEFT-115).
///
/// Holds the identity and crypto session facts after Noise completes.
/// The live byte channel is [`EncryptedChannel`] / [`NoiseChannel`];
/// this struct is the serialisable session descriptor used by pools,
/// diagnostics, and protocol documentation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EncryptedPeer {
    /// Remote node identity.
    pub node_id: String,
    /// Remote static public key (Noise), hex or raw-length bytes as vec.
    pub remote_static_key: Vec<u8>,
    /// Handshake pattern used to establish the session.
    pub pattern: NoisePattern,
    /// Whether a post-quantum KEM upgrade was negotiated.
    pub kem_upgraded: bool,
    /// Wall-clock microseconds when the session was established.
    pub established_at_unix_us: u64,
    /// Peer socket address string (if known).
    pub address: Option<String>,
}

impl EncryptedPeer {
    /// Build session metadata from a completed channel + identity.
    pub fn new(
        node_id: impl Into<String>,
        remote_static_key: Vec<u8>,
        pattern: NoisePattern,
        established_at_unix_us: u64,
    ) -> Self {
        Self {
            node_id: node_id.into(),
            remote_static_key,
            pattern,
            kem_upgraded: false,
            established_at_unix_us,
            address: None,
        }
    }
}

/// Configuration for a Noise handshake.
#[derive(Debug, Clone)]
pub struct NoiseConfig {
    /// Which handshake pattern to use.
    pub pattern: NoisePattern,

    /// Local Ed25519 private key bytes (32 bytes).
    pub local_private_key: [u8; 32],

    /// Remote static public key (required for IK, optional for XX).
    pub remote_static_key: Option<[u8; 32]>,
}

// ── Snow-backed Noise Protocol implementation ───────────────────

/// Real Noise Protocol encrypted channel using `snow`.
///
/// Performs a Noise XX handshake over the underlying `MeshStream`,
/// then encrypts/decrypts all subsequent messages using the established
/// session keys. Uses `Noise_XX_25519_ChaChaPoly_SHA256`.
pub struct NoiseChannel {
    stream: Box<dyn MeshStream>,
    transport: snow::TransportState,
    remote_static: Option<Vec<u8>>,
    handshake_hash: Vec<u8>,
}

impl NoiseChannel {
    /// The session's Noise handshake hash (32 bytes for SHA-256). Equal
    /// on both ends of one session, different on every other session.
    pub fn handshake_hash(&self) -> &[u8] {
        &self.handshake_hash
    }

    /// Perform a Noise XX handshake as the initiator and return an encrypted channel.
    pub async fn initiate(
        mut stream: Box<dyn MeshStream>,
        config: &NoiseConfig,
    ) -> Result<Self, MeshError> {
        let builder = snow::Builder::new(
            "Noise_XX_25519_ChaChaPoly_SHA256"
                .parse()
                .map_err(|e| MeshError::Handshake(format!("bad noise params: {e}")))?,
        )
        .local_private_key(&config.local_private_key)
        .map_err(|e| MeshError::Handshake(format!("bad private key: {e}")))?;

        let mut handshake = builder
            .build_initiator()
            .map_err(|e| MeshError::Handshake(format!("build initiator: {e}")))?;

        let mut buf = vec![0u8; 65535];

        // XX pattern: initiator sends → responder sends → initiator sends

        // -> e
        let len = handshake
            .write_message(&[], &mut buf)
            .map_err(|e| MeshError::Handshake(format!("write msg 1: {e}")))?;
        stream.send(&buf[..len]).await?;

        // <- e, ee, s, es
        let msg = stream.recv().await?;
        handshake
            .read_message(&msg, &mut buf)
            .map_err(|e| MeshError::Handshake(format!("read msg 2: {e}")))?;

        // -> s, se
        let len = handshake
            .write_message(&[], &mut buf)
            .map_err(|e| MeshError::Handshake(format!("write msg 3: {e}")))?;
        stream.send(&buf[..len]).await?;

        let remote_static = handshake.get_remote_static().map(|k| k.to_vec());
        let handshake_hash = handshake.get_handshake_hash().to_vec();

        let transport = handshake
            .into_transport_mode()
            .map_err(|e| MeshError::Handshake(format!("transport mode: {e}")))?;

        tracing::info!("noise XX handshake complete (initiator)");

        Ok(Self {
            stream,
            transport,
            remote_static,
            handshake_hash,
        })
    }

    /// Perform a Noise XX handshake as the responder and return an encrypted channel.
    pub async fn respond(
        mut stream: Box<dyn MeshStream>,
        config: &NoiseConfig,
    ) -> Result<Self, MeshError> {
        let builder = snow::Builder::new(
            "Noise_XX_25519_ChaChaPoly_SHA256"
                .parse()
                .map_err(|e| MeshError::Handshake(format!("bad noise params: {e}")))?,
        )
        .local_private_key(&config.local_private_key)
        .map_err(|e| MeshError::Handshake(format!("bad private key: {e}")))?;

        let mut handshake = builder
            .build_responder()
            .map_err(|e| MeshError::Handshake(format!("build responder: {e}")))?;

        let mut buf = vec![0u8; 65535];

        // XX pattern: initiator sends → responder sends → initiator sends

        // <- e
        let msg = stream.recv().await?;
        handshake
            .read_message(&msg, &mut buf)
            .map_err(|e| MeshError::Handshake(format!("read msg 1: {e}")))?;

        // -> e, ee, s, es
        let len = handshake
            .write_message(&[], &mut buf)
            .map_err(|e| MeshError::Handshake(format!("write msg 2: {e}")))?;
        stream.send(&buf[..len]).await?;

        // <- s, se
        let msg = stream.recv().await?;
        handshake
            .read_message(&msg, &mut buf)
            .map_err(|e| MeshError::Handshake(format!("read msg 3: {e}")))?;

        let remote_static = handshake.get_remote_static().map(|k| k.to_vec());
        let handshake_hash = handshake.get_handshake_hash().to_vec();

        let transport = handshake
            .into_transport_mode()
            .map_err(|e| MeshError::Handshake(format!("transport mode: {e}")))?;

        tracing::info!("noise XX handshake complete (responder)");

        Ok(Self {
            stream,
            transport,
            remote_static,
            handshake_hash,
        })
    }
}

#[async_trait]
impl EncryptedChannel for NoiseChannel {
    async fn send_encrypted(&mut self, plaintext: &[u8]) -> Result<(), MeshError> {
        let mut buf = vec![0u8; plaintext.len() + 16 + 2]; // AEAD tag + length prefix room
        let len = self
            .transport
            .write_message(plaintext, &mut buf)
            .map_err(|e| MeshError::Io(format!("encrypt: {e}")))?;
        self.stream.send(&buf[..len]).await
    }

    async fn recv_encrypted(&mut self) -> Result<Vec<u8>, MeshError> {
        let ciphertext = self.stream.recv().await?;
        let mut buf = vec![0u8; ciphertext.len()];
        let len = self
            .transport
            .read_message(&ciphertext, &mut buf)
            .map_err(|e| MeshError::Io(format!("decrypt: {e}")))?;
        buf.truncate(len);
        Ok(buf)
    }

    fn remote_static_key(&self) -> Option<&[u8]> {
        self.remote_static.as_deref()
    }

    fn handshake_hash(&self) -> Option<&[u8]> {
        Some(&self.handshake_hash)
    }

    async fn close(&mut self) -> Result<(), MeshError> {
        self.stream.close().await
    }
}

// ── Passthrough (test/dev) implementation ────────────────────────

/// Passthrough "encryption" for testing and development.
///
/// No actual crypto — plaintext passes through unchanged.
/// Use `NoiseChannel` for production deployments.
pub struct PassthroughChannel {
    stream: Box<dyn MeshStream>,
}

impl PassthroughChannel {
    /// Wrap a raw stream with passthrough (no-op) encryption.
    pub fn new(stream: Box<dyn MeshStream>) -> Self {
        Self { stream }
    }
}

#[async_trait]
impl EncryptedChannel for PassthroughChannel {
    async fn send_encrypted(&mut self, plaintext: &[u8]) -> Result<(), MeshError> {
        self.stream.send(plaintext).await
    }

    async fn recv_encrypted(&mut self) -> Result<Vec<u8>, MeshError> {
        self.stream.recv().await
    }

    fn remote_static_key(&self) -> Option<&[u8]> {
        None
    }

    async fn close(&mut self) -> Result<(), MeshError> {
        self.stream.close().await
    }
}

/// The Noise static public key for a static private key (Curve25519),
/// i.e. what the remote side sees as this node's remote static. Needed to
/// fill `noise_static_pub` in an admission hello on the initiator side,
/// where the handshake does not expose the local static.
pub fn noise_static_public(private: &[u8; 32]) -> Option<[u8; 32]> {
    use snow::resolvers::{CryptoResolver, DefaultResolver};
    let mut dh = DefaultResolver.resolve_dh(&snow::params::DHChoice::Curve25519)?;
    dh.set(private);
    dh.pubkey().try_into().ok()
}

/// Create an encrypted channel for a mesh connection.
///
/// When `noise_config` is `Some`, performs a real Noise XX handshake.
/// When `None`, returns a passthrough (plaintext) channel.
pub async fn create_encrypted_channel(
    stream: Box<dyn MeshStream>,
    noise_config: Option<&NoiseConfig>,
    is_initiator: bool,
) -> Result<Box<dyn EncryptedChannel>, MeshError> {
    match noise_config {
        Some(config) => {
            let channel = if is_initiator {
                NoiseChannel::initiate(stream, config).await?
            } else {
                NoiseChannel::respond(stream, config).await?
            };
            Ok(Box::new(channel))
        }
        None => Ok(Box::new(PassthroughChannel::new(stream))),
    }
}

// ── Key rotation protocol ───────────────────────────────────────

/// Key rotation state for a mesh connection.
#[derive(Debug, Clone)]
pub struct KeyRotationState {
    /// Current key generation number.
    pub generation: u64,
    /// When the current key was established.
    pub established_at: std::time::Instant,
    /// Maximum key lifetime before rotation.
    pub max_lifetime: std::time::Duration,
    /// Grace period: old key accepted during transition.
    pub grace_period: std::time::Duration,
    /// Whether rotation is in progress.
    pub rotating: bool,
}

impl KeyRotationState {
    /// Create a new key rotation state at generation 0.
    pub fn new(max_lifetime: std::time::Duration, grace_period: std::time::Duration) -> Self {
        Self {
            generation: 0,
            established_at: std::time::Instant::now(),
            max_lifetime,
            grace_period,
            rotating: false,
        }
    }

    /// Check if key rotation is needed based on elapsed lifetime.
    pub fn needs_rotation(&self) -> bool {
        // `>=` so a zero lifetime (and the exact-deadline instant) needs
        // rotation — `>` left a sub-tick race where a freshly established
        // zero-lifetime key read as not-yet-due under parallel load.
        self.established_at.elapsed() >= self.max_lifetime
    }

    /// Start key rotation (marks as in-progress).
    pub fn begin_rotation(&mut self) {
        self.rotating = true;
    }

    /// Complete key rotation, advancing to the next generation.
    pub fn complete_rotation(&mut self) {
        self.generation += 1;
        self.established_at = std::time::Instant::now();
        self.rotating = false;
    }

    /// Check if the old key (previous generation) is still valid.
    /// True during active rotation or within the grace period.
    pub fn old_key_valid(&self) -> bool {
        self.rotating || self.established_at.elapsed() < self.grace_period
    }
}

// ── Post-quantum KEM types and hybrid handshake protocol ──────────

/// Post-quantum Key Encapsulation Mechanism (KEM) configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KemConfig {
    /// Whether this node supports ML-KEM-768.
    pub kem_supported: bool,
    /// KEM public key (1184 bytes for ML-KEM-768).
    pub kem_public_key: Option<Vec<u8>>,
}

/// Result of KEM capability negotiation between two peers.
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum KemNegotiationResult {
    /// Both sides support KEM -- full PQ protection.
    BothSupported,
    /// One side lacks KEM -- graceful fallback to classical.
    GracefulDegradation,
    /// Neither supports KEM -- classical only.
    ClassicalOnly,
}

/// Negotiate KEM capability between two peers.
pub fn negotiate_kem(local: &KemConfig, remote: &KemConfig) -> KemNegotiationResult {
    match (local.kem_supported, remote.kem_supported) {
        (true, true) => KemNegotiationResult::BothSupported,
        (true, false) | (false, true) => KemNegotiationResult::GracefulDegradation,
        (false, false) => KemNegotiationResult::ClassicalOnly,
    }
}

/// Result of a hybrid key exchange (classical Noise XX + optional ML-KEM-768).
#[derive(Debug, Clone)]
pub struct HybridKeyExchange {
    /// Classical shared secret (from Noise XX X25519 DH).
    pub classical_secret: Vec<u8>,
    /// Post-quantum shared secret (from ML-KEM-768, if both sides support).
    pub pq_secret: Option<Vec<u8>>,
    /// Final combined key: SHA-256(classical || pq || "weftos-hybrid-kem-v1").
    pub final_key: Vec<u8>,
    /// Whether PQ was used.
    pub pq_active: bool,
}

impl HybridKeyExchange {
    /// Combine classical and PQ secrets into a final hybrid key.
    ///
    /// The combination is: `SHA-256(classical || pq || "weftos-hybrid-kem-v1")`.
    /// When `pq` is `None`, the domain tag still differentiates the output
    /// from a raw classical key.
    pub fn combine(classical: Vec<u8>, pq: Option<Vec<u8>>) -> Self {
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(&classical);
        if let Some(ref pq_s) = pq {
            hasher.update(pq_s);
        }
        hasher.update(b"weftos-hybrid-kem-v1");
        let final_key = hasher.finalize().to_vec();

        Self {
            pq_active: pq.is_some(),
            classical_secret: classical,
            pq_secret: pq,
            final_key,
        }
    }
}

/// KEM upgrade protocol: runs after Noise XX handshake completes,
/// before sync streams are opened.
///
/// When both peers support ML-KEM-768, the upgrade must complete
/// before any sync stream traffic begins, protecting against
/// store-now-decrypt-later attacks.
pub struct KemUpgradeProtocol {
    /// Local KEM config.
    pub local: KemConfig,
    /// Remote KEM config (from WeftHandshake).
    pub remote: KemConfig,
    /// Negotiation result.
    pub result: KemNegotiationResult,
}

impl KemUpgradeProtocol {
    /// Create a new KEM upgrade protocol from local and remote configs.
    pub fn new(local: KemConfig, remote: KemConfig) -> Self {
        let result = negotiate_kem(&local, &remote);
        Self {
            local,
            remote,
            result,
        }
    }

    /// Whether sync streams should wait for KEM to complete.
    ///
    /// Returns `true` when both sides support KEM, meaning the
    /// PQ key exchange MUST finish before any sync traffic flows.
    pub fn requires_upgrade(&self) -> bool {
        self.result == KemNegotiationResult::BothSupported
    }

    /// Execute the KEM exchange, combining the classical Noise secret
    /// with the post-quantum shared secret (when both sides support it).
    ///
    /// In the current placeholder implementation, the PQ secret is
    /// derived deterministically from the remote KEM public key.
    /// A real implementation would use ML-KEM-768 encapsulate/decapsulate.
    pub fn execute(&self, classical_secret: Vec<u8>) -> HybridKeyExchange {
        match self.result {
            KemNegotiationResult::BothSupported => {
                // In real impl: ML-KEM-768 encapsulate/decapsulate.
                // Placeholder: derive PQ secret from remote public key.
                let pq_secret = {
                    use sha2::{Digest, Sha256};
                    let mut h = Sha256::new();
                    h.update(b"ml-kem-768-simulated");
                    if let Some(ref pk) = self.remote.kem_public_key {
                        h.update(pk);
                    }
                    h.finalize().to_vec()
                };
                HybridKeyExchange::combine(classical_secret, Some(pq_secret))
            }
            _ => HybridKeyExchange::combine(classical_secret, None),
        }
    }
}

// ── In-memory mock stream for testing ───────────────────────────

#[cfg(test)]
mod mock {
    use std::collections::VecDeque;
    use std::net::SocketAddr;
    use std::sync::{Arc, Mutex};

    use async_trait::async_trait;

    use crate::mesh::{MeshError, MeshStream};

    /// Shared buffer backing a pair of mock streams.
    #[derive(Debug, Default)]
    pub struct MockBuffer {
        pub data: VecDeque<Vec<u8>>,
    }

    /// An in-memory mock [`MeshStream`] for unit tests.
    pub struct MockStream {
        /// Outbound buffer (what this side sends).
        pub tx: Arc<Mutex<MockBuffer>>,
        /// Inbound buffer (what this side receives).
        pub rx: Arc<Mutex<MockBuffer>>,
        pub closed: bool,
    }

    impl MockStream {
        /// Create a connected pair of mock streams.
        pub fn pair() -> (Self, Self) {
            let a_to_b = Arc::new(Mutex::new(MockBuffer::default()));
            let b_to_a = Arc::new(Mutex::new(MockBuffer::default()));

            let a = MockStream {
                tx: Arc::clone(&a_to_b),
                rx: Arc::clone(&b_to_a),
                closed: false,
            };
            let b = MockStream {
                tx: b_to_a,
                rx: a_to_b,
                closed: false,
            };
            (a, b)
        }
    }

    #[async_trait]
    impl MeshStream for MockStream {
        async fn send(&mut self, data: &[u8]) -> Result<(), MeshError> {
            if self.closed {
                return Err(MeshError::ConnectionClosed);
            }
            self.tx.lock().unwrap().data.push_back(data.to_vec());
            Ok(())
        }

        async fn recv(&mut self) -> Result<Vec<u8>, MeshError> {
            if self.closed {
                return Err(MeshError::ConnectionClosed);
            }
            self.rx
                .lock()
                .unwrap()
                .data
                .pop_front()
                .ok_or(MeshError::ConnectionClosed)
        }

        async fn close(&mut self) -> Result<(), MeshError> {
            self.closed = true;
            Ok(())
        }

        fn remote_addr(&self) -> Option<SocketAddr> {
            Some("127.0.0.1:0".parse().unwrap())
        }
    }
}

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

    #[test]
    fn noise_pattern_variants() {
        assert_eq!(NoisePattern::XX, NoisePattern::XX);
        assert_ne!(NoisePattern::XX, NoisePattern::IK);
    }

    #[test]
    fn noise_pattern_serde_roundtrip() {
        for pattern in [NoisePattern::XX, NoisePattern::IK] {
            let json = serde_json::to_string(&pattern).unwrap();
            let restored: NoisePattern = serde_json::from_str(&json).unwrap();
            assert_eq!(restored, pattern);
        }
    }

    #[test]
    fn encrypted_peer_serde_roundtrip() {
        let peer = EncryptedPeer {
            node_id: "node-a".into(),
            remote_static_key: vec![1, 2, 3, 4],
            pattern: NoisePattern::XX,
            kem_upgraded: true,
            established_at_unix_us: 1_700_000_000_000_000,
            address: Some("127.0.0.1:9489".into()),
        };
        let json = serde_json::to_string(&peer).unwrap();
        let restored: EncryptedPeer = serde_json::from_str(&json).unwrap();
        assert_eq!(restored.node_id, "node-a");
        assert_eq!(restored.remote_static_key, vec![1, 2, 3, 4]);
        assert_eq!(restored.pattern, NoisePattern::XX);
        assert!(restored.kem_upgraded);
        assert_eq!(restored.established_at_unix_us, 1_700_000_000_000_000);
        assert_eq!(restored.address.as_deref(), Some("127.0.0.1:9489"));
    }

    #[test]
    fn noise_config_fields() {
        let cfg = NoiseConfig {
            pattern: NoisePattern::XX,
            local_private_key: [0x42; 32],
            remote_static_key: None,
        };
        assert_eq!(cfg.pattern, NoisePattern::XX);
        assert_eq!(cfg.local_private_key, [0x42; 32]);
        assert!(cfg.remote_static_key.is_none());

        let cfg_ik = NoiseConfig {
            pattern: NoisePattern::IK,
            local_private_key: [1; 32],
            remote_static_key: Some([2; 32]),
        };
        assert_eq!(cfg_ik.pattern, NoisePattern::IK);
        assert_eq!(cfg_ik.remote_static_key, Some([2; 32]));
    }

    #[tokio::test]
    async fn passthrough_send_recv() {
        let (a, b) = MockStream::pair();
        let mut chan_a = PassthroughChannel::new(Box::new(a));
        let mut chan_b = PassthroughChannel::new(Box::new(b));

        chan_a.send_encrypted(b"hello mesh").await.unwrap();
        let received = chan_b.recv_encrypted().await.unwrap();
        assert_eq!(received, b"hello mesh");
    }

    #[tokio::test]
    async fn passthrough_bidirectional() {
        let (a, b) = MockStream::pair();
        let mut chan_a = PassthroughChannel::new(Box::new(a));
        let mut chan_b = PassthroughChannel::new(Box::new(b));

        chan_a.send_encrypted(b"ping").await.unwrap();
        let msg = chan_b.recv_encrypted().await.unwrap();
        assert_eq!(msg, b"ping");

        chan_b.send_encrypted(b"pong").await.unwrap();
        let msg = chan_a.recv_encrypted().await.unwrap();
        assert_eq!(msg, b"pong");
    }

    #[test]
    fn passthrough_no_remote_key() {
        let (a, _b) = MockStream::pair();
        let chan = PassthroughChannel::new(Box::new(a));
        assert!(chan.remote_static_key().is_none());
    }

    // ── Key rotation tests ────────────────────────────────────────

    #[test]
    fn key_rotation_initial_state() {
        let state = KeyRotationState::new(
            std::time::Duration::from_secs(3600),
            std::time::Duration::from_secs(60),
        );
        assert_eq!(state.generation, 0);
        assert!(!state.rotating);
        assert!(!state.needs_rotation());
    }

    #[test]
    fn key_rotation_needs_rotation_after_lifetime() {
        let mut state = KeyRotationState::new(
            std::time::Duration::from_secs(0), // immediate
            std::time::Duration::from_secs(60),
        );
        // With zero lifetime, needs rotation immediately
        assert!(state.needs_rotation());
        // But old key is still valid within grace period
        assert!(state.old_key_valid());

        state.begin_rotation();
        assert!(state.rotating);
        assert!(state.old_key_valid()); // always valid during rotation
    }

    #[test]
    fn key_rotation_complete_lifecycle() {
        let mut state = KeyRotationState::new(
            std::time::Duration::from_secs(0),
            std::time::Duration::from_secs(3600), // long grace
        );
        assert_eq!(state.generation, 0);

        state.begin_rotation();
        assert!(state.rotating);

        state.complete_rotation();
        assert_eq!(state.generation, 1);
        assert!(!state.rotating);
        // After completion, old key valid within grace period
        assert!(state.old_key_valid());
    }

    #[test]
    fn key_rotation_multiple_generations() {
        let mut state = KeyRotationState::new(
            std::time::Duration::from_secs(0),
            std::time::Duration::from_secs(3600),
        );
        for i in 0..5 {
            assert_eq!(state.generation, i);
            state.begin_rotation();
            state.complete_rotation();
        }
        assert_eq!(state.generation, 5);
    }

    #[tokio::test]
    async fn passthrough_close() {
        let (a, _b) = MockStream::pair();
        let mut chan = PassthroughChannel::new(Box::new(a));
        chan.close().await.unwrap();
        // After close, send should fail
        let result = chan.send_encrypted(b"after close").await;
        assert!(result.is_err());
    }

    // ── KEM negotiation and hybrid handshake tests ───────────────

    #[test]
    fn kem_negotiation_both_supported() {
        let local = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xAA; 1184]),
        };
        let remote = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xBB; 1184]),
        };
        assert_eq!(
            negotiate_kem(&local, &remote),
            KemNegotiationResult::BothSupported,
        );
    }

    #[test]
    fn kem_negotiation_graceful_degradation() {
        // Local supports, remote does not.
        let local = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xAA; 1184]),
        };
        let remote = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        assert_eq!(
            negotiate_kem(&local, &remote),
            KemNegotiationResult::GracefulDegradation,
        );

        // Remote supports, local does not.
        assert_eq!(
            negotiate_kem(&remote, &local),
            KemNegotiationResult::GracefulDegradation,
        );
    }

    #[test]
    fn kem_negotiation_classical_only() {
        let local = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        let remote = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        assert_eq!(
            negotiate_kem(&local, &remote),
            KemNegotiationResult::ClassicalOnly,
        );
    }

    #[test]
    fn hybrid_key_exchange_with_pq() {
        let classical = vec![0x01; 32];
        let pq = vec![0x02; 32];
        let hke = HybridKeyExchange::combine(classical.clone(), Some(pq.clone()));

        assert!(hke.pq_active, "PQ should be active");
        assert_eq!(hke.classical_secret, classical);
        assert_eq!(hke.pq_secret.as_ref().unwrap(), &pq);
        assert_eq!(hke.final_key.len(), 32, "SHA-256 output is 32 bytes");

        // Final key should differ from classical-only derivation.
        let classical_only = HybridKeyExchange::combine(classical.clone(), None);
        assert_ne!(
            hke.final_key, classical_only.final_key,
            "hybrid key must differ from classical-only key"
        );
    }

    #[test]
    fn hybrid_key_exchange_classical_only() {
        let classical = vec![0x01; 32];
        let hke = HybridKeyExchange::combine(classical.clone(), None);

        assert!(!hke.pq_active, "PQ should not be active");
        assert!(hke.pq_secret.is_none());
        assert_eq!(hke.final_key.len(), 32);
    }

    #[test]
    fn kem_upgrade_requires_upgrade_when_both_support() {
        let local = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xAA; 1184]),
        };
        let remote = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xBB; 1184]),
        };
        let protocol = KemUpgradeProtocol::new(local, remote);
        assert!(
            protocol.requires_upgrade(),
            "must require upgrade when both sides support KEM"
        );
        assert_eq!(protocol.result, KemNegotiationResult::BothSupported);
    }

    #[test]
    fn kem_upgrade_no_upgrade_when_degraded() {
        let local = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xAA; 1184]),
        };
        let remote = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        let protocol = KemUpgradeProtocol::new(local, remote);
        assert!(
            !protocol.requires_upgrade(),
            "should not require upgrade with graceful degradation"
        );
    }

    #[test]
    fn kem_upgrade_before_sync_streams() {
        // Both sides support KEM: requires_upgrade is true and execute
        // produces a hybrid key with PQ active.
        let local = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xAA; 1184]),
        };
        let remote = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0xBB; 1184]),
        };
        let protocol = KemUpgradeProtocol::new(local, remote);
        assert!(protocol.requires_upgrade());

        let classical_secret = vec![0xCC; 32];
        let hke = protocol.execute(classical_secret.clone());

        assert!(hke.pq_active, "PQ must be active after KEM upgrade");
        assert!(hke.pq_secret.is_some(), "PQ secret must be present");
        assert_eq!(hke.final_key.len(), 32);

        // Classical-only fallback produces different key.
        let classical_only = HybridKeyExchange::combine(classical_secret, None);
        assert_ne!(
            hke.final_key, classical_only.final_key,
            "hybrid final key must differ from classical-only"
        );
    }

    #[test]
    fn kem_upgrade_classical_fallback() {
        // When degraded, execute produces classical-only key.
        let local = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        let remote = KemConfig {
            kem_supported: false,
            kem_public_key: None,
        };
        let protocol = KemUpgradeProtocol::new(local, remote);
        assert!(!protocol.requires_upgrade());

        let classical_secret = vec![0xCC; 32];
        let hke = protocol.execute(classical_secret);

        assert!(!hke.pq_active);
        assert!(hke.pq_secret.is_none());
    }

    #[test]
    fn kem_config_serde_roundtrip() {
        let cfg = KemConfig {
            kem_supported: true,
            kem_public_key: Some(vec![0x42; 1184]),
        };
        let json = serde_json::to_string(&cfg).unwrap();
        let restored: KemConfig = serde_json::from_str(&json).unwrap();
        assert!(restored.kem_supported);
        assert_eq!(restored.kem_public_key.unwrap().len(), 1184);
    }

    #[test]
    fn kem_negotiation_result_serde_roundtrip() {
        for result in [
            KemNegotiationResult::BothSupported,
            KemNegotiationResult::GracefulDegradation,
            KemNegotiationResult::ClassicalOnly,
        ] {
            let json = serde_json::to_string(&result).unwrap();
            let restored: KemNegotiationResult = serde_json::from_str(&json).unwrap();
            assert_eq!(restored, result);
        }
    }
}