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