lash_core/store/
mod.rs

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

lash_core/store/
mod.rs
