lash_sqlite_store/
blobs.rs

1//! Content-addressed blob, artifact, checkpoint, and usage-ledger storage on
2//! [`Store`].
3//!
4//! Reference module (with `lifecycle.rs`) for the translation pattern. The
5//! `*_conn` helpers here are **synchronous** and take a `&rusqlite::Connection`
6//! so they can be reused from inside any `conn.call`/`conn.write` closure (the
7//! checkpoint/persistence/graph modules call them while already on the
8//! connection thread). The public async methods wrap a single helper call in
9//! `self.conn.call(...)`.
10
11use super::*;
12
13impl Store {
14    pub(crate) fn insert_artifact_blob_conn(
15        conn: &Connection,
16        descriptor: BlobArtifactDescriptor,
17        content: &[u8],
18        profile: BuiltinBlobProfile,
19    ) -> rusqlite::Result<BlobRef> {
20        let hash = format!("{:x}", Sha256::digest(content));
21        let stored = encode_artifact_blob(&descriptor, profile, content);
22        conn.execute(
23            "INSERT OR IGNORE INTO blobs (hash, content) VALUES (?1, ?2)",
24            params![hash, stored],
25        )?;
26        Ok(BlobRef(hash))
27    }
28
29    pub(crate) fn put_typed_artifact_blob_conn<T: serde::Serialize>(
30        conn: &Connection,
31        descriptor: BlobArtifactDescriptor,
32        value: &T,
33        profile: BuiltinBlobProfile,
34    ) -> rusqlite::Result<BlobRef> {
35        let bytes = encode_msgpack(value);
36        Self::insert_artifact_blob_conn(conn, descriptor, &bytes, profile)
37    }
38
39    pub(crate) fn put_checkpoint_conn(
40        conn: &Connection,
41        checkpoint: &HydratedSessionCheckpoint,
42        profile: BuiltinBlobProfile,
43    ) -> rusqlite::Result<StoredSessionCheckpoint> {
44        let tool_state_ref = match checkpoint.tool_state.as_ref() {
45            Some(snapshot) => Some(Self::put_typed_artifact_blob_conn(
46                conn,
47                BlobArtifactDescriptor::tool_state_snapshot(),
48                snapshot,
49                profile,
50            )?),
51            None => checkpoint.tool_state_ref.clone(),
52        };
53        let plugin_snapshot_ref = match checkpoint.plugin_snapshot.as_ref() {
54            Some(snapshot) => Some(Self::put_typed_artifact_blob_conn(
55                conn,
56                BlobArtifactDescriptor::plugin_session_snapshot(),
57                snapshot,
58                profile,
59            )?),
60            None => checkpoint.plugin_snapshot_ref.clone(),
61        };
62        let execution_state_ref = match checkpoint.execution_state.as_ref() {
63            Some(snapshot) => Some(Self::put_typed_artifact_blob_conn(
64                conn,
65                BlobArtifactDescriptor::execution_state_snapshot(),
66                snapshot,
67                profile,
68            )?),
69            None => checkpoint.execution_state_ref.clone(),
70        };
71        let manifest = SessionCheckpoint::new(
72            checkpoint.turn_state.clone(),
73            tool_state_ref,
74            plugin_snapshot_ref,
75            checkpoint.plugin_snapshot_revision,
76            execution_state_ref,
77        );
78        let checkpoint_ref = Self::put_typed_artifact_blob_conn(
79            conn,
80            BlobArtifactDescriptor::checkpoint_manifest(),
81            &manifest,
82            profile,
83        )?;
84        Ok(StoredSessionCheckpoint {
85            checkpoint_ref,
86            manifest,
87        })
88    }
89
90    pub(crate) fn get_blob_conn(conn: &Connection, blob_ref: &BlobRef) -> Option<Vec<u8>> {
91        let bytes: Vec<u8> = conn
92            .query_row(
93                "SELECT content FROM blobs WHERE hash = ?1",
94                params![blob_ref.as_str()],
95                |row| row.get(0),
96            )
97            .optional()
98            .ok()
99            .flatten()?;
100        decode_artifact_blob(&bytes).or(Some(bytes))
101    }
102
103    pub(crate) fn get_typed_blob_conn<T: serde::de::DeserializeOwned>(
104        conn: &Connection,
105        blob_ref: &BlobRef,
106    ) -> Option<T> {
107        let bytes = Self::get_blob_conn(conn, blob_ref)?;
108        decode_msgpack(&bytes)
109    }
110
111    pub(crate) fn get_checkpoint_conn(
112        conn: &Connection,
113        blob_ref: &BlobRef,
114    ) -> Result<Option<HydratedSessionCheckpoint>, StoreError> {
115        let Some(bytes) = Self::get_blob_conn(conn, blob_ref) else {
116            return Ok(None);
117        };
118        let record = decode_checkpoint(&bytes)?;
119        Ok(Some(HydratedSessionCheckpoint {
120            turn_state: record.turn_state,
121            tool_state_ref: record.tool_state_ref.clone(),
122            tool_state: record
123                .tool_state_ref
124                .as_ref()
125                .and_then(|blob_ref| Self::get_typed_blob_conn(conn, blob_ref)),
126            plugin_snapshot_ref: record.plugin_snapshot_ref.clone(),
127            plugin_snapshot: record
128                .plugin_snapshot_ref
129                .as_ref()
130                .and_then(|blob_ref| Self::get_typed_blob_conn(conn, blob_ref)),
131            plugin_snapshot_revision: record.plugin_snapshot_revision,
132            execution_state_ref: record.execution_state_ref.clone(),
133            execution_state: record
134                .execution_state_ref
135                .as_ref()
136                .and_then(|blob_ref| Self::get_typed_blob_conn(conn, blob_ref)),
137        }))
138    }
139
140    pub(crate) fn load_usage_deltas_conn(conn: &Connection) -> Vec<lash_core::TokenLedgerEntry> {
141        let mut stmt = match conn.prepare(
142            "SELECT source, model, input_tokens, output_tokens, cache_read_input_tokens, cache_write_input_tokens, reasoning_output_tokens
143             FROM usage_deltas ORDER BY seq ASC",
144        ) {
145            Ok(stmt) => stmt,
146            Err(_) => return Vec::new(),
147        };
148        let rows = match stmt.query_map([], |row| {
149            Ok(lash_core::TokenLedgerEntry {
150                source: row.get(0)?,
151                model: row.get(1)?,
152                usage: lash_core::TokenUsage {
153                    input_tokens: row.get(2)?,
154                    output_tokens: row.get(3)?,
155                    cache_read_input_tokens: row.get(4)?,
156                    cache_write_input_tokens: row.get(5)?,
157                    reasoning_output_tokens: row.get(6)?,
158                },
159            })
160        }) {
161            Ok(rows) => rows,
162            Err(_) => return Vec::new(),
163        };
164        rows.filter_map(Result::ok).collect()
165    }
166
167    pub async fn put_blob(&self, content: &[u8]) -> BlobRef {
168        let hash = format!("{:x}", Sha256::digest(content));
169        let hash_for_row = hash.clone();
170        let content = content.to_vec();
171        let result = self
172            .conn
173            .call(move |conn| {
174                conn.execute(
175                    "INSERT OR IGNORE INTO blobs (hash, content) VALUES (?1, ?2)",
176                    params![hash_for_row, content],
177                )
178            })
179            .await;
180        if let Err(err) = result {
181            tracing::warn!(error = %err, hash, "failed to persist checkpoint blob");
182        }
183        BlobRef(hash)
184    }
185
186    pub async fn put_artifact_blob(
187        &self,
188        descriptor: BlobArtifactDescriptor,
189        content: &[u8],
190    ) -> BlobRef {
191        let hash = format!("{:x}", Sha256::digest(content));
192        let stored = encode_artifact_blob(&descriptor, self.options.blob_profile, content);
193        let hash_for_row = hash.clone();
194        let result = self
195            .conn
196            .call(move |conn| {
197                conn.execute(
198                    "INSERT OR IGNORE INTO blobs (hash, content) VALUES (?1, ?2)",
199                    params![hash_for_row, stored],
200                )
201            })
202            .await;
203        if let Err(err) = result {
204            tracing::warn!(error = %err, hash, "failed to persist artifact blob");
205        }
206        BlobRef(hash)
207    }
208
209    pub async fn get_blob(&self, blob_ref: &BlobRef) -> Option<Vec<u8>> {
210        let blob_ref = blob_ref.clone();
211        self.conn
212            .call(move |conn| Ok(Self::get_blob_conn(conn, &blob_ref)))
213            .await
214            .ok()
215            .flatten()
216    }
217
218    pub async fn put_typed_blob<T: serde::Serialize>(&self, value: &T) -> BlobRef {
219        let bytes = encode_msgpack(value);
220        self.put_blob(&bytes).await
221    }
222
223    pub async fn put_typed_artifact_blob<T: serde::Serialize>(
224        &self,
225        descriptor: BlobArtifactDescriptor,
226        value: &T,
227    ) -> BlobRef {
228        let bytes = encode_msgpack(value);
229        self.put_artifact_blob(descriptor, &bytes).await
230    }
231
232    pub async fn get_typed_blob<T: serde::de::DeserializeOwned>(
233        &self,
234        blob_ref: &BlobRef,
235    ) -> Option<T> {
236        let bytes = self.get_blob(blob_ref).await?;
237        decode_msgpack(&bytes)
238    }
239
240    pub async fn put_checkpoint(
241        &self,
242        checkpoint: &HydratedSessionCheckpoint,
243    ) -> StoredSessionCheckpoint {
244        let checkpoint = checkpoint.clone();
245        let profile = self.options.blob_profile;
246        self.conn
247            .write(move |tx| Self::put_checkpoint_conn(tx, &checkpoint, profile))
248            .await
249            .expect("checkpoint blob should persist")
250    }
251
252    pub async fn get_checkpoint(&self, blob_ref: &BlobRef) -> Option<HydratedSessionCheckpoint> {
253        let blob_ref = blob_ref.clone();
254        self.conn
255            .call(move |conn| Ok(Self::get_checkpoint_conn(conn, &blob_ref)))
256            .await
257            .ok()
258            .and_then(Result::ok)
259            .flatten()
260    }
261
262    pub async fn append_usage_deltas(&self, entries: &[lash_core::TokenLedgerEntry]) {
263        if entries.is_empty() {
264            return;
265        }
266        let entries = entries.to_vec();
267        let result = self
268            .conn
269            .write(move |tx| {
270                let mut stmt = tx.prepare(
271                    "INSERT INTO usage_deltas (
272                        source, model, input_tokens, output_tokens, cache_read_input_tokens, cache_write_input_tokens, reasoning_output_tokens
273                    ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
274                )?;
275                for entry in &entries {
276                    stmt.execute(params![
277                        entry.source,
278                        entry.model,
279                        entry.usage.input_tokens,
280                        entry.usage.output_tokens,
281                        entry.usage.cache_read_input_tokens,
282                        entry.usage.cache_write_input_tokens,
283                        entry.usage.reasoning_output_tokens,
284                    ])?;
285                }
286                Ok(())
287            })
288            .await;
289        if let Err(err) = result {
290            tracing::warn!(error = %err, "failed to persist usage deltas");
291        }
292    }
293
294    pub async fn load_usage_deltas(&self) -> Vec<lash_core::TokenLedgerEntry> {
295        self.conn
296            .call(|conn| Ok(Self::load_usage_deltas_conn(conn)))
297            .await
298            .unwrap_or_default()
299    }
300}
lash_sqlite_store/blobs.rs

lash_sqlite_store/
blobs.rs
