starweaver_session/store/
contract.rs

1//! Durable session store contract.
2
3use async_trait::async_trait;
4use starweaver_context::ResumableState;
5use starweaver_core::{RunId, SessionId};
6use starweaver_runtime::{AgentCheckpoint, AgentStreamRecord};
7
8use crate::{
9    approval::{ApprovalRecord, DeferredToolRecord},
10    error::SessionStoreResult,
11    records::{
12        EnvironmentStateRef, RunRecord, RunStatus, SessionRecord, SessionStatus, StreamCursorRef,
13    },
14    resume::SessionResumeSnapshot,
15    trace::{CompactRunTrace, CompactSessionTrace},
16};
17
18/// Query filters for listing sessions.
19#[derive(Clone, Debug, Default, Eq, PartialEq)]
20pub struct SessionFilter {
21    /// Required session status.
22    pub status: Option<SessionStatus>,
23    /// Required profile name.
24    pub profile: Option<String>,
25    /// Required workspace identifier or path.
26    pub workspace: Option<String>,
27    /// Maximum number of sessions returned.
28    pub limit: Option<usize>,
29}
30
31/// Durable session store contract.
32#[async_trait]
33pub trait SessionStore: Send + Sync {
34    /// Save a session record.
35    async fn save_session(&self, session: SessionRecord) -> SessionStoreResult<()>;
36
37    /// Load a session record.
38    async fn load_session(&self, session_id: &SessionId) -> SessionStoreResult<SessionRecord>;
39
40    /// List sessions by optional filter.
41    async fn list_sessions(&self, filter: SessionFilter) -> SessionStoreResult<Vec<SessionRecord>>;
42
43    /// Update session status.
44    async fn update_session_status(
45        &self,
46        session_id: &SessionId,
47        status: SessionStatus,
48    ) -> SessionStoreResult<()>;
49
50    /// Save a context state snapshot for a session.
51    async fn save_context_state(
52        &self,
53        session_id: &SessionId,
54        state: ResumableState,
55    ) -> SessionStoreResult<()>;
56
57    /// Save an environment state reference for a session.
58    async fn save_environment_state(
59        &self,
60        session_id: &SessionId,
61        environment_state: EnvironmentStateRef,
62    ) -> SessionStoreResult<()>;
63
64    /// Append or replace a run record.
65    async fn append_run(&self, run: RunRecord) -> SessionStoreResult<()>;
66
67    /// Load a run record.
68    async fn load_run(
69        &self,
70        session_id: &SessionId,
71        run_id: &RunId,
72    ) -> SessionStoreResult<RunRecord>;
73
74    /// List runs for a session.
75    async fn list_runs(&self, session_id: &SessionId) -> SessionStoreResult<Vec<RunRecord>>;
76
77    /// Update run status and optional output preview.
78    async fn update_run_status(
79        &self,
80        session_id: &SessionId,
81        run_id: &RunId,
82        status: RunStatus,
83        output_preview: Option<String>,
84    ) -> SessionStoreResult<()>;
85
86    /// Append a full runtime checkpoint.
87    async fn append_checkpoint(
88        &self,
89        session_id: &SessionId,
90        checkpoint: AgentCheckpoint,
91    ) -> SessionStoreResult<()>;
92
93    /// Load checkpoints for a run in insertion order.
94    async fn load_checkpoints(
95        &self,
96        session_id: &SessionId,
97        run_id: &RunId,
98    ) -> SessionStoreResult<Vec<AgentCheckpoint>>;
99
100    /// Load the latest checkpoint for a run.
101    async fn latest_checkpoint(
102        &self,
103        session_id: &SessionId,
104        run_id: &RunId,
105    ) -> SessionStoreResult<Option<AgentCheckpoint>> {
106        let checkpoints = self.load_checkpoints(session_id, run_id).await?;
107        Ok(checkpoints.into_iter().last())
108    }
109
110    /// Append runtime stream records used as resume evidence.
111    async fn append_stream_records(
112        &self,
113        session_id: &SessionId,
114        run_id: &RunId,
115        records: Vec<AgentStreamRecord>,
116    ) -> SessionStoreResult<()>;
117
118    /// Replay runtime stream records for a run.
119    async fn replay_stream_records(
120        &self,
121        session_id: &SessionId,
122        run_id: &RunId,
123    ) -> SessionStoreResult<Vec<AgentStreamRecord>>;
124
125    /// Replay runtime stream records after a sequence cursor.
126    async fn replay_stream_records_after(
127        &self,
128        session_id: &SessionId,
129        run_id: &RunId,
130        after_sequence: Option<usize>,
131    ) -> SessionStoreResult<Vec<AgentStreamRecord>> {
132        let records = self.replay_stream_records(session_id, run_id).await?;
133        Ok(records
134            .into_iter()
135            .filter(|record| after_sequence.is_none_or(|cursor| record.sequence > cursor))
136            .collect())
137    }
138
139    /// Store a stream cursor reference for a run and session.
140    async fn save_stream_cursor(
141        &self,
142        session_id: &SessionId,
143        run_id: &RunId,
144        cursor: StreamCursorRef,
145    ) -> SessionStoreResult<()>;
146
147    /// Append an approval record.
148    async fn append_approval(&self, approval: ApprovalRecord) -> SessionStoreResult<()>;
149
150    /// Load approval records for a run.
151    async fn load_approvals(
152        &self,
153        session_id: &SessionId,
154        run_id: &RunId,
155    ) -> SessionStoreResult<Vec<ApprovalRecord>>;
156
157    /// Append a deferred tool record.
158    async fn append_deferred_tool(&self, record: DeferredToolRecord) -> SessionStoreResult<()>;
159
160    /// Load deferred tool records for a run.
161    async fn load_deferred_tools(
162        &self,
163        session_id: &SessionId,
164        run_id: &RunId,
165    ) -> SessionStoreResult<Vec<DeferredToolRecord>>;
166
167    /// Load a resume snapshot from session, checkpoint, and stream evidence.
168    async fn resume_snapshot(
169        &self,
170        session_id: &SessionId,
171        run_id: &RunId,
172    ) -> SessionStoreResult<SessionResumeSnapshot> {
173        let session = self.load_session(session_id).await?;
174        let run = self.load_run(session_id, run_id).await?;
175        let latest_checkpoint = self.latest_checkpoint(session_id, run_id).await?;
176        let after_sequence = latest_checkpoint
177            .as_ref()
178            .and_then(|checkpoint| checkpoint.resume.cursor.stream_cursor);
179        let stream_records = self
180            .replay_stream_records_after(session_id, run_id, after_sequence)
181            .await?;
182        let approvals = self.load_approvals(session_id, run_id).await?;
183        let deferred_tools = self.load_deferred_tools(session_id, run_id).await?;
184        let environment_state = run
185            .environment_state
186            .clone()
187            .or_else(|| session.environment_state.clone());
188        let mut stream_cursors = session.stream_cursors.clone();
189        stream_cursors.extend(run.stream_cursors.clone());
190        Ok(SessionResumeSnapshot {
191            state: session.state.clone(),
192            session,
193            run,
194            environment_state,
195            latest_checkpoint,
196            stream_records,
197            approvals,
198            deferred_tools,
199            stream_cursors,
200        })
201    }
202
203    /// Return compact run trace projection.
204    async fn compact_run_trace(
205        &self,
206        session_id: &SessionId,
207        run_id: &RunId,
208    ) -> SessionStoreResult<CompactRunTrace>;
209
210    /// Return compact session trace projection.
211    async fn compact_session_trace(
212        &self,
213        session_id: &SessionId,
214    ) -> SessionStoreResult<CompactSessionTrace>;
215}
starweaver_session/store/contract.rs

starweaver_session/store/
contract.rs
