ninox_core/
store.rs

1use crate::types::*;
2use anyhow::Result;
3use rusqlite::{params, Connection};
4use std::{path::Path, sync::Mutex};
5
6pub struct Store {
7    conn: Mutex<Connection>,
8}
9
10
11impl Store {
12    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
13        let conn = Connection::open(path)?;
14        conn.execute_batch("
15            PRAGMA journal_mode=WAL;
16            PRAGMA busy_timeout=5000;
17            CREATE TABLE IF NOT EXISTS sessions (
18                id TEXT PRIMARY KEY, orchestrator_id TEXT,
19                name TEXT NOT NULL, repo TEXT NOT NULL,
20                status TEXT NOT NULL, agent_type TEXT NOT NULL,
21                cost_usd REAL NOT NULL DEFAULT 0, started_at INTEGER NOT NULL,
22                pr_number INTEGER, pr_id INTEGER,
23                workspace_path TEXT, pid INTEGER,
24                model TEXT, context_tokens INTEGER
25            );
26            CREATE TABLE IF NOT EXISTS orchestrators (
27                id TEXT PRIMARY KEY, name TEXT NOT NULL, created_at INTEGER NOT NULL
28            );
29            CREATE TABLE IF NOT EXISTS prs (
30                id INTEGER PRIMARY KEY, number INTEGER NOT NULL,
31                title TEXT NOT NULL, url TEXT NOT NULL,
32                body TEXT NOT NULL, session_id TEXT NOT NULL
33            );
34            CREATE TABLE IF NOT EXISTS ci_status (
35                pr_id INTEGER PRIMARY KEY, total INTEGER NOT NULL,
36                passing INTEGER NOT NULL, failing INTEGER NOT NULL,
37                pending INTEGER NOT NULL
38            );
39            CREATE TABLE IF NOT EXISTS review_comments (
40                id INTEGER PRIMARY KEY, pr_id INTEGER NOT NULL,
41                author TEXT NOT NULL, body TEXT NOT NULL,
42                path TEXT, line INTEGER, created_at INTEGER NOT NULL
43            );
44        ")?;
45        // Migrations for columns added after initial release — idempotent so
46        // both fresh and pre-existing databases end up with the same schema.
47        for (col, ddl) in [
48            ("model",                "ALTER TABLE sessions ADD COLUMN model TEXT"),
49            ("context_tokens",       "ALTER TABLE sessions ADD COLUMN context_tokens INTEGER"),
50            ("catalogue_path",       "ALTER TABLE sessions ADD COLUMN catalogue_path TEXT"),
51            ("context_used_pct",     "ALTER TABLE sessions ADD COLUMN context_used_pct REAL"),
52            ("context_total_tokens", "ALTER TABLE sessions ADD COLUMN context_total_tokens INTEGER"),
53            ("context_window_size",  "ALTER TABLE sessions ADD COLUMN context_window_size INTEGER"),
54            ("claude_session_id",    "ALTER TABLE sessions ADD COLUMN claude_session_id TEXT"),
55            ("summary",              "ALTER TABLE sessions ADD COLUMN summary TEXT"),
56            ("terminal_at",          "ALTER TABLE sessions ADD COLUMN terminal_at INTEGER"),
57            ("gate_status",          "ALTER TABLE sessions ADD COLUMN gate_status TEXT"),
58        ] {
59            if !Self::column_exists(&conn, "sessions", col)? {
60                conn.execute(ddl, [])?;
61            }
62        }
63        Ok(Self { conn: Mutex::new(conn) })
64    }
65
66    fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool> {
67        let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
68        let exists = stmt
69            .query_map([], |r| r.get::<_, String>(1))?
70            .filter_map(|r| r.ok())
71            .any(|c| c == column);
72        Ok(exists)
73    }
74
75    pub fn upsert_session(&self, s: &Session) -> Result<()> {
76        let status = serde_json::to_string(&s.status)?.replace('"', "");
77        let gate_status = s.gate_status.as_ref()
78            .map(serde_json::to_string)
79            .transpose()?;
80        let conn = self.conn.lock().unwrap();
81        conn.execute(
82            "INSERT INTO sessions (id,orchestrator_id,name,repo,status,agent_type,
83             cost_usd,started_at,pr_number,pr_id,workspace_path,pid,model,context_tokens,
84             catalogue_path,context_used_pct,context_total_tokens,context_window_size,
85             claude_session_id,summary,terminal_at,gate_status)
86             VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20,?21,?22)
87             ON CONFLICT(id) DO UPDATE SET
88             repo=excluded.repo,
89             status=excluded.status,cost_usd=excluded.cost_usd,
90             started_at=excluded.started_at,
91             pr_number=excluded.pr_number,pr_id=excluded.pr_id,
92             workspace_path=excluded.workspace_path,pid=excluded.pid,
93             model=excluded.model,context_tokens=excluded.context_tokens,
94             catalogue_path=excluded.catalogue_path,
95             context_used_pct=excluded.context_used_pct,
96             context_total_tokens=excluded.context_total_tokens,
97             context_window_size=excluded.context_window_size,
98             claude_session_id=excluded.claude_session_id,
99             summary=excluded.summary,
100             terminal_at=excluded.terminal_at,
101             gate_status=excluded.gate_status",
102            params![
103                s.id, s.orchestrator_id, s.name, s.repo, status, s.agent_type,
104                s.cost_usd, s.started_at, s.pr_number, s.pr_id,
105                s.workspace_path, s.pid, s.model, s.context_tokens,
106                s.catalogue_path, s.context_used_pct, s.context_total_tokens,
107                s.context_window_size, s.claude_session_id, s.summary, s.terminal_at,
108                gate_status
109            ],
110        )?;
111        Ok(())
112    }
113
114    pub fn list_sessions(&self) -> Result<Vec<Session>> {
115        let conn = self.conn.lock().unwrap();
116        let mut stmt = conn.prepare(
117            "SELECT id,orchestrator_id,name,repo,status,agent_type,cost_usd,
118             started_at,pr_number,pr_id,workspace_path,pid,model,context_tokens,
119             catalogue_path,context_used_pct,context_total_tokens,context_window_size,
120             claude_session_id,summary,terminal_at,gate_status
121             FROM sessions ORDER BY started_at DESC",
122        )?;
123        let rows = stmt.query_map([], |r| {
124            Ok((
125                r.get::<_, String>(0)?,
126                r.get::<_, Option<String>>(1)?,
127                r.get::<_, String>(2)?,
128                r.get::<_, String>(3)?,
129                r.get::<_, String>(4)?,
130                r.get::<_, String>(5)?,
131                r.get::<_, f64>(6)?,
132                r.get::<_, i64>(7)?,
133                r.get::<_, Option<u64>>(8)?,
134                r.get::<_, Option<i64>>(9)?,
135                r.get::<_, Option<String>>(10)?,
136                r.get::<_, Option<u32>>(11)?,
137                r.get::<_, Option<String>>(12)?,
138                r.get::<_, Option<i64>>(13)?,
139                r.get::<_, Option<String>>(14)?,
140                r.get::<_, Option<f64>>(15)?,
141                r.get::<_, Option<i64>>(16)?,
142                r.get::<_, Option<i64>>(17)?,
143                r.get::<_, Option<String>>(18)?,
144                r.get::<_, Option<String>>(19)?,
145                r.get::<_, Option<i64>>(20)?,
146                r.get::<_, Option<String>>(21)?,
147            ))
148        })?;
149        rows.map(|r| {
150            let (id, orchestrator_id, name, repo, status_str, agent_type,
151                 cost_usd, started_at, pr_number, pr_id, workspace_path, pid,
152                 model, context_tokens, catalogue_path, context_used_pct,
153                 context_total_tokens, context_window_size, claude_session_id,
154                 summary, terminal_at, gate_status_str) = r?;
155            let status = serde_json::from_str(&format!("\"{status_str}\""))
156                .unwrap_or(SessionStatus::Working);
157            let gate_status = gate_status_str
158                .and_then(|s| serde_json::from_str(&s).ok());
159            Ok(Session {
160                id, orchestrator_id, name, repo, status, agent_type,
161                cost_usd, started_at, pr_number, pr_id, workspace_path, pid,
162                model, context_tokens: context_tokens.map(|v| v.max(0) as u64),
163                catalogue_path,
164                context_used_pct,
165                context_total_tokens: context_total_tokens.map(|v| v.max(0) as u64),
166                context_window_size: context_window_size.map(|v| v.max(0) as u64),
167                claude_session_id,
168                summary,
169                terminal_at,
170                gate_status,
171            })
172        })
173        .collect()
174    }
175
176    pub fn get_session(&self, id: &str) -> Result<Option<Session>> {
177        let conn = self.conn.lock().unwrap();
178        let mut stmt = conn.prepare(
179            "SELECT id,orchestrator_id,name,repo,status,agent_type,cost_usd,
180             started_at,pr_number,pr_id,workspace_path,pid,model,context_tokens,
181             catalogue_path,context_used_pct,context_total_tokens,context_window_size,
182             claude_session_id,summary,terminal_at,gate_status
183             FROM sessions WHERE id = ?1",
184        )?;
185        let mut rows = stmt.query_map([id], |r| {
186            Ok((
187                r.get::<_, String>(0)?,
188                r.get::<_, Option<String>>(1)?,
189                r.get::<_, String>(2)?,
190                r.get::<_, String>(3)?,
191                r.get::<_, String>(4)?,
192                r.get::<_, String>(5)?,
193                r.get::<_, f64>(6)?,
194                r.get::<_, i64>(7)?,
195                r.get::<_, Option<u64>>(8)?,
196                r.get::<_, Option<i64>>(9)?,
197                r.get::<_, Option<String>>(10)?,
198                r.get::<_, Option<u32>>(11)?,
199                r.get::<_, Option<String>>(12)?,
200                r.get::<_, Option<i64>>(13)?,
201                r.get::<_, Option<String>>(14)?,
202                r.get::<_, Option<f64>>(15)?,
203                r.get::<_, Option<i64>>(16)?,
204                r.get::<_, Option<i64>>(17)?,
205                r.get::<_, Option<String>>(18)?,
206                r.get::<_, Option<String>>(19)?,
207                r.get::<_, Option<i64>>(20)?,
208                r.get::<_, Option<String>>(21)?,
209            ))
210        })?;
211        match rows.next() {
212            None => Ok(None),
213            Some(r) => {
214                let (id, orchestrator_id, name, repo, status_str, agent_type,
215                     cost_usd, started_at, pr_number, pr_id, workspace_path, pid,
216                     model, context_tokens, catalogue_path, context_used_pct,
217                     context_total_tokens, context_window_size, claude_session_id,
218                     summary, terminal_at, gate_status_str) = r?;
219                let status = serde_json::from_str(&format!("\"{status_str}\""))
220                    .unwrap_or(SessionStatus::Working);
221                let gate_status = gate_status_str
222                    .and_then(|s| serde_json::from_str(&s).ok());
223                Ok(Some(Session {
224                    id, orchestrator_id, name, repo, status, agent_type,
225                    cost_usd, started_at, pr_number, pr_id, workspace_path, pid,
226                    model, context_tokens: context_tokens.map(|v| v.max(0) as u64),
227                    catalogue_path,
228                    context_used_pct,
229                    context_total_tokens: context_total_tokens.map(|v| v.max(0) as u64),
230                    context_window_size: context_window_size.map(|v| v.max(0) as u64),
231                    claude_session_id,
232                    summary,
233                    terminal_at,
234                    gate_status,
235                }))
236            }
237        }
238    }
239
240    /// Non-zero `cost_usd` samples recorded for sessions matching the given
241    /// agent harness (`agent_type`) and model — used to compute a
242    /// data-driven spawn-modal cost estimate once enough history exists for
243    /// a given preset. Read-only; built on `list_sessions` like
244    /// `sessions_by_orchestrator`.
245    pub fn cost_samples(&self, agent_type: &str, model: Option<&str>) -> Result<Vec<f64>> {
246        let sessions = self.list_sessions()?;
247        Ok(sessions
248            .into_iter()
249            .filter(|s| {
250                s.agent_type == agent_type
251                    && s.model.as_deref() == model
252                    && s.cost_usd > 0.0
253            })
254            .map(|s| s.cost_usd)
255            .collect())
256    }
257
258    pub fn upsert_orchestrator(&self, o: &Orchestrator) -> Result<()> {
259        let conn = self.conn.lock().unwrap();
260        conn.execute(
261            "INSERT INTO orchestrators(id,name,created_at) VALUES(?1,?2,?3)
262             ON CONFLICT(id) DO UPDATE SET name=excluded.name",
263            params![o.id, o.name, o.created_at],
264        )?;
265        Ok(())
266    }
267
268    pub fn sessions_by_orchestrator(&self, orchestrator_id: &str) -> Result<Vec<Session>> {
269        let sessions = self.list_sessions()?;
270        Ok(sessions.into_iter().filter(|s| s.orchestrator_id.as_deref() == Some(orchestrator_id)).collect())
271    }
272
273    pub fn delete_session(&self, id: &str) -> Result<()> {
274        let conn = self.conn.lock().unwrap();
275        conn.execute("DELETE FROM sessions WHERE id = ?1", [id])?;
276        Ok(())
277    }
278
279    pub fn delete_orchestrator(&self, id: &str) -> Result<()> {
280        let conn = self.conn.lock().unwrap();
281        conn.execute("DELETE FROM sessions WHERE orchestrator_id = ?1", [id])?;
282        conn.execute("DELETE FROM sessions WHERE id = ?1", [id])?;
283        conn.execute("DELETE FROM orchestrators WHERE id = ?1", [id])?;
284        Ok(())
285    }
286
287    pub fn upsert_pr(&self, pr: &PR) -> Result<()> {
288        let conn = self.conn.lock().unwrap();
289        conn.execute(
290            "INSERT OR REPLACE INTO prs
291             (id, number, title, url, body, session_id)
292             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
293            params![pr.id, pr.number, pr.title, pr.url, pr.body, pr.session_id],
294        )?;
295        Ok(())
296    }
297
298    pub fn get_pr(&self, id: PrId) -> Result<Option<PR>> {
299        let conn = self.conn.lock().unwrap();
300        let mut stmt = conn.prepare(
301            "SELECT id, number, title, url, body, session_id FROM prs WHERE id = ?1",
302        )?;
303        let mut rows = stmt.query_map([id], |r| {
304            Ok(PR {
305                id:         r.get(0)?,
306                number:     r.get(1)?,
307                title:      r.get(2)?,
308                url:        r.get(3)?,
309                body:       r.get(4)?,
310                session_id: r.get(5)?,
311            })
312        })?;
313        rows.next().transpose().map_err(Into::into)
314    }
315
316    pub fn upsert_ci_status(&self, ci: &CIStatus) -> Result<()> {
317        let conn = self.conn.lock().unwrap();
318        conn.execute(
319            "INSERT OR REPLACE INTO ci_status
320             (pr_id, total, passing, failing, pending)
321             VALUES (?1, ?2, ?3, ?4, ?5)",
322            params![ci.pr_id, ci.total, ci.passing, ci.failing, ci.pending],
323        )?;
324        Ok(())
325    }
326
327    pub fn upsert_comment(&self, c: &Comment) -> Result<()> {
328        let conn = self.conn.lock().unwrap();
329        conn.execute(
330            "INSERT OR REPLACE INTO review_comments
331             (id, pr_id, author, body, path, line, created_at)
332             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
333            params![c.id, c.pr_id, c.author, c.body, c.path, c.line, c.created_at],
334        )?;
335        Ok(())
336    }
337
338    /// All persisted review/issue comments, ordered by `created_at` — used to
339    /// hydrate `App`'s in-memory comment feed on startup so a restart doesn't
340    /// lose comments a previous run already fetched.
341    pub fn list_comments(&self) -> Result<Vec<Comment>> {
342        let conn = self.conn.lock().unwrap();
343        let mut stmt = conn.prepare(
344            "SELECT id, pr_id, author, body, path, line, created_at
345             FROM review_comments ORDER BY created_at ASC",
346        )?;
347        let rows = stmt.query_map([], |r| {
348            Ok(Comment {
349                id:         r.get(0)?,
350                pr_id:      r.get(1)?,
351                author:     r.get(2)?,
352                body:       r.get(3)?,
353                path:       r.get(4)?,
354                line:       r.get(5)?,
355                created_at: r.get(6)?,
356            })
357        })?;
358        rows.collect::<rusqlite::Result<Vec<_>>>().map_err(Into::into)
359    }
360
361    pub fn list_orchestrators(&self) -> Result<Vec<Orchestrator>> {
362        let conn = self.conn.lock().unwrap();
363        let mut stmt = conn.prepare(
364            "SELECT id,name,created_at FROM orchestrators ORDER BY created_at DESC",
365        )?;
366        let rows = stmt.query_map([], |r| {
367            Ok(Orchestrator {
368                id: r.get(0)?,
369                name: r.get(1)?,
370                created_at: r.get(2)?,
371            })
372        })?;
373        rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
374    }
375}
376
377#[cfg(test)]
378mod tests {
379    use super::*;
380    use tempfile::tempdir;
381
382    fn test_store() -> Store {
383        let dir = tempdir().unwrap();
384        let path = dir.path().join("t.db");
385        // keep dir alive for the lifetime of the test by leaking it
386        std::mem::forget(dir);
387        Store::open(path).unwrap()
388    }
389
390    #[test]
391    fn upsert_and_list_session() {
392        let store = test_store();
393        let session = Session {
394            id: "s1".into(), orchestrator_id: None, name: "worker-1".into(),
395            repo: "slievr/Athene".into(), status: SessionStatus::Working,
396            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
397            pr_number: None, pr_id: None, workspace_path: None, pid: None,
398            model: None, context_tokens: None, catalogue_path: None,
399            context_used_pct: None, context_total_tokens: None, context_window_size: None,
400            claude_session_id: None,
401            summary: None,
402            terminal_at: None, gate_status: None,
403        };
404        store.upsert_session(&session).unwrap();
405        let list = store.list_sessions().unwrap();
406        assert_eq!(list.len(), 1);
407        assert_eq!(list[0].id, "s1");
408    }
409
410    #[test]
411    fn upsert_updates_status() {
412        let store = test_store();
413        let mut s = Session {
414            id: "s1".into(), orchestrator_id: None, name: "w".into(),
415            repo: "r".into(), status: SessionStatus::Working,
416            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
417            pr_number: None, pr_id: None, workspace_path: None, pid: None,
418            model: None, context_tokens: None, catalogue_path: None,
419            context_used_pct: None, context_total_tokens: None, context_window_size: None,
420            claude_session_id: None,
421            summary: None,
422            terminal_at: None, gate_status: None,
423        };
424        store.upsert_session(&s).unwrap();
425        s.status = SessionStatus::Done;
426        store.upsert_session(&s).unwrap();
427        let list = store.list_sessions().unwrap();
428        assert_eq!(list.len(), 1);
429        assert!(matches!(list[0].status, SessionStatus::Done));
430    }
431
432    /// The poller's dual-remote self-heal (`session.repo` corrected once a
433    /// PR is found against a different remote than the one on record)
434    /// depends on `repo` being an updatable column, not just a write-once
435    /// field set at insert time.
436    #[test]
437    fn upsert_updates_repo() {
438        let store = test_store();
439        let mut s = Session {
440            id: "s1".into(), orchestrator_id: None, name: "w".into(),
441            repo: "OwnerA/repoA".into(), status: SessionStatus::Working,
442            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
443            pr_number: None, pr_id: None, workspace_path: None, pid: None,
444            model: None, context_tokens: None, catalogue_path: None,
445            context_used_pct: None, context_total_tokens: None, context_window_size: None,
446            claude_session_id: None,
447            summary: None,
448            terminal_at: None, gate_status: None,
449        };
450        store.upsert_session(&s).unwrap();
451        s.repo = "OwnerB/repoB".into();
452        store.upsert_session(&s).unwrap();
453        let updated = store.get_session("s1").unwrap().unwrap();
454        assert_eq!(updated.repo, "OwnerB/repoB");
455    }
456
457    #[test]
458    fn get_session_by_id() {
459        let store = test_store();
460        let s = Session {
461            id: "s1".into(), orchestrator_id: None, name: "w".into(),
462            repo: "r".into(), status: SessionStatus::Working,
463            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
464            pr_number: None, pr_id: None, workspace_path: None, pid: None,
465            model: None, context_tokens: None, catalogue_path: None,
466            context_used_pct: None, context_total_tokens: None, context_window_size: None,
467            claude_session_id: None,
468            summary: None,
469            terminal_at: None, gate_status: None,
470        };
471        store.upsert_session(&s).unwrap();
472        let found = store.get_session("s1").unwrap();
473        assert!(found.is_some());
474        assert_eq!(found.unwrap().name, "w");
475        assert!(store.get_session("missing").unwrap().is_none());
476    }
477
478    #[test]
479    fn model_and_context_tokens_round_trip() {
480        let store = test_store();
481        let s = Session {
482            id: "s1".into(), orchestrator_id: None, name: "w".into(),
483            repo: "r".into(), status: SessionStatus::Working,
484            agent_type: "claude-code".into(), cost_usd: 1.5, started_at: 0,
485            pr_number: None, pr_id: None, workspace_path: None, pid: None,
486            model: Some("claude-fable-5".into()), context_tokens: Some(214_000), catalogue_path: None,
487            context_used_pct: None, context_total_tokens: None, context_window_size: None,
488            claude_session_id: None,
489            summary: None,
490            terminal_at: None, gate_status: None,
491        };
492        store.upsert_session(&s).unwrap();
493        let found = store.get_session("s1").unwrap().unwrap();
494        assert_eq!(found.model.as_deref(), Some("claude-fable-5"));
495        assert_eq!(found.context_tokens, Some(214_000));
496    }
497
498    #[test]
499    fn catalogue_path_round_trips() {
500        let store = test_store();
501        let s = Session {
502            id: "s2".into(), orchestrator_id: None, name: "w".into(),
503            repo: "r".into(), status: SessionStatus::Working,
504            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
505            pr_number: None, pr_id: None, workspace_path: None, pid: None,
506            model: None, context_tokens: None,
507            catalogue_path: Some("/brains/x".into()),
508            context_used_pct: None, context_total_tokens: None, context_window_size: None,
509            claude_session_id: None,
510            summary: None,
511            terminal_at: None, gate_status: None,
512        };
513        store.upsert_session(&s).unwrap();
514        let found = store.get_session("s2").unwrap().unwrap();
515        assert_eq!(found.catalogue_path.as_deref(), Some("/brains/x"));
516        // list path decodes it too
517        assert_eq!(store.list_sessions().unwrap()[0].catalogue_path.as_deref(), Some("/brains/x"));
518    }
519
520    #[test]
521    fn summary_round_trips() {
522        let store = test_store();
523        let s = Session {
524            id: "s2b".into(), orchestrator_id: None, name: "w".into(),
525            repo: "r".into(), status: SessionStatus::Working,
526            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
527            pr_number: None, pr_id: None, workspace_path: None, pid: None,
528            model: None, context_tokens: None, catalogue_path: None,
529            context_used_pct: None, context_total_tokens: None, context_window_size: None,
530            claude_session_id: None,
531            summary: Some("Fix flaky CI on the auth suite".into()),
532            terminal_at: None, gate_status: None,
533        };
534        store.upsert_session(&s).unwrap();
535        let found = store.get_session("s2b").unwrap().unwrap();
536        assert_eq!(found.summary.as_deref(), Some("Fix flaky CI on the auth suite"));
537        // list path decodes it too
538        assert_eq!(
539            store.list_sessions().unwrap().iter().find(|x| x.id == "s2b").unwrap().summary.as_deref(),
540            Some("Fix flaky CI on the auth suite"),
541        );
542
543        // None round-trips as None, not "" or "none"
544        let mut s2 = s.clone();
545        s2.id = "s2c".into();
546        s2.summary = None;
547        store.upsert_session(&s2).unwrap();
548        assert_eq!(store.get_session("s2c").unwrap().unwrap().summary, None);
549    }
550
551    #[test]
552    fn claude_session_id_round_trips() {
553        let store = test_store();
554        let s = Session {
555            id: "s3".into(), orchestrator_id: None, name: "w".into(),
556            repo: "r".into(), status: SessionStatus::Working,
557            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
558            pr_number: None, pr_id: None, workspace_path: None, pid: None,
559            model: None, context_tokens: None, catalogue_path: None,
560            context_used_pct: None, context_total_tokens: None, context_window_size: None,
561            claude_session_id: Some("b7e0b3a0-0000-4000-8000-000000000001".into()),
562            summary: None,
563            terminal_at: None, gate_status: None,
564        };
565        store.upsert_session(&s).unwrap();
566        let found = store.get_session("s3").unwrap().unwrap();
567        assert_eq!(found.claude_session_id.as_deref(), Some("b7e0b3a0-0000-4000-8000-000000000001"));
568        // list path decodes it too
569        assert_eq!(
570            store.list_sessions().unwrap().iter().find(|x| x.id == "s3").unwrap().claude_session_id.as_deref(),
571            Some("b7e0b3a0-0000-4000-8000-000000000001"),
572        );
573
574        // None round-trips as None, not "" or "none"
575        let mut s2 = s.clone();
576        s2.id = "s4".into();
577        s2.claude_session_id = None;
578        store.upsert_session(&s2).unwrap();
579        assert_eq!(store.get_session("s4").unwrap().unwrap().claude_session_id, None);
580    }
581
582    #[test]
583    fn terminal_at_round_trips() {
584        let store = test_store();
585        let s = Session {
586            id: "s5".into(), orchestrator_id: None, name: "w".into(),
587            repo: "r".into(), status: SessionStatus::Done,
588            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
589            pr_number: None, pr_id: None, workspace_path: None, pid: None,
590            model: None, context_tokens: None, catalogue_path: None,
591            context_used_pct: None, context_total_tokens: None, context_window_size: None,
592            claude_session_id: None,
593            summary: None,
594            terminal_at: Some(1_720_000_000_000), gate_status: None,
595        };
596        store.upsert_session(&s).unwrap();
597        let found = store.get_session("s5").unwrap().unwrap();
598        assert_eq!(found.terminal_at, Some(1_720_000_000_000));
599        // list path decodes it too
600        assert_eq!(
601            store.list_sessions().unwrap().iter().find(|x| x.id == "s5").unwrap().terminal_at,
602            Some(1_720_000_000_000),
603        );
604
605        // None round-trips as None — non-terminal / not-yet-retired sessions.
606        let mut s2 = s.clone();
607        s2.id = "s6".into();
608        s2.terminal_at = None;
609        store.upsert_session(&s2).unwrap();
610        assert_eq!(store.get_session("s6").unwrap().unwrap().terminal_at, None);
611    }
612
613    /// A Re-file respawns the same session id with a fresh `started_at` —
614    /// the conflict-update path must persist it, or the in-memory time
615    /// silently reverts to the original spawn time on app restart.
616    #[test]
617    fn upsert_conflict_updates_started_at() {
618        let store = test_store();
619        let mut s = Session {
620            id: "s3".into(), orchestrator_id: None, name: "w".into(),
621            repo: "r".into(), status: SessionStatus::Working,
622            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 100,
623            pr_number: None, pr_id: None, workspace_path: None, pid: None,
624            model: None, context_tokens: None, catalogue_path: None,
625            context_used_pct: None, context_total_tokens: None, context_window_size: None,
626            claude_session_id: None,
627            summary: None,
628            terminal_at: None, gate_status: None,
629        };
630        store.upsert_session(&s).unwrap();
631        s.started_at = 200;
632        store.upsert_session(&s).unwrap();
633        assert_eq!(store.get_session("s3").unwrap().unwrap().started_at, 200);
634    }
635
636    #[test]
637    fn get_pr_round_trips_and_misses_cleanly() {
638        let store = test_store();
639        assert!(store.get_pr(9).unwrap().is_none());
640        let pr = PR {
641            id: 9, number: 9, title: "t".into(),
642            url: "https://github.com/org/repo/pull/9".into(),
643            body: String::new(), session_id: "s1".into(),
644        };
645        store.upsert_pr(&pr).unwrap();
646        let found = store.get_pr(9).unwrap().unwrap();
647        assert_eq!(found.number, 9);
648        assert_eq!(found.session_id, "s1");
649        assert_eq!(found.url, pr.url);
650    }
651
652    /// `list_comments` is what hydrates `App::review_threads` on restart —
653    /// it must come back ordered by `created_at` so the UI feed doesn't need
654    /// to re-sort, and `upsert_comment`'s INSERT OR REPLACE must not produce
655    /// duplicate rows for a comment seen twice.
656    #[test]
657    fn list_comments_orders_by_created_at_and_upsert_dedupes() {
658        let store = test_store();
659        let later = Comment {
660            id: 2, pr_id: 1, author: "bob".into(), body: "second".into(),
661            path: None, line: None, created_at: 2_000,
662        };
663        let earlier = Comment {
664            id: 1, pr_id: 1, author: "alice".into(), body: "first".into(),
665            path: Some("src/lib.rs".into()), line: Some(10), created_at: 1_000,
666        };
667        store.upsert_comment(&later).unwrap();
668        store.upsert_comment(&earlier).unwrap();
669
670        let comments = store.list_comments().unwrap();
671        assert_eq!(comments.len(), 2);
672        assert_eq!(comments[0].id, 1, "earlier created_at sorts first");
673        assert_eq!(comments[1].id, 2);
674
675        // Re-upserting the same id (e.g. a repeated poll) must replace, not duplicate.
676        store.upsert_comment(&earlier).unwrap();
677        assert_eq!(store.list_comments().unwrap().len(), 2);
678    }
679
680    #[test]
681    fn context_fields_round_trip() {
682        let store = test_store();
683        let s = Session {
684            id: "s1".into(), orchestrator_id: None, name: "w".into(),
685            repo: "r".into(), status: SessionStatus::Working,
686            agent_type: "claude-code".into(), cost_usd: 2.6, started_at: 0,
687            pr_number: None, pr_id: None, workspace_path: None, pid: None,
688            model: None, context_tokens: None, catalogue_path: None,
689            context_used_pct: Some(62.0),
690            context_total_tokens: Some(124_000),
691            context_window_size: Some(200_000),
692            claude_session_id: None,
693            summary: None,
694            terminal_at: None, gate_status: None,
695        };
696        store.upsert_session(&s).unwrap();
697        let found = store.get_session("s1").unwrap().unwrap();
698        assert_eq!(found.context_used_pct, Some(62.0));
699        assert_eq!(found.context_total_tokens, Some(124_000));
700        assert_eq!(found.context_window_size, Some(200_000));
701        // list path decodes it too
702        assert_eq!(store.list_sessions().unwrap()[0].context_used_pct, Some(62.0));
703    }
704
705    #[test]
706    fn context_fields_default_to_none() {
707        let store = test_store();
708        let s = Session {
709            id: "s2".into(), orchestrator_id: None, name: "w".into(),
710            repo: "r".into(), status: SessionStatus::Working,
711            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
712            pr_number: None, pr_id: None, workspace_path: None, pid: None,
713            model: None, context_tokens: None, catalogue_path: None,
714            context_used_pct: None, context_total_tokens: None, context_window_size: None,
715            claude_session_id: None,
716            summary: None,
717            terminal_at: None, gate_status: None,
718        };
719        store.upsert_session(&s).unwrap();
720        let found = store.get_session("s2").unwrap().unwrap();
721        assert_eq!(found.context_used_pct, None);
722        assert_eq!(found.context_total_tokens, None);
723        assert_eq!(found.context_window_size, None);
724    }
725
726    #[test]
727    fn cost_samples_filters_by_agent_and_model_and_excludes_zero() {
728        let store = test_store();
729        for (id, agent_type, model, cost) in [
730            ("a", "claude-code", Some("claude-fable-5"), 3.0),
731            ("b", "claude-code", Some("claude-fable-5"), 5.0),
732            ("c", "claude-code", Some("claude-fable-5"), 0.0), // excluded: zero cost
733            ("d", "claude-code", Some("claude-opus-4-8"), 2.0), // excluded: different model
734            ("e", "codex",       Some("claude-fable-5"), 4.0),  // excluded: different harness
735        ] {
736            store.upsert_session(&Session {
737                id: id.into(), orchestrator_id: None, name: id.into(),
738                repo: "r".into(), status: SessionStatus::Working,
739                agent_type: agent_type.into(), cost_usd: cost, started_at: 0,
740                pr_number: None, pr_id: None, workspace_path: None, pid: None,
741                model: model.map(String::from), context_tokens: None, catalogue_path: None,
742                context_used_pct: None, context_total_tokens: None, context_window_size: None,
743                claude_session_id: None, summary: None,
744                terminal_at: None, gate_status: None,
745            }).unwrap();
746        }
747        let samples = store.cost_samples("claude-code", Some("claude-fable-5")).unwrap();
748        assert_eq!(samples.len(), 2);
749        assert!((samples.iter().sum::<f64>() - 8.0).abs() < f64::EPSILON);
750    }
751
752    #[test]
753    fn upsert_and_fetch_session_round_trips_gate_status() {
754        let store = Store::open(tempdir().unwrap().keep().join("t.db")).unwrap();
755        let mut session = crate::types::Session {
756            id: "s1".into(), orchestrator_id: None, name: "w".into(),
757            repo: "r".into(), status: crate::types::SessionStatus::PrOpen,
758            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
759            pr_number: Some(1), pr_id: Some(1), workspace_path: None, pid: None,
760            model: None, context_tokens: None, catalogue_path: None,
761            context_used_pct: None, context_total_tokens: None, context_window_size: None,
762            claude_session_id: None, summary: None, terminal_at: None,
763            gate_status: None,
764        };
765        session.gate_status = Some(crate::types::GateStatus {
766            ci: crate::types::GateCheck::Failing,
767            review: crate::types::GateCheck::Passing,
768            mergeable: crate::types::GateCheck::Unknown,
769            since: 12_345,
770        });
771        store.upsert_session(&session).unwrap();
772
773        let fetched = store.get_session("s1").unwrap().unwrap();
774        assert_eq!(fetched.gate_status, session.gate_status);
775
776        let listed = store.list_sessions().unwrap();
777        assert_eq!(listed[0].gate_status, session.gate_status);
778    }
779
780    #[test]
781    fn legacy_row_without_gate_status_column_defaults_to_none() {
782        // A row written before this migration has no gate_status column
783        // value — column_exists-gated ALTER TABLE means the column is added
784        // but existing rows get SQL NULL, which must deserialize to `None`.
785        let store = Store::open(tempdir().unwrap().keep().join("t.db")).unwrap();
786        let session = crate::types::Session {
787            id: "s2".into(), orchestrator_id: None, name: "w".into(),
788            repo: "r".into(), status: crate::types::SessionStatus::Working,
789            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
790            pr_number: None, pr_id: None, workspace_path: None, pid: None,
791            model: None, context_tokens: None, catalogue_path: None,
792            context_used_pct: None, context_total_tokens: None, context_window_size: None,
793            claude_session_id: None, summary: None, terminal_at: None,
794            gate_status: None,
795        };
796        store.upsert_session(&session).unwrap();
797        let fetched = store.get_session("s2").unwrap().unwrap();
798        assert_eq!(fetched.gate_status, None);
799    }
800}
ninox_core/store.rs

ninox_core/
store.rs
