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

lash_core/store/
mod.rs
