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

lash_core/store/
mod.rs
