lash_core/store/
mod.rs

1//! The runtime's settled-session persistence contract and shared store types.
2
3pub mod queued_work;
4
5const PROC_BOOT_ID_PATH: &str = "/proc/sys/kernel/random/boot_id";
6
7fn default_root_session_id() -> String {
8    "root".to_string()
9}
10
11#[cfg(test)]
12mod persisted_state_tests {
13    use super::*;
14
15    #[test]
16    fn persisted_state_hydrates_provider_id_without_live_provider_rebinding() {
17        let state = persisted_session_state_from_head(
18            SessionHead {
19                session_id: "stored".to_string(),
20                head_revision: 7,
21                agent_frames: Vec::new(),
22                current_agent_frame_id: String::new(),
23                graph: crate::SessionGraph::default(),
24                config: crate::PersistedSessionConfig {
25                    provider_id: "stored-provider".to_string(),
26                    model: crate::ModelSpec::default(),
27                },
28                checkpoint_ref: None,
29                token_ledger: Vec::new(),
30            },
31            None,
32        );
33
34        assert_eq!(state.policy.recorded_provider_id(), "stored-provider");
35        assert!(
36            state
37                .agent_frames
38                .iter()
39                .all(|frame| frame.assignment.policy.recorded_provider_id() == "stored-provider")
40        );
41        assert_eq!(state.head_revision, Some(7));
42    }
43}
44
45#[derive(Debug, thiserror::Error)]
46pub enum StoreError {
47    #[error(
48        "store is already bound to session `{bound_session_id}` and cannot be reused for `{attempted_session_id}`"
49    )]
50    SessionBindingMismatch {
51        bound_session_id: String,
52        attempted_session_id: String,
53    },
54    #[error("store does not support read scope {0:?}")]
55    UnsupportedReadScope(SessionReadScope),
56    #[error("store head revision conflict: expected {expected:?}, actual {actual}")]
57    HeadRevisionConflict { expected: Option<u64>, actual: u64 },
58    #[error(
59        "runtime turn `{turn_id}` for session `{session_id}` was already committed with a different commit hash"
60    )]
61    RuntimeTurnCommitConflict { session_id: String, turn_id: String },
62    #[error("queued work claim `{claim_id}` for session `{session_id}` is missing or expired")]
63    QueuedWorkClaimExpired {
64        session_id: String,
65        claim_id: String,
66    },
67    #[error("turn input claim `{claim_id}` for session `{session_id}` is missing or expired")]
68    TurnInputClaimExpired {
69        session_id: String,
70        claim_id: String,
71    },
72    #[error("session execution lease for session `{session_id}` is missing or expired")]
73    SessionExecutionLeaseExpired { session_id: String },
74    #[error(
75        "{record_kind} schema_version {actual} is not supported by this binary (expected {expected})"
76    )]
77    UnsupportedRecordSchemaVersion {
78        record_kind: &'static str,
79        actual: u32,
80        expected: u32,
81    },
82    #[error("store backend error: {0}")]
83    Backend(String),
84}
85
86#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
87pub struct SessionMeta {
88    pub session_id: String,
89    pub session_name: String,
90    pub created_at: String,
91    pub model: String,
92    pub cwd: Option<String>,
93    pub relation: crate::SessionRelation,
94}
95
96impl SessionMeta {
97    /// Returns the parent session id, if any, derived from the canonical
98    /// [`SessionRelation`] field.
99    pub fn parent_session_id(&self) -> Option<&str> {
100        self.relation.parent_session_id()
101    }
102}
103
104/// Lightweight session info for the resume picker.
105#[derive(Clone, Debug)]
106pub struct SessionPickerInfo {
107    pub session_id: String,
108    pub cwd: Option<String>,
109    pub relation: crate::SessionRelation,
110    pub first_user_message: String,
111    pub user_message_count: usize,
112}
113
114impl SessionPickerInfo {
115    pub fn parent_session_id(&self) -> Option<&str> {
116        self.relation.parent_session_id()
117    }
118}
119
120#[derive(Clone, Debug, Default, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
121#[serde(transparent)]
122pub struct BlobRef(pub String);
123
124impl BlobRef {
125    pub fn as_str(&self) -> &str {
126        &self.0
127    }
128}
129
130impl std::fmt::Display for BlobRef {
131    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        f.write_str(&self.0)
133    }
134}
135
136impl From<String> for BlobRef {
137    fn from(value: String) -> Self {
138        Self(value)
139    }
140}
141
142#[derive(Clone, Debug, Default, PartialEq, Eq)]
143pub struct GcReport {
144    pub root_count: usize,
145    pub retained_blob_count: usize,
146    pub deleted_blob_count: usize,
147}
148
149/// Result of a `RuntimePersistence::vacuum()` call.
150/// `removed_node_count` counts the tombstoned graph-node rows that were
151/// physically deleted from the store. Returned so hosts can emit metrics.
152#[derive(Clone, Debug, Default, PartialEq, Eq)]
153pub struct VacuumReport {
154    pub removed_node_count: usize,
155}
156
157#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
158pub struct SessionCheckpoint {
159    #[serde(default)]
160    pub turn_state: crate::PersistedTurnState,
161    #[serde(default, skip_serializing_if = "Option::is_none")]
162    pub tool_state_ref: Option<BlobRef>,
163    #[serde(default, skip_serializing_if = "Option::is_none")]
164    pub plugin_snapshot_ref: Option<BlobRef>,
165    #[serde(default, skip_serializing_if = "Option::is_none")]
166    pub plugin_snapshot_revision: Option<u64>,
167    #[serde(default, skip_serializing_if = "Option::is_none")]
168    pub execution_state_ref: Option<BlobRef>,
169}
170
171#[derive(Clone, Debug, Default, serde::Serialize, serde::Deserialize)]
172pub struct HydratedSessionCheckpoint {
173    pub turn_state: crate::PersistedTurnState,
174    pub tool_state_ref: Option<BlobRef>,
175    pub tool_state: Option<crate::ToolState>,
176    pub plugin_snapshot_ref: Option<BlobRef>,
177    pub plugin_snapshot: Option<crate::PluginSessionSnapshot>,
178    pub plugin_snapshot_revision: Option<u64>,
179    pub execution_state_ref: Option<BlobRef>,
180    pub execution_state: Option<Vec<u8>>,
181}
182
183#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
184pub struct SessionHead {
185    #[serde(default = "default_root_session_id")]
186    pub session_id: String,
187    #[serde(default)]
188    pub head_revision: u64,
189    #[serde(default)]
190    pub agent_frames: Vec<crate::AgentFrameRecord>,
191    #[serde(default, skip_serializing_if = "String::is_empty")]
192    pub current_agent_frame_id: crate::AgentFrameId,
193    pub graph: crate::SessionGraph,
194    pub config: crate::PersistedSessionConfig,
195    #[serde(default, skip_serializing_if = "Option::is_none")]
196    pub checkpoint_ref: Option<BlobRef>,
197    #[serde(default, skip_serializing_if = "Vec::is_empty")]
198    pub token_ledger: Vec<crate::TokenLedgerEntry>,
199}
200
201#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
202pub struct SessionHeadMeta {
203    #[serde(default = "default_root_session_id")]
204    pub session_id: String,
205    #[serde(default)]
206    pub head_revision: u64,
207    pub config: crate::PersistedSessionConfig,
208    #[serde(default)]
209    pub agent_frames: Vec<crate::AgentFrameRecord>,
210    #[serde(default, skip_serializing_if = "String::is_empty")]
211    pub current_agent_frame_id: crate::AgentFrameId,
212    #[serde(default, skip_serializing_if = "Option::is_none")]
213    pub checkpoint_ref: Option<BlobRef>,
214    #[serde(default, skip_serializing_if = "Option::is_none")]
215    pub leaf_node_id: Option<String>,
216    #[serde(default)]
217    pub graph_node_count: usize,
218    #[serde(default, skip_serializing_if = "Vec::is_empty")]
219    pub token_ledger: Vec<crate::TokenLedgerEntry>,
220}
221
222fn persisted_session_config_from_state(
223    state: &crate::RuntimeSessionState,
224) -> crate::PersistedSessionConfig {
225    crate::PersistedSessionConfig {
226        provider_id: state.policy.recorded_provider_id().to_string(),
227        model: state.policy.model.clone(),
228    }
229}
230
231#[derive(Clone, Debug, PartialEq, Eq)]
232pub enum SessionReadScope {
233    FullGraph,
234    ActivePath { leaf_node_id: Option<String> },
235}
236
237#[derive(Clone, Debug)]
238pub struct PersistedSessionRead {
239    pub session_id: String,
240    pub head_revision: u64,
241    pub config: crate::PersistedSessionConfig,
242    pub agent_frames: Vec<crate::AgentFrameRecord>,
243    pub current_agent_frame_id: crate::AgentFrameId,
244    pub graph: crate::SessionGraph,
245    pub checkpoint_ref: Option<BlobRef>,
246    pub checkpoint: Option<HydratedSessionCheckpoint>,
247    pub token_ledger: Vec<crate::TokenLedgerEntry>,
248}
249
250#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
251pub enum GraphCommitDelta {
252    Unchanged {
253        leaf_node_id: Option<String>,
254    },
255    Append {
256        nodes: Vec<crate::SessionNodeRecord>,
257        leaf_node_id: Option<String>,
258    },
259    ReplaceFull(crate::SessionGraph),
260}
261
262impl GraphCommitDelta {
263    pub fn leaf_node_id(&self) -> Option<&String> {
264        match self {
265            Self::Unchanged { leaf_node_id } | Self::Append { leaf_node_id, .. } => {
266                leaf_node_id.as_ref()
267            }
268            Self::ReplaceFull(graph) => graph.leaf_node_id.as_ref(),
269        }
270    }
271}
272
273#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
274pub struct RuntimeCommit {
275    pub session_id: String,
276    pub expected_head_revision: Option<u64>,
277    #[serde(default, skip_serializing_if = "Option::is_none")]
278    pub session_execution_lease: Option<SessionExecutionLeaseFence>,
279    #[serde(default, skip_serializing_if = "Option::is_none")]
280    pub release_session_execution_lease: Option<SessionExecutionLeaseCompletion>,
281    pub config: crate::PersistedSessionConfig,
282    pub agent_frames: Vec<crate::AgentFrameRecord>,
283    pub current_agent_frame_id: crate::AgentFrameId,
284    pub graph: GraphCommitDelta,
285    pub checkpoint: HydratedSessionCheckpoint,
286    pub usage_deltas: Vec<crate::TokenLedgerEntry>,
287    pub turn_commit: Option<RuntimeTurnCommitStamp>,
288    pub completed_queue_claims: Vec<crate::QueuedWorkCompletion>,
289    pub completed_turn_input_claims: Vec<crate::TurnInputCompletion>,
290    #[serde(default, skip_serializing_if = "Option::is_none")]
291    pub interrupted_turn_input_turn_id: Option<String>,
292    /// Attachment ids whose bytes are referenced by this commit and
293    /// should be stamped `committed` in the write-ahead manifest as
294    /// part of the same SQL transaction. The backend marks each id
295    /// committed via [`AttachmentManifest::commit_refs`] before the
296    /// commit returns success. Hosts populate this from the
297    /// attachments emitted by tool calls and inline LLM-request
298    /// attachments produced during the turn.
299    pub committed_attachment_ids: Vec<crate::AttachmentId>,
300}
301
302#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
303pub struct RuntimeCommitResult {
304    pub head_revision: u64,
305    pub checkpoint_ref: BlobRef,
306    pub manifest: SessionCheckpoint,
307}
308
309#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
310pub struct LeaseOwnerIdentity {
311    pub owner_id: String,
312    pub incarnation_id: String,
313    #[serde(default)]
314    pub liveness: LeaseOwnerLiveness,
315}
316
317impl LeaseOwnerIdentity {
318    pub fn opaque(
319        owner_id: impl Into<String>,
320        incarnation_id: impl Into<String>,
321    ) -> LeaseOwnerIdentity {
322        LeaseOwnerIdentity {
323            owner_id: owner_id.into(),
324            incarnation_id: incarnation_id.into(),
325            liveness: LeaseOwnerLiveness::Opaque,
326        }
327    }
328
329    pub fn local_process(
330        owner_id: impl Into<String>,
331        incarnation_id: impl Into<String>,
332        host_id: impl Into<String>,
333    ) -> LeaseOwnerIdentity {
334        let liveness = LeaseOwnerLiveness::current_local_process(host_id.into())
335            .unwrap_or(LeaseOwnerLiveness::Opaque);
336        LeaseOwnerIdentity {
337            owner_id: owner_id.into(),
338            incarnation_id: incarnation_id.into(),
339            liveness,
340        }
341    }
342
343    pub fn same_incarnation(&self, other: &LeaseOwnerIdentity) -> bool {
344        self.owner_id == other.owner_id && self.incarnation_id == other.incarnation_id
345    }
346
347    pub fn is_definitely_dead_for_claimant(&self, claimant: &LeaseOwnerIdentity) -> bool {
348        self.liveness
349            .is_definitely_dead_for_claimant(&claimant.liveness)
350    }
351}
352
353#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize, Default)]
354#[serde(tag = "kind", rename_all = "snake_case")]
355pub enum LeaseOwnerLiveness {
356    LocalProcess {
357        host_id: String,
358        boot_id: String,
359        pid: u32,
360        process_start: String,
361    },
362    #[default]
363    Opaque,
364}
365
366impl LeaseOwnerLiveness {
367    pub fn current_local_process(host_id: impl Into<String>) -> Option<LeaseOwnerLiveness> {
368        let boot_id = std::fs::read_to_string(PROC_BOOT_ID_PATH)
369            .ok()
370            .map(|value| value.trim().to_string())
371            .filter(|value| !value.is_empty())?;
372        let pid = std::process::id();
373        let process_start = read_linux_process_start(pid)?;
374        Some(LeaseOwnerLiveness::LocalProcess {
375            host_id: host_id.into(),
376            boot_id,
377            pid,
378            process_start,
379        })
380    }
381
382    pub fn local_process_for_test(
383        host_id: impl Into<String>,
384        boot_id: impl Into<String>,
385        pid: u32,
386        process_start: impl Into<String>,
387    ) -> LeaseOwnerLiveness {
388        LeaseOwnerLiveness::LocalProcess {
389            host_id: host_id.into(),
390            boot_id: boot_id.into(),
391            pid,
392            process_start: process_start.into(),
393        }
394    }
395
396    pub fn is_definitely_dead_for_claimant(&self, claimant: &LeaseOwnerLiveness) -> bool {
397        let (
398            LeaseOwnerLiveness::LocalProcess {
399                host_id,
400                boot_id,
401                pid,
402                process_start,
403            },
404            LeaseOwnerLiveness::LocalProcess {
405                host_id: claimant_host_id,
406                boot_id: claimant_boot_id,
407                ..
408            },
409        ) = (self, claimant)
410        else {
411            return false;
412        };
413        if host_id != claimant_host_id || boot_id != claimant_boot_id {
414            return false;
415        }
416        matches!(linux_process_is_live(*pid, process_start), Some(false))
417    }
418}
419
420fn read_linux_process_start(pid: u32) -> Option<String> {
421    let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
422    parse_linux_process_start(&stat)
423}
424
425fn linux_process_is_live(pid: u32, expected_process_start: &str) -> Option<bool> {
426    match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
427        Ok(stat) => parse_linux_process_start(&stat).map(|start| start == expected_process_start),
428        Err(err) if err.kind() == std::io::ErrorKind::NotFound => Some(false),
429        Err(_) => None,
430    }
431}
432
433fn parse_linux_process_start(stat: &str) -> Option<String> {
434    let after_comm = stat.rsplit_once(") ")?.1;
435    after_comm.split_whitespace().nth(19).map(ToOwned::to_owned)
436}
437
438#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
439pub struct SessionExecutionLease {
440    pub session_id: String,
441    pub owner: LeaseOwnerIdentity,
442    pub lease_token: String,
443    pub fencing_token: u64,
444    pub claimed_at_epoch_ms: u64,
445    pub expires_at_epoch_ms: u64,
446}
447
448#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
449pub struct SessionExecutionLeaseFence {
450    pub session_id: String,
451    pub owner: LeaseOwnerIdentity,
452    pub lease_token: String,
453    pub fencing_token: u64,
454}
455
456#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
457pub struct SessionExecutionLeaseCompletion {
458    pub session_id: String,
459    pub owner: LeaseOwnerIdentity,
460    pub lease_token: String,
461    pub fencing_token: u64,
462}
463
464impl SessionExecutionLease {
465    pub fn fence(&self) -> SessionExecutionLeaseFence {
466        SessionExecutionLeaseFence {
467            session_id: self.session_id.clone(),
468            owner: self.owner.clone(),
469            lease_token: self.lease_token.clone(),
470            fencing_token: self.fencing_token,
471        }
472    }
473
474    pub fn completion(&self) -> SessionExecutionLeaseCompletion {
475        SessionExecutionLeaseCompletion {
476            session_id: self.session_id.clone(),
477            owner: self.owner.clone(),
478            lease_token: self.lease_token.clone(),
479            fencing_token: self.fencing_token,
480        }
481    }
482}
483
484impl SessionExecutionLeaseCompletion {
485    pub fn from_lease(lease: &SessionExecutionLease) -> Self {
486        lease.completion()
487    }
488}
489
490#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
491pub enum SessionExecutionLeaseClaimOutcome {
492    Acquired(SessionExecutionLease),
493    Busy { holder: SessionExecutionLease },
494}
495
496impl SessionExecutionLeaseClaimOutcome {
497    pub fn acquired(self) -> Option<SessionExecutionLease> {
498        match self {
499            Self::Acquired(lease) => Some(lease),
500            Self::Busy { .. } => None,
501        }
502    }
503}
504
505// =============================================================================
506// Attachment write-ahead manifest
507// =============================================================================
508
509/// A pending attachment write recorded *before* the bytes hit the
510/// [`AttachmentStore`](crate::AttachmentStore) backend.
511///
512/// The runtime calls [`AttachmentManifest::record_intent`] from the
513/// [`SessionScopedAttachmentStore`](crate::SessionScopedAttachmentStore)
514/// wrapper before each `put`, so the manifest is a durable record that
515/// "some bytes are about to land at this URI." When the turn that
516/// references the attachment commits successfully via
517/// [`RuntimePersistence::commit_runtime_state`], the same transaction
518/// stamps `committed_at_epoch_ms`. Periodic GC sweeps manifest rows
519/// whose intent has aged past a host-chosen threshold without ever
520/// being committed and deletes the corresponding bytes — that's how we
521/// reconcile orphaned files left behind by crashes between `put` and
522/// the next turn commit.
523#[derive(Clone, Debug)]
524pub struct AttachmentIntent {
525    pub attachment_id: crate::AttachmentId,
526    pub session_id: String,
527    /// Canonical URI for the attachment payload in the backing store.
528    /// For file-backed stores this is the absolute on-disk path; for
529    /// blob-backed stores it can be any stable identifier the host
530    /// uses to clean the payload up.
531    pub canonical_uri: String,
532    pub intent_at_epoch_ms: u64,
533}
534
535#[derive(Clone, Debug)]
536pub struct AttachmentManifestEntry {
537    pub attachment_id: crate::AttachmentId,
538    pub session_id: String,
539    pub canonical_uri: String,
540    pub intent_at_epoch_ms: u64,
541    pub committed_at_epoch_ms: Option<u64>,
542}
543
544/// Trait alias for the synchronous attachment-manifest surface on
545/// [`RuntimePersistence`]. Used by
546/// [`SessionScopedAttachmentStore`](crate::SessionScopedAttachmentStore)
547/// to record intent rows before `put` and by GC sweeps to reconcile
548/// orphans. See the [`AttachmentIntent`] doc comment for the full
549/// crash-safety story.
550///
551/// Backends with no attachment story (in-memory tests, mock stores)
552/// inherit the default no-op impls on [`RuntimePersistence`] and
553/// participate transparently — `record_intent` is a no-op, the
554/// scoped wrapper still works, and GC sweeps return empty.
555pub trait AttachmentManifest: Send + Sync {
556    fn record_intent(&self, intent: AttachmentIntent) -> Result<(), StoreError>;
557
558    /// Mark a set of attachment ids as committed (i.e. now referenced
559    /// by a durable session-graph commit). Backends that store
560    /// commits and manifest in the same database stamp this inside
561    /// the commit transaction; the trait-level method is the
562    /// out-of-band entry point for hosts that want to commit an id
563    /// outside the normal turn-commit flow.
564    fn commit_refs(
565        &self,
566        session_id: &str,
567        attachment_ids: &[crate::AttachmentId],
568    ) -> Result<(), StoreError>;
569
570    /// Return manifest entries whose intent has aged past
571    /// `older_than_epoch_ms` without ever being committed. Hosts run
572    /// this periodically to find orphans left by crashes between
573    /// `record_intent` and the next turn commit.
574    fn list_uncommitted(
575        &self,
576        older_than_epoch_ms: u64,
577    ) -> Result<Vec<AttachmentManifestEntry>, StoreError>;
578
579    /// Remove a manifest row entirely. Called by the GC coordinator
580    /// after the corresponding bytes have been removed from the
581    /// backing [`AttachmentStore`](crate::AttachmentStore).
582    fn forget(&self, attachment_id: &crate::AttachmentId) -> Result<(), StoreError>;
583}
584
585/// Mixin macro for [`RuntimePersistence`] implementors that have no
586/// attachment-write story (mock backends, in-memory test stores,
587/// runtime-perf harnesses). Pastes no-op impls of every
588/// [`AttachmentManifest`] method.
589#[macro_export]
590macro_rules! impl_noop_attachment_manifest {
591    ($ty:ty) => {
592        impl $crate::AttachmentManifest for $ty {
593            fn record_intent(
594                &self,
595                _intent: $crate::AttachmentIntent,
596            ) -> ::std::result::Result<(), $crate::StoreError> {
597                Ok(())
598            }
599
600            fn commit_refs(
601                &self,
602                _session_id: &str,
603                _attachment_ids: &[$crate::AttachmentId],
604            ) -> ::std::result::Result<(), $crate::StoreError> {
605                Ok(())
606            }
607
608            fn list_uncommitted(
609                &self,
610                _older_than_epoch_ms: u64,
611            ) -> ::std::result::Result<Vec<$crate::AttachmentManifestEntry>, $crate::StoreError>
612            {
613                Ok(Vec::new())
614            }
615
616            fn forget(
617                &self,
618                _attachment_id: &$crate::AttachmentId,
619            ) -> ::std::result::Result<(), $crate::StoreError> {
620                Ok(())
621            }
622        }
623    };
624}
625
626/// Reject a persisted record whose `schema_version` does not match the
627/// version this binary supports. Backends call this immediately after
628/// deserializing a record from durable storage.
629pub fn ensure_supported_schema_version(
630    record_kind: &'static str,
631    actual: u32,
632    expected: u32,
633) -> Result<(), StoreError> {
634    if actual == expected {
635        Ok(())
636    } else {
637        Err(StoreError::UnsupportedRecordSchemaVersion {
638            record_kind,
639            actual,
640            expected,
641        })
642    }
643}
644
645#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
646pub struct RuntimeTurnCommitStamp {
647    pub session_id: String,
648    pub turn_id: String,
649    pub turn_commit_hash: String,
650}
651
652impl RuntimeTurnCommitStamp {
653    pub fn new(
654        session_id: impl Into<String>,
655        turn_id: impl Into<String>,
656        turn_commit_hash: impl Into<String>,
657    ) -> Self {
658        Self {
659            session_id: session_id.into(),
660            turn_id: turn_id.into(),
661            turn_commit_hash: turn_commit_hash.into(),
662        }
663    }
664}
665
666fn build_persisted_turn_state(state: &crate::RuntimeSessionState) -> crate::PersistedTurnState {
667    crate::PersistedTurnState {
668        turn_index: state.turn_index,
669        token_usage: state.token_usage.clone(),
670        last_prompt_usage: state.last_prompt_usage.clone(),
671        protocol_turn_options: state.protocol_turn_options.clone(),
672    }
673}
674
675fn build_checkpoint_from_persisted_state(
676    state: &crate::RuntimeSessionState,
677) -> HydratedSessionCheckpoint {
678    HydratedSessionCheckpoint {
679        turn_state: build_persisted_turn_state(state),
680        tool_state_ref: state.tool_state_ref.clone(),
681        tool_state: state.tool_state_snapshot.clone(),
682        plugin_snapshot_ref: state.plugin_snapshot_ref.clone(),
683        plugin_snapshot_revision: state.plugin_snapshot_revision,
684        plugin_snapshot: state.plugin_snapshot.clone(),
685        execution_state_ref: state.execution_state_ref.clone(),
686        execution_state: state.execution_state_snapshot.clone(),
687    }
688}
689
690impl RuntimeCommit {
691    pub fn turn_commit_hash(&self) -> Result<String, StoreError> {
692        let mut semantic_commit = self.clone();
693        semantic_commit.expected_head_revision = None;
694        semantic_commit.session_execution_lease = None;
695        semantic_commit.release_session_execution_lease = None;
696        semantic_commit.turn_commit = None;
697        let mut semantic_commit = serde_json::to_value(&semantic_commit).map_err(|err| {
698            StoreError::Backend(format!("failed to serialize runtime turn commit: {err}"))
699        })?;
700        scrub_turn_commit_hash_value(&mut semantic_commit);
701        crate::stable_hash::stable_json_sha256_hex(&semantic_commit).map_err(|err| {
702            StoreError::Backend(format!(
703                "failed to serialize runtime turn commit hash: {err}"
704            ))
705        })
706    }
707
708    pub fn persisted_state(
709        state: &crate::RuntimeSessionState,
710        usage_deltas: &[crate::TokenLedgerEntry],
711    ) -> Self {
712        Self {
713            session_id: state.session_id.clone(),
714            expected_head_revision: state.head_revision,
715            session_execution_lease: None,
716            release_session_execution_lease: None,
717            config: persisted_session_config_from_state(state),
718            agent_frames: state.agent_frames.clone(),
719            current_agent_frame_id: state.current_agent_frame_id.clone(),
720            graph: if state.graph_replace_required || state.head_revision.is_none() {
721                GraphCommitDelta::ReplaceFull(state.session_graph.clone())
722            } else {
723                GraphCommitDelta::Unchanged {
724                    leaf_node_id: state.session_graph.leaf_node_id.clone(),
725                }
726            },
727            checkpoint: build_checkpoint_from_persisted_state(state),
728            usage_deltas: usage_deltas.to_vec(),
729            turn_commit: None,
730            completed_queue_claims: Vec::new(),
731            completed_turn_input_claims: Vec::new(),
732            interrupted_turn_input_turn_id: None,
733            committed_attachment_ids: Vec::new(),
734        }
735    }
736
737    pub(crate) fn persisted_state_with_graph_commit(
738        state: &crate::RuntimeSessionState,
739        graph: GraphCommitDelta,
740        usage_deltas: &[crate::TokenLedgerEntry],
741    ) -> Self {
742        Self {
743            session_id: state.session_id.clone(),
744            expected_head_revision: state.head_revision,
745            session_execution_lease: None,
746            release_session_execution_lease: None,
747            config: persisted_session_config_from_state(state),
748            agent_frames: state.agent_frames.clone(),
749            current_agent_frame_id: state.current_agent_frame_id.clone(),
750            graph,
751            checkpoint: build_checkpoint_from_persisted_state(state),
752            usage_deltas: usage_deltas.to_vec(),
753            turn_commit: None,
754            completed_queue_claims: Vec::new(),
755            completed_turn_input_claims: Vec::new(),
756            interrupted_turn_input_turn_id: None,
757            committed_attachment_ids: Vec::new(),
758        }
759    }
760
761    pub fn with_turn_commit(mut self, turn_commit: RuntimeTurnCommitStamp) -> Self {
762        self.turn_commit = Some(turn_commit);
763        self
764    }
765
766    pub fn with_session_execution_lease(mut self, lease: SessionExecutionLeaseFence) -> Self {
767        self.session_execution_lease = Some(lease);
768        self
769    }
770
771    pub fn releasing_session_execution_lease(
772        mut self,
773        completion: SessionExecutionLeaseCompletion,
774    ) -> Self {
775        self.release_session_execution_lease = Some(completion);
776        self
777    }
778
779    pub fn completing_queue_claim(
780        mut self,
781        completed_queue_claim: crate::QueuedWorkCompletion,
782    ) -> Self {
783        self.completed_queue_claims.push(completed_queue_claim);
784        self
785    }
786
787    pub fn completing_queue_claims(
788        mut self,
789        completed_queue_claims: impl IntoIterator<Item = crate::QueuedWorkCompletion>,
790    ) -> Self {
791        self.completed_queue_claims.extend(completed_queue_claims);
792        self
793    }
794
795    pub fn completing_turn_input_claim(
796        mut self,
797        completed_turn_input_claim: crate::TurnInputCompletion,
798    ) -> Self {
799        self.completed_turn_input_claims
800            .push(completed_turn_input_claim);
801        self
802    }
803
804    pub fn completing_turn_input_claims(
805        mut self,
806        completed_turn_input_claims: impl IntoIterator<Item = crate::TurnInputCompletion>,
807    ) -> Self {
808        self.completed_turn_input_claims
809            .extend(completed_turn_input_claims);
810        self
811    }
812
813    pub fn deferring_interrupted_turn_inputs(mut self, turn_id: impl Into<String>) -> Self {
814        self.interrupted_turn_input_turn_id = Some(turn_id.into());
815        self
816    }
817
818    pub fn with_committed_attachments(
819        mut self,
820        attachment_ids: impl IntoIterator<Item = crate::AttachmentId>,
821    ) -> Self {
822        self.committed_attachment_ids = attachment_ids.into_iter().collect();
823        self
824    }
825}
826
827fn scrub_turn_commit_hash_value(value: &mut serde_json::Value) {
828    match value {
829        serde_json::Value::Object(map) => {
830            let is_message = map.contains_key("role") && map.contains_key("parts");
831            let is_message_part = map.contains_key("kind")
832                && map.contains_key("content")
833                && map.contains_key("prune_state");
834            if is_message || is_message_part {
835                map.remove("id");
836            }
837            for volatile_key in ["node_id", "parent_node_id", "leaf_node_id", "timestamp"] {
838                map.remove(volatile_key);
839            }
840            for child in map.values_mut() {
841                scrub_turn_commit_hash_value(child);
842            }
843        }
844        serde_json::Value::Array(items) => {
845            for item in items {
846                scrub_turn_commit_hash_value(item);
847            }
848        }
849        _ => {}
850    }
851}
852
853fn persisted_session_state_from_head(
854    head: SessionHead,
855    checkpoint: Option<HydratedSessionCheckpoint>,
856) -> crate::RuntimeSessionState {
857    let mut state = crate::RuntimeSessionState {
858        session_id: head.session_id,
859        policy: crate::SessionPolicy::default(),
860        agent_frames: head.agent_frames,
861        current_agent_frame_id: head.current_agent_frame_id,
862        session_graph: head.graph,
863        turn_index: 0,
864        token_usage: crate::TokenUsage::default(),
865        last_prompt_usage: None,
866        protocol_turn_options: crate::ProtocolTurnOptions::default(),
867        tool_state_ref: None,
868        tool_state_generation: None,
869        tool_state_snapshot: None,
870        plugin_snapshot_ref: None,
871        plugin_snapshot_revision: None,
872        plugin_snapshot: None,
873        execution_state_ref: None,
874        execution_state_snapshot: None,
875        token_ledger: head.token_ledger,
876        checkpoint_ref: head.checkpoint_ref.clone(),
877        head_revision: Some(head.head_revision),
878        graph_replace_required: false,
879    };
880    state.policy.model = head.config.model.clone();
881    state.policy.provider_id = head.config.provider_id.clone();
882    if let Some(checkpoint) = checkpoint {
883        state.turn_index = checkpoint.turn_state.turn_index;
884        state.token_usage = checkpoint.turn_state.token_usage;
885        state.last_prompt_usage = checkpoint.turn_state.last_prompt_usage;
886        state.protocol_turn_options = checkpoint.turn_state.protocol_turn_options;
887        state.tool_state_ref = checkpoint.tool_state_ref.clone();
888        state.tool_state_generation = checkpoint
889            .tool_state
890            .as_ref()
891            .map(|snapshot| snapshot.generation());
892        state.tool_state_snapshot = checkpoint.tool_state;
893        state.plugin_snapshot_ref = checkpoint.plugin_snapshot_ref.clone();
894        state.plugin_snapshot_revision = checkpoint.plugin_snapshot_revision;
895        state.plugin_snapshot = checkpoint.plugin_snapshot;
896        state.execution_state_ref = checkpoint.execution_state_ref.clone();
897        state.execution_state_snapshot = checkpoint.execution_state;
898    }
899    state.ensure_agent_frame_initialized();
900    state
901}
902
903impl Default for SessionHead {
904    fn default() -> Self {
905        Self {
906            session_id: default_root_session_id(),
907            head_revision: 0,
908            agent_frames: Vec::new(),
909            current_agent_frame_id: String::new(),
910            graph: crate::SessionGraph::default(),
911            config: crate::PersistedSessionConfig::default(),
912            checkpoint_ref: None,
913            token_ledger: Vec::new(),
914        }
915    }
916}
917
918impl Default for SessionHeadMeta {
919    fn default() -> Self {
920        Self {
921            session_id: default_root_session_id(),
922            head_revision: 0,
923            config: crate::PersistedSessionConfig::default(),
924            agent_frames: Vec::new(),
925            current_agent_frame_id: String::new(),
926            checkpoint_ref: None,
927            leaf_node_id: None,
928            graph_node_count: 0,
929            token_ledger: Vec::new(),
930        }
931    }
932}
933
934/// Exact settled-session persistence protocol required by the runtime.
935///
936/// This is the runtime's atomic transaction facade for visible session state:
937/// session graph/head commits, queued-work ingress and completion, final
938/// turn-commit idempotency, metadata, usage, and the attachment write-ahead
939/// manifest. In-flight nondeterministic work belongs to the active
940/// [`EffectHost`](crate::EffectHost), not to the store contract.
941///
942/// The [`AttachmentManifest`] supertrait is required so the runtime can wrap
943/// any persistence backend with a
944/// [`SessionScopedAttachmentStore`](crate::SessionScopedAttachmentStore)
945/// without dual-trait casting. Backends with no attachment-write story can
946/// implement the manifest methods as no-ops via
947/// [`NoopAttachmentManifest`]'s blanket helpers.
948#[async_trait::async_trait]
949pub trait RuntimePersistence: AttachmentManifest + Send + Sync {
950    /// Durability tier this session store provides; defaults to
951    /// [`DurabilityTier::Inline`].
952    fn durability_tier(&self) -> crate::DurabilityTier {
953        crate::DurabilityTier::Inline
954    }
955
956    async fn load_session(
957        &self,
958        scope: SessionReadScope,
959    ) -> Result<Option<PersistedSessionRead>, StoreError>;
960
961    async fn load_node(
962        &self,
963        node_id: &str,
964    ) -> Result<Option<crate::SessionNodeRecord>, StoreError>;
965
966    async fn commit_runtime_state(
967        &self,
968        commit: RuntimeCommit,
969    ) -> Result<RuntimeCommitResult, StoreError>;
970
971    /// Persist model-visible user input into the pending turn-input lifecycle.
972    ///
973    /// Active-turn ingress is claimed only by the matching live turn at a
974    /// checkpoint. Next-turn ingress is claimed only by idle dispatch. User
975    /// input must not be represented as generic queued work.
976    async fn enqueue_pending_turn_input(
977        &self,
978        _input: crate::PendingTurnInputDraft,
979    ) -> Result<crate::PendingTurnInput, StoreError> {
980        Err(StoreError::Backend(
981            "pending turn input is not supported by this test store".to_string(),
982        ))
983    }
984
985    /// List pending user inputs for UI reconciliation and queue preview.
986    ///
987    /// This excludes completed/cancelled rows and rows currently held by a live
988    /// claim. Expired claims are visible again according to their state.
989    async fn list_pending_turn_inputs(
990        &self,
991        _session_id: &str,
992    ) -> Result<Vec<crate::PendingTurnInput>, StoreError> {
993        Ok(Vec::new())
994    }
995
996    /// Cancel an unclaimed pending user input by id.
997    async fn cancel_pending_turn_input(
998        &self,
999        _session_id: &str,
1000        _input_id: &str,
1001    ) -> Result<Option<crate::PendingTurnInput>, StoreError> {
1002        Ok(None)
1003    }
1004
1005    /// Claim active-turn input at a checkpoint for the live turn id.
1006    async fn claim_active_turn_inputs(
1007        &self,
1008        session_id: &str,
1009        _session_execution_lease: &SessionExecutionLeaseFence,
1010        _owner: &LeaseOwnerIdentity,
1011        _turn_id: &str,
1012        _checkpoint: crate::CheckpointKind,
1013        _lease_ttl_ms: u64,
1014        _max_inputs: usize,
1015    ) -> Result<Option<crate::TurnInputClaim>, StoreError> {
1016        Err(StoreError::Backend(format!(
1017            "pending turn input is not supported for session `{session_id}` by this test store"
1018        )))
1019    }
1020
1021    /// Claim queued next-turn input at idle.
1022    async fn claim_next_turn_inputs(
1023        &self,
1024        session_id: &str,
1025        _session_execution_lease: &SessionExecutionLeaseFence,
1026        _owner: &LeaseOwnerIdentity,
1027        _lease_ttl_ms: u64,
1028        _max_inputs: usize,
1029    ) -> Result<Option<crate::TurnInputClaim>, StoreError> {
1030        Err(StoreError::Backend(format!(
1031            "pending turn input is not supported for session `{session_id}` by this test store"
1032        )))
1033    }
1034
1035    /// Abandon a held pending-turn-input claim so it can be reclaimed.
1036    async fn abandon_turn_input_claim(
1037        &self,
1038        _claim: &crate::TurnInputClaim,
1039    ) -> Result<(), StoreError> {
1040        Ok(())
1041    }
1042
1043    /// Try to claim the durable single-writer execution lane for `session_id`.
1044    ///
1045    /// Returns [`SessionExecutionLeaseClaimOutcome::Busy`] when another owner
1046    /// holds an unexpired lease. Expired or released leases may be reclaimed
1047    /// and receive a higher fencing token. An unexpired lease held by the same
1048    /// owner id but a different incarnation is busy.
1049    async fn try_claim_session_execution_lease(
1050        &self,
1051        session_id: &str,
1052        owner: &LeaseOwnerIdentity,
1053        lease_ttl_ms: u64,
1054    ) -> Result<SessionExecutionLeaseClaimOutcome, StoreError>;
1055
1056    /// Reclaim an unexpired session execution lease whose observed holder is
1057    /// definitely dead according to persisted local-process liveness metadata.
1058    ///
1059    /// Backends must CAS on `observed_holder` so a stale claimant cannot clear
1060    /// a newer live lease that won the race after the busy observation.
1061    async fn reclaim_session_execution_lease(
1062        &self,
1063        session_id: &str,
1064        owner: &LeaseOwnerIdentity,
1065        observed_holder: &SessionExecutionLeaseFence,
1066        lease_ttl_ms: u64,
1067    ) -> Result<SessionExecutionLeaseClaimOutcome, StoreError>;
1068
1069    /// Extend a live session execution lease owned by the caller.
1070    ///
1071    /// Backends must reject stale, released, superseded, or expired fences with
1072    /// [`StoreError::SessionExecutionLeaseExpired`].
1073    async fn renew_session_execution_lease(
1074        &self,
1075        fence: &SessionExecutionLeaseFence,
1076        lease_ttl_ms: u64,
1077    ) -> Result<SessionExecutionLease, StoreError>;
1078
1079    /// Release a session execution lease fenced by its completion token.
1080    ///
1081    /// This operation is idempotent and must not clear a newer owner's lease.
1082    async fn release_session_execution_lease(
1083        &self,
1084        completion: &SessionExecutionLeaseCompletion,
1085    ) -> Result<(), StoreError>;
1086
1087    /// Persist a queued-work batch for later claiming.
1088    ///
1089    /// The default implementation rejects the batch: backends that do not
1090    /// support queued work inherit it and stay loud rather than silently
1091    /// dropping work.
1092    async fn enqueue_queued_work(
1093        &self,
1094        _batch: crate::QueuedWorkBatchDraft,
1095    ) -> Result<crate::QueuedWorkBatch, StoreError> {
1096        Err(StoreError::Backend(
1097            "queued work is not supported by this test store".to_string(),
1098        ))
1099    }
1100
1101    /// Claim a leading ready session-command batch for `owner_id`.
1102    ///
1103    /// A command claim is returned only when the earliest ready claimable batch
1104    /// is classified as [`crate::runtime::QueuedWorkClass::SessionCommand`].
1105    /// Backends derive the class from queued payloads; no schema column is
1106    /// required.
1107    /// The default implementation reports queued work as unsupported.
1108    async fn claim_leading_ready_session_command(
1109        &self,
1110        session_id: &str,
1111        _session_execution_lease: &SessionExecutionLeaseFence,
1112        _owner: &LeaseOwnerIdentity,
1113        _lease_ttl_ms: u64,
1114    ) -> Result<Option<crate::QueuedWorkClaim>, StoreError> {
1115        Err(StoreError::Backend(format!(
1116            "queued work is not supported for session `{session_id}` by this test store"
1117        )))
1118    }
1119
1120    /// Claim the next ready turn-work group for `owner_id`.
1121    ///
1122    /// A turn-work claim is returned only when the earliest ready claimable
1123    /// batch is classified as [`crate::runtime::QueuedWorkClass::TurnWork`].
1124    /// Earlier ready session commands are not skipped and are never
1125    /// materialized as turn input.
1126    ///
1127    /// The default implementation reports queued work as unsupported.
1128    async fn claim_ready_queued_work(
1129        &self,
1130        session_id: &str,
1131        _session_execution_lease: &SessionExecutionLeaseFence,
1132        _owner: &LeaseOwnerIdentity,
1133        _boundary: crate::QueuedWorkClaimBoundary,
1134        _lease_ttl_ms: u64,
1135        _max_batches: usize,
1136    ) -> Result<Option<crate::QueuedWorkClaim>, StoreError> {
1137        Err(StoreError::Backend(format!(
1138            "queued work is not supported for session `{session_id}` by this test store"
1139        )))
1140    }
1141
1142    /// Claim a specific ready batch set selected from the durable queue.
1143    ///
1144    /// This is the host-facing counterpart to [`claim_ready_queued_work`]:
1145    /// callers that project queued work into a UI can claim the exact batch ids
1146    /// they rendered instead of reconstructing authority from local draft state.
1147    /// The default implementation preserves the ordered queue contract by
1148    /// claiming the next ready group and returning it only when the durable ids
1149    /// match exactly.
1150    async fn claim_ready_queued_work_by_batch_ids(
1151        &self,
1152        session_id: &str,
1153        session_execution_lease: &SessionExecutionLeaseFence,
1154        owner: &LeaseOwnerIdentity,
1155        boundary: crate::QueuedWorkClaimBoundary,
1156        lease_ttl_ms: u64,
1157        batch_ids: &[String],
1158    ) -> Result<Option<crate::QueuedWorkClaim>, StoreError> {
1159        if batch_ids.is_empty() {
1160            return Ok(None);
1161        }
1162        let Some(claim) = self
1163            .claim_ready_queued_work(
1164                session_id,
1165                session_execution_lease,
1166                owner,
1167                boundary,
1168                lease_ttl_ms,
1169                batch_ids.len(),
1170            )
1171            .await?
1172        else {
1173            return Ok(None);
1174        };
1175        let claimed_ids = claim
1176            .batches
1177            .iter()
1178            .map(|batch| batch.batch_id.as_str())
1179            .collect::<Vec<_>>();
1180        if claimed_ids == batch_ids.iter().map(String::as_str).collect::<Vec<_>>() {
1181            return Ok(Some(claim));
1182        }
1183        self.abandon_queued_work_claim(&claim).await?;
1184        Ok(None)
1185    }
1186
1187    /// Extend the lease on a held queued-work claim.
1188    ///
1189    /// The default implementation reports the claim as expired, matching a
1190    /// backend that never granted one.
1191    async fn renew_queued_work_claim(
1192        &self,
1193        claim: &crate::QueuedWorkClaim,
1194        _lease_ttl_ms: u64,
1195    ) -> Result<crate::QueuedWorkClaim, StoreError> {
1196        Err(StoreError::QueuedWorkClaimExpired {
1197            session_id: claim.session_id.clone(),
1198            claim_id: claim.claim_id.clone(),
1199        })
1200    }
1201
1202    /// Release a held queued-work claim without completing it.
1203    ///
1204    /// The default implementation is a no-op: with no queued work there is
1205    /// nothing to release.
1206    async fn abandon_queued_work_claim(
1207        &self,
1208        _claim: &crate::QueuedWorkClaim,
1209    ) -> Result<(), StoreError> {
1210        Ok(())
1211    }
1212
1213    /// Remove an unclaimed queued-work batch from durable ingress.
1214    ///
1215    /// Returns the removed batch when cancellation won the race. Returns `None`
1216    /// when the batch is missing or currently held by a live claim; callers must
1217    /// treat that as "already claimed or completed" and must not restore any
1218    /// stale local draft state.
1219    ///
1220    /// The default implementation reports `None` (nothing queued, nothing to
1221    /// cancel).
1222    async fn cancel_queued_work_batch(
1223        &self,
1224        _session_id: &str,
1225        _batch_id: &str,
1226    ) -> Result<Option<crate::QueuedWorkBatch>, StoreError> {
1227        Ok(None)
1228    }
1229
1230    /// List all queued-work batches for a session.
1231    ///
1232    /// The default implementation reports an empty queue.
1233    async fn list_queued_work(
1234        &self,
1235        _session_id: &str,
1236    ) -> Result<Vec<crate::QueuedWorkBatch>, StoreError> {
1237        Ok(Vec::new())
1238    }
1239
1240    /// List queued-work batches that are still pending presentation/editing.
1241    ///
1242    /// This excludes batches currently held by a live claim. Expired claims are
1243    /// considered pending again because they can be reclaimed or cancelled.
1244    async fn list_pending_queued_work(
1245        &self,
1246        session_id: &str,
1247    ) -> Result<Vec<crate::QueuedWorkBatch>, StoreError> {
1248        self.list_queued_work(session_id).await
1249    }
1250
1251    async fn save_session_meta(&self, meta: SessionMeta) -> Result<(), StoreError>;
1252    async fn load_session_meta(&self) -> Result<Option<SessionMeta>, StoreError>;
1253
1254    async fn tombstone_nodes(&self, ids: &[String]) -> Result<(), StoreError>;
1255    async fn vacuum(&self) -> Result<VacuumReport, StoreError>;
1256    async fn gc_unreachable(&self) -> Result<GcReport, StoreError>;
1257}
1258
1259fn persisted_session_state_from_read(read: PersistedSessionRead) -> crate::RuntimeSessionState {
1260    persisted_session_state_from_head(
1261        SessionHead {
1262            session_id: read.session_id,
1263            head_revision: read.head_revision,
1264            agent_frames: read.agent_frames,
1265            current_agent_frame_id: read.current_agent_frame_id,
1266            graph: read.graph,
1267            config: read.config,
1268            checkpoint_ref: read.checkpoint_ref,
1269            token_ledger: read.token_ledger,
1270        },
1271        read.checkpoint,
1272    )
1273}
1274
1275pub async fn load_persisted_session_state(
1276    store: &(dyn RuntimePersistence + '_),
1277) -> Result<Option<crate::RuntimeSessionState>, StoreError> {
1278    Ok(store
1279        .load_session(SessionReadScope::FullGraph)
1280        .await?
1281        .map(persisted_session_state_from_read))
1282}
1283
1284pub async fn load_persisted_session_state_active_path(
1285    store: &(dyn RuntimePersistence + '_),
1286    leaf_node_id: Option<String>,
1287) -> Result<Option<crate::RuntimeSessionState>, StoreError> {
1288    Ok(store
1289        .load_session(SessionReadScope::ActivePath { leaf_node_id })
1290        .await?
1291        .map(persisted_session_state_from_read))
1292}
1293
1294pub async fn refresh_persisted_session_state(
1295    store: &(dyn RuntimePersistence + '_),
1296    state: &mut crate::RuntimeSessionState,
1297) -> Result<(), StoreError> {
1298    if let Some(mut fresh) = load_persisted_session_state(store).await? {
1299        fresh.policy.session_id = state.policy.session_id.clone();
1300        fresh.policy.max_turns = state.policy.max_turns;
1301        *state = fresh;
1302    }
1303    Ok(())
1304}
1305
1306#[cfg(test)]
1307mod tests {
1308    use super::{LeaseOwnerIdentity, LeaseOwnerLiveness};
1309
1310    fn local_liveness(
1311        host_id: &str,
1312        boot_id: &str,
1313        pid: u32,
1314        process_start: &str,
1315    ) -> LeaseOwnerLiveness {
1316        LeaseOwnerLiveness::local_process_for_test(host_id, boot_id, pid, process_start)
1317    }
1318
1319    #[test]
1320    fn lease_owner_identity_requires_same_incarnation() {
1321        let first = LeaseOwnerIdentity::opaque("owner", "incarnation-a");
1322        let same = LeaseOwnerIdentity::opaque("owner", "incarnation-a");
1323        let next = LeaseOwnerIdentity::opaque("owner", "incarnation-b");
1324
1325        assert!(first.same_incarnation(&same));
1326        assert!(!first.same_incarnation(&next));
1327    }
1328
1329    #[test]
1330    fn local_liveness_only_proves_same_host_boot_dead_processes() {
1331        let holder = local_liveness(
1332            "host-a",
1333            "boot-a",
1334            std::process::id(),
1335            "not-the-current-process-start",
1336        );
1337        let same_host_boot = local_liveness("host-a", "boot-a", std::process::id(), "claimant");
1338        let other_host = local_liveness("host-b", "boot-a", std::process::id(), "claimant");
1339        let other_boot = local_liveness("host-a", "boot-b", std::process::id(), "claimant");
1340
1341        assert!(holder.is_definitely_dead_for_claimant(&same_host_boot));
1342        assert!(!holder.is_definitely_dead_for_claimant(&other_host));
1343        assert!(!holder.is_definitely_dead_for_claimant(&other_boot));
1344        assert!(!holder.is_definitely_dead_for_claimant(&LeaseOwnerLiveness::Opaque));
1345        assert!(!LeaseOwnerLiveness::Opaque.is_definitely_dead_for_claimant(&same_host_boot));
1346    }
1347}
lash_core/store/mod.rs

lash_core/store/
mod.rs
