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

lash_core/store/
mod.rs
