1pub mod memory;
7#[cfg(feature = "sqlite-store")]
8pub mod sqlite;
9
10use meerkat_core::lifecycle::{InputId, RunBoundaryReceipt, RunId};
11
12use crate::identifiers::LogicalRuntimeId;
13use crate::input_state::{InputStatePersistenceRecord, StoredInputState};
14use crate::runtime_state::RuntimeState;
15
16const LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 1;
17const MACHINE_LIFECYCLE_STORE_RECORD_VERSION: u16 = 2;
18
19#[derive(Debug, Clone, thiserror::Error)]
21#[non_exhaustive]
22pub enum RuntimeStoreError {
23 #[error("Store write failed: {0}")]
25 WriteFailed(String),
26 #[error("Store read failed: {0}")]
28 ReadFailed(String),
29 #[error("Session store key mismatch: expected {expected}, actual {actual}")]
31 SessionKeyMismatch {
32 expected: meerkat_core::types::SessionId,
33 actual: meerkat_core::types::SessionId,
34 },
35 #[error("Not found: {0}")]
37 NotFound(String),
38 #[error("Unsupported store operation: {0}")]
40 Unsupported(String),
41 #[error("Transcript revision conflict: expected {expected}, actual {actual}")]
43 TranscriptRevisionConflict { expected: String, actual: String },
44 #[error("Internal error: {0}")]
46 Internal(String),
47}
48
49pub type AuthOAuthFlowSnapshotUpdate<'a> =
51 dyn FnMut(Option<&[u8]>) -> Result<Vec<u8>, RuntimeStoreError> + 'a;
52
53#[derive(Debug, Clone)]
55pub struct SessionDelta {
56 pub session_snapshot: Vec<u8>,
58}
59
60#[derive(Debug, Clone, Default, PartialEq, Eq)]
67pub struct MachineLifecycleBindingFacts {
68 agent_runtime_id: Option<String>,
69 fence_token: Option<u64>,
70 runtime_generation: Option<u64>,
71 runtime_epoch_id: Option<String>,
72}
73
74#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct RevokedSupervisorReceipt {
82 peer_id: String,
83 signing_public_key: String,
84 epoch: u64,
85}
86
87#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct SupervisorBindingReceipt {
91 name: String,
92 peer_id: String,
93 address: String,
94 signing_public_key: String,
95 epoch: u64,
96}
97
98#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct SupervisorRevocationPendingReceipt {
106 name: String,
107 peer_id: String,
108 address: String,
109 signing_public_key: String,
110 epoch: u64,
111}
112
113#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
114#[serde(rename_all = "snake_case")]
115pub enum SupervisorRotationPersistencePhase {
116 PreviousRevokePending,
117 NextPublishPending,
118 Completed,
119 Rejected,
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
123#[serde(rename_all = "snake_case")]
124pub enum SupervisorRotationRejection {
125 OperationConflict,
126 NotBound,
127 SenderMismatch,
128 TargetEpochNotAdvanced,
129 InvalidTarget,
130 UnsupportedProtocolVersion,
131}
132
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct SupervisorRotationReceipt {
135 operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
136 phase: SupervisorRotationPersistencePhase,
137 rejection: Option<SupervisorRotationRejection>,
138 previous: SupervisorBindingReceipt,
139 next: SupervisorBindingReceipt,
140}
141
142impl SupervisorRotationReceipt {
143 pub(crate) fn new(
144 operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
145 phase: SupervisorRotationPersistencePhase,
146 rejection: Option<SupervisorRotationRejection>,
147 previous: SupervisorBindingReceipt,
148 next: SupervisorBindingReceipt,
149 ) -> Self {
150 Self {
151 operation_id,
152 phase,
153 rejection,
154 previous,
155 next,
156 }
157 }
158
159 pub fn operation_id(
160 &self,
161 ) -> meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId {
162 self.operation_id
163 }
164
165 pub fn phase(&self) -> SupervisorRotationPersistencePhase {
166 self.phase
167 }
168
169 pub fn rejection(&self) -> Option<SupervisorRotationRejection> {
170 self.rejection
171 }
172
173 pub fn previous(&self) -> &SupervisorBindingReceipt {
174 &self.previous
175 }
176
177 pub fn next(&self) -> &SupervisorBindingReceipt {
178 &self.next
179 }
180}
181
182impl SupervisorBindingReceipt {
183 pub(crate) fn new(
184 name: String,
185 peer_id: String,
186 address: String,
187 signing_public_key: String,
188 epoch: u64,
189 ) -> Self {
190 Self {
191 name,
192 peer_id,
193 address,
194 signing_public_key,
195 epoch,
196 }
197 }
198
199 pub fn name(&self) -> &str {
200 &self.name
201 }
202
203 pub fn peer_id(&self) -> &str {
204 &self.peer_id
205 }
206
207 pub fn address(&self) -> &str {
208 &self.address
209 }
210
211 pub fn signing_public_key(&self) -> &str {
212 &self.signing_public_key
213 }
214
215 pub fn epoch(&self) -> u64 {
216 self.epoch
217 }
218}
219
220impl RevokedSupervisorReceipt {
221 pub(crate) fn new(peer_id: String, signing_public_key: String, epoch: u64) -> Self {
222 Self {
223 peer_id,
224 signing_public_key,
225 epoch,
226 }
227 }
228
229 pub fn peer_id(&self) -> &str {
230 &self.peer_id
231 }
232
233 pub fn signing_public_key(&self) -> &str {
234 &self.signing_public_key
235 }
236
237 pub fn epoch(&self) -> u64 {
238 self.epoch
239 }
240}
241
242impl SupervisorRevocationPendingReceipt {
243 pub(crate) fn new(
244 name: String,
245 peer_id: String,
246 address: String,
247 signing_public_key: String,
248 epoch: u64,
249 ) -> Self {
250 Self {
251 name,
252 peer_id,
253 address,
254 signing_public_key,
255 epoch,
256 }
257 }
258
259 pub fn name(&self) -> &str {
260 &self.name
261 }
262
263 pub fn peer_id(&self) -> &str {
264 &self.peer_id
265 }
266
267 pub fn address(&self) -> &str {
268 &self.address
269 }
270
271 pub fn signing_public_key(&self) -> &str {
272 &self.signing_public_key
273 }
274
275 pub fn epoch(&self) -> u64 {
276 self.epoch
277 }
278}
279
280#[derive(Debug, Clone, Default, PartialEq, Eq)]
284pub enum SupervisorAuthoritySnapshot {
285 #[default]
286 UnboundNoReceipt,
287 Bound(SupervisorBindingReceipt),
288 RevocationPending(SupervisorRevocationPendingReceipt),
289 RotationOperation(SupervisorRotationReceipt),
290 RevokedReceipt(RevokedSupervisorReceipt),
291 WithRotationHistory {
292 current: Box<SupervisorAuthoritySnapshot>,
293 terminal_receipts: std::collections::BTreeMap<
294 meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
295 SupervisorRotationReceipt,
296 >,
297 },
298}
299
300impl MachineLifecycleBindingFacts {
301 pub(crate) fn new(
302 agent_runtime_id: Option<String>,
303 fence_token: Option<u64>,
304 runtime_generation: Option<u64>,
305 runtime_epoch_id: Option<String>,
306 ) -> Self {
307 Self {
308 agent_runtime_id,
309 fence_token,
310 runtime_generation,
311 runtime_epoch_id,
312 }
313 }
314
315 pub fn agent_runtime_id(&self) -> Option<&str> {
316 self.agent_runtime_id.as_deref()
317 }
318
319 pub fn fence_token(&self) -> Option<u64> {
320 self.fence_token
321 }
322
323 pub fn runtime_generation(&self) -> Option<u64> {
324 self.runtime_generation
325 }
326
327 pub fn runtime_epoch_id(&self) -> Option<&str> {
328 self.runtime_epoch_id.as_deref()
329 }
330}
331
332#[derive(Debug, Clone, PartialEq, Eq)]
334pub struct MachineLifecycleSnapshot {
335 runtime_state: RuntimeState,
336 binding: MachineLifecycleBindingFacts,
337 supervisor_authority: SupervisorAuthoritySnapshot,
338}
339
340impl MachineLifecycleSnapshot {
341 pub(crate) fn new(
342 runtime_state: RuntimeState,
343 binding: MachineLifecycleBindingFacts,
344 supervisor_authority: SupervisorAuthoritySnapshot,
345 ) -> Self {
346 Self {
347 runtime_state,
348 binding,
349 supervisor_authority,
350 }
351 }
352
353 pub fn runtime_state(&self) -> RuntimeState {
355 self.runtime_state
356 }
357
358 pub fn binding(&self) -> &MachineLifecycleBindingFacts {
360 &self.binding
361 }
362
363 pub fn supervisor_authority(&self) -> &SupervisorAuthoritySnapshot {
364 &self.supervisor_authority
365 }
366}
367
368#[allow(
369 clippy::option_option,
370 reason = "serde distinguishes missing from explicit null"
371)]
372fn deserialize_present_nullable<'de, D, T>(deserializer: D) -> Result<Option<Option<T>>, D::Error>
373where
374 D: serde::Deserializer<'de>,
375 T: serde::Deserialize<'de>,
376{
377 <Option<T> as serde::Deserialize>::deserialize(deserializer).map(Some)
378}
379
380#[allow(
381 clippy::option_option,
382 reason = "serde distinguishes missing from explicit null"
383)]
384fn require_present_nullable<T>(
385 value: Option<Option<T>>,
386 field: &str,
387) -> Result<Option<T>, RuntimeStoreError> {
388 value.ok_or_else(|| {
389 RuntimeStoreError::ReadFailed(format!(
390 "machine lifecycle v2 field {field} is required (explicit null is allowed)"
391 ))
392 })
393}
394
395#[derive(serde::Serialize, serde::Deserialize)]
396#[serde(deny_unknown_fields)]
397struct MachineLifecycleBindingFactsStoreWire {
398 #[allow(
399 clippy::option_option,
400 reason = "serde distinguishes missing from explicit null"
401 )]
402 #[serde(default, deserialize_with = "deserialize_present_nullable")]
403 agent_runtime_id: Option<Option<String>>,
404 #[allow(
405 clippy::option_option,
406 reason = "serde distinguishes missing from explicit null"
407 )]
408 #[serde(default, deserialize_with = "deserialize_present_nullable")]
409 fence_token: Option<Option<u64>>,
410 #[allow(
411 clippy::option_option,
412 reason = "serde distinguishes missing from explicit null"
413 )]
414 #[serde(default, deserialize_with = "deserialize_present_nullable")]
415 runtime_generation: Option<Option<u64>>,
416 #[allow(
417 clippy::option_option,
418 reason = "serde distinguishes missing from explicit null"
419 )]
420 #[serde(default, deserialize_with = "deserialize_present_nullable")]
421 runtime_epoch_id: Option<Option<String>>,
422}
423
424#[derive(serde::Deserialize)]
425#[serde(deny_unknown_fields)]
426struct MachineLifecycleBindingFactsStoreWireV1 {
427 agent_runtime_id: Option<String>,
428 fence_token: Option<u64>,
429 runtime_generation: Option<u64>,
430 runtime_epoch_id: Option<String>,
431}
432
433impl From<&MachineLifecycleBindingFacts> for MachineLifecycleBindingFactsStoreWire {
434 fn from(binding: &MachineLifecycleBindingFacts) -> Self {
435 Self {
436 agent_runtime_id: Some(binding.agent_runtime_id().map(ToOwned::to_owned)),
437 fence_token: Some(binding.fence_token()),
438 runtime_generation: Some(binding.runtime_generation()),
439 runtime_epoch_id: Some(binding.runtime_epoch_id().map(ToOwned::to_owned)),
440 }
441 }
442}
443
444impl TryFrom<MachineLifecycleBindingFactsStoreWire> for MachineLifecycleBindingFacts {
445 type Error = RuntimeStoreError;
446
447 fn try_from(binding: MachineLifecycleBindingFactsStoreWire) -> Result<Self, Self::Error> {
448 Ok(Self::new(
449 require_present_nullable(binding.agent_runtime_id, "binding.agent_runtime_id")?,
450 require_present_nullable(binding.fence_token, "binding.fence_token")?,
451 require_present_nullable(binding.runtime_generation, "binding.runtime_generation")?,
452 require_present_nullable(binding.runtime_epoch_id, "binding.runtime_epoch_id")?,
453 ))
454 }
455}
456
457impl From<MachineLifecycleBindingFactsStoreWireV1> for MachineLifecycleBindingFacts {
458 fn from(binding: MachineLifecycleBindingFactsStoreWireV1) -> Self {
459 Self::new(
460 binding.agent_runtime_id,
461 binding.fence_token,
462 binding.runtime_generation,
463 binding.runtime_epoch_id,
464 )
465 }
466}
467
468#[derive(serde::Serialize, serde::Deserialize)]
469#[serde(deny_unknown_fields)]
470struct MachineLifecycleSnapshotStoreWire {
471 record_version: u16,
472 runtime_state: RuntimeState,
473 binding: MachineLifecycleBindingFactsStoreWire,
474 supervisor_authority: SupervisorAuthoritySnapshotStoreWire,
475}
476
477#[derive(serde::Deserialize)]
481#[serde(deny_unknown_fields)]
482struct MachineLifecycleSnapshotStoreWireV1 {
483 record_version: u16,
484 runtime_state: RuntimeState,
485 binding: MachineLifecycleBindingFactsStoreWireV1,
486}
487
488#[derive(serde::Deserialize)]
489struct MachineLifecycleSnapshotStoreVersionProbe {
490 record_version: u16,
491}
492
493#[derive(Default, serde::Serialize, serde::Deserialize)]
494#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
495enum SupervisorAuthoritySnapshotStoreWire {
496 #[default]
497 UnboundNoReceipt,
498 Bound {
499 binding: SupervisorBindingReceiptStoreWire,
500 },
501 RevocationPending {
502 pending: SupervisorRevocationPendingReceiptStoreWire,
503 },
504 RotationOperation {
505 rotation: SupervisorRotationReceiptStoreWire,
506 },
507 RevokedReceipt {
508 receipt: RevokedSupervisorReceiptStoreWire,
509 },
510 WithRotationHistory {
511 current: Box<SupervisorAuthoritySnapshotStoreWire>,
512 terminal_receipts: Vec<SupervisorRotationReceiptStoreWire>,
513 },
514}
515
516#[derive(serde::Serialize, serde::Deserialize)]
517#[serde(deny_unknown_fields)]
518struct SupervisorBindingReceiptStoreWire {
519 name: String,
520 peer_id: String,
521 address: String,
522 signing_public_key: String,
523 epoch: u64,
524}
525
526impl From<&SupervisorBindingReceipt> for SupervisorBindingReceiptStoreWire {
527 fn from(receipt: &SupervisorBindingReceipt) -> Self {
528 Self {
529 name: receipt.name().to_owned(),
530 peer_id: receipt.peer_id().to_owned(),
531 address: receipt.address().to_owned(),
532 signing_public_key: receipt.signing_public_key().to_owned(),
533 epoch: receipt.epoch(),
534 }
535 }
536}
537
538impl From<SupervisorBindingReceiptStoreWire> for SupervisorBindingReceipt {
539 fn from(receipt: SupervisorBindingReceiptStoreWire) -> Self {
540 Self::new(
541 receipt.name,
542 receipt.peer_id,
543 receipt.address,
544 receipt.signing_public_key,
545 receipt.epoch,
546 )
547 }
548}
549
550#[derive(serde::Serialize, serde::Deserialize)]
551#[serde(deny_unknown_fields)]
552struct RevokedSupervisorReceiptStoreWire {
553 peer_id: String,
554 signing_public_key: String,
555 epoch: u64,
556}
557
558#[derive(serde::Serialize, serde::Deserialize)]
559#[serde(deny_unknown_fields)]
560struct SupervisorRevocationPendingReceiptStoreWire {
561 name: String,
562 peer_id: String,
563 address: String,
564 signing_public_key: String,
565 epoch: u64,
566}
567
568#[derive(serde::Serialize, serde::Deserialize)]
569#[serde(deny_unknown_fields)]
570struct SupervisorRotationReceiptStoreWire {
571 operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
572 phase: SupervisorRotationPersistencePhase,
573 #[allow(
574 clippy::option_option,
575 reason = "serde distinguishes missing from explicit null"
576 )]
577 #[serde(default, deserialize_with = "deserialize_present_nullable")]
578 rejection: Option<Option<SupervisorRotationRejection>>,
579 previous: SupervisorBindingReceiptStoreWire,
580 next: SupervisorBindingReceiptStoreWire,
581}
582
583impl From<&SupervisorRotationReceipt> for SupervisorRotationReceiptStoreWire {
584 fn from(receipt: &SupervisorRotationReceipt) -> Self {
585 Self {
586 operation_id: receipt.operation_id(),
587 phase: receipt.phase(),
588 rejection: Some(receipt.rejection()),
589 previous: receipt.previous().into(),
590 next: receipt.next().into(),
591 }
592 }
593}
594
595impl TryFrom<SupervisorRotationReceiptStoreWire> for SupervisorRotationReceipt {
596 type Error = RuntimeStoreError;
597
598 fn try_from(receipt: SupervisorRotationReceiptStoreWire) -> Result<Self, Self::Error> {
599 Ok(Self::new(
600 receipt.operation_id,
601 receipt.phase,
602 require_present_nullable(receipt.rejection, "supervisor_authority.rotation.rejection")?,
603 receipt.previous.into(),
604 receipt.next.into(),
605 ))
606 }
607}
608
609impl From<&SupervisorRevocationPendingReceipt> for SupervisorRevocationPendingReceiptStoreWire {
610 fn from(receipt: &SupervisorRevocationPendingReceipt) -> Self {
611 Self {
612 name: receipt.name().to_owned(),
613 peer_id: receipt.peer_id().to_owned(),
614 address: receipt.address().to_owned(),
615 signing_public_key: receipt.signing_public_key().to_owned(),
616 epoch: receipt.epoch(),
617 }
618 }
619}
620
621impl From<SupervisorRevocationPendingReceiptStoreWire> for SupervisorRevocationPendingReceipt {
622 fn from(receipt: SupervisorRevocationPendingReceiptStoreWire) -> Self {
623 Self::new(
624 receipt.name,
625 receipt.peer_id,
626 receipt.address,
627 receipt.signing_public_key,
628 receipt.epoch,
629 )
630 }
631}
632
633impl From<&RevokedSupervisorReceipt> for RevokedSupervisorReceiptStoreWire {
634 fn from(receipt: &RevokedSupervisorReceipt) -> Self {
635 Self {
636 peer_id: receipt.peer_id().to_owned(),
637 signing_public_key: receipt.signing_public_key().to_owned(),
638 epoch: receipt.epoch(),
639 }
640 }
641}
642
643impl From<RevokedSupervisorReceiptStoreWire> for RevokedSupervisorReceipt {
644 fn from(receipt: RevokedSupervisorReceiptStoreWire) -> Self {
645 Self::new(receipt.peer_id, receipt.signing_public_key, receipt.epoch)
646 }
647}
648
649impl From<&SupervisorAuthoritySnapshot> for SupervisorAuthoritySnapshotStoreWire {
650 fn from(snapshot: &SupervisorAuthoritySnapshot) -> Self {
651 match snapshot {
652 SupervisorAuthoritySnapshot::UnboundNoReceipt => Self::UnboundNoReceipt,
653 SupervisorAuthoritySnapshot::Bound(binding) => Self::Bound {
654 binding: binding.into(),
655 },
656 SupervisorAuthoritySnapshot::RevocationPending(pending) => Self::RevocationPending {
657 pending: pending.into(),
658 },
659 SupervisorAuthoritySnapshot::RotationOperation(rotation) => Self::RotationOperation {
660 rotation: rotation.into(),
661 },
662 SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => Self::RevokedReceipt {
663 receipt: receipt.into(),
664 },
665 SupervisorAuthoritySnapshot::WithRotationHistory {
666 current,
667 terminal_receipts,
668 } => Self::WithRotationHistory {
669 current: Box::new(current.as_ref().into()),
670 terminal_receipts: terminal_receipts.values().map(Into::into).collect(),
671 },
672 }
673 }
674}
675
676fn supervisor_authority_read_error(
677 context: &str,
678 detail: impl std::fmt::Display,
679) -> RuntimeStoreError {
680 RuntimeStoreError::ReadFailed(format!("{context}: {detail}"))
681}
682
683fn validate_supervisor_descriptor(
684 name: &str,
685 peer_id: &str,
686 address: &str,
687 signing_public_key: &str,
688 context: &str,
689) -> Result<(), RuntimeStoreError> {
690 let pubkey = crate::comms_drain::decode_supervisor_signing_public_key(signing_public_key)
691 .map_err(|error| supervisor_authority_read_error(context, error))?;
692 let spec = meerkat_contracts::wire::supervisor_bridge::BridgePeerSpec {
693 name: name.to_owned(),
694 peer_id: peer_id.to_owned(),
695 address: address.to_owned(),
696 pubkey,
697 };
698 meerkat_core::comms::TrustedPeerDescriptor::try_from(&spec)
699 .map(|_| ())
700 .map_err(|error| supervisor_authority_read_error(context, error))
701}
702
703fn validate_supervisor_binding_receipt(
704 receipt: &SupervisorBindingReceipt,
705 context: &str,
706) -> Result<(), RuntimeStoreError> {
707 validate_supervisor_descriptor(
708 receipt.name(),
709 receipt.peer_id(),
710 receipt.address(),
711 receipt.signing_public_key(),
712 context,
713 )
714}
715
716fn validate_revoked_supervisor_receipt(
717 receipt: &RevokedSupervisorReceipt,
718 context: &str,
719) -> Result<(), RuntimeStoreError> {
720 let pubkey =
721 crate::comms_drain::decode_supervisor_signing_public_key(receipt.signing_public_key())
722 .map_err(|error| supervisor_authority_read_error(context, error))?;
723 if pubkey.iter().all(|byte| *byte == 0) {
724 return Err(supervisor_authority_read_error(
725 context,
726 "supervisor signing public key must be non-zero",
727 ));
728 }
729 let peer_id = meerkat_core::comms::PeerId::parse(receipt.peer_id())
730 .map_err(|error| supervisor_authority_read_error(context, error))?;
731 let derived = meerkat_core::comms::PeerId::from_ed25519_pubkey(&pubkey);
732 if peer_id != derived {
733 return Err(supervisor_authority_read_error(
734 context,
735 format!("peer id {peer_id} does not match signing-key-derived id {derived}"),
736 ));
737 }
738 Ok(())
739}
740
741fn validate_supervisor_rotation_receipt(
742 receipt: &SupervisorRotationReceipt,
743 terminal_history: bool,
744) -> Result<(), RuntimeStoreError> {
745 let operation_id = receipt.operation_id();
746 if operation_id.as_uuid().is_nil() {
747 return Err(supervisor_authority_read_error(
748 "supervisor rotation operation",
749 "operation id must not be the nil UUID",
750 ));
751 }
752 validate_supervisor_binding_receipt(
753 receipt.previous(),
754 &format!("supervisor rotation {operation_id} previous authority is invalid"),
755 )?;
756
757 let rejection_matches = matches!(
758 (receipt.phase(), receipt.rejection()),
759 (
760 SupervisorRotationPersistencePhase::PreviousRevokePending
761 | SupervisorRotationPersistencePhase::NextPublishPending
762 | SupervisorRotationPersistencePhase::Completed,
763 None
764 ) | (SupervisorRotationPersistencePhase::Rejected, Some(_))
765 );
766 if !rejection_matches {
767 return Err(supervisor_authority_read_error(
768 "supervisor rotation operation",
769 format!("{operation_id} has inconsistent rejection state"),
770 ));
771 }
772 if terminal_history
773 && !matches!(
774 receipt.phase(),
775 SupervisorRotationPersistencePhase::Completed
776 | SupervisorRotationPersistencePhase::Rejected
777 )
778 {
779 return Err(supervisor_authority_read_error(
780 "supervisor rotation history",
781 format!("{operation_id} is not terminal"),
782 ));
783 }
784
785 match receipt.phase() {
786 SupervisorRotationPersistencePhase::PreviousRevokePending
787 | SupervisorRotationPersistencePhase::NextPublishPending => {
788 validate_supervisor_binding_receipt(
789 receipt.next(),
790 &format!("supervisor rotation {operation_id} target is invalid"),
791 )?;
792 if receipt.next().epoch() <= receipt.previous().epoch() {
793 return Err(supervisor_authority_read_error(
794 "supervisor rotation operation",
795 format!(
796 "{operation_id} target epoch {} does not advance previous epoch {}",
797 receipt.next().epoch(),
798 receipt.previous().epoch()
799 ),
800 ));
801 }
802 }
803 SupervisorRotationPersistencePhase::Completed => {
804 validate_supervisor_binding_receipt(
805 receipt.next(),
806 &format!("supervisor rotation {operation_id} target is invalid"),
807 )?;
808 let exact_current_adoption = receipt.previous() == receipt.next();
812 if !exact_current_adoption && receipt.next().epoch() <= receipt.previous().epoch() {
813 return Err(supervisor_authority_read_error(
814 "supervisor rotation operation",
815 format!(
816 "{operation_id} completed target epoch {} does not advance previous epoch {}",
817 receipt.next().epoch(),
818 receipt.previous().epoch()
819 ),
820 ));
821 }
822 }
823 SupervisorRotationPersistencePhase::Rejected => {
824 let Some(rejection) = receipt.rejection() else {
825 return Err(supervisor_authority_read_error(
826 "supervisor rotation operation",
827 format!("{operation_id} rejected without a rejection class"),
828 ));
829 };
830 match rejection {
831 SupervisorRotationRejection::InvalidTarget
832 | SupervisorRotationRejection::UnsupportedProtocolVersion => {
833 }
837 SupervisorRotationRejection::TargetEpochNotAdvanced => {
838 validate_supervisor_binding_receipt(
839 receipt.next(),
840 &format!("supervisor rotation {operation_id} rejected target is invalid"),
841 )?;
842 if receipt.next().epoch() > receipt.previous().epoch() {
843 return Err(supervisor_authority_read_error(
844 "supervisor rotation operation",
845 format!(
846 "{operation_id} rejected as non-advancing but target epoch {} advances previous epoch {}",
847 receipt.next().epoch(),
848 receipt.previous().epoch()
849 ),
850 ));
851 }
852 }
853 SupervisorRotationRejection::OperationConflict
854 | SupervisorRotationRejection::NotBound
855 | SupervisorRotationRejection::SenderMismatch => {
856 return Err(supervisor_authority_read_error(
857 "supervisor rotation operation",
858 format!(
859 "{operation_id} transient rejection {rejection:?} must not be persisted as a durable receipt"
860 ),
861 ));
862 }
863 }
864 }
865 }
866 Ok(())
867}
868
869type SupervisorEpochKeyIndex = std::collections::BTreeMap<u64, [u8; 32]>;
870
871fn record_supervisor_epoch_key(
872 epochs: &mut SupervisorEpochKeyIndex,
873 epoch: u64,
874 signing_public_key: &str,
875 context: &str,
876) -> Result<(), RuntimeStoreError> {
877 let key = crate::comms_drain::decode_supervisor_signing_public_key(signing_public_key)
878 .map_err(|error| supervisor_authority_read_error(context, error))?;
879 if let Some(existing) = epochs.get(&epoch) {
880 if existing != &key {
881 return Err(supervisor_authority_read_error(
882 context,
883 format!("epoch {epoch} is bound to conflicting supervisor signing keys"),
884 ));
885 }
886 } else {
887 epochs.insert(epoch, key);
888 }
889 Ok(())
890}
891
892fn record_supervisor_binding_epoch(
893 epochs: &mut SupervisorEpochKeyIndex,
894 receipt: &SupervisorBindingReceipt,
895 context: &str,
896) -> Result<(), RuntimeStoreError> {
897 record_supervisor_epoch_key(
898 epochs,
899 receipt.epoch(),
900 receipt.signing_public_key(),
901 context,
902 )
903}
904
905fn record_rotation_authoritative_epochs(
906 epochs: &mut SupervisorEpochKeyIndex,
907 receipt: &SupervisorRotationReceipt,
908 context: &str,
909) -> Result<(), RuntimeStoreError> {
910 record_supervisor_binding_epoch(epochs, receipt.previous(), context)?;
911 if matches!(
912 receipt.phase(),
913 SupervisorRotationPersistencePhase::PreviousRevokePending
914 | SupervisorRotationPersistencePhase::NextPublishPending
915 | SupervisorRotationPersistencePhase::Completed
916 ) {
917 record_supervisor_binding_epoch(epochs, receipt.next(), context)?;
918 }
919 Ok(())
920}
921
922fn record_current_authoritative_epochs(
923 epochs: &mut SupervisorEpochKeyIndex,
924 current: &SupervisorAuthoritySnapshot,
925) -> Result<(), RuntimeStoreError> {
926 match current {
927 SupervisorAuthoritySnapshot::UnboundNoReceipt => Ok(()),
928 SupervisorAuthoritySnapshot::Bound(binding) => {
929 record_supervisor_binding_epoch(epochs, binding, "current supervisor authority")
930 }
931 SupervisorAuthoritySnapshot::RevocationPending(pending) => record_supervisor_epoch_key(
932 epochs,
933 pending.epoch(),
934 pending.signing_public_key(),
935 "current pending supervisor revocation authority",
936 ),
937 SupervisorAuthoritySnapshot::RotationOperation(rotation) => {
938 record_rotation_authoritative_epochs(
939 epochs,
940 rotation,
941 "current supervisor rotation authority",
942 )
943 }
944 SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => record_supervisor_epoch_key(
945 epochs,
946 receipt.epoch(),
947 receipt.signing_public_key(),
948 "current revoked supervisor authority",
949 ),
950 SupervisorAuthoritySnapshot::WithRotationHistory { .. } => {
951 Err(RuntimeStoreError::ReadFailed(
952 "nested supervisor rotation history is not canonical".to_string(),
953 ))
954 }
955 }
956}
957
958fn current_supervisor_epoch(current: &SupervisorAuthoritySnapshot) -> Option<u64> {
959 match current {
960 SupervisorAuthoritySnapshot::UnboundNoReceipt => None,
961 SupervisorAuthoritySnapshot::Bound(binding) => Some(binding.epoch()),
962 SupervisorAuthoritySnapshot::RevocationPending(pending) => Some(pending.epoch()),
963 SupervisorAuthoritySnapshot::RotationOperation(rotation) => Some(match rotation.phase() {
964 SupervisorRotationPersistencePhase::PreviousRevokePending
965 | SupervisorRotationPersistencePhase::Rejected => rotation.previous().epoch(),
966 SupervisorRotationPersistencePhase::NextPublishPending
967 | SupervisorRotationPersistencePhase::Completed => rotation.next().epoch(),
968 }),
969 SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => Some(receipt.epoch()),
970 SupervisorAuthoritySnapshot::WithRotationHistory { .. } => None,
971 }
972}
973
974fn terminal_rotation_authority_epoch(receipt: &SupervisorRotationReceipt) -> u64 {
975 match receipt.phase() {
976 SupervisorRotationPersistencePhase::Completed => receipt.next().epoch(),
977 SupervisorRotationPersistencePhase::Rejected => receipt.previous().epoch(),
978 SupervisorRotationPersistencePhase::PreviousRevokePending
979 | SupervisorRotationPersistencePhase::NextPublishPending => receipt.previous().epoch(),
980 }
981}
982
983fn validate_supervisor_rotation_history_coherence(
984 current: &SupervisorAuthoritySnapshot,
985 terminal_receipts: &std::collections::BTreeMap<
986 meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
987 SupervisorRotationReceipt,
988 >,
989) -> Result<(), RuntimeStoreError> {
990 let Some(current_epoch) = current_supervisor_epoch(current) else {
991 return Err(RuntimeStoreError::ReadFailed(
992 "supervisor rotation history requires a current authority epoch".to_string(),
993 ));
994 };
995
996 let mut epochs = SupervisorEpochKeyIndex::new();
997 record_current_authoritative_epochs(&mut epochs, current)?;
998 let mut history_high_water = 0;
999 for receipt in terminal_receipts.values() {
1000 record_rotation_authoritative_epochs(
1001 &mut epochs,
1002 receipt,
1003 "supervisor rotation history authority",
1004 )?;
1005 history_high_water = history_high_water.max(terminal_rotation_authority_epoch(receipt));
1006 }
1007 if current_epoch < history_high_water {
1008 return Err(RuntimeStoreError::ReadFailed(format!(
1009 "current supervisor epoch {current_epoch} is below terminal rotation history high-water {history_high_water}"
1010 )));
1011 }
1012 Ok(())
1013}
1014
1015fn validate_supervisor_authority_snapshot(
1016 snapshot: &SupervisorAuthoritySnapshot,
1017) -> Result<(), RuntimeStoreError> {
1018 match snapshot {
1019 SupervisorAuthoritySnapshot::UnboundNoReceipt => Ok(()),
1020 SupervisorAuthoritySnapshot::Bound(binding) => {
1021 validate_supervisor_binding_receipt(binding, "bound supervisor is invalid")
1022 }
1023 SupervisorAuthoritySnapshot::RevocationPending(pending) => validate_supervisor_descriptor(
1024 pending.name(),
1025 pending.peer_id(),
1026 pending.address(),
1027 pending.signing_public_key(),
1028 "pending supervisor revocation authority is invalid",
1029 ),
1030 SupervisorAuthoritySnapshot::RotationOperation(rotation) => {
1031 validate_supervisor_rotation_receipt(rotation, false)
1032 }
1033 SupervisorAuthoritySnapshot::RevokedReceipt(receipt) => {
1034 validate_revoked_supervisor_receipt(receipt, "revoked supervisor receipt is invalid")
1035 }
1036 SupervisorAuthoritySnapshot::WithRotationHistory {
1037 current,
1038 terminal_receipts,
1039 } => {
1040 if matches!(
1041 current.as_ref(),
1042 SupervisorAuthoritySnapshot::WithRotationHistory { .. }
1043 ) {
1044 return Err(RuntimeStoreError::ReadFailed(
1045 "nested supervisor rotation history is not canonical".to_string(),
1046 ));
1047 }
1048 if terminal_receipts.is_empty() {
1049 return Err(RuntimeStoreError::ReadFailed(
1050 "empty supervisor rotation history wrapper is not canonical".to_string(),
1051 ));
1052 }
1053 validate_supervisor_authority_snapshot(current)?;
1054 for (operation_id, receipt) in terminal_receipts {
1055 if operation_id != &receipt.operation_id() {
1056 return Err(RuntimeStoreError::ReadFailed(format!(
1057 "supervisor rotation history key {operation_id} does not match receipt id {}",
1058 receipt.operation_id()
1059 )));
1060 }
1061 validate_supervisor_rotation_receipt(receipt, true)?;
1062 }
1063 if let SupervisorAuthoritySnapshot::RotationOperation(active) = current.as_ref()
1064 && terminal_receipts.contains_key(&active.operation_id())
1065 {
1066 return Err(RuntimeStoreError::ReadFailed(
1067 "active supervisor rotation is duplicated in terminal history".to_string(),
1068 ));
1069 }
1070 validate_supervisor_rotation_history_coherence(current, terminal_receipts)
1071 }
1072 }
1073}
1074
1075impl TryFrom<SupervisorAuthoritySnapshotStoreWire> for SupervisorAuthoritySnapshot {
1076 type Error = RuntimeStoreError;
1077
1078 fn try_from(snapshot: SupervisorAuthoritySnapshotStoreWire) -> Result<Self, Self::Error> {
1079 match snapshot {
1080 SupervisorAuthoritySnapshotStoreWire::UnboundNoReceipt => Ok(Self::UnboundNoReceipt),
1081 SupervisorAuthoritySnapshotStoreWire::Bound { binding } => {
1082 let binding = binding.into();
1083 validate_supervisor_binding_receipt(&binding, "bound supervisor is invalid")?;
1084 Ok(Self::Bound(binding))
1085 }
1086 SupervisorAuthoritySnapshotStoreWire::RevocationPending { pending } => {
1087 let pending: SupervisorRevocationPendingReceipt = pending.into();
1088 validate_supervisor_descriptor(
1089 pending.name(),
1090 pending.peer_id(),
1091 pending.address(),
1092 pending.signing_public_key(),
1093 "pending supervisor revocation authority is invalid",
1094 )?;
1095 Ok(Self::RevocationPending(pending))
1096 }
1097 SupervisorAuthoritySnapshotStoreWire::RotationOperation { rotation } => {
1098 let receipt: SupervisorRotationReceipt = rotation.try_into()?;
1099 validate_supervisor_rotation_receipt(&receipt, false)?;
1100 Ok(Self::RotationOperation(receipt))
1101 }
1102 SupervisorAuthoritySnapshotStoreWire::RevokedReceipt { receipt } => {
1103 let receipt = receipt.into();
1104 validate_revoked_supervisor_receipt(
1105 &receipt,
1106 "revoked supervisor receipt is invalid",
1107 )?;
1108 Ok(Self::RevokedReceipt(receipt))
1109 }
1110 SupervisorAuthoritySnapshotStoreWire::WithRotationHistory {
1111 current,
1112 terminal_receipts,
1113 } => {
1114 if terminal_receipts.is_empty() {
1115 return Err(RuntimeStoreError::ReadFailed(
1116 "empty supervisor rotation history wrapper is not canonical".to_string(),
1117 ));
1118 }
1119 let current = Self::try_from(*current)?;
1120 if matches!(current, Self::WithRotationHistory { .. }) {
1121 return Err(RuntimeStoreError::ReadFailed(
1122 "nested supervisor rotation history is not canonical".to_string(),
1123 ));
1124 }
1125 let mut receipts = std::collections::BTreeMap::new();
1126 for wire in terminal_receipts {
1127 let receipt: SupervisorRotationReceipt = wire.try_into()?;
1128 validate_supervisor_rotation_receipt(&receipt, true)?;
1129 if receipts.insert(receipt.operation_id(), receipt).is_some() {
1130 return Err(RuntimeStoreError::ReadFailed(
1131 "supervisor rotation history contains a duplicate operation id"
1132 .to_string(),
1133 ));
1134 }
1135 }
1136 if let Self::RotationOperation(active) = ¤t
1137 && receipts.contains_key(&active.operation_id())
1138 {
1139 return Err(RuntimeStoreError::ReadFailed(
1140 "active supervisor rotation is duplicated in terminal history".to_string(),
1141 ));
1142 }
1143 let snapshot = Self::WithRotationHistory {
1144 current: Box::new(current),
1145 terminal_receipts: receipts,
1146 };
1147 validate_supervisor_authority_snapshot(&snapshot)?;
1148 Ok(snapshot)
1149 }
1150 }
1151 }
1152}
1153
1154impl From<&MachineLifecycleSnapshot> for MachineLifecycleSnapshotStoreWire {
1155 fn from(snapshot: &MachineLifecycleSnapshot) -> Self {
1156 Self {
1157 record_version: MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
1158 runtime_state: snapshot.runtime_state(),
1159 binding: snapshot.binding().into(),
1160 supervisor_authority: snapshot.supervisor_authority().into(),
1161 }
1162 }
1163}
1164
1165impl TryFrom<MachineLifecycleSnapshotStoreWire> for MachineLifecycleSnapshot {
1166 type Error = RuntimeStoreError;
1167
1168 fn try_from(record: MachineLifecycleSnapshotStoreWire) -> Result<Self, Self::Error> {
1169 if record.record_version != MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
1170 return Err(RuntimeStoreError::ReadFailed(format!(
1171 "unsupported machine lifecycle store record version {}",
1172 record.record_version
1173 )));
1174 }
1175 Ok(Self::new(
1176 record.runtime_state,
1177 record.binding.try_into()?,
1178 record.supervisor_authority.try_into()?,
1179 ))
1180 }
1181}
1182
1183fn decode_machine_lifecycle_store_record(
1184 bytes: &[u8],
1185) -> Result<MachineLifecycleSnapshot, RuntimeStoreError> {
1186 let version = serde_json::from_slice::<MachineLifecycleSnapshotStoreVersionProbe>(bytes)
1187 .map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
1188 match version.record_version {
1189 LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
1190 let record = serde_json::from_slice::<MachineLifecycleSnapshotStoreWireV1>(bytes)
1191 .map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
1192 if record.record_version != LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION {
1193 return Err(RuntimeStoreError::ReadFailed(format!(
1194 "unsupported machine lifecycle store record version {}",
1195 record.record_version
1196 )));
1197 }
1198 Ok(MachineLifecycleSnapshot::new(
1199 record.runtime_state,
1200 record.binding.into(),
1201 SupervisorAuthoritySnapshot::UnboundNoReceipt,
1202 ))
1203 }
1204 MACHINE_LIFECYCLE_STORE_RECORD_VERSION => {
1205 let record = serde_json::from_slice::<MachineLifecycleSnapshotStoreWire>(bytes)
1206 .map_err(|err| RuntimeStoreError::ReadFailed(err.to_string()))?;
1207 MachineLifecycleSnapshot::try_from(record)
1208 }
1209 unsupported => Err(RuntimeStoreError::ReadFailed(format!(
1210 "unsupported machine lifecycle store record version {unsupported}"
1211 ))),
1212 }
1213}
1214
1215pub async fn load_runtime_state(
1223 store: &dyn RuntimeStore,
1224 runtime_id: &LogicalRuntimeId,
1225) -> Result<Option<RuntimeState>, RuntimeStoreError> {
1226 Ok(load_machine_lifecycle(store, runtime_id)
1227 .await?
1228 .map(|snapshot| snapshot.runtime_state()))
1229}
1230
1231pub(crate) async fn load_machine_lifecycle(
1232 store: &dyn RuntimeStore,
1233 runtime_id: &LogicalRuntimeId,
1234) -> Result<Option<MachineLifecycleSnapshot>, RuntimeStoreError> {
1235 store
1236 .load_machine_lifecycle_record(runtime_id)
1237 .await?
1238 .map(|bytes| decode_machine_lifecycle_store_record(&bytes))
1239 .transpose()
1240}
1241
1242#[derive(Debug, Clone, PartialEq, Eq)]
1248pub struct MachineLifecycleStoreRecord {
1249 snapshot: MachineLifecycleSnapshot,
1250}
1251
1252impl MachineLifecycleStoreRecord {
1253 pub(crate) fn from_snapshot(snapshot: &MachineLifecycleSnapshot) -> Self {
1254 Self {
1255 snapshot: snapshot.clone(),
1256 }
1257 }
1258
1259 pub fn encode(&self) -> Result<Vec<u8>, RuntimeStoreError> {
1260 validate_supervisor_authority_snapshot(self.snapshot.supervisor_authority())
1261 .map_err(|error| RuntimeStoreError::WriteFailed(error.to_string()))?;
1262 let wire = MachineLifecycleSnapshotStoreWire::from(&self.snapshot);
1263 serde_json::to_vec(&wire).map_err(|err| RuntimeStoreError::WriteFailed(err.to_string()))
1264 }
1265}
1266
1267#[derive(Debug, Clone, PartialEq, Eq)]
1273pub struct MachineLifecycleCommit {
1274 snapshot: MachineLifecycleSnapshot,
1275}
1276
1277impl MachineLifecycleCommit {
1278 pub(crate) fn new_with_binding(
1279 runtime_state: RuntimeState,
1280 binding: MachineLifecycleBindingFacts,
1281 supervisor_authority: SupervisorAuthoritySnapshot,
1282 ) -> Self {
1283 Self {
1284 snapshot: MachineLifecycleSnapshot::new(runtime_state, binding, supervisor_authority),
1285 }
1286 }
1287
1288 pub fn runtime_state(&self) -> RuntimeState {
1290 self.snapshot.runtime_state()
1291 }
1292
1293 pub fn snapshot(&self) -> &MachineLifecycleSnapshot {
1295 &self.snapshot
1296 }
1297
1298 pub fn store_record(&self) -> MachineLifecycleStoreRecord {
1300 MachineLifecycleStoreRecord::from_snapshot(&self.snapshot)
1301 }
1302
1303 pub(crate) fn into_snapshot(self) -> MachineLifecycleSnapshot {
1304 self.snapshot
1305 }
1306}
1307
1308#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
1314#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
1315pub trait RuntimeStore: Send + Sync {
1316 fn auth_authority_key(&self) -> Option<String> {
1319 None
1320 }
1321
1322 fn persist_auth_oauth_flow_snapshot(
1328 &self,
1329 snapshot_json: &[u8],
1330 ) -> Result<(), RuntimeStoreError> {
1331 let _ = snapshot_json;
1332 Err(RuntimeStoreError::Unsupported(
1333 "persist_auth_oauth_flow_snapshot".into(),
1334 ))
1335 }
1336
1337 fn load_auth_oauth_flow_snapshot(&self) -> Result<Option<Vec<u8>>, RuntimeStoreError> {
1339 Err(RuntimeStoreError::Unsupported(
1340 "load_auth_oauth_flow_snapshot".into(),
1341 ))
1342 }
1343
1344 fn update_auth_oauth_flow_snapshot(
1350 &self,
1351 _update: &mut AuthOAuthFlowSnapshotUpdate<'_>,
1352 ) -> Result<(), RuntimeStoreError> {
1353 Err(RuntimeStoreError::Unsupported(
1354 "update_auth_oauth_flow_snapshot".into(),
1355 ))
1356 }
1357
1358 async fn commit_session_snapshot(
1363 &self,
1364 runtime_id: &LogicalRuntimeId,
1365 session_delta: SessionDelta,
1366 ) -> Result<(), RuntimeStoreError>;
1367
1368 async fn commit_session_transcript_rewrite_snapshot(
1374 &self,
1375 runtime_id: &LogicalRuntimeId,
1376 session_delta: SessionDelta,
1377 commit: &meerkat_core::TranscriptRewriteCommit,
1378 ) -> Result<(), RuntimeStoreError> {
1379 let _ = (runtime_id, session_delta, commit);
1380 Err(RuntimeStoreError::Unsupported(
1381 "commit_session_transcript_rewrite_snapshot".into(),
1382 ))
1383 }
1384
1385 async fn atomic_apply(
1398 &self,
1399 runtime_id: &LogicalRuntimeId,
1400 session_delta: Option<SessionDelta>,
1401 receipt: RunBoundaryReceipt,
1402 input_updates: Vec<InputStatePersistenceRecord>,
1403 session_store_key: Option<meerkat_core::types::SessionId>,
1404 ) -> Result<(), RuntimeStoreError>;
1405
1406 async fn load_input_states(
1408 &self,
1409 runtime_id: &LogicalRuntimeId,
1410 ) -> Result<Vec<StoredInputState>, RuntimeStoreError>;
1411
1412 async fn load_boundary_receipt(
1414 &self,
1415 runtime_id: &LogicalRuntimeId,
1416 run_id: &RunId,
1417 sequence: u64,
1418 ) -> Result<Option<RunBoundaryReceipt>, RuntimeStoreError>;
1419
1420 async fn load_session_snapshot(
1422 &self,
1423 runtime_id: &LogicalRuntimeId,
1424 ) -> Result<Option<Vec<u8>>, RuntimeStoreError>;
1425
1426 async fn clear_session_snapshot(
1432 &self,
1433 runtime_id: &LogicalRuntimeId,
1434 ) -> Result<(), RuntimeStoreError>;
1435
1436 async fn replace_session_snapshot_if_current(
1443 &self,
1444 runtime_id: &LogicalRuntimeId,
1445 expected_current: &[u8],
1446 replacement: Vec<u8>,
1447 ) -> Result<bool, RuntimeStoreError>;
1448
1449 async fn clear_session_snapshot_if_current(
1454 &self,
1455 runtime_id: &LogicalRuntimeId,
1456 expected_current: &[u8],
1457 ) -> Result<bool, RuntimeStoreError>;
1458
1459 async fn is_runtime_projection_quarantined(
1471 &self,
1472 runtime_id: &LogicalRuntimeId,
1473 ) -> Result<bool, RuntimeStoreError> {
1474 let _ = runtime_id;
1475 Ok(false)
1476 }
1477
1478 async fn persist_input_state(
1480 &self,
1481 runtime_id: &LogicalRuntimeId,
1482 state: &InputStatePersistenceRecord,
1483 ) -> Result<(), RuntimeStoreError>;
1484
1485 async fn load_input_state(
1487 &self,
1488 runtime_id: &LogicalRuntimeId,
1489 input_id: &InputId,
1490 ) -> Result<Option<StoredInputState>, RuntimeStoreError>;
1491
1492 async fn load_machine_lifecycle_record(
1500 &self,
1501 runtime_id: &LogicalRuntimeId,
1502 ) -> Result<Option<Vec<u8>>, RuntimeStoreError>;
1503
1504 async fn commit_machine_lifecycle(
1511 &self,
1512 runtime_id: &LogicalRuntimeId,
1513 commit: MachineLifecycleCommit,
1514 input_states: &[InputStatePersistenceRecord],
1515 ) -> Result<(), RuntimeStoreError>;
1516
1517 async fn persist_ops_lifecycle(
1519 &self,
1520 runtime_id: &LogicalRuntimeId,
1521 snapshot: &crate::ops_lifecycle::PersistedOpsSnapshot,
1522 ) -> Result<(), RuntimeStoreError> {
1523 let _ = (runtime_id, snapshot);
1524 Err(RuntimeStoreError::Unsupported(
1525 "persist_ops_lifecycle".into(),
1526 ))
1527 }
1528
1529 async fn load_ops_lifecycle(
1531 &self,
1532 runtime_id: &LogicalRuntimeId,
1533 ) -> Result<Option<crate::ops_lifecycle::PersistedOpsSnapshot>, RuntimeStoreError> {
1534 let _ = runtime_id;
1535 Err(RuntimeStoreError::Unsupported("load_ops_lifecycle".into()))
1536 }
1537
1538 async fn delete_ops_lifecycle(
1540 &self,
1541 runtime_id: &LogicalRuntimeId,
1542 ) -> Result<(), RuntimeStoreError> {
1543 let _ = runtime_id;
1544 Err(RuntimeStoreError::Unsupported(
1545 "delete_ops_lifecycle".into(),
1546 ))
1547 }
1548}
1549
1550pub use memory::InMemoryRuntimeStore;
1551#[cfg(feature = "sqlite-store")]
1552pub use sqlite::SqliteRuntimeStore;
1553
1554#[cfg(test)]
1555mod lifecycle_record_compatibility_tests {
1556 use super::*;
1557
1558 fn operation_id(
1559 value: u128,
1560 ) -> meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId {
1561 meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId::from_uuid(
1562 uuid::Uuid::from_u128(value),
1563 )
1564 }
1565
1566 fn binding(seed: u8, name: &str, epoch: u64) -> SupervisorBindingReceipt {
1567 let pubkey = [seed; 32];
1568 SupervisorBindingReceipt::new(
1569 name.to_string(),
1570 meerkat_core::comms::PeerId::from_ed25519_pubkey(&pubkey).as_str(),
1571 format!("inproc://{name}"),
1572 crate::comms_drain::encode_supervisor_signing_public_key(pubkey),
1573 epoch,
1574 )
1575 }
1576
1577 fn rotation(
1578 operation_id: meerkat_contracts::wire::supervisor_bridge::SupervisorRotationOperationId,
1579 phase: SupervisorRotationPersistencePhase,
1580 rejection: Option<SupervisorRotationRejection>,
1581 previous: SupervisorBindingReceipt,
1582 next: SupervisorBindingReceipt,
1583 ) -> SupervisorRotationReceipt {
1584 SupervisorRotationReceipt::new(operation_id, phase, rejection, previous, next)
1585 }
1586
1587 fn snapshot(authority: SupervisorAuthoritySnapshot) -> MachineLifecycleSnapshot {
1588 MachineLifecycleSnapshot::new(
1589 RuntimeState::Idle,
1590 MachineLifecycleBindingFacts::new(None, None, None, None),
1591 authority,
1592 )
1593 }
1594
1595 fn encode_snapshot(snapshot: &MachineLifecycleSnapshot) -> Vec<u8> {
1596 MachineLifecycleStoreRecord::from_snapshot(snapshot)
1597 .encode()
1598 .expect("encode lifecycle snapshot")
1599 }
1600
1601 fn encode_unvalidated_snapshot(snapshot: &MachineLifecycleSnapshot) -> Vec<u8> {
1602 serde_json::to_vec(&MachineLifecycleSnapshotStoreWire::from(snapshot))
1603 .expect("serialize deliberately corrupt lifecycle snapshot")
1604 }
1605
1606 fn encoded_value(snapshot: &MachineLifecycleSnapshot) -> serde_json::Value {
1607 serde_json::from_slice(&encode_snapshot(snapshot)).expect("decode encoded snapshot as JSON")
1608 }
1609
1610 fn assert_decode_fails(value: serde_json::Value) {
1611 let bytes = serde_json::to_vec(&value).expect("serialize corrupt lifecycle record");
1612 assert!(
1613 decode_machine_lifecycle_store_record(&bytes).is_err(),
1614 "corrupt lifecycle record must fail closed: {value}"
1615 );
1616 }
1617
1618 #[test]
1619 fn version_one_record_without_supervisor_authority_migrates_explicitly_to_unbound() {
1620 let bytes = serde_json::to_vec(&serde_json::json!({
1621 "record_version": LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
1622 "runtime_state": RuntimeState::Retired,
1623 "binding": {
1624 "agent_runtime_id": "rt:session:legacy-v1",
1625 "fence_token": 19,
1626 "runtime_generation": 4,
1627 "runtime_epoch_id": "epoch-legacy-v1"
1628 }
1629 }))
1630 .expect("serialize legacy v1 lifecycle record");
1631
1632 let decoded = decode_machine_lifecycle_store_record(&bytes)
1633 .expect("valid v1 record without the additive field must decode");
1634 assert_eq!(decoded.runtime_state(), RuntimeState::Retired);
1635 assert_eq!(
1636 decoded.supervisor_authority(),
1637 &SupervisorAuthoritySnapshot::UnboundNoReceipt
1638 );
1639 }
1640
1641 #[test]
1642 fn version_two_record_requires_supervisor_authority() {
1643 assert_decode_fails(serde_json::json!({
1644 "record_version": MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
1645 "runtime_state": RuntimeState::Idle,
1646 "binding": {
1647 "agent_runtime_id": null,
1648 "fence_token": null,
1649 "runtime_generation": null,
1650 "runtime_epoch_id": null
1651 }
1652 }));
1653 }
1654
1655 #[test]
1656 fn version_two_nullable_fields_require_presence_but_accept_explicit_null() {
1657 let unbound = snapshot(SupervisorAuthoritySnapshot::UnboundNoReceipt);
1658 let encoded = encoded_value(&unbound);
1659 assert_eq!(
1660 decode_machine_lifecycle_store_record(
1661 &serde_json::to_vec(&encoded).expect("serialize valid v2 record")
1662 )
1663 .expect("explicit-null v2 binding fields must decode"),
1664 unbound
1665 );
1666
1667 for field in [
1668 "agent_runtime_id",
1669 "fence_token",
1670 "runtime_generation",
1671 "runtime_epoch_id",
1672 ] {
1673 let mut partial = encoded.clone();
1674 partial["binding"]
1675 .as_object_mut()
1676 .expect("binding object")
1677 .remove(field);
1678 assert_decode_fails(partial);
1679 }
1680
1681 let completed = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1682 operation_id(101),
1683 SupervisorRotationPersistencePhase::Completed,
1684 None,
1685 binding(30, "required-null-previous", 4),
1686 binding(31, "required-null-next", 5),
1687 )));
1688 let mut missing_rejection = encoded_value(&completed);
1689 assert!(missing_rejection["supervisor_authority"]["rotation"]["rejection"].is_null());
1690 missing_rejection["supervisor_authority"]["rotation"]
1691 .as_object_mut()
1692 .expect("rotation object")
1693 .remove("rejection");
1694 assert_decode_fails(missing_rejection);
1695 }
1696
1697 #[test]
1698 fn version_one_migration_rejects_current_authority_fields() {
1699 assert_decode_fails(serde_json::json!({
1700 "record_version": LEGACY_MACHINE_LIFECYCLE_STORE_RECORD_VERSION,
1701 "runtime_state": RuntimeState::Idle,
1702 "binding": {
1703 "agent_runtime_id": null,
1704 "fence_token": null,
1705 "runtime_generation": null,
1706 "runtime_epoch_id": null
1707 },
1708 "supervisor_authority": { "kind": "unbound_no_receipt" }
1709 }));
1710 }
1711
1712 #[test]
1713 fn mixed_or_unknown_supervisor_authority_fields_fail_closed() {
1714 let current = binding(1, "current-supervisor", 7);
1715 let mut value = encoded_value(&snapshot(SupervisorAuthoritySnapshot::Bound(current)));
1716 value["supervisor_authority"]["rotation"] = serde_json::json!({});
1717 assert_decode_fails(value);
1718 }
1719
1720 #[test]
1721 fn completed_rotation_operation_receipt_round_trips_for_cold_observation() {
1722 let snapshot = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1723 operation_id(1),
1724 SupervisorRotationPersistencePhase::Completed,
1725 None,
1726 binding(1, "previous-supervisor", 7),
1727 binding(2, "next-supervisor", 8),
1728 )));
1729
1730 let encoded = encode_snapshot(&snapshot);
1731 let decoded = decode_machine_lifecycle_store_record(&encoded)
1732 .expect("decode completed rotation receipt");
1733
1734 assert_eq!(decoded, snapshot);
1735 }
1736
1737 #[test]
1738 fn exact_current_completed_adoption_round_trips_but_other_equal_epoch_completion_fails() {
1739 let current = binding(3, "already-rotated-supervisor", 9);
1740 let adoption = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1741 operation_id(2),
1742 SupervisorRotationPersistencePhase::Completed,
1743 None,
1744 current.clone(),
1745 current,
1746 )));
1747 assert_eq!(
1748 decode_machine_lifecycle_store_record(&encode_snapshot(&adoption))
1749 .expect("exact-current legacy adoption receipt must decode"),
1750 adoption
1751 );
1752
1753 let non_advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1754 operation_id(3),
1755 SupervisorRotationPersistencePhase::Completed,
1756 None,
1757 binding(3, "previous-supervisor", 9),
1758 binding(4, "different-supervisor", 9),
1759 )));
1760 assert!(
1761 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&non_advancing))
1762 .is_err()
1763 );
1764 }
1765
1766 #[test]
1767 fn malformed_rotation_descriptors_epochs_and_operation_ids_fail_closed() {
1768 let invalid_previous = SupervisorBindingReceipt::new(
1769 String::new(),
1770 "not-a-uuid".to_string(),
1771 "not-an-address".to_string(),
1772 "not-a-key".to_string(),
1773 1,
1774 );
1775 let invalid_previous_receipt =
1776 snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1777 operation_id(4),
1778 SupervisorRotationPersistencePhase::Rejected,
1779 Some(SupervisorRotationRejection::InvalidTarget),
1780 invalid_previous,
1781 binding(5, "raw-target", 2),
1782 )));
1783 assert!(
1784 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
1785 &invalid_previous_receipt,
1786 ))
1787 .is_err()
1788 );
1789
1790 let invalid_next = SupervisorBindingReceipt::new(
1791 "invalid-target".to_string(),
1792 "not-a-uuid".to_string(),
1793 "not-an-address".to_string(),
1794 "not-a-key".to_string(),
1795 2,
1796 );
1797 let invalid_completed_target =
1798 snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1799 operation_id(5),
1800 SupervisorRotationPersistencePhase::Completed,
1801 None,
1802 binding(6, "previous-supervisor", 1),
1803 invalid_next,
1804 )));
1805 assert!(
1806 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
1807 &invalid_completed_target,
1808 ))
1809 .is_err()
1810 );
1811
1812 let mut invalid_id = encoded_value(&snapshot(
1813 SupervisorAuthoritySnapshot::RotationOperation(rotation(
1814 operation_id(6),
1815 SupervisorRotationPersistencePhase::PreviousRevokePending,
1816 None,
1817 binding(7, "previous-supervisor", 1),
1818 binding(8, "next-supervisor", 2),
1819 )),
1820 ));
1821 invalid_id["supervisor_authority"]["rotation"]["operation_id"] =
1822 serde_json::json!("not-a-uuid");
1823 assert_decode_fails(invalid_id);
1824
1825 let nil_id = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1826 operation_id(0),
1827 SupervisorRotationPersistencePhase::PreviousRevokePending,
1828 None,
1829 binding(7, "previous-supervisor", 1),
1830 binding(8, "next-supervisor", 2),
1831 )));
1832 assert!(
1833 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&nil_id)).is_err()
1834 );
1835
1836 let non_advancing_pending =
1837 snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1838 operation_id(13),
1839 SupervisorRotationPersistencePhase::PreviousRevokePending,
1840 None,
1841 binding(7, "previous-supervisor", 4),
1842 binding(8, "next-supervisor", 4),
1843 )));
1844 assert!(
1845 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
1846 &non_advancing_pending,
1847 ))
1848 .is_err()
1849 );
1850 }
1851
1852 #[test]
1853 fn rejected_invalid_or_unsupported_target_preserves_raw_evidence() {
1854 for (id, rejection) in [
1855 (7, SupervisorRotationRejection::InvalidTarget),
1856 (14, SupervisorRotationRejection::UnsupportedProtocolVersion),
1857 ] {
1858 let raw_invalid_target = SupervisorBindingReceipt::new(
1859 "".to_string(),
1860 "not-a-peer-id".to_string(),
1861 "not-an-address".to_string(),
1862 "not-a-signing-key".to_string(),
1863 0,
1864 );
1865 let snapshot = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1866 operation_id(id),
1867 SupervisorRotationPersistencePhase::Rejected,
1868 Some(rejection),
1869 binding(9, "retained-supervisor", 11),
1870 raw_invalid_target,
1871 )));
1872 assert_eq!(
1873 decode_machine_lifecycle_store_record(&encode_snapshot(&snapshot))
1874 .expect("rejected raw target evidence must remain durable"),
1875 snapshot
1876 );
1877 }
1878 }
1879
1880 #[test]
1881 fn only_raw_target_rejections_are_durable_and_epoch_rejection_must_be_genuine() {
1882 for (id, rejection) in [
1883 (102, SupervisorRotationRejection::OperationConflict),
1884 (103, SupervisorRotationRejection::NotBound),
1885 (104, SupervisorRotationRejection::SenderMismatch),
1886 ] {
1887 let impossible = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1888 operation_id(id),
1889 SupervisorRotationPersistencePhase::Rejected,
1890 Some(rejection),
1891 binding(32, "retained-supervisor", 7),
1892 binding(33, "requested-supervisor", 8),
1893 )));
1894 assert!(
1895 MachineLifecycleStoreRecord::from_snapshot(&impossible)
1896 .encode()
1897 .is_err()
1898 );
1899 assert!(
1900 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&impossible))
1901 .is_err()
1902 );
1903 }
1904
1905 let advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1906 operation_id(105),
1907 SupervisorRotationPersistencePhase::Rejected,
1908 Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
1909 binding(34, "retained-supervisor", 9),
1910 binding(35, "advancing-target", 10),
1911 )));
1912 assert!(
1913 MachineLifecycleStoreRecord::from_snapshot(&advancing)
1914 .encode()
1915 .is_err()
1916 );
1917 assert!(
1918 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&advancing))
1919 .is_err()
1920 );
1921
1922 let non_advancing = snapshot(SupervisorAuthoritySnapshot::RotationOperation(rotation(
1923 operation_id(106),
1924 SupervisorRotationPersistencePhase::Rejected,
1925 Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
1926 binding(36, "retained-supervisor", 11),
1927 binding(37, "non-advancing-target", 11),
1928 )));
1929 assert_eq!(
1930 decode_machine_lifecycle_store_record(&encode_snapshot(&non_advancing))
1931 .expect("genuine target-epoch rejection must remain durable"),
1932 non_advancing
1933 );
1934 }
1935
1936 #[test]
1937 fn malformed_current_authority_variants_fail_closed() {
1938 let malformed = SupervisorBindingReceipt::new(
1939 String::new(),
1940 "not-a-peer-id".to_string(),
1941 "not-an-address".to_string(),
1942 "not-a-signing-key".to_string(),
1943 1,
1944 );
1945 let bound = snapshot(SupervisorAuthoritySnapshot::Bound(malformed.clone()));
1946 assert!(
1947 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&bound)).is_err()
1948 );
1949
1950 let pending = snapshot(SupervisorAuthoritySnapshot::RevocationPending(
1951 SupervisorRevocationPendingReceipt::new(
1952 malformed.name().to_owned(),
1953 malformed.peer_id().to_owned(),
1954 malformed.address().to_owned(),
1955 malformed.signing_public_key().to_owned(),
1956 malformed.epoch(),
1957 ),
1958 ));
1959 assert!(
1960 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&pending)).is_err()
1961 );
1962
1963 let revoked = snapshot(SupervisorAuthoritySnapshot::RevokedReceipt(
1964 RevokedSupervisorReceipt::new(
1965 malformed.peer_id().to_owned(),
1966 malformed.signing_public_key().to_owned(),
1967 malformed.epoch(),
1968 ),
1969 ));
1970 assert!(
1971 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&revoked)).is_err()
1972 );
1973 }
1974
1975 #[test]
1976 fn partial_and_nonterminal_history_records_fail_closed() {
1977 let receipt = rotation(
1978 operation_id(8),
1979 SupervisorRotationPersistencePhase::Completed,
1980 None,
1981 binding(10, "history-previous", 1),
1982 binding(11, "history-next", 2),
1983 );
1984 let history = std::collections::BTreeMap::from([(receipt.operation_id(), receipt)]);
1985 let snapshot = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
1986 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
1987 12,
1988 "current-supervisor",
1989 3,
1990 ))),
1991 terminal_receipts: history,
1992 });
1993
1994 let mut partial = encoded_value(&snapshot);
1995 partial["supervisor_authority"]["terminal_receipts"][0]
1996 .as_object_mut()
1997 .expect("history receipt object")
1998 .remove("next");
1999 assert_decode_fails(partial);
2000
2001 let mut nonterminal = encoded_value(&snapshot);
2002 nonterminal["supervisor_authority"]["terminal_receipts"][0]["phase"] =
2003 serde_json::json!("next_publish_pending");
2004 assert_decode_fails(nonterminal);
2005 }
2006
2007 #[test]
2008 fn duplicate_nested_and_active_history_conflicts_fail_closed() {
2009 let history_receipt = rotation(
2010 operation_id(9),
2011 SupervisorRotationPersistencePhase::Completed,
2012 None,
2013 binding(13, "history-previous", 1),
2014 binding(14, "history-next", 2),
2015 );
2016 let history = std::collections::BTreeMap::from([(
2017 history_receipt.operation_id(),
2018 history_receipt.clone(),
2019 )]);
2020 let wrapper = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2021 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2022 15,
2023 "current-supervisor",
2024 3,
2025 ))),
2026 terminal_receipts: history,
2027 });
2028
2029 let mut duplicate = encoded_value(&wrapper);
2030 let receipt = duplicate["supervisor_authority"]["terminal_receipts"][0].clone();
2031 duplicate["supervisor_authority"]["terminal_receipts"]
2032 .as_array_mut()
2033 .expect("history receipt array")
2034 .push(receipt);
2035 assert_decode_fails(duplicate);
2036
2037 let mut nested = encoded_value(&wrapper);
2038 let nested_current = nested["supervisor_authority"].clone();
2039 nested["supervisor_authority"]["current"] = nested_current;
2040 assert_decode_fails(nested);
2041
2042 let active_conflict = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2043 current: Box::new(SupervisorAuthoritySnapshot::RotationOperation(
2044 history_receipt.clone(),
2045 )),
2046 terminal_receipts: std::collections::BTreeMap::from([(
2047 history_receipt.operation_id(),
2048 history_receipt,
2049 )]),
2050 });
2051 assert!(
2052 MachineLifecycleStoreRecord::from_snapshot(&active_conflict)
2053 .encode()
2054 .is_err()
2055 );
2056
2057 let empty_history = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2058 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2059 20,
2060 "current-supervisor",
2061 4,
2062 ))),
2063 terminal_receipts: std::collections::BTreeMap::new(),
2064 });
2065 assert!(
2066 MachineLifecycleStoreRecord::from_snapshot(&empty_history)
2067 .encode()
2068 .is_err()
2069 );
2070 assert!(
2071 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&empty_history))
2072 .is_err()
2073 );
2074
2075 let mismatched_key = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2076 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2077 21,
2078 "current-supervisor",
2079 4,
2080 ))),
2081 terminal_receipts: std::collections::BTreeMap::from([(
2082 operation_id(99),
2083 rotation(
2084 operation_id(98),
2085 SupervisorRotationPersistencePhase::Completed,
2086 None,
2087 binding(22, "history-previous", 2),
2088 binding(23, "history-next", 3),
2089 ),
2090 )]),
2091 });
2092 assert!(
2093 MachineLifecycleStoreRecord::from_snapshot(&mismatched_key)
2094 .encode()
2095 .is_err()
2096 );
2097 }
2098
2099 #[test]
2100 fn history_current_epoch_and_same_epoch_identity_must_cohere() {
2101 let previous = binding(38, "history-previous", 12);
2102 let next = binding(39, "history-next", 13);
2103 let completed = rotation(
2104 operation_id(107),
2105 SupervisorRotationPersistencePhase::Completed,
2106 None,
2107 previous.clone(),
2108 next.clone(),
2109 );
2110 let history =
2111 std::collections::BTreeMap::from([(completed.operation_id(), completed.clone())]);
2112
2113 let stale_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2114 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2115 38,
2116 "refreshed-history-previous",
2117 12,
2118 ))),
2119 terminal_receipts: history.clone(),
2120 });
2121 assert!(
2122 MachineLifecycleStoreRecord::from_snapshot(&stale_current)
2123 .encode()
2124 .is_err()
2125 );
2126 assert!(
2127 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(&stale_current))
2128 .is_err()
2129 );
2130
2131 let conflicting_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2132 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2133 40,
2134 "conflicting-current",
2135 13,
2136 ))),
2137 terminal_receipts: history.clone(),
2138 });
2139 assert!(
2140 MachineLifecycleStoreRecord::from_snapshot(&conflicting_current)
2141 .encode()
2142 .is_err()
2143 );
2144 assert!(
2145 decode_machine_lifecycle_store_record(&encode_unvalidated_snapshot(
2146 &conflicting_current,
2147 ))
2148 .is_err()
2149 );
2150
2151 let route_refreshed_current = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2152 current: Box::new(SupervisorAuthoritySnapshot::Bound(binding(
2153 39,
2154 "route-refreshed-history-next",
2155 13,
2156 ))),
2157 terminal_receipts: history,
2158 });
2159 assert_eq!(
2160 decode_machine_lifecycle_store_record(&encode_snapshot(&route_refreshed_current))
2161 .expect("same identity may refresh route metadata within one epoch"),
2162 route_refreshed_current
2163 );
2164 }
2165
2166 #[test]
2167 fn terminal_history_survives_later_rotation_and_recovery() {
2168 let first = rotation(
2169 operation_id(10),
2170 SupervisorRotationPersistencePhase::Completed,
2171 None,
2172 binding(16, "first-supervisor", 1),
2173 binding(17, "second-supervisor", 2),
2174 );
2175 let rejected = rotation(
2176 operation_id(11),
2177 SupervisorRotationPersistencePhase::Rejected,
2178 Some(SupervisorRotationRejection::TargetEpochNotAdvanced),
2179 binding(17, "second-supervisor", 2),
2180 binding(18, "rejected-supervisor", 2),
2181 );
2182 let later = rotation(
2183 operation_id(12),
2184 SupervisorRotationPersistencePhase::Completed,
2185 None,
2186 binding(17, "second-supervisor", 2),
2187 binding(19, "current-supervisor", 3),
2188 );
2189 let snapshot = snapshot(SupervisorAuthoritySnapshot::WithRotationHistory {
2190 current: Box::new(SupervisorAuthoritySnapshot::RotationOperation(later)),
2191 terminal_receipts: std::collections::BTreeMap::from([
2192 (first.operation_id(), first),
2193 (rejected.operation_id(), rejected),
2194 ]),
2195 });
2196
2197 let decoded = decode_machine_lifecycle_store_record(&encode_snapshot(&snapshot))
2198 .expect("later rotation and old terminal history must recover together");
2199 assert_eq!(decoded, snapshot);
2200 }
2201}