Skip to main content

meerkat_runtime/store/
mod.rs

1//! RuntimeStore — atomic persistence for runtime state.
2//!
3//! Machine-owned runtime commands durably persist [`RunBoundaryReceipt`] values
4//! atomically with their session and input-state effects.
5
6pub 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/// Errors from RuntimeStore operations.
20#[derive(Debug, Clone, thiserror::Error)]
21#[non_exhaustive]
22pub enum RuntimeStoreError {
23    /// Write failed.
24    #[error("Store write failed: {0}")]
25    WriteFailed(String),
26    /// Read failed.
27    #[error("Store read failed: {0}")]
28    ReadFailed(String),
29    /// The explicit session-store key does not match the serialized session.
30    #[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    /// Not found.
36    #[error("Not found: {0}")]
37    NotFound(String),
38    /// Operation is not supported by this store implementation.
39    #[error("Unsupported store operation: {0}")]
40    Unsupported(String),
41    /// Runtime snapshot CAS rejected a stale transcript rewrite.
42    #[error("Transcript revision conflict: expected {expected}, actual {actual}")]
43    TranscriptRevisionConflict { expected: String, actual: String },
44    /// Internal error.
45    #[error("Internal error: {0}")]
46    Internal(String),
47}
48
49/// Transactional updater for the runtime-owned OAuth login-flow payload snapshot.
50pub type AuthOAuthFlowSnapshotUpdate<'a> =
51    dyn FnMut(Option<&[u8]>) -> Result<Vec<u8>, RuntimeStoreError> + 'a;
52
53/// Describes a serialized session snapshot for boundary and snapshot-only commits.
54#[derive(Debug, Clone)]
55pub struct SessionDelta {
56    /// Serialized session snapshot (opaque to RuntimeStore).
57    pub session_snapshot: Vec<u8>,
58}
59
60/// Runtime binding facts selected by generated MeerkatMachine authority.
61///
62/// RuntimeStore implementations persist and read these facts as part of a
63/// machine lifecycle snapshot. The commit token that writes these facts stays
64/// crate-private so compatibility callers cannot mint replacement lifecycle
65/// truth.
66#[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/// Durable identity receipt for the last completed supervisor revoke.
75///
76/// This is not a live supervisor binding and carries no route address. It is
77/// only the identity/key/epoch witness needed to authorize an exact duplicate
78/// revoke response after a cold restart; the current authenticated request
79/// supplies its current route.
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct RevokedSupervisorReceipt {
82    peer_id: String,
83    signing_public_key: String,
84    epoch: u64,
85}
86
87/// Durable current supervisor binding used to authenticate terminal retry
88/// traffic after a cold runtime restart.
89#[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/// Durable in-flight supervisor revocation receipt.
99///
100/// This is the closed-world hand-off between generated machine authority and
101/// the concrete router mutation.  It deliberately retains the complete prior
102/// route so a cold runtime can authenticate an exact retry and re-materialize
103/// the generated remove obligation without resurrecting a live binding.
104#[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/// Closed durable supervisor authority state. Each variant owns one complete
281/// recovery shape; terminal rotation receipts retain their exact operation and
282/// participant descriptors for idempotent submission and later observation.
283#[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/// Durable read-back shape for machine-owned lifecycle state.
333#[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    /// Runtime state selected by the owning MeerkatMachine transition.
354    pub fn runtime_state(&self) -> RuntimeState {
355        self.runtime_state
356    }
357
358    /// Runtime binding facts selected by the owning MeerkatMachine transition.
359    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/// Exact pre-supervisor-authority lifecycle shape. Version 1 is decoded only
478/// through this migration carrier so a missing authority on a current record
479/// cannot be confused with legacy data.
480#[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            // A legacy member may already have the exact target installed
809            // before the operation protocol assigns an id. Its adoption
810            // receipt is Completed with an exact previous == next witness.
811            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                    // These two rejection classes retain the undecodable target
834                    // fields as raw evidence. They are deliberately exempt from
835                    // target descriptor validation and epoch comparison.
836                }
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) = &current
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
1215/// Load the last persisted runtime-state projection from a generated lifecycle
1216/// record.
1217///
1218/// This is a projection of [`MachineLifecycleCommit`] authority. Store
1219/// implementations provide only opaque record bytes; the runtime crate owns the
1220/// decoding and rejects compatibility rows that are not machine lifecycle
1221/// records.
1222pub 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/// Declared durable store record for generated machine lifecycle truth.
1243///
1244/// Stores receive this record from [`MachineLifecycleCommit`] and may persist
1245/// its encoded form. Loading must decode this exact record shape; compatibility
1246/// runtime-state projections are not lifecycle authority.
1247#[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/// Machine-owned lifecycle commit token.
1268///
1269/// This token has no public constructor. RuntimeStore implementors can persist
1270/// the selected state and binding facts, but callers outside the machine/driver
1271/// commit path cannot select arbitrary lifecycle truth.
1272#[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    /// Runtime state selected by the owning MeerkatMachine transition.
1289    pub fn runtime_state(&self) -> RuntimeState {
1290        self.snapshot.runtime_state()
1291    }
1292
1293    /// Durable lifecycle snapshot selected by the owning MeerkatMachine transition.
1294    pub fn snapshot(&self) -> &MachineLifecycleSnapshot {
1295        &self.snapshot
1296    }
1297
1298    /// Durable record selected by the owning MeerkatMachine transition.
1299    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/// Atomic persistence interface for runtime state.
1309///
1310/// Implementations:
1311/// - `InMemoryRuntimeStore` — in-memory, no durability (ephemeral/testing)
1312/// - `SqliteRuntimeStore` — SQLite-backed durable runtime state
1313#[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    /// Stable key for process-local auth/OAuth authority reuse across reopened
1317    /// handles for the same durable store.
1318    fn auth_authority_key(&self) -> Option<String> {
1319        None
1320    }
1321
1322    /// Persist the runtime-owned OAuth login-flow payload snapshot.
1323    ///
1324    /// The AuthMachine owns admission/consume semantics; this payload snapshot
1325    /// carries the PKCE verifier and device-code correlation data needed to
1326    /// rehydrate active flows after a persistent runtime process restart.
1327    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    /// Load the runtime-owned OAuth login-flow payload snapshot, if present.
1338    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    /// Atomically update the runtime-owned OAuth login-flow payload snapshot.
1345    ///
1346    /// Stores that support OAuth snapshots must override this with a lock,
1347    /// transaction, or compare-and-swap boundary. A load/compute/persist
1348    /// fallback is not safe for admission, capacity, or consume claims.
1349    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    /// Atomically persist a session snapshot that is not a run boundary.
1359    ///
1360    /// Session-control snapshots update durable session authority without
1361    /// producing a [`RunBoundaryReceipt`].
1362    async fn commit_session_snapshot(
1363        &self,
1364        runtime_id: &LogicalRuntimeId,
1365        session_delta: SessionDelta,
1366    ) -> Result<(), RuntimeStoreError>;
1367
1368    /// Atomically persist a same-session transcript rewrite snapshot.
1369    ///
1370    /// Store implementations that support transcript edits must compare the
1371    /// currently persisted session transcript revision with `commit.parent_revision`
1372    /// inside the same lock or transaction that writes `session_delta`.
1373    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    /// Atomically persist session delta + receipt + input state updates.
1386    ///
1387    /// All three writes MUST commit in a single atomic operation.
1388    /// If any write fails, none should be visible.
1389    /// Atomically persist session delta + receipt + input state updates.
1390    ///
1391    /// All writes MUST commit in a single atomic operation.
1392    /// If `session_store_key` is `Some`, validates that the snapshot belongs
1393    /// to that session and, for stores that physically share a `SessionStore`
1394    /// table, writes that table in the same transaction. Runtime snapshot
1395    /// authority remains keyed only by `runtime_id`; `session_store_key` must
1396    /// not create a raw session UUID runtime alias.
1397    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    /// Load all input states for a runtime.
1407    async fn load_input_states(
1408        &self,
1409        runtime_id: &LogicalRuntimeId,
1410    ) -> Result<Vec<StoredInputState>, RuntimeStoreError>;
1411
1412    /// Load a specific boundary receipt.
1413    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    /// Load the latest committed session snapshot for a runtime, if any.
1421    async fn load_session_snapshot(
1422        &self,
1423        runtime_id: &LogicalRuntimeId,
1424    ) -> Result<Option<Vec<u8>>, RuntimeStoreError>;
1425
1426    /// Remove the latest committed session snapshot for a runtime.
1427    ///
1428    /// This is used only as a fail-closed quarantine path after a compatibility
1429    /// projection write rejects a runtime snapshot that was already staged as
1430    /// runtime authority and the service cannot restore the previous snapshot.
1431    async fn clear_session_snapshot(
1432        &self,
1433        runtime_id: &LogicalRuntimeId,
1434    ) -> Result<(), RuntimeStoreError>;
1435
1436    /// Replace the latest committed session snapshot only if it still matches
1437    /// `expected_current`.
1438    ///
1439    /// Used by fail-closed recovery after a compatibility projection rejected
1440    /// a runtime snapshot. Implementations must compare and write atomically so
1441    /// recovery cannot overwrite a newer runtime-authoritative snapshot.
1442    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    /// Remove the latest committed session snapshot only if it still matches
1450    /// `expected_current`.
1451    ///
1452    /// This is the conditional variant of the fail-closed quarantine path.
1453    async fn clear_session_snapshot_if_current(
1454        &self,
1455        runtime_id: &LogicalRuntimeId,
1456        expected_current: &[u8],
1457    ) -> Result<bool, RuntimeStoreError>;
1458
1459    /// Report whether the runtime-projection fallback for `runtime_id` is
1460    /// quarantined.
1461    ///
1462    /// This is a durable single-owner fact: when
1463    /// [`clear_session_snapshot_if_current`](Self::clear_session_snapshot_if_current)
1464    /// matches and DELETEs a rejected runtime snapshot, the same atomic boundary
1465    /// records the quarantine marker. A subsequent live snapshot write clears it.
1466    /// Recovery reads this to decide whether a store-only projection may stand in
1467    /// for an absent runtime snapshot. The default is fail-safe (`false`): stores
1468    /// that cannot record the marker durably never claim a snapshot is
1469    /// quarantined.
1470    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    /// Persist a single input state (for durable-before-ack).
1479    async fn persist_input_state(
1480        &self,
1481        runtime_id: &LogicalRuntimeId,
1482        state: &InputStatePersistenceRecord,
1483    ) -> Result<(), RuntimeStoreError>;
1484
1485    /// Load a single input state.
1486    async fn load_input_state(
1487        &self,
1488        runtime_id: &LogicalRuntimeId,
1489        input_id: &InputId,
1490    ) -> Result<Option<StoredInputState>, RuntimeStoreError>;
1491
1492    /// Load the last persisted machine lifecycle record bytes, if any.
1493    ///
1494    /// Implementations return only the opaque bytes previously obtained from
1495    /// [`MachineLifecycleCommit::store_record`]. The runtime crate decodes
1496    /// these bytes through `load_runtime_state` or internal recovery helpers;
1497    /// stores must not promote compatibility rows or bare runtime states into
1498    /// lifecycle authority.
1499    async fn load_machine_lifecycle_record(
1500        &self,
1501        runtime_id: &LogicalRuntimeId,
1502    ) -> Result<Option<Vec<u8>>, RuntimeStoreError>;
1503
1504    /// Atomically commit machine-owned lifecycle state changes.
1505    ///
1506    /// Writes runtime state, generated runtime binding facts, and all input
1507    /// state updates in a single atomic operation. `MachineLifecycleCommit` has
1508    /// no public constructor, so this cannot be used by compatibility callers
1509    /// to pick runtime truth.
1510    async fn commit_machine_lifecycle(
1511        &self,
1512        runtime_id: &LogicalRuntimeId,
1513        commit: MachineLifecycleCommit,
1514        input_states: &[InputStatePersistenceRecord],
1515    ) -> Result<(), RuntimeStoreError>;
1516
1517    /// Persist a snapshot of the ops lifecycle registry state.
1518    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    /// Load a previously persisted ops lifecycle snapshot.
1530    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    /// Delete a previously persisted ops lifecycle snapshot.
1539    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}