protocol-droid 0.5.1

A Model Context Protocol toolkit that speaks every published spec revision, with one canonical IR behind many per-revision codecs.
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
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
//! The `Mcp-Session-Id` map: the server's half of the session lifecycle.
//!
//! # What this owns
//!
//! Three transitions, and the specification names all three:
//!
//! - `initialize` **inserts**. The server mints an id, hands it back in a
//!   response header, and every later request quotes it.
//! - every later request **looks up**.
//! - `DELETE` **removes**, and the specification says the server "MUST NOT
//!   respond to requests containing that session ID" afterwards.
//!
//! Between those, a lookup miss is [`None`](SessionStore::lookup), and the
//! router turns that into `404`. That is deliberately a different outcome from
//! a header that will not parse: `McpSessionId::from_str` fails *before* the
//! store is consulted, so a malformed id is `400` and never becomes a `404`.
//! The two are not interchangeable — the specification's own recovery is
//! "on 404, start a new session", and a client whose header is malformed has
//! a different problem to fix.
//!
//! # Why the lock never escapes
//!
//! Every method takes the lock, does one map operation, and returns something
//! that is *not* a guard. Lookups hand back an [`Arc`] cloned out of the map,
//! which is what makes it structurally impossible for a caller to hold the
//! lock across an `.await`: the guard is a local, dropped at the end of the
//! function, and the returned value has no lifetime tied to `self`. A handler
//! that receives an `Arc<S>` can hold it for as long as it likes, including
//! across suspension points, and the store stays available to every other
//! request. Returning `&S` instead would push the same constraint on every
//! caller and be wrong on exactly the one that matters.
//!
//! [`concurrent_lookups_never_block_one_another`](self::tests::concurrent_lookups_never_block_one_another)
//! is the test that holds the `Send + Sync` claim — the router puts this in
//! shared state and a type that is neither would not compile there.
//!
//! # Why there is no capacity ceiling
//!
//! Because a ceiling on this map has exactly one honest eviction rule — drop
//! the oldest session — and that rule is indistinguishable, from the client,
//! from a server that forgot the session. Both produce `404`, and both reach
//! the client as "start a new session", so a load spike silently destroys live
//! state that the client believes it still owns. Refusing to open the
//! thousand-and-first session is loud and fixable; evicting the first is quiet
//! and is not.
//!
//! Idle reaping, by contrast, is a different shape: a session that has been
//! quiet for longer than an operator-chosen idle window is not a session a
//! client is still driving, and a server that evicts it produces a `404` on
//! the *next* request — which is exactly what an evicted-after-DELETE
//! session already produces. The two are not distinguishable from the
//! client and were never going to be. A deployment that wants a bound
//! constructs the store with
//! [`with_idle_reaper`](SessionStore::with_idle_reaper) and calls
//! [`reap_idle`](SessionStore::reap_idle) from a background task with the
//! operator-chosen `max_age`; the store itself does not start a task,
//! because a background task started inside a constructor is one a
//! deployment cannot stop.
//!
//! Note what that leaves out: unlike
//! [`InMemoryIdempotencyStore`](crate::adapters::secondary::InMemoryIdempotencyStore),
//! these keys are not peer-chosen. Nobody can make the server allocate a
//! session except by successfully completing an `initialize`, which is the
//! one operation that already does work worth bounding elsewhere.
//!
//! # Why the id comes from the engine's `IdSource`
//!
//! Because the alternative is a process-global RNG call inside an adapter,
//! which is the determinism rule. A test that wants to know which id a
//! `initialize` will produce cannot arrange it if the adapter reaches for
//! `rand::thread_rng()`; a test that runs the whole session lifecycle twice
//! cannot compare anything if the second run's ids differ from the first's.
//! [`IdSource`](crate::application::session::IdSource) is the seam the engine
//! already owns for exactly this, and sharing it means an HTTP session and an
//! engine session are numbered from the same sequence rather than two
//! unrelated ones.
//!
//! # Protocol neutrality
//!
//! The store names a *header*, not a method. Nothing here knows what
//! `initialize` is called, what it negotiates, or what a session does once it
//! exists — the payload is a type parameter, so a protocol whose sessions are
//! shaped differently instantiates this with a different `S` and changes
//! nothing. See [`SessionStore`].

use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use thiserror::Error;

use crate::application::session::{Clock, IdSource};
use crate::domain::mcp::McpSessionId;
use crate::domain::session::{Instant, SessionId};

/// The sessions this server currently holds open.
///
/// Generic over what a session *is*, because this type is a map and a map does
/// not need to know. The Streamable HTTP adapter instantiates it with whatever
/// the router keeps per session — an engine handle, a protocol state machine, a
/// cancellation token — and the `Arc` in [`lookup`](Self::lookup) is what lets
/// a handler hold that state across an `.await` without borrowing the store.
///
/// # Example
///
/// ```
/// use std::sync::Arc;
/// use protocol_droid::adapters::primary::transport::http::SessionStore;
/// use protocol_droid::application::session::IdSource;
/// use protocol_droid::domain::session::state::SessionId;
///
/// // A source that hands out a fixed sequence, so the test can say which id it
/// // expects instead of discovering it.
/// struct Counting(core::sync::atomic::AtomicU64);
/// impl IdSource for Counting {
///     fn next_session_id(&self) -> SessionId {
///         SessionId::new(
///             self.0.fetch_add(1, core::sync::atomic::Ordering::SeqCst) + 1,
///         )
///     }
/// }
/// let ids: Arc<dyn IdSource + Send + Sync> =
///     Arc::new(Counting(core::sync::atomic::AtomicU64::new(0)));
///
/// let store = SessionStore::new(ids);
/// let id = store.open(Arc::new("the negotiated state"))?;
/// // The id is a header value, so it round-trips through the header parser.
/// assert_eq!(id.to_string().parse(), Ok(id));
/// assert!(store.lookup(&id)?.is_some());
///
/// // DELETE removes it, and a later request is a miss rather than a leftover.
/// assert!(store.terminate(&id)?.is_some());
/// assert!(store.lookup(&id)?.is_none());
/// # Ok::<(), protocol_droid::adapters::primary::transport::http::SessionStoreError>(())
/// ```
pub struct SessionStore<S> {
    sessions: Mutex<HashMap<McpSessionId, Arc<S>>>,
    /// The last time each session was opened or looked up.
    ///
    /// Populated only when the store was built through
    /// [`with_idle_reaper`](Self::with_idle_reaper); a `None` here means
    /// the store has no reaper and [`reap_idle`](Self::reap_idle) is a
    /// no-op. A separate map, rather than a wrapper around `Arc<S>`, so
    /// the existing `Arc<S>` shape is preserved for the common case
    /// where reaping is off.
    last_touched: Option<Mutex<HashMap<McpSessionId, Instant>>>,
    ids: Arc<dyn IdSource + Send + Sync>,
    /// The clock used to record `last_touched`. `None` when the reaper
    /// is off, and reading it would force a `clock` to exist that
    /// nothing uses.
    clock: Option<Arc<dyn Clock + Send + Sync>>,
}

/// The two pieces of state a [`SessionStore`] with a reaper carries
/// together: the timestamps map and the clock the reaper reads.
///
/// A type alias rather than spelling the pair at every call site,
/// because the alternative is a long `&Mutex<HashMap<...>>` and a
/// `&Arc<dyn Clock + ...>` on the same line, and clippy's
/// `type_complexity` will rightly call that out the first time
/// the signature is reused.
type ReaperPair<'a> = (
    &'a Mutex<HashMap<McpSessionId, Instant>>,
    &'a Arc<dyn Clock + Send + Sync>,
);

impl<S> SessionStore<S> {
    /// A store with no sessions, numbering them from `ids`.
    ///
    /// Takes the source rather than constructing one: the engine already has an
    /// `IdSource` for the same reason, and two sources means two sequences for
    /// what a reader will reasonably assume is one numbering.
    ///
    /// The store built through this constructor is unbounded. A deployment
    /// that needs a bound constructs with
    /// [`with_idle_reaper`](Self::with_idle_reaper) instead, which adds
    /// the bookkeeping `reap_idle` reads.
    #[must_use]
    pub fn new(ids: Arc<dyn IdSource + Send + Sync>) -> Self {
        Self {
            sessions: Mutex::new(HashMap::new()),
            last_touched: None,
            ids,
            clock: None,
        }
    }

    /// A store that tracks when each session was last touched, so a
    /// caller-driven [`reap_idle`](Self::reap_idle) can evict the ones
    /// a client has stopped driving.
    ///
    /// The store does not start a reaper task on its own: a
    /// background task started inside a constructor is one a deployment
    /// cannot stop, and reaping intervals are an operator decision. The
    /// caller is expected to call [`reap_idle`](Self::reap_idle) from
    /// whichever loop it owns (a periodic timer, a `select!` on shutdown,
    /// an external cron) with the `max_age` it considers idle.
    ///
    /// `clock` is the source of time the store reads. It is
    /// injected rather than read from a global so a test can drive
    /// reaping without sleeping; the rest of the crate already takes
    /// the same shape for [`IdSource`](crate::application::session::IdSource).
    #[must_use]
    pub fn with_idle_reaper(
        ids: Arc<dyn IdSource + Send + Sync>,
        clock: Arc<dyn Clock + Send + Sync>,
    ) -> Self {
        Self {
            sessions: Mutex::new(HashMap::new()),
            last_touched: Some(Mutex::new(HashMap::new())),
            ids,
            clock: Some(clock),
        }
    }

    /// Registers `session` under a freshly minted id and returns that id.
    ///
    /// The caller puts the id in the `Mcp-Session-Id` response header; this type
    /// never formats it, because the buffer-based
    /// [`McpSessionId::as_bytes`](crate::domain::mcp::McpSessionId::as_bytes) is
    /// how the header gets filled without an allocation per response.
    ///
    /// When the store was built with
    /// [`with_idle_reaper`](Self::with_idle_reaper), the new entry's
    /// `last_touched` is the clock's current reading. The recording
    /// happens under the same lock as the insertion, so a `reap_idle`
    /// that runs concurrently with an `open` cannot see the session
    /// without also seeing the timestamp that would spare it.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::DuplicateId`] if the injected source returned an id
    /// this store is already holding. [`IdSource`] promises not to repeat one,
    /// so a source that has wrapped or one that was seeded twice has broken its
    /// contract, and overwriting would hand the second caller a session
    /// belonging to the first. One id is tried: retrying a source that is
    /// provably out of ids is a livelock with better manners, and the error
    /// says what to do.
    ///
    /// [`SessionStoreError::Unavailable`] if the lock is poisoned.
    pub fn open(&self, session: Arc<S>) -> Result<McpSessionId, SessionStoreError> {
        let id = McpSessionId::new(self.ids.next_session_id());
        self.insert(id, session)?;
        Ok(id)
    }

    /// Opens a session under an id the caller has already taken.
    ///
    /// Separate from [`Self::open`] because a session's *peer* has to know its
    /// own id before it is inserted — the peer is part of the value being
    /// inserted, and everything the peer keys on (idempotency, in-flight
    /// requests) reads the id from inside it. A store that allocated the id on
    /// the way in could only hand the peer that id afterwards, by which point
    /// the peer has already been constructed with the wrong one.
    ///
    /// # Errors
    ///
    /// As [`Self::open`]: [`SessionStoreError::DuplicateId`] if the id is
    /// already held, [`SessionStoreError::Unavailable`] if the lock is poisoned.
    pub fn open_with_id(
        &self,
        session: Arc<S>,
        id: SessionId,
    ) -> Result<McpSessionId, SessionStoreError> {
        let id = McpSessionId::new(id);
        self.insert(id, session)?;
        Ok(id)
    }

    /// The next id this store would hand out.
    ///
    /// For a caller that must build a session *around* a known id before it
    /// can hand that session to [`Self::open_with_id`]. The contract is the
    /// same as the one `open` relies on: a source that repeats an id is caught
    /// by the duplicate check there, so reading an id early is safe even
    /// though nothing reserves it in the meantime.
    ///
    /// Two callers racing here get two different ids and both open; the loser
    /// is refused by the duplicate check, which is the answer for a source that
    /// repeats one.
    pub fn next_id(&self) -> SessionId {
        self.ids.next_session_id()
    }

    /// Inserts a session under an id already wrapped as a header id.
    fn insert(&self, id: McpSessionId, session: Arc<S>) -> Result<(), SessionStoreError> {
        // The guard is scoped to the block so it is released before the
        // `Ok`, not at the end of the function: a caller that has just opened a
        // session should not still be holding the lock by the time it starts
        // doing the work the session exists for. The `record_touched` call
        // happens after the sessions map's lock is dropped, so a poisoned
        // touched-map lock cannot deadlock an `open` against another
        // request that is mid-lookup on a different session.
        {
            let mut sessions = self.locked()?;
            if sessions.contains_key(&id) {
                return Err(SessionStoreError::DuplicateId);
            }
            sessions.insert(id, session);
        }
        self.record_touched(id)?;
        Ok(())
    }

    /// The session under `id`, if this server still has it.
    ///
    /// A returned [`Arc`] is a snapshot of *reachability*, not of the session's
    /// contents: a concurrent `DELETE` can remove the map entry while this
    /// clone is still alive and being used. That is the intended semantics —
    /// the in-flight request finishes against the session it already found,
    /// and only the *next* request sees the `404` — so callers must not treat
    /// a `Some` as a promise that the entry will still be there in a moment.
    ///
    /// When the store was built with
    /// [`with_idle_reaper`](Self::with_idle_reaper), a successful
    /// lookup updates the entry's `last_touched` to the clock's
    /// current reading. The bookkeeping is what makes a session
    /// "idle" mean "no request has hit it for the window", and a
    /// request hitting it is the only signal the store has that
    /// the client is still driving it.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the lock is poisoned.
    pub fn lookup(&self, id: &McpSessionId) -> Result<Option<Arc<S>>, SessionStoreError> {
        let found = self.locked()?.get(id).cloned();
        if found.is_some() {
            self.record_touched(*id)?;
        }
        Ok(found)
    }

    /// Removes the session under `id`, returning it if it was there.
    ///
    /// A `None` here is the same `None` [`lookup`](Self::lookup) returns, and
    /// the router maps both to the same outcome. Deciding between "terminate
    /// this session" and "this endpoint does not offer termination here" is
    /// the router's, because only the router knows what the endpoint offers.
    ///
    /// When the store was built with
    /// [`with_idle_reaper`](Self::with_idle_reaper), the corresponding
    /// `last_touched` is also removed: leaving it behind would be a
    /// slow leak of a different shape (timestamps whose sessions
    /// have already gone), and the next `reap_idle` would walk
    /// past an entry that no longer has a session to evict.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the lock is poisoned.
    pub fn terminate(&self, id: &McpSessionId) -> Result<Option<Arc<S>>, SessionStoreError> {
        let removed = self.locked()?.remove(id);
        if removed.is_some() {
            self.forget_touched(*id)?;
        }
        Ok(removed)
    }

    /// Evicts every session whose `last_touched` is older than `max_age`,
    /// returning the ids that were reaped.
    ///
    /// The store built through [`new`](Self::new) has no reaper and
    /// `reap_idle` is a no-op that returns an empty vector. The
    /// store built through
    /// [`with_idle_reaper`](Self::with_idle_reaper) records the time
    /// on every `open` and `lookup`, and `reap_idle` compares it
    /// against the clock's current reading.
    ///
    /// The caller drives reaping: a periodic timer, a `select!` on
    /// shutdown, an external cron. There is no background task
    /// started inside the store, because a background task started
    /// inside a constructor is one a deployment cannot stop.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the reaper's lock is
    /// poisoned, which the same rule that refuses to recover the
    /// sessions map also applies here: answering from state this
    /// process cannot vouch for is how one request ends up served
    /// another's.
    pub fn reap_idle(&self, max_age: Duration) -> Result<Vec<McpSessionId>, SessionStoreError> {
        let Some((touched, clock)) = self.reaper_pair() else {
            return Ok(Vec::new());
        };
        let now = clock.now();
        let mut stale: Vec<McpSessionId> = Vec::new();
        // The two lock acquisitions in this function are
        // deliberately separated: the touched map is read
        // (to find stale ids), released, then the sessions map
        // is held to do the eviction, then released, then the
        // touched map is taken again to drop the timestamps
        // for the ids that were evicted. Holding both at once
        // would risk a deadlock against any future caller that
        // takes them in the other order; the cost of the
        // separation is two extra lock acquisitions per sweep,
        // which is a budget the operator chose when they
        // scheduled the sweep.
        {
            let touched = touched.lock().map_err(|_| SessionStoreError::Unavailable)?;
            for (id, when) in touched.iter() {
                // A session touched at `now` is not stale at
                // `now`, even with a zero `max_age` — `>` is
                // the right comparison, not `>=`, because the
                // operator's "idle window" is a window, not a
                // "before" cut. `saturating_sub` clamps a
                // backwards clock to zero, so a clock that
                // stepped back to before `when` cannot produce
                // a negative elapsed time and is never stale.
                if now.as_duration().saturating_sub(when.as_duration()) > max_age {
                    stale.push(*id);
                }
            }
        }
        let mut reaped: Vec<McpSessionId> = Vec::with_capacity(stale.len());
        {
            let mut sessions = self.locked()?;
            for id in stale {
                if sessions.remove(&id).is_some() {
                    reaped.push(id);
                }
            }
        }
        {
            let mut touched = touched.lock().map_err(|_| SessionStoreError::Unavailable)?;
            for id in &reaped {
                touched.remove(id);
            }
        }
        Ok(reaped)
    }

    /// How many sessions are open.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the lock is poisoned. A made-up
    /// number here would be worse than the error: this is the count an
    /// operator reads to decide whether something is wrong.
    pub fn len(&self) -> Result<usize, SessionStoreError> {
        Ok(self.locked()?.len())
    }

    /// Whether no session is open.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the lock is poisoned.
    pub fn is_empty(&self) -> Result<bool, SessionStoreError> {
        Ok(self.locked()?.is_empty())
    }

    /// One lock acquisition, shared by every operation.
    ///
    /// The guard's lifetime is this function's return type and nothing longer.
    /// That is the whole of the "no lock across an await" rule: the guard is
    /// consumed by the caller of `locked` inside a single expression, so it
    /// cannot outlive the statement it was needed for.
    fn locked(&self) -> Result<MutexGuard<'_, HashMap<McpSessionId, Arc<S>>>, SessionStoreError> {
        self.sessions
            .lock()
            .map_err(|_| SessionStoreError::Unavailable)
    }

    /// The reaper's two pieces of state, when the store has a reaper.
    ///
    /// `None` when the store was built through [`new`](Self::new);
    /// `Some` when built through
    /// [`with_idle_reaper`](Self::with_idle_reaper). The two
    /// references are returned together because the touched map and
    /// the clock are always either both present or both absent —
    /// a state the constructor's two-field literal pins at
    /// construction time, so a `Some(touched)` here is also a
    /// `Some(clock)`, and the `unreachable!` is unreachable.
    fn reaper_pair(&self) -> Option<ReaperPair<'_>> {
        let (touched, clock) = match (&self.last_touched, &self.clock) {
            (Some(t), Some(c)) => (t, c),
            (None, None) => return None,
            // The two fields are set together in the constructor;
            // any other combination is a violation of the
            // invariant the constructor maintains.
            _ => unreachable!("last_touched and clock must agree"),
        };
        Some((touched, clock))
    }

    /// Records `id` as touched at the clock's current reading.
    ///
    /// A no-op when the store has no reaper, so a caller can record
    /// unconditionally without a `cfg(feature = ...)` or a runtime
    /// check on the constructor.
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the touched map's lock
    /// is poisoned.
    fn record_touched(&self, id: McpSessionId) -> Result<(), SessionStoreError> {
        let Some((touched, clock)) = self.reaper_pair() else {
            return Ok(());
        };
        // The lock is scoped to this block so the guard is
        // dropped before the `Ok` — clippy's
        // `significant_drop_tightening` lint tracks a `MutexGuard`
        // through the whole function and complains that the
        // surrounding function (which does no further work) is
        // holding a lock it does not need.
        {
            let mut touched = touched.lock().map_err(|_| SessionStoreError::Unavailable)?;
            touched.insert(id, clock.now());
        }
        Ok(())
    }

    /// Forgets the `last_touched` entry for `id`, if one was kept.
    ///
    /// Called from [`terminate`](Self::terminate) so a session that
    /// was explicitly removed does not leave a timestamp behind.
    /// A no-op when the store has no reaper, by the same reasoning
    /// as [`record_touched`](Self::record_touched).
    ///
    /// # Errors
    ///
    /// [`SessionStoreError::Unavailable`] if the touched map's lock
    /// is poisoned.
    fn forget_touched(&self, id: McpSessionId) -> Result<(), SessionStoreError> {
        let Some((touched, _)) = self.reaper_pair() else {
            return Ok(());
        };
        {
            let mut touched = touched.lock().map_err(|_| SessionStoreError::Unavailable)?;
            touched.remove(&id);
        }
        Ok(())
    }
}

impl<S> core::fmt::Debug for SessionStore<S> {
    /// Prints the type's name and nothing else.
    ///
    /// Deliberately opaque, and the omission is the point: a derived `Debug`
    /// on this struct would print the whole map, which is a table of session
    /// ids — and the project does not put a full session id in a log line.
    /// A `{:?}` in a log statement must not become a way around that.
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        formatter
            .debug_struct("SessionStore")
            .finish_non_exhaustive()
    }
}

/// Why a [`SessionStore`] operation could not be answered.
///
/// Not an `Error` for its own sake — it is one so a handler can `?` it into the
/// same failure path as everything else the adapter refuses, and so
/// `thiserror` writes the text that reaches a log line.
///
/// # Why no id appears in either message
///
/// Because `Display` output here is very likely to end up in a log, and the
/// project does not log a full session id. Carrying the value inside the error
/// would make the leak one careless `tracing::error!("{err}")` away, and the
/// operator debugging a duplicate id wants to know that it happened far more
/// than they want to know which id it was — the source is the thing to inspect.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum SessionStoreError {
    /// A panic elsewhere unwound through the lock, so the map may be missing
    /// an insert it was midway through.
    ///
    /// The store refuses rather than recovering, matching the idempotency
    /// store's rule: answering from state this process cannot vouch for is how
    /// one request ends up served another request's session.
    #[error("the session store's lock is poisoned")]
    Unavailable,

    /// The injected [`IdSource`] returned an id that is already registered.
    ///
    /// [`IdSource`] requires that an id is not reused while either session can
    /// still appear in the event log, so this means the source has wrapped or
    /// was seeded twice. Replacing the source is the fix; the store will not
    /// overwrite, because doing so silently transfers an established session to
    /// whoever asked next.
    #[error("the session id source returned an id that is already in use")]
    DuplicateId,
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::Duration;

    use super::{SessionStore, SessionStoreError};
    use crate::application::session::{Clock, IdSource};
    use crate::domain::mcp::{McpSessionId, VISIBLE_ASCII_RANGE};
    use crate::domain::session::{Instant, SessionId};

    /// A source that hands out a fixed sequence and counts what it produced.
    struct Counting(AtomicU64);

    impl Counting {
        fn new() -> Self {
            Self(AtomicU64::new(0))
        }

        fn shared() -> Arc<dyn IdSource + Send + Sync> {
            Arc::new(Self::new())
        }

        /// How many ids this source has been asked for.
        fn issued(&self) -> u64 {
            self.0.load(Ordering::SeqCst)
        }
    }

    impl IdSource for Counting {
        fn next_session_id(&self) -> SessionId {
            SessionId::new(self.0.fetch_add(1, Ordering::SeqCst))
        }
    }

    /// A source that always returns the same id, to exercise the collision
    /// path that a well-behaved source can never reach.
    struct Stuck(SessionId);

    impl IdSource for Stuck {
        fn next_session_id(&self) -> SessionId {
            self.0
        }
    }

    type TestError = Box<dyn std::error::Error>;

    /// What a lookup thread can report.
    ///
    /// A named error rather than a boxed `dyn Error`, because the two failures
    /// a thread has — the store refusing, and a session that was there at open
    /// time being absent at lookup time — are different bugs and the message
    /// has to survive the thread boundary to say which.
    #[derive(Debug)]
    enum ThreadFailure {
        /// The store refused an operation.
        Store(SessionStoreError),
        /// A session that was open was not found.
        Vanished,
    }

    impl From<SessionStoreError> for ThreadFailure {
        fn from(error: SessionStoreError) -> Self {
            Self::Store(error)
        }
    }

    impl core::fmt::Display for ThreadFailure {
        fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
            match self {
                Self::Store(error) => write!(formatter, "the store refused a lookup: {error}"),
                Self::Vanished => write!(formatter, "a live session was not found"),
            }
        }
    }

    impl std::error::Error for ThreadFailure {}

    /// The two failure modes name themselves in their `Display`, so a
    /// `?`-chained test can distinguish them without a match. If this ever
    /// fails, the two messages have been swapped or made identical.
    #[test]
    fn the_error_variants_are_distinguishable() {
        assert_ne!(
            SessionStoreError::Unavailable.to_string(),
            SessionStoreError::DuplicateId.to_string()
        );
        assert!(
            SessionStoreError::DuplicateId
                .to_string()
                .contains("already in use")
        );
    }

    /// The store's ids must be ids the header parser accepts.
    ///
    /// The store and the wire format are separate types, and nothing in the
    /// type system connects them: `McpSessionId` is constructed from a
    /// `SessionId` by a `const fn` that cannot fail, so the only thing standing
    /// between "the store minted an id" and "the client can quote it" is that
    /// the two agree. This is the assertion that says they do — parse the
    /// rendered header value back and compare the whole `Result`, so a failure
    /// prints what the parser objected to rather than just a mismatch.
    #[test]
    fn a_minted_id_round_trips_through_the_header_parser() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        let id = store.open(Arc::new(42usize))?;
        assert_eq!(id.to_string().parse::<McpSessionId>(), Ok(id));
        Ok(())
    }

    /// The rendered header value is fixed-width and entirely legal in a header.
    ///
    /// Two separate claims, and they fail differently. A width that varied with
    /// the value would let a length-sensitive intermediary treat `7` and
    /// `0000000000000007` as different sessions; a byte outside visible ASCII
    /// would not be a header at all. So the assertion checks both the exact
    /// spelling and the byte range, rather than "it parsed".
    #[test]
    fn the_header_value_is_fixed_width_and_visible_ascii() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        for _ in 0..4 {
            let id = store.open(Arc::new(()))?;
            let rendered = id.to_string();
            assert_eq!(
                rendered.len(),
                16,
                "session id {rendered:?} is not sixteen characters"
            );
            assert!(
                rendered
                    .bytes()
                    .all(|byte| VISIBLE_ASCII_RANGE.contains(&byte)),
                "session id {rendered:?} carries a byte no header may contain"
            );
            assert!(
                rendered.bytes().all(|byte| byte.is_ascii_hexdigit()),
                "session id {rendered:?} is not hexadecimal"
            );
        }
        Ok(())
    }

    /// Ids come from the injected source, and the store never numbers on its own.
    ///
    /// The claim is *substitution*: if the adapter reached for a process-global
    /// RNG instead of the injected port, the sequence below would not be the
    /// one the source produced. Asserting the exact first id is what makes a
    /// `thread_rng` mutant fail here rather than merely being unlikely to be
    /// noticed.
    #[test]
    fn ids_come_from_the_injected_source() -> Result<(), TestError> {
        let source = Arc::new(Counting::new());
        let store = SessionStore::new(source.clone());
        let first = store.open(Arc::new(()))?;
        let second = store.open(Arc::new(()))?;
        assert_eq!(first, McpSessionId::new(SessionId::new(0)));
        assert_eq!(second, McpSessionId::new(SessionId::new(1)));
        assert_eq!(source.issued(), 2, "the store asked for a different count");
        Ok(())
    }

    /// A lookup miss is `None`, and it is reached only after the header parsed.
    ///
    /// This is the 404-versus-400 boundary. `McpSessionId` is produced by
    /// `from_str`, so a client that sends garbage never gets as far as
    /// `lookup` — and the test says so by parsing a malformed header and
    /// checking the failure is a parse error, *then* checking that a
    /// well-formed header naming nothing is a miss. Conflating the two would
    /// answer a broken header with "start a new session", which is advice that
    /// cannot possibly help.
    #[test]
    fn a_miss_and_a_malformed_header_are_different_outcomes() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        let live = store.open(Arc::new(()))?;
        store.terminate(&live)?;

        // Well-formed, and unknown: a miss, which the router answers 404.
        assert_eq!(store.lookup(&live)?, None);

        // Not well-formed: the header parser refuses before the store is asked,
        // which is the router's 400.
        assert!("not-a-session".parse::<McpSessionId>().is_err());
        assert!("".parse::<McpSessionId>().is_err());
        Ok(())
    }

    /// `DELETE` of a session that is not there is also a miss, not a panic.
    ///
    /// The specification lets a server answer `405` on a `DELETE` it does not
    /// offer, and either way the store's answer is the same `None` the router
    /// turns into a decision. The claim under test is that removing twice is
    /// well-defined and the second removal changes nothing.
    #[test]
    fn terminating_an_absent_session_is_a_miss() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        let id = store.open(Arc::new(()))?;
        assert!(store.terminate(&id)?.is_some());
        assert_eq!(store.terminate(&id)?, None);
        assert!(store.is_empty()?);
        Ok(())
    }

    /// A source that repeats an id must not transfer a live session to a new
    /// caller.
    ///
    /// Overwriting here would be a session hijack: the second `initialize`
    /// would be handed the first client's negotiated state, its event ring,
    /// and its id, and neither side would find out until a request came back
    /// addressed to someone else's conversation. `IdSource` forbids this, so
    /// the store's job is to notice rather than to comply.
    #[test]
    fn a_repeated_id_refuses_instead_of_overwriting() -> Result<(), TestError> {
        let store = SessionStore::new(Arc::new(Stuck(SessionId::new(7))));
        let first = store.open(Arc::new("first"))?;
        let second = store.open(Arc::new("second"));
        assert_eq!(second, Err(SessionStoreError::DuplicateId));

        // And the original session is untouched: one entry, still the first.
        assert_eq!(store.len()?, 1);
        let survivor = store.lookup(&first)?;
        assert_eq!(survivor.as_deref(), Some(&"first"));
        Ok(())
    }

    /// The store is usable from many threads at once, and no lookup waits on
    /// another.
    ///
    /// Two claims in one test because they have the same failure mode to
    /// detect: if the guard were part of a returned value, or if `S` were not
    /// `Send + Sync`, this would not compile — and the router putting the store
    /// in shared state needs it to. The assertion that every thread finds every
    /// session also catches a store that accidentally hands out a shared
    /// mutable view instead of a clone.
    #[test]
    fn concurrent_lookups_never_block_one_another() -> Result<(), TestError> {
        const SESSIONS: usize = 8;
        const THREADS: usize = 4;

        let store = Arc::new(SessionStore::new(Counting::shared()));
        let mut ids = Vec::new();
        for _ in 0..SESSIONS {
            ids.push(store.open(Arc::new(SESSIONS))?);
        }

        let threads: Vec<_> = (0..THREADS)
            .map(|_| {
                let store = Arc::clone(&store);
                let ids = ids.clone();
                std::thread::spawn(move || -> Result<(), ThreadFailure> {
                    for _ in 0..64 {
                        for id in &ids {
                            if store.lookup(id)?.is_none() {
                                return Err(ThreadFailure::Vanished);
                            }
                        }
                    }
                    Ok(())
                })
            })
            .collect();
        for thread in threads {
            match thread.join() {
                Ok(result) => result?,
                Err(_) => return Err("a lookup thread panicked".into()),
            }
        }

        assert_eq!(store.len()?, SESSIONS);
        Ok(())
    }

    /// `len` and `is_empty` describe the same map the other methods mutate.
    ///
    /// Without this, an adapter could report a healthy zero-session server
    /// while refusing every request, and the operator would be looking at the
    /// wrong number.
    #[test]
    fn the_counters_track_the_operations() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        assert!(store.is_empty()?);
        assert_eq!(store.len()?, 0);

        let first = store.open(Arc::new(()))?;
        let second = store.open(Arc::new(()))?;
        assert_eq!(store.len()?, 2);
        assert!(!store.is_empty()?);

        store.terminate(&first)?;
        assert_eq!(store.len()?, 1);
        store.terminate(&second)?;
        assert!(store.is_empty()?);
        Ok(())
    }

    /// A clock the test drives by hand, so reaping can expire
    /// without a real-time wait.
    ///
    /// The clock is `Arc<Stepping>`-shaped: the test owns the inner
    /// `AtomicU64`, the store holds an `Arc<dyn Clock>` to the same
    /// value, and `advance` mutates the time the store reads.
    struct Stepping {
        millis: AtomicU64,
    }

    impl Stepping {
        fn new() -> Arc<Self> {
            Arc::new(Self {
                millis: AtomicU64::new(0),
            })
        }

        fn advance(&self, by: Duration) {
            let added = u64::try_from(by.as_millis()).unwrap_or(u64::MAX);
            self.millis.fetch_add(added, Ordering::Relaxed);
        }
    }

    impl Clock for Stepping {
        fn now(&self) -> Instant {
            Instant::from_duration(Duration::from_millis(self.millis.load(Ordering::Relaxed)))
        }
    }

    impl Clock for Arc<Stepping> {
        fn now(&self) -> Instant {
            <Stepping as Clock>::now(self.as_ref())
        }
    }

    /// A store built through `new` has no reaper.
    ///
    /// `reap_idle` returns an empty vector rather than tripping on
    /// the absent clock, so an operator who never wired one in
    /// does not get a panic on a sweep call.
    #[test]
    fn a_store_without_a_reaper_is_a_no_op_for_reap_idle() -> Result<(), TestError> {
        let store = SessionStore::new(Counting::shared());
        let id = store.open(Arc::new(()))?;
        assert!(store.reap_idle(Duration::from_millis(0))?.is_empty());
        assert_eq!(store.len()?, 1, "no reaper means no eviction");
        let _ = id;
        Ok(())
    }

    /// Sessions whose `last_touched` is older than the threshold
    /// are removed; live ones are not.
    ///
    /// Three sessions are opened, the clock advances past the
    /// threshold, only one is touched (a `lookup` updates the
    /// timestamp), and `reap_idle` removes the other two.
    #[test]
    fn reap_idle_evicts_sessions_older_than_the_threshold() -> Result<(), TestError> {
        let clock = Stepping::new();
        let store = SessionStore::with_idle_reaper(Counting::shared(), clock.clone());
        let stale_a = store.open(Arc::new(()))?;
        let stale_b = store.open(Arc::new(()))?;
        let live = store.open(Arc::new(()))?;

        // The clock now reads 0; everything opened at 0 is fresh.
        assert!(store.reap_idle(Duration::from_secs(60))?.is_empty());
        assert_eq!(store.len()?, 3);

        // Move the clock past the threshold for `stale_a` and
        // `stale_b`, then touch `live` to refresh it.
        clock.advance(Duration::from_secs(120));
        assert!(
            store.lookup(&live)?.is_some(),
            "the live session should still be in the store"
        );

        // Now the threshold is 60s; `stale_a` and `stale_b` are
        // 120s old, `live` is 0s old.
        let reaped = store.reap_idle(Duration::from_secs(60))?;
        let reaped_set: std::collections::HashSet<McpSessionId> = reaped.into_iter().collect();
        assert!(reaped_set.contains(&stale_a));
        assert!(reaped_set.contains(&stale_b));
        assert!(!reaped_set.contains(&live));
        assert_eq!(store.len()?, 1);
        assert!(
            store.lookup(&live)?.is_some(),
            "the live session must survive the sweep"
        );
        assert!(store.lookup(&stale_a)?.is_none());
        assert!(store.lookup(&stale_b)?.is_none());
        Ok(())
    }

    /// A subsequent `reap_idle` after the clock has not moved
    /// is a no-op, even with a zero threshold.
    ///
    /// The first sweep removes what was stale; the second one
    /// has nothing left to remove, and the count of returns
    /// (the number reaped) reflects that.
    #[test]
    fn reap_idle_is_idempotent_under_a_steady_clock() -> Result<(), TestError> {
        let clock = Stepping::new();
        let store = SessionStore::with_idle_reaper(Counting::shared(), clock.clone());
        let _id = store.open(Arc::new(()))?;
        clock.advance(Duration::from_secs(120));
        let first = store.reap_idle(Duration::from_secs(60))?;
        assert_eq!(first.len(), 1);
        let second = store.reap_idle(Duration::from_secs(60))?;
        assert!(second.is_empty(), "the second sweep has nothing to do");
        Ok(())
    }

    /// `terminate` removes the session *and* its `last_touched`
    /// entry, so a later `reap_idle` does not walk a stale
    /// timestamp that no longer has a session to evict.
    #[test]
    fn terminate_clears_the_touched_entry() -> Result<(), TestError> {
        let clock = Stepping::new();
        let store = SessionStore::with_idle_reaper(Counting::shared(), clock.clone());
        let id = store.open(Arc::new(()))?;
        store.terminate(&id)?;
        clock.advance(Duration::from_secs(120));
        // A reaper that walked timestamps only would find a
        // `last_touched` for `id` here and try to evict a
        // session that is already gone. The bookkeeping that
        // `terminate` does is what prevents the dangling
        // timestamp.
        let reaped = store.reap_idle(Duration::from_secs(60))?;
        assert!(reaped.is_empty());
        assert_eq!(store.len()?, 0);
        Ok(())
    }

    /// `lookup` is the signal the store has that the client is
    /// still driving a session. A session that is repeatedly
    /// looked up survives indefinitely, no matter how long the
    /// clock runs.
    #[test]
    fn a_looked_up_session_never_becomes_stale() -> Result<(), TestError> {
        let clock = Stepping::new();
        let store = SessionStore::with_idle_reaper(Counting::shared(), clock.clone());
        let id = store.open(Arc::new(()))?;
        for _ in 0..16 {
            clock.advance(Duration::from_secs(60));
            assert!(store.lookup(&id)?.is_some());
        }
        // The session is 16 × 60s = 960s old by the clock, but
        // every step touched it. A zero-threshold sweep is the
        // harshest test: a session that is not strictly newer
        // than the previous reading is stale. The latest
        // `lookup` is the one the reaper sees.
        let reaped = store.reap_idle(Duration::from_secs(0))?;
        assert!(
            reaped.is_empty(),
            "a session touched at `now` is not stale at `now`"
        );
        Ok(())
    }
}