1use std::collections::BTreeMap;
2use std::fmt;
3use std::path::{Path, PathBuf};
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::{Arc, Mutex};
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use rusqlite::{params, Connection, OptionalExtension};
9use serde_json::Value;
10
11use crate::app_server::protocol::{
12 AppItem, AppThread, AppTurn, ThreadStartParams, ThreadStatus, TurnStatus, UserInput,
13};
14
15#[derive(Clone)]
16pub struct SqliteThreadStore {
17 connection: Arc<Mutex<Connection>>,
18 next_thread_id: Arc<AtomicU64>,
19 next_turn_id: Arc<AtomicU64>,
20}
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub enum ItemAppendOutcome {
24 Inserted,
25 AlreadyPresent,
26}
27
28impl SqliteThreadStore {
29 pub fn in_memory() -> Result<Self, ThreadStoreError> {
30 let connection = Connection::open_in_memory().map_err(ThreadStoreError::sql)?;
31 Self::from_connection(connection)
32 }
33
34 pub fn open(path: impl AsRef<Path>) -> Result<Self, ThreadStoreError> {
35 let connection = Connection::open(path).map_err(ThreadStoreError::sql)?;
36 Self::from_connection(connection)
37 }
38
39 fn from_connection(connection: Connection) -> Result<Self, ThreadStoreError> {
40 let store = Self {
41 connection: Arc::new(Mutex::new(connection)),
42 next_thread_id: Arc::new(AtomicU64::new(1)),
43 next_turn_id: Arc::new(AtomicU64::new(1)),
44 };
45 store.migrate()?;
46 store.recover_interrupted_threads()?;
47 store.seed_sequences()?;
48 Ok(store)
49 }
50
51 pub fn create_thread(&self, params: ThreadStartParams) -> Result<AppThread, ThreadStoreError> {
52 let sequence = self.next_thread_id.fetch_add(1, Ordering::Relaxed);
53 let now = timestamp_seconds();
54 let thread = AppThread {
55 thread_id: format!("thread_{sequence}"),
56 agent_key: params.agent_key,
57 cwd: params.cwd,
58 created_at: now,
59 updated_at: now,
60 archived_at: None,
61 status: ThreadStatus::Idle,
62 metadata: params.metadata,
63 };
64 self.insert_thread(&thread)?;
65 Ok(thread)
66 }
67
68 pub fn get_thread(&self, thread_id: &str) -> Result<Option<AppThread>, ThreadStoreError> {
69 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
70 connection
71 .query_row(
72 "SELECT thread_id, agent_key, cwd, created_at, updated_at, archived_at, status, metadata_json
73 FROM app_server_threads WHERE thread_id = ?1",
74 params![thread_id],
75 row_to_thread,
76 )
77 .optional()
78 .map_err(ThreadStoreError::sql)
79 }
80
81 pub fn list_threads(&self, include_archived: bool) -> Result<Vec<AppThread>, ThreadStoreError> {
82 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
83 let sql = if include_archived {
84 "SELECT thread_id, agent_key, cwd, created_at, updated_at, archived_at, status, metadata_json
85 FROM app_server_threads ORDER BY rowid ASC"
86 } else {
87 "SELECT thread_id, agent_key, cwd, created_at, updated_at, archived_at, status, metadata_json
88 FROM app_server_threads WHERE archived_at IS NULL ORDER BY rowid ASC"
89 };
90 let mut statement = connection.prepare(sql).map_err(ThreadStoreError::sql)?;
91 let rows = statement
92 .query_map([], row_to_thread)
93 .map_err(ThreadStoreError::sql)?;
94 rows.collect::<Result<Vec<_>, _>>()
95 .map_err(ThreadStoreError::sql)
96 }
97
98 pub fn archive_thread(&self, thread_id: &str) -> Result<(), ThreadStoreError> {
99 let now = timestamp_seconds();
100 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
101 let changed = connection
102 .execute(
103 "UPDATE app_server_threads
104 SET archived_at = ?2, status = 'archived', updated_at = ?2
105 WHERE thread_id = ?1",
106 params![thread_id, now],
107 )
108 .map_err(ThreadStoreError::sql)?;
109 if changed == 0 {
110 return Err(ThreadStoreError::not_found("thread", thread_id));
111 }
112 Ok(())
113 }
114
115 pub fn set_active_turn(
116 &self,
117 thread_id: &str,
118 active_turn_id: Option<&str>,
119 status: ThreadStatus,
120 ) -> Result<(), ThreadStoreError> {
121 let now = timestamp_seconds();
122 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
123 let changed = connection
124 .execute(
125 "UPDATE app_server_threads
126 SET active_turn_id = ?2, status = ?3, updated_at = ?4
127 WHERE thread_id = ?1",
128 params![thread_id, active_turn_id, thread_status_to_str(status), now],
129 )
130 .map_err(ThreadStoreError::sql)?;
131 if changed == 0 {
132 return Err(ThreadStoreError::not_found("thread", thread_id));
133 }
134 Ok(())
135 }
136
137 pub fn create_turn(
138 &self,
139 thread_id: &str,
140 input: Vec<UserInput>,
141 ) -> Result<AppTurn, ThreadStoreError> {
142 let sequence = self.next_turn_id.fetch_add(1, Ordering::Relaxed);
143 let turn = AppTurn {
144 turn_id: format!("turn_{sequence}"),
145 thread_id: thread_id.to_string(),
146 run_id: None,
147 status: TurnStatus::Running,
148 started_at: timestamp_seconds(),
149 completed_at: None,
150 input,
151 result: BTreeMap::new(),
152 };
153 let input_json = serde_json::to_string(&turn.input).map_err(ThreadStoreError::json)?;
154 let result_json = serde_json::to_string(&turn.result).map_err(ThreadStoreError::json)?;
155 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
156 connection
157 .execute(
158 "INSERT INTO app_server_turns
159 (turn_id, thread_id, run_id, status, started_at, completed_at, input_json, result_json)
160 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
161 params![
162 turn.turn_id,
163 turn.thread_id,
164 turn.run_id,
165 turn_status_to_str(turn.status),
166 turn.started_at,
167 turn.completed_at,
168 input_json,
169 result_json,
170 ],
171 )
172 .map_err(ThreadStoreError::sql)?;
173 Ok(turn)
174 }
175
176 pub fn get_turn(
177 &self,
178 thread_id: &str,
179 turn_id: &str,
180 ) -> Result<Option<AppTurn>, ThreadStoreError> {
181 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
182 connection
183 .query_row(
184 "SELECT turn_id, thread_id, run_id, status, started_at, completed_at, input_json, result_json
185 FROM app_server_turns WHERE thread_id = ?1 AND turn_id = ?2",
186 params![thread_id, turn_id],
187 row_to_turn,
188 )
189 .optional()
190 .map_err(ThreadStoreError::sql)
191 }
192
193 pub fn mark_turn_running(
194 &self,
195 thread_id: &str,
196 turn_id: &str,
197 run_id: &str,
198 ) -> Result<AppTurn, ThreadStoreError> {
199 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
200 let changed = connection
201 .execute(
202 "UPDATE app_server_turns
203 SET status = 'running', run_id = ?3, completed_at = NULL, result_json = '{}'
204 WHERE thread_id = ?1 AND turn_id = ?2",
205 params![thread_id, turn_id, run_id],
206 )
207 .map_err(ThreadStoreError::sql)?;
208 if changed == 0 {
209 return Err(ThreadStoreError::not_found("turn", turn_id));
210 }
211 query_turn(&connection, turn_id)?
212 .ok_or_else(|| ThreadStoreError::not_found("turn", turn_id))
213 }
214
215 pub fn update_turn(
216 &self,
217 turn_id: &str,
218 status: TurnStatus,
219 run_id: Option<&str>,
220 result: &BTreeMap<String, Value>,
221 ) -> Result<AppTurn, ThreadStoreError> {
222 let completed_at = timestamp_seconds();
223 let result_json = serde_json::to_string(result).map_err(ThreadStoreError::json)?;
224 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
225 let changed = connection
226 .execute(
227 "UPDATE app_server_turns
228 SET status = ?2, run_id = COALESCE(?3, run_id), completed_at = ?4, result_json = ?5
229 WHERE turn_id = ?1",
230 params![
231 turn_id,
232 turn_status_to_str(status),
233 run_id,
234 completed_at,
235 result_json,
236 ],
237 )
238 .map_err(ThreadStoreError::sql)?;
239 if changed == 0 {
240 return Err(ThreadStoreError::not_found("turn", turn_id));
241 }
242 query_turn(&connection, turn_id)?
243 .ok_or_else(|| ThreadStoreError::not_found("turn", turn_id))
244 }
245
246 pub fn list_turns(&self, thread_id: &str) -> Result<Vec<AppTurn>, ThreadStoreError> {
247 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
248 let mut statement = connection
249 .prepare(
250 "SELECT turn_id, thread_id, run_id, status, started_at, completed_at, input_json, result_json
251 FROM app_server_turns WHERE thread_id = ?1 ORDER BY rowid ASC",
252 )
253 .map_err(ThreadStoreError::sql)?;
254 let rows = statement
255 .query_map(params![thread_id], row_to_turn)
256 .map_err(ThreadStoreError::sql)?;
257 rows.collect::<Result<Vec<_>, _>>()
258 .map_err(ThreadStoreError::sql)
259 }
260
261 pub fn append_item(
262 &self,
263 thread_id: &str,
264 turn_id: &str,
265 item: AppItem,
266 ) -> Result<ItemAppendOutcome, ThreadStoreError> {
267 if item.thread_id != thread_id || item.turn_id != turn_id {
268 return Err(ThreadStoreError::item_identity_conflict(&item.item_id));
269 }
270 let payload_json = serde_json::to_string(&item).map_err(ThreadStoreError::json)?;
271 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
272 let sequence: i64 = connection
273 .query_row(
274 "SELECT COALESCE(MAX(sequence), 0) + 1 FROM app_server_items WHERE thread_id = ?1",
275 params![thread_id],
276 |row| row.get(0),
277 )
278 .map_err(ThreadStoreError::sql)?;
279 let inserted = connection
280 .execute(
281 "INSERT OR IGNORE INTO app_server_items
282 (item_id, thread_id, turn_id, sequence, payload_json)
283 VALUES (?1, ?2, ?3, ?4, ?5)",
284 params![&item.item_id, thread_id, turn_id, sequence, payload_json],
285 )
286 .map_err(ThreadStoreError::sql)?;
287 if inserted == 1 {
288 return Ok(ItemAppendOutcome::Inserted);
289 }
290
291 let existing = connection
292 .query_row(
293 "SELECT thread_id, turn_id, payload_json
294 FROM app_server_items WHERE item_id = ?1",
295 params![&item.item_id],
296 |row| {
297 Ok((
298 row.get::<_, String>(0)?,
299 row.get::<_, String>(1)?,
300 row.get::<_, String>(2)?,
301 ))
302 },
303 )
304 .optional()
305 .map_err(ThreadStoreError::sql)?;
306 let Some((existing_thread_id, existing_turn_id, existing_payload)) = existing else {
307 return Err(ThreadStoreError::item_identity_conflict(&item.item_id));
308 };
309 let existing_item = serde_json::from_str::<AppItem>(&existing_payload)
310 .map_err(|_| ThreadStoreError::item_identity_conflict(&item.item_id))?;
311 if existing_thread_id == thread_id && existing_turn_id == turn_id && existing_item == item {
312 Ok(ItemAppendOutcome::AlreadyPresent)
313 } else {
314 Err(ThreadStoreError::item_identity_conflict(&item.item_id))
315 }
316 }
317
318 pub fn replay_items(&self, thread_id: &str) -> Result<Vec<AppItem>, ThreadStoreError> {
319 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
320 let mut statement = connection
321 .prepare(
322 "SELECT payload_json FROM app_server_items
323 WHERE thread_id = ?1 ORDER BY sequence ASC",
324 )
325 .map_err(ThreadStoreError::sql)?;
326 let rows = statement
327 .query_map(params![thread_id], |row| row.get::<_, String>(0))
328 .map_err(ThreadStoreError::sql)?;
329 let mut items = Vec::new();
330 for row in rows {
331 let payload_json = row.map_err(ThreadStoreError::sql)?;
332 let mut item: AppItem =
333 serde_json::from_str(&payload_json).map_err(ThreadStoreError::json)?;
334 item.created_at = normalize_timestamp(item.created_at);
335 item.updated_at = normalize_timestamp(item.updated_at);
336 items.push(item);
337 }
338 Ok(items)
339 }
340
341 fn insert_thread(&self, thread: &AppThread) -> Result<(), ThreadStoreError> {
342 let cwd = thread
343 .cwd
344 .as_ref()
345 .map(|path| path_to_string(path.as_path()));
346 let metadata_json =
347 serde_json::to_string(&thread.metadata).map_err(ThreadStoreError::json)?;
348 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
349 connection
350 .execute(
351 "INSERT INTO app_server_threads
352 (thread_id, agent_key, cwd, created_at, updated_at, archived_at, status, metadata_json, active_turn_id)
353 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL)",
354 params![
355 thread.thread_id,
356 thread.agent_key,
357 cwd,
358 thread.created_at,
359 thread.updated_at,
360 thread.archived_at,
361 thread_status_to_str(thread.status),
362 metadata_json,
363 ],
364 )
365 .map_err(ThreadStoreError::sql)?;
366 Ok(())
367 }
368
369 fn migrate(&self) -> Result<(), ThreadStoreError> {
370 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
371 connection
372 .execute_batch(
373 r#"
374 CREATE TABLE IF NOT EXISTS app_server_threads (
375 thread_id TEXT PRIMARY KEY,
376 agent_key TEXT NOT NULL,
377 cwd TEXT,
378 created_at REAL NOT NULL,
379 updated_at REAL NOT NULL,
380 archived_at REAL,
381 status TEXT NOT NULL,
382 metadata_json TEXT NOT NULL,
383 active_turn_id TEXT
384 );
385
386 CREATE TABLE IF NOT EXISTS app_server_turns (
387 turn_id TEXT PRIMARY KEY,
388 thread_id TEXT NOT NULL,
389 run_id TEXT,
390 status TEXT NOT NULL,
391 started_at REAL NOT NULL,
392 completed_at REAL,
393 input_json TEXT NOT NULL,
394 result_json TEXT NOT NULL
395 );
396
397 CREATE TABLE IF NOT EXISTS app_server_items (
398 item_id TEXT PRIMARY KEY,
399 thread_id TEXT NOT NULL,
400 turn_id TEXT NOT NULL,
401 sequence INTEGER NOT NULL,
402 payload_json TEXT NOT NULL
403 );
404
405 CREATE INDEX IF NOT EXISTS idx_app_server_items_thread_sequence
406 ON app_server_items(thread_id, sequence);
407 CREATE INDEX IF NOT EXISTS idx_app_server_turns_thread
408 ON app_server_turns(thread_id);
409 "#,
410 )
411 .map_err(ThreadStoreError::sql)?;
412 ensure_column(
413 &connection,
414 "app_server_threads",
415 "status",
416 "TEXT NOT NULL DEFAULT 'idle'",
417 )?;
418 ensure_column(&connection, "app_server_threads", "active_turn_id", "TEXT")?;
419 Ok(())
420 }
421
422 fn recover_interrupted_threads(&self) -> Result<(), ThreadStoreError> {
423 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
424 connection
425 .execute(
426 "UPDATE app_server_threads
427 SET status = 'idle', active_turn_id = NULL
428 WHERE status = 'running'",
429 [],
430 )
431 .map_err(ThreadStoreError::sql)?;
432 Ok(())
433 }
434
435 fn seed_sequences(&self) -> Result<(), ThreadStoreError> {
436 let connection = self.connection.lock().map_err(ThreadStoreError::poisoned)?;
437 let thread_count: i64 = connection
438 .query_row("SELECT COUNT(*) FROM app_server_threads", [], |row| {
439 row.get(0)
440 })
441 .map_err(ThreadStoreError::sql)?;
442 let turn_count: i64 = connection
443 .query_row("SELECT COUNT(*) FROM app_server_turns", [], |row| {
444 row.get(0)
445 })
446 .map_err(ThreadStoreError::sql)?;
447 self.next_thread_id
448 .store(thread_count.max(0) as u64 + 1, Ordering::Relaxed);
449 self.next_turn_id
450 .store(turn_count.max(0) as u64 + 1, Ordering::Relaxed);
451 Ok(())
452 }
453}
454
455#[derive(Debug, Clone, PartialEq, Eq)]
456pub struct ThreadStoreError {
457 message: String,
458}
459
460impl ThreadStoreError {
461 fn sql(error: rusqlite::Error) -> Self {
462 Self {
463 message: error.to_string(),
464 }
465 }
466
467 fn json(error: serde_json::Error) -> Self {
468 Self {
469 message: error.to_string(),
470 }
471 }
472
473 fn poisoned<T>(_: std::sync::PoisonError<T>) -> Self {
474 Self {
475 message: "thread store lock poisoned".to_string(),
476 }
477 }
478
479 fn not_found(kind: &str, id: &str) -> Self {
480 Self {
481 message: format!("unknown {kind}: {id}"),
482 }
483 }
484
485 fn item_identity_conflict(item_id: &str) -> Self {
486 Self {
487 message: format!("app_server_item_identity_conflict: {item_id}"),
488 }
489 }
490}
491
492impl fmt::Display for ThreadStoreError {
493 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
494 formatter.write_str(&self.message)
495 }
496}
497
498impl std::error::Error for ThreadStoreError {}
499
500fn row_to_thread(row: &rusqlite::Row<'_>) -> rusqlite::Result<AppThread> {
501 let cwd: Option<String> = row.get("cwd")?;
502 let status: String = row.get("status")?;
503 let created_at = read_timestamp(row, "created_at")?;
504 let updated_at = read_timestamp(row, "updated_at")?;
505 let archived_at = read_optional_timestamp(row, "archived_at")?;
506 let metadata_json: String = row.get("metadata_json")?;
507 Ok(AppThread {
508 thread_id: row.get("thread_id")?,
509 agent_key: row.get("agent_key")?,
510 cwd: cwd.map(PathBuf::from),
511 created_at,
512 updated_at,
513 archived_at,
514 status: thread_status_from_str(&status).map_err(conversion_error)?,
515 metadata: serde_json::from_str(&metadata_json).map_err(json_conversion_error)?,
516 })
517}
518
519fn row_to_turn(row: &rusqlite::Row<'_>) -> rusqlite::Result<AppTurn> {
520 let status: String = row.get("status")?;
521 let started_at = read_timestamp(row, "started_at")?;
522 let completed_at = read_optional_timestamp(row, "completed_at")?;
523 let input_json: String = row.get("input_json")?;
524 let result_json: String = row.get("result_json")?;
525 Ok(AppTurn {
526 turn_id: row.get("turn_id")?,
527 thread_id: row.get("thread_id")?,
528 run_id: row.get("run_id")?,
529 status: turn_status_from_str(&status).map_err(conversion_error)?,
530 started_at,
531 completed_at,
532 input: serde_json::from_str(&input_json).map_err(json_conversion_error)?,
533 result: serde_json::from_str(&result_json).map_err(json_conversion_error)?,
534 })
535}
536
537fn query_turn(connection: &Connection, turn_id: &str) -> Result<Option<AppTurn>, ThreadStoreError> {
538 connection
539 .query_row(
540 "SELECT turn_id, thread_id, run_id, status, started_at, completed_at, input_json, result_json
541 FROM app_server_turns WHERE turn_id = ?1",
542 params![turn_id],
543 row_to_turn,
544 )
545 .optional()
546 .map_err(ThreadStoreError::sql)
547}
548
549fn thread_status_to_str(status: ThreadStatus) -> &'static str {
550 match status {
551 ThreadStatus::Idle => "idle",
552 ThreadStatus::Running => "running",
553 ThreadStatus::Archived => "archived",
554 ThreadStatus::Closed => "closed",
555 }
556}
557
558fn thread_status_from_str(status: &str) -> Result<ThreadStatus, String> {
559 match status {
560 "idle" => Ok(ThreadStatus::Idle),
561 "running" => Ok(ThreadStatus::Running),
562 "archived" => Ok(ThreadStatus::Archived),
563 "closed" => Ok(ThreadStatus::Closed),
564 other => Err(format!("unknown thread status: {other}")),
565 }
566}
567
568fn turn_status_to_str(status: TurnStatus) -> &'static str {
569 match status {
570 TurnStatus::Queued => "queued",
571 TurnStatus::Running => "running",
572 TurnStatus::Completed => "completed",
573 TurnStatus::Failed => "failed",
574 TurnStatus::Interrupted => "interrupted",
575 }
576}
577
578fn turn_status_from_str(status: &str) -> Result<TurnStatus, String> {
579 match status {
580 "queued" => Ok(TurnStatus::Queued),
581 "running" => Ok(TurnStatus::Running),
582 "completed" => Ok(TurnStatus::Completed),
583 "failed" => Ok(TurnStatus::Failed),
584 "interrupted" => Ok(TurnStatus::Interrupted),
585 other => Err(format!("unknown turn status: {other}")),
586 }
587}
588
589fn conversion_error(message: String) -> rusqlite::Error {
590 rusqlite::Error::FromSqlConversionFailure(
591 0,
592 rusqlite::types::Type::Text,
593 Box::new(ThreadStoreError { message }),
594 )
595}
596
597fn json_conversion_error(error: serde_json::Error) -> rusqlite::Error {
598 conversion_error(error.to_string())
599}
600
601fn path_to_string(path: &Path) -> String {
602 path.to_string_lossy().to_string()
603}
604
605fn timestamp_seconds() -> f64 {
606 SystemTime::now()
607 .duration_since(UNIX_EPOCH)
608 .map(|duration| duration.as_secs_f64())
609 .unwrap_or_default()
610}
611
612fn read_timestamp(row: &rusqlite::Row<'_>, column: &str) -> rusqlite::Result<f64> {
613 use rusqlite::types::ValueRef;
614
615 let value = match row.get_ref(column)? {
616 ValueRef::Integer(value) => value as f64,
617 ValueRef::Real(value) => value,
618 other => {
619 return Err(rusqlite::Error::InvalidColumnType(
620 0,
621 column.to_string(),
622 other.data_type(),
623 ))
624 }
625 };
626 Ok(normalize_timestamp(value))
627}
628
629fn read_optional_timestamp(row: &rusqlite::Row<'_>, column: &str) -> rusqlite::Result<Option<f64>> {
630 use rusqlite::types::ValueRef;
631
632 match row.get_ref(column)? {
633 ValueRef::Null => Ok(None),
634 ValueRef::Integer(value) => Ok(Some(normalize_timestamp(value as f64))),
635 ValueRef::Real(value) => Ok(Some(normalize_timestamp(value))),
636 other => Err(rusqlite::Error::InvalidColumnType(
637 0,
638 column.to_string(),
639 other.data_type(),
640 )),
641 }
642}
643
644fn normalize_timestamp(value: f64) -> f64 {
645 if value.abs() >= 100_000_000_000.0 {
646 value / 1000.0
647 } else {
648 value
649 }
650}
651
652fn ensure_column(
653 connection: &Connection,
654 table: &str,
655 column: &str,
656 declaration: &str,
657) -> Result<(), ThreadStoreError> {
658 let mut statement = connection
659 .prepare(&format!("PRAGMA table_info({table})"))
660 .map_err(ThreadStoreError::sql)?;
661 let columns = statement
662 .query_map([], |row| row.get::<_, String>(1))
663 .map_err(ThreadStoreError::sql)?
664 .collect::<Result<Vec<_>, _>>()
665 .map_err(ThreadStoreError::sql)?;
666 drop(statement);
667 if !columns.iter().any(|existing| existing == column) {
668 connection
669 .execute(
670 &format!("ALTER TABLE {table} ADD COLUMN {column} {declaration}"),
671 [],
672 )
673 .map_err(ThreadStoreError::sql)?;
674 }
675 Ok(())
676}
677
678#[cfg(test)]
679mod tests {
680 use super::*;
681 use crate::app_server::protocol::{AppItemKind, AppItemStatus};
682
683 #[test]
684 fn item_append_redelivery_survives_restart_and_rejects_conflicts() {
685 let directory = tempfile::tempdir().expect("tempdir");
686 let path = directory.path().join("thread-store.sqlite");
687 let store = SqliteThreadStore::open(&path).expect("store");
688 let thread = store
689 .create_thread(ThreadStartParams {
690 agent_key: "default".to_string(),
691 cwd: None,
692 metadata: Default::default(),
693 })
694 .expect("thread");
695 let turn = store
696 .create_turn(&thread.thread_id, Vec::new())
697 .expect("turn");
698 let item = AppItem {
699 item_id: "item_1".to_string(),
700 thread_id: thread.thread_id.clone(),
701 turn_id: turn.turn_id.clone(),
702 kind: AppItemKind::AgentMessage,
703 status: AppItemStatus::Completed,
704 payload: serde_json::json!({"text": "original"}),
705 created_at: 1.0,
706 updated_at: 1.0,
707 };
708 assert_eq!(
709 store
710 .append_item(&thread.thread_id, &turn.turn_id, item.clone())
711 .expect("first append"),
712 ItemAppendOutcome::Inserted
713 );
714 drop(store);
715
716 let store = SqliteThreadStore::open(&path).expect("reopened store");
717 assert_eq!(
718 store
719 .append_item(&thread.thread_id, &turn.turn_id, item.clone())
720 .expect("same event redelivery after restart"),
721 ItemAppendOutcome::AlreadyPresent
722 );
723 let mut replacement = item.clone();
724 replacement.payload = serde_json::json!({"text": "replacement"});
725
726 let error = store
727 .append_item(&thread.thread_id, &turn.turn_id, replacement)
728 .expect_err("conflicting projection");
729 assert_eq!(
730 error.to_string(),
731 "app_server_item_identity_conflict: item_1"
732 );
733 let other_turn = store
734 .create_turn(&thread.thread_id, Vec::new())
735 .expect("other turn");
736 let mut misplaced = item.clone();
737 misplaced.turn_id = other_turn.turn_id.clone();
738 let error = store
739 .append_item(&thread.thread_id, &other_turn.turn_id, misplaced)
740 .expect_err("same identity cannot move to another turn");
741 assert_eq!(
742 error.to_string(),
743 "app_server_item_identity_conflict: item_1"
744 );
745
746 let mut next = item.clone();
747 next.item_id = "item_2".to_string();
748 next.payload = serde_json::json!({"text": "next"});
749 assert_eq!(
750 store
751 .append_item(&thread.thread_id, &turn.turn_id, next)
752 .expect("next append"),
753 ItemAppendOutcome::Inserted
754 );
755 let replay = store.replay_items(&thread.thread_id).expect("replay");
756 assert_eq!(replay.len(), 2);
757 assert_eq!(replay[0], item);
758 assert_eq!(replay[1].item_id, "item_2");
759 }
760
761 #[test]
762 fn opening_a_legacy_database_adds_thread_lifecycle_columns() {
763 let directory = tempfile::tempdir().expect("tempdir");
764 let path = directory.path().join("legacy.sqlite");
765 let connection = Connection::open(&path).expect("legacy connection");
766 connection
767 .execute_batch(
768 r#"
769 CREATE TABLE app_server_threads (
770 thread_id TEXT PRIMARY KEY,
771 agent_key TEXT NOT NULL,
772 cwd TEXT,
773 created_at REAL NOT NULL,
774 updated_at REAL NOT NULL,
775 archived_at REAL,
776 metadata_json TEXT NOT NULL
777 );
778 INSERT INTO app_server_threads (
779 thread_id, agent_key, cwd, created_at, updated_at, archived_at, metadata_json
780 ) VALUES ('thread_1', 'default', NULL, 1.0, 1.0, NULL, '{}');
781 "#,
782 )
783 .expect("legacy schema");
784 drop(connection);
785
786 let store = SqliteThreadStore::open(&path).expect("migrated store");
787 let thread = store
788 .get_thread("thread_1")
789 .expect("read thread")
790 .expect("thread exists");
791 assert_eq!(thread.status, ThreadStatus::Idle);
792 }
793}