1use crate::error::MacpError;
2use crate::mode::ModeResponse;
3use crate::policy::PolicyDefinition;
4use macp_pb::pb::SessionStartPayload;
5use prost::Message;
6use std::collections::{HashMap, HashSet};
7
8pub const MAX_TTL_MS: i64 = 24 * 60 * 60 * 1000;
9
10pub const MAX_SUSPEND_MS: i64 = 7 * 24 * 60 * 60 * 1000;
17
18pub const CURRENT_SEMANTICS_REV: u32 = 1;
29
30#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
31pub enum SessionState {
32 Open,
33 Suspended,
37 Resolved,
38 Expired,
39 Cancelled,
41}
42
43impl SessionState {
44 pub fn is_terminal(&self) -> bool {
46 matches!(
47 self,
48 SessionState::Resolved | SessionState::Expired | SessionState::Cancelled
49 )
50 }
51}
52
53#[non_exhaustive]
58#[derive(Clone, Debug)]
59pub struct Session {
60 pub session_id: String,
61 pub state: SessionState,
62 pub ttl_expiry: i64,
63 pub ttl_ms: i64,
64 pub started_at_unix_ms: i64,
65 pub resolution: Option<Vec<u8>>,
66 pub mode: String,
67 pub mode_state: Vec<u8>,
68 pub participants: Vec<String>,
69 pub seen_message_ids: HashSet<String>,
70 pub intent: String,
71 pub mode_version: String,
72 pub configuration_version: String,
73 pub policy_version: String,
74 pub context_id: String,
75 pub extensions: HashMap<String, Vec<u8>>,
76 pub roots: Vec<macp_pb::pb::Root>,
77 pub initiator_sender: String,
78 pub participant_message_counts: HashMap<String, u32>,
79 pub participant_last_seen: HashMap<String, i64>,
80 pub policy_definition: Option<PolicyDefinition>,
81 pub suspended_at_ms: Option<i64>,
84 pub accumulated_suspended_ms: i64,
87 pub semantics_rev: u32,
90 pub max_suspend_ms: i64,
96}
97
98impl Session {
99 pub fn builder(
103 session_id: impl Into<String>,
104 mode: impl Into<String>,
105 initiator_sender: impl Into<String>,
106 ) -> SessionBuilder {
107 SessionBuilder {
108 inner: Session {
109 session_id: session_id.into(),
110 state: SessionState::Open,
111 ttl_expiry: i64::MAX,
115 ttl_ms: 0,
116 started_at_unix_ms: 0,
117 resolution: None,
118 mode: mode.into(),
119 mode_state: vec![],
120 participants: vec![],
121 seen_message_ids: HashSet::new(),
122 intent: String::new(),
123 mode_version: String::new(),
124 configuration_version: String::new(),
125 policy_version: String::new(),
126 context_id: String::new(),
127 extensions: HashMap::new(),
128 roots: vec![],
129 initiator_sender: initiator_sender.into(),
130 participant_message_counts: HashMap::new(),
131 participant_last_seen: HashMap::new(),
132 policy_definition: None,
133 suspended_at_ms: None,
134 accumulated_suspended_ms: 0,
135 semantics_rev: CURRENT_SEMANTICS_REV,
136 max_suspend_ms: 0,
137 },
138 }
139 }
140
141 pub fn record_participant_activity(&mut self, sender: &str, timestamp_ms: i64) {
142 *self
143 .participant_message_counts
144 .entry(sender.to_string())
145 .or_insert(0) += 1;
146 self.participant_last_seen
147 .insert(sender.to_string(), timestamp_ms);
148 }
149
150 pub fn suspend(&mut self, now_ms: i64) -> Result<(), MacpError> {
154 if self.state != SessionState::Open {
155 return Err(MacpError::SessionNotOpen);
156 }
157 self.state = SessionState::Suspended;
158 self.suspended_at_ms = Some(now_ms);
159 Ok(())
160 }
161
162 pub fn effective_max_suspend_ms(&self) -> i64 {
169 if self.max_suspend_ms > 0 {
170 self.max_suspend_ms
171 } else {
172 MAX_SUSPEND_MS
173 }
174 }
175
176 pub fn resume(&mut self, now_ms: i64) -> Result<(), MacpError> {
177 if self.state != SessionState::Suspended {
178 return Err(MacpError::SessionNotOpen);
179 }
180 let suspended_at = self.suspended_at_ms.unwrap_or(now_ms);
181 let banked = (now_ms - suspended_at).max(0);
182 self.accumulated_suspended_ms = self.accumulated_suspended_ms.saturating_add(banked);
183 self.suspended_at_ms = None;
184 if self.accumulated_suspended_ms > self.effective_max_suspend_ms() {
185 self.state = SessionState::Expired;
186 return Err(MacpError::TtlExpired);
187 }
188 self.ttl_expiry = self.ttl_expiry.saturating_add(banked);
189 self.state = SessionState::Open;
190 Ok(())
191 }
192
193 pub fn cancel(&mut self) -> Result<(), MacpError> {
196 if self.state.is_terminal() {
197 return Err(MacpError::SessionNotOpen);
198 }
199 self.state = SessionState::Cancelled;
200 self.suspended_at_ms = None;
201 Ok(())
202 }
203
204 pub fn suspend_cap_exceeded(&self, now_ms: i64) -> bool {
207 match self.suspended_at_ms {
208 Some(at) => {
209 self.accumulated_suspended_ms
210 .saturating_add((now_ms - at).max(0))
211 > self.effective_max_suspend_ms()
212 }
213 None => self.accumulated_suspended_ms > self.effective_max_suspend_ms(),
214 }
215 }
216
217 pub fn apply_mode_response(&mut self, response: ModeResponse) {
218 match response {
219 ModeResponse::NoOp => {}
220 ModeResponse::PersistState(state) => self.mode_state = state,
221 ModeResponse::Resolve(resolution) => {
222 self.state = SessionState::Resolved;
223 self.resolution = Some(resolution);
224 }
225 ModeResponse::PersistAndResolve { state, resolution } => {
226 self.mode_state = state;
227 self.state = SessionState::Resolved;
228 self.resolution = Some(resolution);
229 }
230 }
231 }
232}
233
234#[derive(Clone, Debug)]
240pub struct SessionBuilder {
241 inner: Session,
242}
243
244macro_rules! builder_setters {
245 ($($(#[$doc:meta])* $name:ident: $ty:ty),* $(,)?) => {
246 $(
247 $(#[$doc])*
248 pub fn $name(mut self, value: $ty) -> Self {
249 self.inner.$name = value;
250 self
251 }
252 )*
253 };
254}
255
256impl SessionBuilder {
257 builder_setters! {
258 state: SessionState,
259 ttl_expiry: i64,
260 ttl_ms: i64,
261 started_at_unix_ms: i64,
262 resolution: Option<Vec<u8>>,
263 mode_state: Vec<u8>,
264 participants: Vec<String>,
265 seen_message_ids: HashSet<String>,
266 extensions: HashMap<String, Vec<u8>>,
267 roots: Vec<macp_pb::pb::Root>,
268 participant_message_counts: HashMap<String, u32>,
269 participant_last_seen: HashMap<String, i64>,
270 policy_definition: Option<crate::policy::PolicyDefinition>,
271 suspended_at_ms: Option<i64>,
272 accumulated_suspended_ms: i64,
273 semantics_rev: u32,
274 max_suspend_ms: i64,
277 }
278
279 pub fn intent(mut self, value: impl Into<String>) -> Self {
280 self.inner.intent = value.into();
281 self
282 }
283
284 pub fn mode_version(mut self, value: impl Into<String>) -> Self {
285 self.inner.mode_version = value.into();
286 self
287 }
288
289 pub fn configuration_version(mut self, value: impl Into<String>) -> Self {
290 self.inner.configuration_version = value.into();
291 self
292 }
293
294 pub fn policy_version(mut self, value: impl Into<String>) -> Self {
295 self.inner.policy_version = value.into();
296 self
297 }
298
299 pub fn context_id(mut self, value: impl Into<String>) -> Self {
300 self.inner.context_id = value.into();
301 self
302 }
303
304 pub fn build(self) -> Session {
305 self.inner
306 }
307}
308
309pub fn requires_strict_session_start(mode: &str) -> bool {
310 matches!(
311 mode,
312 "macp.mode.decision.v1"
313 | "macp.mode.proposal.v1"
314 | "macp.mode.task.v1"
315 | "macp.mode.handoff.v1"
316 | "macp.mode.quorum.v1"
317 | "ext.multi_round.v1"
318 )
319}
320
321pub fn parse_session_start_payload(payload: &[u8]) -> Result<SessionStartPayload, MacpError> {
323 if payload.is_empty() {
324 return Err(MacpError::InvalidPayload);
325 }
326 SessionStartPayload::decode(payload).map_err(|_| MacpError::InvalidPayload)
327}
328
329pub fn extract_ttl_ms(payload: &SessionStartPayload) -> Result<i64, MacpError> {
331 if !(1..=MAX_TTL_MS).contains(&payload.ttl_ms) {
332 return Err(MacpError::InvalidTtl);
333 }
334 Ok(payload.ttl_ms)
335}
336
337pub fn validate_canonical_session_start_payload(
339 payload: &SessionStartPayload,
340) -> Result<(), MacpError> {
341 extract_ttl_ms(payload)?;
342
343 if payload.mode_version.trim().is_empty() || payload.configuration_version.trim().is_empty() {
344 return Err(MacpError::InvalidPayload);
345 }
346
347 if payload.participants.is_empty() {
348 return Err(MacpError::InvalidPayload);
349 }
350
351 const MAX_PARTICIPANTS: usize = 1000;
353 if payload.participants.len() > MAX_PARTICIPANTS {
354 return Err(MacpError::InvalidPayload);
355 }
356
357 let mut seen = HashSet::new();
358 for participant in &payload.participants {
359 let participant = participant.trim();
360 if participant.is_empty() || !seen.insert(participant.to_string()) {
361 return Err(MacpError::InvalidPayload);
362 }
363 }
364
365 if payload.max_suspend_ms < 0 {
368 return Err(MacpError::InvalidPayload);
369 }
370
371 Ok(())
372}
373
374pub fn validate_strict_session_start_payload(
376 mode: &str,
377 payload: &SessionStartPayload,
378) -> Result<(), MacpError> {
379 if !requires_strict_session_start(mode) {
380 return Ok(());
381 }
382
383 validate_canonical_session_start_payload(payload)
384}
385
386pub fn validate_session_id_for_acceptance(session_id: &str) -> Result<(), MacpError> {
394 if session_id.is_empty() {
395 return Err(MacpError::InvalidSessionId);
396 }
397
398 if session_id.len() == 36 && session_id.contains('-') {
404 if let Ok(parsed) = uuid::Uuid::parse_str(session_id) {
405 if parsed.as_hyphenated().to_string() == session_id {
406 match parsed.get_version() {
407 Some(uuid::Version::Random) | Some(uuid::Version::SortRand) => {
408 return Ok(());
409 }
410 _ => {}
411 }
412 }
413 return Err(MacpError::InvalidSessionId);
414 }
415 }
416
417 if session_id.len() >= 22
419 && session_id
420 .chars()
421 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
422 {
423 return Ok(());
424 }
425
426 Err(MacpError::InvalidSessionId)
427}
428
429#[cfg(test)]
430mod tests {
431 use super::*;
432 use prost::Message;
433
434 fn encode_payload(ttl_ms: i64, participants: Vec<String>) -> Vec<u8> {
435 let payload = SessionStartPayload {
436 intent: String::new(),
437 participants,
438 mode_version: "1.0.0".into(),
439 configuration_version: "cfg-1".into(),
440 policy_version: String::new(),
441 ttl_ms,
442 context_id: String::new(),
443 extensions: std::collections::HashMap::new(),
444 roots: vec![],
445 max_suspend_ms: 0,
446 };
447 payload.encode_to_vec()
448 }
449
450 #[test]
451 fn parse_empty_payload_is_invalid() {
452 let err = parse_session_start_payload(b"").unwrap_err();
453 assert_eq!(err.to_string(), "InvalidPayload");
454 }
455
456 #[test]
457 fn parse_valid_protobuf_payload() {
458 let bytes = encode_payload(5000, vec!["alice".into(), "bob".into()]);
459 let result = parse_session_start_payload(&bytes).unwrap();
460 assert_eq!(result.ttl_ms, 5000);
461 assert_eq!(result.participants, vec!["alice", "bob"]);
462 }
463
464 #[test]
465 fn extract_ttl_requires_explicit_positive_value() {
466 let payload = SessionStartPayload::default();
467 assert_eq!(
468 extract_ttl_ms(&payload).unwrap_err().to_string(),
469 "InvalidTtl"
470 );
471
472 let payload = SessionStartPayload {
473 ttl_ms: 5000,
474 ..Default::default()
475 };
476 assert_eq!(extract_ttl_ms(&payload).unwrap(), 5000);
477 }
478
479 #[test]
480 fn standard_mode_requires_explicit_versions_and_participants() {
481 let payload = SessionStartPayload {
482 participants: vec!["alice".into()],
483 mode_version: String::new(),
484 configuration_version: "cfg-1".into(),
485 ttl_ms: 1000,
486 ..Default::default()
487 };
488 assert_eq!(
489 validate_strict_session_start_payload("macp.mode.decision.v1", &payload)
490 .unwrap_err()
491 .to_string(),
492 "InvalidPayload"
493 );
494
495 let payload = SessionStartPayload {
496 participants: vec![],
497 mode_version: "1.0.0".into(),
498 configuration_version: "cfg-1".into(),
499 ttl_ms: 1000,
500 ..Default::default()
501 };
502 assert_eq!(
503 validate_strict_session_start_payload("macp.mode.decision.v1", &payload)
504 .unwrap_err()
505 .to_string(),
506 "InvalidPayload"
507 );
508 }
509
510 fn open_session(ttl_expiry: i64) -> Session {
511 Session {
512 session_id: "s1".into(),
513 state: SessionState::Open,
514 ttl_expiry,
515 ttl_ms: 60_000,
516 started_at_unix_ms: 0,
517 resolution: None,
518 mode: "macp.mode.decision.v1".into(),
519 mode_state: vec![],
520 participants: vec![],
521 seen_message_ids: HashSet::new(),
522 intent: String::new(),
523 mode_version: "1.0.0".into(),
524 configuration_version: "cfg-1".into(),
525 policy_version: String::new(),
526 context_id: String::new(),
527 extensions: HashMap::new(),
528 roots: vec![],
529 initiator_sender: "agent://a".into(),
530 participant_message_counts: HashMap::new(),
531 participant_last_seen: HashMap::new(),
532 policy_definition: None,
533 suspended_at_ms: None,
534 accumulated_suspended_ms: 0,
535 semantics_rev: CURRENT_SEMANTICS_REV,
536 max_suspend_ms: 0,
537 }
538 }
539
540 #[test]
541 fn suspend_then_resume_banks_ttl() {
542 let mut s = open_session(10_000);
543 s.suspend(2_000).unwrap();
544 assert_eq!(s.state, SessionState::Suspended);
545 assert_eq!(s.suspended_at_ms, Some(2_000));
546 s.resume(5_000).unwrap();
548 assert_eq!(s.state, SessionState::Open);
549 assert_eq!(s.ttl_expiry, 13_000);
550 assert_eq!(s.accumulated_suspended_ms, 3_000);
551 assert_eq!(s.suspended_at_ms, None);
552 }
553
554 #[test]
555 fn suspend_requires_open_and_resume_requires_suspended() {
556 let mut s = open_session(10_000);
557 assert!(matches!(
559 s.resume(1).unwrap_err(),
560 MacpError::SessionNotOpen
561 ));
562 s.suspend(1).unwrap();
563 assert!(matches!(
565 s.suspend(2).unwrap_err(),
566 MacpError::SessionNotOpen
567 ));
568 }
569
570 #[test]
571 fn resume_exceeding_max_suspend_expires() {
572 let mut s = open_session(10_000);
573 s.suspend(0).unwrap();
574 let err = s.resume(MAX_SUSPEND_MS + 1).unwrap_err();
576 assert!(matches!(err, MacpError::TtlExpired));
577 assert_eq!(s.state, SessionState::Expired);
578 }
579
580 #[test]
584 fn bound_cap_overrides_default_on_resume() {
585 let mut s = open_session(10_000);
586 s.max_suspend_ms = 500;
587 s.suspend(0).unwrap();
588 let err = s.resume(501).unwrap_err();
589 assert!(matches!(err, MacpError::TtlExpired));
590 assert_eq!(s.state, SessionState::Expired);
591 }
592
593 #[test]
594 fn bound_cap_within_limit_resumes_and_banks_ttl() {
595 let mut s = open_session(10_000);
596 s.max_suspend_ms = 500;
597 s.suspend(0).unwrap();
598 s.resume(400).unwrap();
599 assert_eq!(s.state, SessionState::Open);
600 assert_eq!(s.ttl_expiry, 10_400);
601 }
602
603 #[test]
604 fn suspend_cap_exceeded_uses_bound_cap() {
605 let mut s = open_session(10_000);
606 s.max_suspend_ms = 500;
607 s.suspend(0).unwrap();
608 assert!(!s.suspend_cap_exceeded(400));
609 assert!(s.suspend_cap_exceeded(501));
610 }
611
612 #[test]
613 fn unbound_session_uses_default_cap() {
614 let s = open_session(10_000);
615 assert_eq!(s.max_suspend_ms, 0);
616 assert_eq!(s.effective_max_suspend_ms(), MAX_SUSPEND_MS);
617 }
618
619 #[test]
620 fn negative_max_suspend_ms_rejected_in_canonical_payload() {
621 let payload = SessionStartPayload {
622 participants: vec!["a".into()],
623 mode_version: "1.0.0".into(),
624 configuration_version: "cfg-1".into(),
625 ttl_ms: 60_000,
626 max_suspend_ms: -1,
627 ..Default::default()
628 };
629 assert_eq!(
630 validate_canonical_session_start_payload(&payload)
631 .unwrap_err()
632 .to_string(),
633 "InvalidPayload"
634 );
635 let ok0 = SessionStartPayload {
637 max_suspend_ms: 0,
638 ..payload.clone()
639 };
640 validate_canonical_session_start_payload(&ok0).unwrap();
641 let ok_pos = SessionStartPayload {
642 max_suspend_ms: 60_000,
643 ..payload
644 };
645 validate_canonical_session_start_payload(&ok_pos).unwrap();
646 }
647
648 #[test]
649 fn cancel_from_open_or_suspended_then_terminal_is_rejected() {
650 let mut s = open_session(10_000);
651 s.suspend(1).unwrap();
652 s.cancel().unwrap();
653 assert_eq!(s.state, SessionState::Cancelled);
654 assert_eq!(s.suspended_at_ms, None);
655 assert!(matches!(s.cancel().unwrap_err(), MacpError::SessionNotOpen));
657
658 let mut open = open_session(10_000);
659 open.cancel().unwrap();
660 assert_eq!(open.state, SessionState::Cancelled);
661 }
662
663 #[test]
664 fn standard_mode_rejects_duplicate_participants() {
665 let payload = SessionStartPayload {
666 participants: vec!["alice".into(), "alice".into()],
667 mode_version: "1.0.0".into(),
668 configuration_version: "cfg-1".into(),
669 ttl_ms: 1000,
670 ..Default::default()
671 };
672 assert_eq!(
673 validate_strict_session_start_payload("macp.mode.proposal.v1", &payload)
674 .unwrap_err()
675 .to_string(),
676 "InvalidPayload"
677 );
678 }
679
680 #[test]
681 fn multi_round_requires_strict_session_start() {
682 let payload = SessionStartPayload::default();
683 assert!(validate_strict_session_start_payload("ext.multi_round.v1", &payload).is_err());
684 }
685
686 #[test]
687 fn valid_uuid_v4_accepted() {
688 let id = uuid::Uuid::new_v4().as_hyphenated().to_string();
689 validate_session_id_for_acceptance(&id).unwrap();
690 }
691
692 #[test]
693 fn valid_base64url_accepted() {
694 validate_session_id_for_acceptance("abcdefghijklmnopqrstuv").unwrap();
696 validate_session_id_for_acceptance("abc-def_ghi-jkl_mno-pqr").unwrap();
698 }
699
700 #[test]
701 fn empty_id_rejected() {
702 assert_eq!(
703 validate_session_id_for_acceptance("")
704 .unwrap_err()
705 .to_string(),
706 "InvalidSessionId"
707 );
708 }
709
710 #[test]
711 fn short_weak_id_rejected() {
712 assert_eq!(
713 validate_session_id_for_acceptance("s1")
714 .unwrap_err()
715 .to_string(),
716 "InvalidSessionId"
717 );
718 assert_eq!(
719 validate_session_id_for_acceptance("decision-demo-1")
720 .unwrap_err()
721 .to_string(),
722 "InvalidSessionId"
723 );
724 }
725
726 #[test]
727 fn uppercase_uuid_rejected() {
728 let id = uuid::Uuid::new_v4()
729 .as_hyphenated()
730 .to_string()
731 .to_uppercase();
732 assert_eq!(
733 validate_session_id_for_acceptance(&id)
734 .unwrap_err()
735 .to_string(),
736 "InvalidSessionId"
737 );
738 }
739
740 #[test]
741 fn base64url_36_chars_with_hyphen_accepted() {
742 let id = "Zx-abcdefghijklmnopqrstuvwxyz_ABCDE-";
747 assert_eq!(id.len(), 36);
748 assert!(uuid::Uuid::parse_str(id).is_err());
749 validate_session_id_for_acceptance(id).unwrap();
750 }
751
752 #[test]
753 fn uuid_shaped_but_wrong_version_does_not_fall_through() {
754 let v4 = uuid::Uuid::new_v4();
757 let mut bytes = *v4.as_bytes();
758 bytes[6] = (bytes[6] & 0x0F) | 0x10;
759 bytes[8] = (bytes[8] & 0x3F) | 0x80;
760 let v1_id = uuid::Uuid::from_bytes(bytes).as_hyphenated().to_string();
761 assert!(validate_session_id_for_acceptance(&v1_id).is_err());
762 }
763
764 #[test]
765 fn base64url_too_short_rejected() {
766 assert_eq!(
767 validate_session_id_for_acceptance("abcdefghij")
768 .unwrap_err()
769 .to_string(),
770 "InvalidSessionId"
771 );
772 }
773
774 #[test]
775 fn valid_uuid_v7_accepted() {
776 let v4 = uuid::Uuid::new_v4();
778 let mut bytes = *v4.as_bytes();
779 bytes[6] = (bytes[6] & 0x0F) | 0x70;
781 bytes[8] = (bytes[8] & 0x3F) | 0x80;
783 let v7_id = uuid::Uuid::from_bytes(bytes).as_hyphenated().to_string();
784 assert!(validate_session_id_for_acceptance(&v7_id).is_ok());
785 }
786
787 #[test]
788 fn uuid_v1_rejected() {
789 let v4 = uuid::Uuid::new_v4();
791 let mut bytes = *v4.as_bytes();
792 bytes[6] = (bytes[6] & 0x0F) | 0x10;
794 bytes[8] = (bytes[8] & 0x3F) | 0x80;
796 let v1_id = uuid::Uuid::from_bytes(bytes).as_hyphenated().to_string();
797 assert_eq!(
798 validate_session_id_for_acceptance(&v1_id)
799 .unwrap_err()
800 .to_string(),
801 "InvalidSessionId"
802 );
803 }
804
805 #[test]
806 fn too_many_participants_rejected() {
807 let participants: Vec<String> = (0..1001).map(|i| format!("agent://p{i}")).collect();
808 let bytes = encode_payload(5000, participants);
809 let payload = parse_session_start_payload(&bytes).unwrap();
810 assert_eq!(
811 validate_canonical_session_start_payload(&payload)
812 .unwrap_err()
813 .to_string(),
814 "InvalidPayload"
815 );
816 }
817
818 #[test]
819 fn max_participants_accepted() {
820 let participants: Vec<String> = (0..1000).map(|i| format!("agent://p{i}")).collect();
821 let bytes = encode_payload(5000, participants);
822 let payload = parse_session_start_payload(&bytes).unwrap();
823 validate_canonical_session_start_payload(&payload).unwrap();
824 }
825}