starweaver_cli/local_store/
session_store.rs

1//! CLI adapter over the shared `SQLite` session store.
2
3use async_trait::async_trait;
4use starweaver_context::ResumableState;
5use starweaver_core::{RunId, SessionId};
6use starweaver_runtime::{AgentCheckpoint, AgentStreamRecord};
7use starweaver_session::{
8    ApprovalRecord, CompactRunTrace, CompactSessionTrace, DeferredToolRecord, EnvironmentStateRef,
9    HitlResumeClaim, PendingStreamPublication, RunEvidenceCommit, RunRecord, RunStatus,
10    SessionFilter, SessionRecord, SessionStatus, SessionStore, SessionStoreResult, StreamCursorRef,
11    StreamPublicationTarget,
12};
13use starweaver_storage::SqliteSessionStore;
14
15use crate::config::CliConfig;
16
17/// Session store bound to a resolved CLI database path.
18#[derive(Clone, Debug)]
19pub struct LocalSessionStore {
20    store: SqliteSessionStore,
21}
22
23impl LocalSessionStore {
24    /// Open a CLI session-store adapter before it is used by async runtime code.
25    pub fn new(config: CliConfig) -> SessionStoreResult<Self> {
26        crate::config::ensure_config_dirs(&config)
27            .map_err(|error| starweaver_session::SessionStoreError::Failed(error.to_string()))?;
28        let database_path = config.database_path;
29        Ok(Self {
30            store: SqliteSessionStore::open(&database_path)?,
31        })
32    }
33}
34
35#[async_trait]
36impl SessionStore for LocalSessionStore {
37    async fn commit_run_evidence(
38        &self,
39        commit: RunEvidenceCommit,
40    ) -> SessionStoreResult<RunRecord> {
41        self.store.commit_run_evidence(commit).await
42    }
43
44    async fn commit_checkpoint(
45        &self,
46        session_id: &SessionId,
47        checkpoint: AgentCheckpoint,
48    ) -> SessionStoreResult<()> {
49        self.store.commit_checkpoint(session_id, checkpoint).await
50    }
51
52    async fn claim_hitl_resume(&self, claim: HitlResumeClaim) -> SessionStoreResult<()> {
53        self.store.claim_hitl_resume(claim).await
54    }
55
56    async fn mark_hitl_resume_started(
57        &self,
58        session_id: &SessionId,
59        run_id: &RunId,
60        claim_id: &str,
61    ) -> SessionStoreResult<()> {
62        self.store
63            .mark_hitl_resume_started(session_id, run_id, claim_id)
64            .await
65    }
66
67    async fn release_hitl_resume_claim(
68        &self,
69        session_id: &SessionId,
70        run_id: &RunId,
71        claim_id: &str,
72    ) -> SessionStoreResult<()> {
73        self.store
74            .release_hitl_resume_claim(session_id, run_id, claim_id)
75            .await
76    }
77
78    async fn pending_stream_publications(
79        &self,
80        session_id: &SessionId,
81    ) -> SessionStoreResult<Vec<PendingStreamPublication>> {
82        self.store.pending_stream_publications(session_id).await
83    }
84
85    async fn acknowledge_stream_publication(
86        &self,
87        publication_id: &str,
88        target: StreamPublicationTarget,
89    ) -> SessionStoreResult<()> {
90        self.store
91            .acknowledge_stream_publication(publication_id, target)
92            .await
93    }
94
95    async fn save_session(&self, session: SessionRecord) -> SessionStoreResult<()> {
96        self.store.save_session(session).await
97    }
98
99    async fn load_session(&self, session_id: &SessionId) -> SessionStoreResult<SessionRecord> {
100        self.store.load_session(session_id).await
101    }
102
103    async fn list_sessions(&self, filter: SessionFilter) -> SessionStoreResult<Vec<SessionRecord>> {
104        self.store.list_sessions(filter).await
105    }
106
107    async fn update_session_status(
108        &self,
109        session_id: &SessionId,
110        status: SessionStatus,
111    ) -> SessionStoreResult<()> {
112        self.store.update_session_status(session_id, status).await
113    }
114
115    async fn save_context_state(
116        &self,
117        session_id: &SessionId,
118        state: ResumableState,
119    ) -> SessionStoreResult<()> {
120        self.store.save_context_state(session_id, state).await
121    }
122
123    async fn save_environment_state(
124        &self,
125        session_id: &SessionId,
126        environment_state: EnvironmentStateRef,
127    ) -> SessionStoreResult<()> {
128        self.store
129            .save_environment_state(session_id, environment_state)
130            .await
131    }
132
133    async fn append_run(&self, run: RunRecord) -> SessionStoreResult<()> {
134        self.store.append_run(run).await
135    }
136
137    async fn load_run(
138        &self,
139        session_id: &SessionId,
140        run_id: &RunId,
141    ) -> SessionStoreResult<RunRecord> {
142        self.store.load_run(session_id, run_id).await
143    }
144
145    async fn list_runs(&self, session_id: &SessionId) -> SessionStoreResult<Vec<RunRecord>> {
146        self.store.list_runs(session_id).await
147    }
148
149    async fn update_run_status(
150        &self,
151        session_id: &SessionId,
152        run_id: &RunId,
153        status: RunStatus,
154        output_preview: Option<String>,
155    ) -> SessionStoreResult<()> {
156        self.store
157            .update_run_status(session_id, run_id, status, output_preview)
158            .await
159    }
160
161    async fn append_checkpoint(
162        &self,
163        session_id: &SessionId,
164        checkpoint: AgentCheckpoint,
165    ) -> SessionStoreResult<()> {
166        self.store.append_checkpoint(session_id, checkpoint).await
167    }
168
169    async fn load_checkpoints(
170        &self,
171        session_id: &SessionId,
172        run_id: &RunId,
173    ) -> SessionStoreResult<Vec<AgentCheckpoint>> {
174        self.store.load_checkpoints(session_id, run_id).await
175    }
176
177    async fn append_stream_records(
178        &self,
179        session_id: &SessionId,
180        run_id: &RunId,
181        records: Vec<AgentStreamRecord>,
182    ) -> SessionStoreResult<()> {
183        self.store
184            .append_stream_records(session_id, run_id, records)
185            .await
186    }
187
188    async fn replay_stream_records(
189        &self,
190        session_id: &SessionId,
191        run_id: &RunId,
192    ) -> SessionStoreResult<Vec<AgentStreamRecord>> {
193        self.store.replay_stream_records(session_id, run_id).await
194    }
195
196    async fn replay_stream_records_after(
197        &self,
198        session_id: &SessionId,
199        run_id: &RunId,
200        after_sequence: Option<usize>,
201    ) -> SessionStoreResult<Vec<AgentStreamRecord>> {
202        self.store
203            .replay_stream_records_after(session_id, run_id, after_sequence)
204            .await
205    }
206
207    async fn save_stream_cursor(
208        &self,
209        session_id: &SessionId,
210        run_id: &RunId,
211        cursor: StreamCursorRef,
212    ) -> SessionStoreResult<()> {
213        self.store
214            .save_stream_cursor(session_id, run_id, cursor)
215            .await
216    }
217
218    async fn append_approval(&self, approval: ApprovalRecord) -> SessionStoreResult<()> {
219        self.store.append_approval(approval).await
220    }
221
222    async fn load_approvals(
223        &self,
224        session_id: &SessionId,
225        run_id: &RunId,
226    ) -> SessionStoreResult<Vec<ApprovalRecord>> {
227        self.store.load_approvals(session_id, run_id).await
228    }
229
230    async fn append_deferred_tool(&self, record: DeferredToolRecord) -> SessionStoreResult<()> {
231        self.store.append_deferred_tool(record).await
232    }
233
234    async fn load_deferred_tools(
235        &self,
236        session_id: &SessionId,
237        run_id: &RunId,
238    ) -> SessionStoreResult<Vec<DeferredToolRecord>> {
239        self.store.load_deferred_tools(session_id, run_id).await
240    }
241
242    async fn compact_run_trace(
243        &self,
244        session_id: &SessionId,
245        run_id: &RunId,
246    ) -> SessionStoreResult<CompactRunTrace> {
247        self.store.compact_run_trace(session_id, run_id).await
248    }
249
250    async fn compact_session_trace(
251        &self,
252        session_id: &SessionId,
253    ) -> SessionStoreResult<CompactSessionTrace> {
254        self.store.compact_session_trace(session_id).await
255    }
256}
starweaver_cli/local_store/session_store.rs

starweaver_cli/local_store/
session_store.rs
