1use std::fs;
2use std::path::{Path, PathBuf};
3use std::time::{SystemTime, UNIX_EPOCH};
4
5use anyhow::{Context, Result};
6use rusqlite::{Connection, OptionalExtension, params};
7use uuid::Uuid;
8
9use crate::model::{ContentBlock, ConversationItem, Role, Usage};
10
11const SCHEMA: &str = r#"
12CREATE TABLE IF NOT EXISTS sessions (
13 id TEXT PRIMARY KEY,
14 cwd TEXT NOT NULL,
15 client_session_key TEXT,
16 title TEXT,
17 head_item_id TEXT,
18 provider_type TEXT NOT NULL,
19 model TEXT NOT NULL,
20 model_profile TEXT,
21 effort TEXT,
22 created_at INTEGER NOT NULL,
23 updated_at INTEGER NOT NULL
24);
25CREATE INDEX IF NOT EXISTS idx_sessions_cwd_updated ON sessions(cwd, updated_at DESC);
26
27CREATE TABLE IF NOT EXISTS conversation_items (
28 id TEXT PRIMARY KEY,
29 session_id TEXT NOT NULL,
30 parent_id TEXT,
31 role TEXT NOT NULL,
32 kind TEXT NOT NULL,
33 content_json TEXT NOT NULL,
34 usage_json TEXT,
35 created_at INTEGER NOT NULL,
36 FOREIGN KEY(session_id) REFERENCES sessions(id)
37);
38CREATE INDEX IF NOT EXISTS idx_items_session ON conversation_items(session_id);
39CREATE INDEX IF NOT EXISTS idx_items_parent ON conversation_items(parent_id);
40
41CREATE TABLE IF NOT EXISTS provider_state (
42 session_id TEXT NOT NULL,
43 key TEXT NOT NULL,
44 value_json TEXT NOT NULL,
45 PRIMARY KEY(session_id, key),
46 FOREIGN KEY(session_id) REFERENCES sessions(id)
47);
48
49CREATE TABLE IF NOT EXISTS shell_history (
50 id INTEGER PRIMARY KEY AUTOINCREMENT,
51 cwd TEXT NOT NULL,
52 client_session_key TEXT,
53 command TEXT NOT NULL,
54 exit_code INTEGER,
55 pipe_status TEXT,
56 started_at INTEGER NOT NULL,
57 duration_ms INTEGER
58);
59CREATE INDEX IF NOT EXISTS idx_shell_cwd_time ON shell_history(cwd, started_at DESC);
60
61CREATE TABLE IF NOT EXISTS input_history (
62 id INTEGER PRIMARY KEY AUTOINCREMENT,
63 text TEXT NOT NULL,
64 created_at INTEGER NOT NULL
65);
66"#;
67
68pub const TURN_INTERRUPTED_NOTICE: &str = "[The user interrupted the previous turn. Do not continue or retry its unfinished task unless the user explicitly asks you to.]";
69pub const CONVERSATION_SUMMARY_PREFIX: &str = "[Compacted conversation summary]";
70
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct Session {
73 pub id: String,
74 pub cwd: String,
75 pub title: Option<String>,
76 pub head_item_id: Option<String>,
77 pub provider_type: String,
78 pub model: String,
79 pub model_profile: Option<String>,
80 pub effort: Option<String>,
81 pub created_at: i64,
82 pub updated_at: i64,
83}
84
85#[derive(Debug, Clone)]
86pub struct NewSession {
87 pub cwd: String,
88 pub provider_type: String,
89 pub model: String,
90 pub client_session_key: Option<String>,
91 pub model_profile: Option<String>,
92 pub effort: Option<String>,
93}
94
95impl NewSession {
96 pub fn new(
97 cwd: impl Into<String>,
98 provider_type: impl Into<String>,
99 model: impl Into<String>,
100 ) -> Self {
101 Self {
102 cwd: cwd.into(),
103 provider_type: provider_type.into(),
104 model: model.into(),
105 client_session_key: None,
106 model_profile: None,
107 effort: None,
108 }
109 }
110
111 pub fn with_client_session_key(mut self, key: impl Into<String>) -> Self {
112 self.client_session_key = Some(key.into());
113 self
114 }
115
116 pub fn with_model_selection(
117 mut self,
118 profile: impl Into<String>,
119 effort: Option<&str>,
120 ) -> Self {
121 self.model_profile = Some(profile.into());
122 self.effort = effort.map(str::to_owned);
123 self
124 }
125}
126
127#[derive(Debug, Clone, PartialEq, Eq)]
128pub struct ShellHistoryItem {
129 pub id: i64,
130 pub cwd: String,
131 pub command: String,
132 pub exit_code: Option<i32>,
133 pub pipe_status: Option<String>,
134 pub started_at: i64,
135 pub duration_ms: Option<i64>,
136}
137
138pub struct SessionStore {
139 connection: Connection,
140}
141
142impl SessionStore {
143 pub fn open(path: &Path) -> Result<Self> {
144 if let Some(parent) = path.parent() {
145 fs::create_dir_all(parent)
146 .with_context(|| format!("create state directory {}", parent.display()))?;
147 }
148 let connection = Connection::open(path)
149 .with_context(|| format!("open session database {}", path.display()))?;
150 Self::initialize(connection)
151 }
152
153 pub fn open_in_memory() -> Result<Self> {
154 Self::initialize(Connection::open_in_memory()?)
155 }
156
157 fn initialize(connection: Connection) -> Result<Self> {
158 connection.busy_timeout(std::time::Duration::from_millis(1000))?;
159 connection.pragma_update(None, "foreign_keys", "ON")?;
160 connection.pragma_update(None, "synchronous", "NORMAL")?;
161 connection.pragma_update(None, "journal_mode", "WAL")?;
162 connection.execute_batch(SCHEMA)?;
163 migrate_schema(&connection)?;
164 Ok(Self { connection })
165 }
166
167 pub fn create_session(&mut self, new: NewSession) -> Result<Session> {
168 let id = format!("a_{}", Uuid::new_v4().simple());
169 let now = now_millis();
170 self.connection.execute(
171 "INSERT INTO sessions (id, cwd, client_session_key, provider_type, model, model_profile, effort, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?8)",
172 params![id, new.cwd, new.client_session_key, new.provider_type, new.model, new.model_profile, new.effort, now],
173 )?;
174 self.get_session(&id)?.context("new session disappeared")
175 }
176
177 pub fn get_session(&self, id: &str) -> Result<Option<Session>> {
178 self.connection
179 .query_row(
180 "SELECT id, cwd, title, head_item_id, provider_type, model, model_profile, effort, created_at, updated_at FROM sessions WHERE id = ?1",
181 [id],
182 row_to_session,
183 )
184 .optional()
185 .map_err(Into::into)
186 }
187
188 pub fn find_latest_session(&self, cwd: &str) -> Result<Option<Session>> {
189 self.connection
190 .query_row(
191 "SELECT id, cwd, title, head_item_id, provider_type, model, model_profile, effort, created_at, updated_at FROM sessions WHERE cwd = ?1 ORDER BY updated_at DESC, rowid DESC LIMIT 1",
192 [cwd],
193 row_to_session,
194 )
195 .optional()
196 .map_err(Into::into)
197 }
198
199 pub fn find_client_session(&self, cwd: &str, key: &str) -> Result<Option<Session>> {
200 self.connection
201 .query_row(
202 "SELECT id, cwd, title, head_item_id, provider_type, model, model_profile, effort, created_at, updated_at FROM sessions WHERE cwd = ?1 AND client_session_key = ?2 LIMIT 1",
203 params![cwd, key],
204 row_to_session,
205 )
206 .optional()
207 .map_err(Into::into)
208 }
209
210 pub fn recent_sessions(&self, cwd: &str, limit: usize) -> Result<Vec<Session>> {
211 let mut statement = self.connection.prepare(
212 "SELECT id, cwd, title, head_item_id, provider_type, model, model_profile, effort, created_at, updated_at FROM sessions WHERE cwd = ?1 AND EXISTS (SELECT 1 FROM conversation_items WHERE conversation_items.session_id = sessions.id AND conversation_items.kind = 'user_message') ORDER BY updated_at DESC, rowid DESC LIMIT ?2",
213 )?;
214 let rows = statement.query_map(params![cwd, limit], row_to_session)?;
215 rows.collect::<rusqlite::Result<Vec<_>>>()
216 .map_err(Into::into)
217 }
218
219 pub fn first_user_prompt(&self, session_id: &str) -> Result<Option<String>> {
220 let content = self
221 .connection
222 .query_row(
223 "SELECT content_json FROM conversation_items WHERE session_id = ?1 AND kind = 'user_message' ORDER BY created_at ASC, rowid ASC LIMIT 1",
224 [session_id],
225 |row| row.get::<_, String>(0),
226 )
227 .optional()?;
228 content
229 .map(|content| {
230 let blocks: Vec<ContentBlock> = serde_json::from_str(&content)?;
231 Ok(blocks.into_iter().find_map(|block| match block {
232 ContentBlock::Text(text) => Some(text),
233 _ => None,
234 }))
235 })
236 .transpose()
237 .map(Option::flatten)
238 }
239
240 pub fn rebind_client_session_key(
241 &mut self,
242 cwd: &str,
243 key: &str,
244 target_session_id: &str,
245 ) -> Result<()> {
246 let transaction = self.connection.transaction()?;
247 let (target_cwd, target_key): (String, Option<String>) = transaction
248 .query_row(
249 "SELECT cwd, client_session_key FROM sessions WHERE id = ?1",
250 [target_session_id],
251 |row| Ok((row.get(0)?, row.get(1)?)),
252 )
253 .optional()?
254 .with_context(|| format!("session not found: {target_session_id}"))?;
255 if target_cwd != cwd {
256 anyhow::bail!("cannot resume a session from a different cwd");
257 }
258 if target_key.as_deref().is_some_and(|target| target != key) {
259 anyhow::bail!("session is active in another Fish process");
260 }
261 transaction.execute(
262 "UPDATE sessions SET client_session_key = NULL WHERE cwd = ?1 AND client_session_key = ?2",
263 params![cwd, key],
264 )?;
265 transaction.execute(
266 "UPDATE sessions SET client_session_key = ?1, updated_at = ?2 WHERE id = ?3",
267 params![key, now_millis(), target_session_id],
268 )?;
269 transaction.commit()?;
270 Ok(())
271 }
272
273 pub fn append_item(
274 &mut self,
275 session_id: &str,
276 role: Role,
277 blocks: Vec<ContentBlock>,
278 ) -> Result<ConversationItem> {
279 let kind = item_kind(role, &blocks);
280 self.append_item_with_kind(session_id, role, blocks, kind, None)
281 }
282
283 pub fn append_assistant_item(
284 &mut self,
285 session_id: &str,
286 blocks: Vec<ContentBlock>,
287 usage: Option<Usage>,
288 ) -> Result<ConversationItem> {
289 let kind = item_kind(Role::Assistant, &blocks);
290 self.append_item_with_kind(session_id, Role::Assistant, blocks, kind, usage)
291 }
292
293 pub fn append_turn_interrupted(&mut self, session_id: &str) -> Result<ConversationItem> {
294 self.append_item_with_kind(
295 session_id,
296 Role::User,
297 vec![ContentBlock::Text(TURN_INTERRUPTED_NOTICE.into())],
298 "turn_interrupted",
299 None,
300 )
301 }
302
303 pub fn replace_branch_with_summary(
304 &mut self,
305 session_id: &str,
306 summary: &str,
307 ) -> Result<ConversationItem> {
308 self.get_session(session_id)?
309 .with_context(|| format!("session not found: {session_id}"))?;
310 let id = format!("i_{}", Uuid::new_v4().simple());
311 let created_at = now_millis();
312 let blocks = vec![ContentBlock::Text(format!(
313 "{CONVERSATION_SUMMARY_PREFIX}\n{summary}"
314 ))];
315 let content_json = serde_json::to_string(&blocks)?;
316 let transaction = self.connection.transaction()?;
317 transaction.execute(
318 "INSERT INTO conversation_items (id, session_id, parent_id, role, kind, content_json, usage_json, created_at) VALUES (?1, ?2, NULL, 'user', 'conversation_summary', ?3, NULL, ?4)",
319 params![id, session_id, content_json, created_at],
320 )?;
321 transaction.execute(
322 "UPDATE sessions SET head_item_id = ?1, updated_at = ?2 WHERE id = ?3",
323 params![id, created_at, session_id],
324 )?;
325 transaction.commit()?;
326 Ok(ConversationItem {
327 id,
328 session_id: session_id.into(),
329 parent_id: None,
330 role: Role::User,
331 blocks,
332 usage: None,
333 created_at,
334 })
335 }
336
337 fn append_item_with_kind(
338 &mut self,
339 session_id: &str,
340 role: Role,
341 blocks: Vec<ContentBlock>,
342 kind: &str,
343 usage: Option<Usage>,
344 ) -> Result<ConversationItem> {
345 let parent_id = self
346 .get_session(session_id)?
347 .with_context(|| format!("session not found: {session_id}"))?
348 .head_item_id;
349 let id = format!("i_{}", Uuid::new_v4().simple());
350 let created_at = now_millis();
351 let content_json = serde_json::to_string(&blocks)?;
352 let usage_json = usage
353 .map(|usage| serde_json::to_string(&usage))
354 .transpose()?;
355 let role_text = role_to_str(role);
356 let transaction = self.connection.transaction()?;
357 transaction.execute(
358 "INSERT INTO conversation_items (id, session_id, parent_id, role, kind, content_json, usage_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
359 params![id, session_id, parent_id, role_text, kind, content_json, usage_json, created_at],
360 )?;
361 transaction.execute(
362 "UPDATE sessions SET head_item_id = ?1, updated_at = ?2 WHERE id = ?3",
363 params![id, created_at, session_id],
364 )?;
365 transaction.commit()?;
366 Ok(ConversationItem {
367 id,
368 session_id: session_id.into(),
369 parent_id,
370 role,
371 blocks,
372 usage,
373 created_at,
374 })
375 }
376
377 pub fn get_item(&self, id: &str) -> Result<Option<ConversationItem>> {
378 self.connection
379 .query_row(
380 "SELECT id, session_id, parent_id, role, content_json, usage_json, created_at FROM conversation_items WHERE id = ?1",
381 [id],
382 row_to_item,
383 )
384 .optional()
385 .map_err(Into::into)
386 }
387
388 pub fn active_branch(&self, session_id: &str) -> Result<Vec<ConversationItem>> {
389 let mut current = self
390 .get_session(session_id)?
391 .with_context(|| format!("session not found: {session_id}"))?
392 .head_item_id;
393 let mut branch = Vec::new();
394 while let Some(id) = current {
395 let item = self
396 .get_item(&id)?
397 .with_context(|| format!("conversation item not found: {id}"))?;
398 current = item.parent_id.clone();
399 branch.push(item);
400 }
401 branch.reverse();
402 Ok(branch)
403 }
404
405 pub fn rewind(&mut self, session_id: &str, item_id: &str) -> Result<()> {
406 let item = self
407 .get_item(item_id)?
408 .with_context(|| format!("rewind item not found: {item_id}"))?;
409 if item.session_id != session_id {
410 anyhow::bail!("rewind item does not belong to session {session_id}");
411 }
412 self.connection.execute(
413 "UPDATE sessions SET head_item_id = ?1, updated_at = ?2 WHERE id = ?3",
414 params![item_id, now_millis(), session_id],
415 )?;
416 Ok(())
417 }
418
419 pub fn user_checkpoints(&self, session_id: &str) -> Result<Vec<ConversationItem>> {
420 let mut statement = self.connection.prepare(
421 "SELECT id, session_id, parent_id, role, content_json, usage_json, created_at FROM conversation_items WHERE session_id = ?1 AND kind = 'user_message' ORDER BY created_at ASC, rowid ASC",
422 )?;
423 let rows = statement.query_map([session_id], row_to_item)?;
424 rows.collect::<rusqlite::Result<Vec<_>>>()
425 .map_err(Into::into)
426 }
427
428 pub fn count_items(&self, session_id: &str) -> Result<usize> {
429 Ok(self.connection.query_row(
430 "SELECT COUNT(*) FROM conversation_items WHERE session_id = ?1",
431 [session_id],
432 |row| row.get(0),
433 )?)
434 }
435
436 pub fn update_model_selection(
437 &self,
438 session_id: &str,
439 provider_type: &str,
440 model: &str,
441 model_profile: &str,
442 effort: Option<&str>,
443 ) -> Result<()> {
444 self.connection.execute(
445 "UPDATE sessions SET provider_type = ?1, model = ?2, model_profile = ?3, effort = ?4, updated_at = ?5 WHERE id = ?6",
446 params![provider_type, model, model_profile, effort, now_millis(), session_id],
447 )?;
448 Ok(())
449 }
450
451 pub fn clear_session(&self, session_id: &str) -> Result<()> {
452 self.connection.execute(
453 "UPDATE sessions SET head_item_id = NULL, updated_at = ?1 WHERE id = ?2",
454 params![now_millis(), session_id],
455 )?;
456 Ok(())
457 }
458
459 pub fn record_input_history(&self, text: &str) -> Result<()> {
460 self.connection.execute(
461 "INSERT INTO input_history (text, created_at) VALUES (?1, ?2)",
462 params![text, now_millis()],
463 )?;
464 Ok(())
465 }
466
467 pub fn recent_input_history(&self, limit: usize) -> Result<Vec<String>> {
468 let mut statement = self
469 .connection
470 .prepare("SELECT text FROM input_history ORDER BY id DESC LIMIT ?1")?;
471 let rows = statement.query_map([limit], |row| row.get::<_, String>(0))?;
472 let mut entries = rows.collect::<rusqlite::Result<Vec<_>>>()?;
473 entries.reverse();
474 Ok(entries)
475 }
476
477 pub fn prune_input_history(&self, maximum: usize) -> Result<()> {
478 self.connection.execute(
479 "DELETE FROM input_history WHERE id NOT IN (SELECT id FROM input_history ORDER BY id DESC LIMIT ?1)",
480 [maximum],
481 )?;
482 Ok(())
483 }
484
485 pub fn set_provider_state(
486 &self,
487 session_id: &str,
488 key: &str,
489 value: &serde_json::Value,
490 ) -> Result<()> {
491 self.connection.execute(
492 "INSERT INTO provider_state (session_id, key, value_json) VALUES (?1, ?2, ?3) ON CONFLICT(session_id, key) DO UPDATE SET value_json = excluded.value_json",
493 params![session_id, key, serde_json::to_string(value)?],
494 )?;
495 Ok(())
496 }
497
498 pub fn provider_state(&self, session_id: &str, key: &str) -> Result<Option<serde_json::Value>> {
499 let value = self
500 .connection
501 .query_row(
502 "SELECT value_json FROM provider_state WHERE session_id = ?1 AND key = ?2",
503 params![session_id, key],
504 |row| row.get::<_, String>(0),
505 )
506 .optional()?;
507 value
508 .map(|value| serde_json::from_str(&value).map_err(Into::into))
509 .transpose()
510 }
511
512 #[allow(clippy::too_many_arguments)]
513 pub fn record_shell_history(
514 &self,
515 cwd: &str,
516 client_session_key: Option<&str>,
517 command: &str,
518 exit_code: Option<i32>,
519 started_at: i64,
520 duration_ms: Option<i64>,
521 pipe_status: Option<&str>,
522 ) -> Result<()> {
523 self.connection.execute(
524 "INSERT INTO shell_history (cwd, client_session_key, command, exit_code, pipe_status, started_at, duration_ms) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
525 params![cwd, client_session_key, command, exit_code, pipe_status, started_at, duration_ms],
526 )?;
527 Ok(())
528 }
529
530 pub fn recent_shell_history(
531 &self,
532 cwd: &str,
533 client_session_key: Option<&str>,
534 limit: usize,
535 ) -> Result<Vec<ShellHistoryItem>> {
536 let query = if client_session_key.is_some() {
537 "SELECT id, cwd, command, exit_code, pipe_status, started_at, duration_ms FROM shell_history WHERE cwd = ?1 AND client_session_key = ?2 ORDER BY started_at DESC, id DESC LIMIT ?3"
538 } else {
539 "SELECT id, cwd, command, exit_code, pipe_status, started_at, duration_ms FROM shell_history WHERE cwd = ?1 ORDER BY started_at DESC, id DESC LIMIT ?3"
540 };
541 let mut statement = self.connection.prepare(query)?;
542 let rows = statement.query_map(params![cwd, client_session_key, limit], |row| {
543 Ok(ShellHistoryItem {
544 id: row.get(0)?,
545 cwd: row.get(1)?,
546 command: row.get(2)?,
547 exit_code: row.get(3)?,
548 pipe_status: row.get(4)?,
549 started_at: row.get(5)?,
550 duration_ms: row.get(6)?,
551 })
552 })?;
553 rows.collect::<rusqlite::Result<Vec<_>>>()
554 .map_err(Into::into)
555 }
556
557 pub fn prune_shell_history(&self, maximum: usize) -> Result<()> {
558 self.connection.execute(
559 "DELETE FROM shell_history WHERE id NOT IN (SELECT id FROM shell_history ORDER BY started_at DESC, id DESC LIMIT ?1)",
560 [maximum],
561 )?;
562 Ok(())
563 }
564
565 pub fn shell_history_count(&self) -> Result<usize> {
566 Ok(self
567 .connection
568 .query_row("SELECT COUNT(*) FROM shell_history", [], |row| row.get(0))?)
569 }
570}
571
572pub fn default_database_path(home: &Path) -> PathBuf {
573 std::env::var_os("XDG_STATE_HOME")
574 .map(PathBuf::from)
575 .unwrap_or_else(|| home.join(".local/state"))
576 .join("a/sessions.db")
577}
578
579fn migrate_schema(connection: &Connection) -> Result<()> {
580 ensure_column(connection, "sessions", "client_session_key", "TEXT")?;
581 ensure_column(connection, "sessions", "model_profile", "TEXT")?;
582 ensure_column(connection, "sessions", "effort", "TEXT")?;
583 ensure_column(connection, "shell_history", "client_session_key", "TEXT")?;
584 ensure_column(connection, "conversation_items", "usage_json", "TEXT")?;
585 connection.execute_batch(
586 "CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_client_key ON sessions(cwd, client_session_key) WHERE client_session_key IS NOT NULL;
587 CREATE INDEX IF NOT EXISTS idx_shell_client_time ON shell_history(cwd, client_session_key, started_at DESC);
588 PRAGMA user_version = 4;",
589 )?;
590 Ok(())
591}
592
593fn ensure_column(connection: &Connection, table: &str, column: &str, kind: &str) -> Result<()> {
594 let mut statement = connection.prepare(&format!("PRAGMA table_info({table})"))?;
595 let columns = statement
596 .query_map([], |row| row.get::<_, String>(1))?
597 .collect::<rusqlite::Result<Vec<_>>>()?;
598 if !columns.iter().any(|existing| existing == column) {
599 connection.execute(
600 &format!("ALTER TABLE {table} ADD COLUMN {column} {kind}"),
601 [],
602 )?;
603 }
604 Ok(())
605}
606
607fn row_to_session(row: &rusqlite::Row<'_>) -> rusqlite::Result<Session> {
608 Ok(Session {
609 id: row.get(0)?,
610 cwd: row.get(1)?,
611 title: row.get(2)?,
612 head_item_id: row.get(3)?,
613 provider_type: row.get(4)?,
614 model: row.get(5)?,
615 model_profile: row.get(6)?,
616 effort: row.get(7)?,
617 created_at: row.get(8)?,
618 updated_at: row.get(9)?,
619 })
620}
621
622fn row_to_item(row: &rusqlite::Row<'_>) -> rusqlite::Result<ConversationItem> {
623 let role: String = row.get(3)?;
624 let content: String = row.get(4)?;
625 let usage: Option<String> = row.get(5)?;
626 Ok(ConversationItem {
627 id: row.get(0)?,
628 session_id: row.get(1)?,
629 parent_id: row.get(2)?,
630 role: str_to_role(&role).map_err(|error| {
631 rusqlite::Error::FromSqlConversionFailure(3, rusqlite::types::Type::Text, error.into())
632 })?,
633 blocks: serde_json::from_str(&content).map_err(|error| {
634 rusqlite::Error::FromSqlConversionFailure(4, rusqlite::types::Type::Text, error.into())
635 })?,
636 usage: usage
637 .map(|usage| serde_json::from_str(&usage))
638 .transpose()
639 .map_err(|error| {
640 rusqlite::Error::FromSqlConversionFailure(
641 5,
642 rusqlite::types::Type::Text,
643 error.into(),
644 )
645 })?,
646 created_at: row.get(6)?,
647 })
648}
649
650fn role_to_str(role: Role) -> &'static str {
651 match role {
652 Role::System => "system",
653 Role::User => "user",
654 Role::Assistant => "assistant",
655 Role::Tool => "tool",
656 }
657}
658
659fn str_to_role(role: &str) -> Result<Role, String> {
660 match role {
661 "system" => Ok(Role::System),
662 "user" => Ok(Role::User),
663 "assistant" => Ok(Role::Assistant),
664 "tool" => Ok(Role::Tool),
665 _ => Err(format!("unknown conversation role: {role}")),
666 }
667}
668
669fn item_kind(role: Role, blocks: &[ContentBlock]) -> &'static str {
670 match role {
671 Role::User => "user_message",
672 Role::Tool => "tool_result",
673 Role::System => "system_checkpoint",
674 Role::Assistant
675 if blocks
676 .iter()
677 .any(|block| matches!(block, ContentBlock::ToolCall(_))) =>
678 {
679 "tool_call"
680 }
681 Role::Assistant
682 if blocks
683 .iter()
684 .any(|block| matches!(block, ContentBlock::Reasoning(_))) =>
685 {
686 "assistant_reasoning"
687 }
688 Role::Assistant => "assistant_text",
689 }
690}
691
692fn now_millis() -> i64 {
693 SystemTime::now()
694 .duration_since(UNIX_EPOCH)
695 .unwrap_or_default()
696 .as_millis() as i64
697}