starweaver_session/store/
contract.rs

1//! Durable session store contract.
2
3use async_trait::async_trait;
4use starweaver_context::{AgentCheckpoint, ResumableState};
5use starweaver_core::{RunId, SessionId};
6use starweaver_stream::AgentStreamRecord;
7
8use crate::{
9    approval::{ApprovalRecord, DeferredToolRecord},
10    claim::HitlResumeClaim,
11    error::{SessionStoreError, SessionStoreResult},
12    evidence::RunEvidenceCommit,
13    publication::{PendingStreamPublication, StreamPublicationTarget},
14    records::{
15        EnvironmentStateRef, RunRecord, RunStatus, SessionRecord, SessionStatus, StreamCursorRef,
16    },
17    resume::SessionResumeSnapshot,
18    trace::{CompactRunTrace, CompactSessionTrace},
19};
20
21/// Query filters for listing sessions.
22#[derive(Clone, Debug, Default, Eq, PartialEq)]
23pub struct SessionFilter {
24    /// Required session status.
25    pub status: Option<SessionStatus>,
26    /// Required profile name.
27    pub profile: Option<String>,
28    /// Required workspace identifier or path.
29    pub workspace: Option<String>,
30    /// Maximum number of sessions returned.
31    pub limit: Option<usize>,
32}
33
34/// Durable session store contract.
35#[async_trait]
36pub trait SessionStore: Send + Sync {
37    /// Atomically persist a complete run evidence bundle.
38    ///
39    /// Implementations must leave either the complete previous state or the complete committed
40    /// state visible. Identical retries are idempotent and conflicting retries must fail.
41    async fn commit_run_evidence(&self, commit: RunEvidenceCommit)
42    -> SessionStoreResult<RunRecord>;
43
44    /// Atomically bootstrap missing session/run records and persist one runtime checkpoint.
45    ///
46    /// This is the executor write path. Implementations must not expose a session or run without
47    /// the checkpoint when the operation fails. Exact checkpoint retries are idempotent and
48    /// conflicting retries fail.
49    async fn commit_checkpoint(
50        &self,
51        session_id: &SessionId,
52        checkpoint: AgentCheckpoint,
53    ) -> SessionStoreResult<()>;
54
55    /// Acquire exclusive ownership of a waiting run before any continuation side effect.
56    ///
57    /// A run may have at most one active claim. Claims are consumed by the related-run update in
58    /// [`Self::commit_run_evidence`].
59    async fn claim_hitl_resume(&self, _claim: HitlResumeClaim) -> SessionStoreResult<()> {
60        Err(SessionStoreError::Failed(
61            "session store does not support exclusive HITL resume claims".to_string(),
62        ))
63    }
64
65    /// Mark a claim started immediately before the first continuation hook or tool executes.
66    async fn mark_hitl_resume_started(
67        &self,
68        _session_id: &SessionId,
69        _run_id: &RunId,
70        _claim_id: &str,
71    ) -> SessionStoreResult<()> {
72        Err(SessionStoreError::Failed(
73            "session store does not support exclusive HITL resume claims".to_string(),
74        ))
75    }
76
77    /// Release a preflight claim. Stores must reject release after execution has started.
78    async fn release_hitl_resume_claim(
79        &self,
80        _session_id: &SessionId,
81        _run_id: &RunId,
82        _claim_id: &str,
83    ) -> SessionStoreResult<()> {
84        Err(SessionStoreError::Failed(
85            "session store does not support exclusive HITL resume claims".to_string(),
86        ))
87    }
88
89    /// List transactionally enqueued stream publications still awaiting at least one sink.
90    async fn pending_stream_publications(
91        &self,
92        _session_id: &SessionId,
93    ) -> SessionStoreResult<Vec<PendingStreamPublication>> {
94        Err(SessionStoreError::Failed(
95            "session store does not support transactional stream publication".to_string(),
96        ))
97    }
98
99    /// Acknowledge one sink only after its complete idempotent delivery succeeds.
100    async fn acknowledge_stream_publication(
101        &self,
102        _publication_id: &str,
103        _target: StreamPublicationTarget,
104    ) -> SessionStoreResult<()> {
105        Err(SessionStoreError::Failed(
106            "session store does not support transactional stream publication".to_string(),
107        ))
108    }
109
110    /// Save a session record.
111    async fn save_session(&self, session: SessionRecord) -> SessionStoreResult<()>;
112
113    /// Load a session record.
114    async fn load_session(&self, session_id: &SessionId) -> SessionStoreResult<SessionRecord>;
115
116    /// List sessions by optional filter.
117    async fn list_sessions(&self, filter: SessionFilter) -> SessionStoreResult<Vec<SessionRecord>>;
118
119    /// Update session status.
120    async fn update_session_status(
121        &self,
122        session_id: &SessionId,
123        status: SessionStatus,
124    ) -> SessionStoreResult<()>;
125
126    /// Save a context state snapshot for a session.
127    async fn save_context_state(
128        &self,
129        session_id: &SessionId,
130        state: ResumableState,
131    ) -> SessionStoreResult<()>;
132
133    /// Save an environment state reference for a session.
134    async fn save_environment_state(
135        &self,
136        session_id: &SessionId,
137        environment_state: EnvironmentStateRef,
138    ) -> SessionStoreResult<()>;
139
140    /// Append or replace a run record.
141    ///
142    /// A zero `sequence_no` requests atomic session-local allocation. Replacing an existing run
143    /// must preserve its assigned sequence; an explicit attempt to change that sequence fails.
144    async fn append_run(&self, run: RunRecord) -> SessionStoreResult<()>;
145
146    /// Load a run record.
147    async fn load_run(
148        &self,
149        session_id: &SessionId,
150        run_id: &RunId,
151    ) -> SessionStoreResult<RunRecord>;
152
153    /// List runs for a session.
154    async fn list_runs(&self, session_id: &SessionId) -> SessionStoreResult<Vec<RunRecord>>;
155
156    /// Update run status and optional output preview.
157    async fn update_run_status(
158        &self,
159        session_id: &SessionId,
160        run_id: &RunId,
161        status: RunStatus,
162        output_preview: Option<String>,
163    ) -> SessionStoreResult<()>;
164
165    /// Append a full runtime checkpoint.
166    async fn append_checkpoint(
167        &self,
168        session_id: &SessionId,
169        checkpoint: AgentCheckpoint,
170    ) -> SessionStoreResult<()>;
171
172    /// Load checkpoints for a run in insertion order.
173    async fn load_checkpoints(
174        &self,
175        session_id: &SessionId,
176        run_id: &RunId,
177    ) -> SessionStoreResult<Vec<AgentCheckpoint>>;
178
179    /// Load the latest checkpoint for a run.
180    async fn latest_checkpoint(
181        &self,
182        session_id: &SessionId,
183        run_id: &RunId,
184    ) -> SessionStoreResult<Option<AgentCheckpoint>> {
185        let checkpoints = self.load_checkpoints(session_id, run_id).await?;
186        Ok(checkpoints.into_iter().last())
187    }
188
189    /// Append runtime stream records used as resume evidence.
190    async fn append_stream_records(
191        &self,
192        session_id: &SessionId,
193        run_id: &RunId,
194        records: Vec<AgentStreamRecord>,
195    ) -> SessionStoreResult<()>;
196
197    /// Replay runtime stream records for a run.
198    async fn replay_stream_records(
199        &self,
200        session_id: &SessionId,
201        run_id: &RunId,
202    ) -> SessionStoreResult<Vec<AgentStreamRecord>>;
203
204    /// Replay runtime stream records after a sequence cursor.
205    async fn replay_stream_records_after(
206        &self,
207        session_id: &SessionId,
208        run_id: &RunId,
209        after_sequence: Option<usize>,
210    ) -> SessionStoreResult<Vec<AgentStreamRecord>> {
211        let records = self.replay_stream_records(session_id, run_id).await?;
212        Ok(records
213            .into_iter()
214            .filter(|record| after_sequence.is_none_or(|cursor| record.sequence > cursor))
215            .collect())
216    }
217
218    /// Store a stream cursor reference for a run and session.
219    async fn save_stream_cursor(
220        &self,
221        session_id: &SessionId,
222        run_id: &RunId,
223        cursor: StreamCursorRef,
224    ) -> SessionStoreResult<()>;
225
226    /// Append an approval record.
227    async fn append_approval(&self, approval: ApprovalRecord) -> SessionStoreResult<()>;
228
229    /// Load approval records for a run.
230    async fn load_approvals(
231        &self,
232        session_id: &SessionId,
233        run_id: &RunId,
234    ) -> SessionStoreResult<Vec<ApprovalRecord>>;
235
236    /// Append a deferred tool record.
237    async fn append_deferred_tool(&self, record: DeferredToolRecord) -> SessionStoreResult<()>;
238
239    /// Load deferred tool records for a run.
240    async fn load_deferred_tools(
241        &self,
242        session_id: &SessionId,
243        run_id: &RunId,
244    ) -> SessionStoreResult<Vec<DeferredToolRecord>>;
245
246    /// Load a resume snapshot from session, checkpoint, and stream evidence.
247    async fn resume_snapshot(
248        &self,
249        _session_id: &SessionId,
250        _run_id: &RunId,
251    ) -> SessionStoreResult<SessionResumeSnapshot> {
252        Err(SessionStoreError::Failed(
253            "session store does not support per-run resume snapshots".to_string(),
254        ))
255    }
256
257    /// Return compact run trace projection.
258    async fn compact_run_trace(
259        &self,
260        session_id: &SessionId,
261        run_id: &RunId,
262    ) -> SessionStoreResult<CompactRunTrace>;
263
264    /// Return compact session trace projection.
265    async fn compact_session_trace(
266        &self,
267        session_id: &SessionId,
268    ) -> SessionStoreResult<CompactSessionTrace>;
269}
starweaver_session/store/contract.rs

starweaver_session/store/
contract.rs
