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    pub fn list_orchestrators(&self) -> Result<Vec<Orchestrator>> {
339        let conn = self.conn.lock().unwrap();
340        let mut stmt = conn.prepare(
341            "SELECT id,name,created_at FROM orchestrators ORDER BY created_at DESC",
342        )?;
343        let rows = stmt.query_map([], |r| {
344            Ok(Orchestrator {
345                id: r.get(0)?,
346                name: r.get(1)?,
347                created_at: r.get(2)?,
348            })
349        })?;
350        rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
351    }
352}
353
354#[cfg(test)]
355mod tests {
356    use super::*;
357    use tempfile::tempdir;
358
359    fn test_store() -> Store {
360        let dir = tempdir().unwrap();
361        let path = dir.path().join("t.db");
362        // keep dir alive for the lifetime of the test by leaking it
363        std::mem::forget(dir);
364        Store::open(path).unwrap()
365    }
366
367    #[test]
368    fn upsert_and_list_session() {
369        let store = test_store();
370        let session = Session {
371            id: "s1".into(), orchestrator_id: None, name: "worker-1".into(),
372            repo: "slievr/Athene".into(), status: SessionStatus::Working,
373            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
374            pr_number: None, pr_id: None, workspace_path: None, pid: None,
375            model: None, context_tokens: None, catalogue_path: None,
376            context_used_pct: None, context_total_tokens: None, context_window_size: None,
377            claude_session_id: None,
378            summary: None,
379            terminal_at: None, gate_status: None,
380        };
381        store.upsert_session(&session).unwrap();
382        let list = store.list_sessions().unwrap();
383        assert_eq!(list.len(), 1);
384        assert_eq!(list[0].id, "s1");
385    }
386
387    #[test]
388    fn upsert_updates_status() {
389        let store = test_store();
390        let mut s = Session {
391            id: "s1".into(), orchestrator_id: None, name: "w".into(),
392            repo: "r".into(), status: SessionStatus::Working,
393            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
394            pr_number: None, pr_id: None, workspace_path: None, pid: None,
395            model: None, context_tokens: None, catalogue_path: None,
396            context_used_pct: None, context_total_tokens: None, context_window_size: None,
397            claude_session_id: None,
398            summary: None,
399            terminal_at: None, gate_status: None,
400        };
401        store.upsert_session(&s).unwrap();
402        s.status = SessionStatus::Done;
403        store.upsert_session(&s).unwrap();
404        let list = store.list_sessions().unwrap();
405        assert_eq!(list.len(), 1);
406        assert!(matches!(list[0].status, SessionStatus::Done));
407    }
408
409    /// The poller's dual-remote self-heal (`session.repo` corrected once a
410    /// PR is found against a different remote than the one on record)
411    /// depends on `repo` being an updatable column, not just a write-once
412    /// field set at insert time.
413    #[test]
414    fn upsert_updates_repo() {
415        let store = test_store();
416        let mut s = Session {
417            id: "s1".into(), orchestrator_id: None, name: "w".into(),
418            repo: "OwnerA/repoA".into(), status: SessionStatus::Working,
419            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
420            pr_number: None, pr_id: None, workspace_path: None, pid: None,
421            model: None, context_tokens: None, catalogue_path: None,
422            context_used_pct: None, context_total_tokens: None, context_window_size: None,
423            claude_session_id: None,
424            summary: None,
425            terminal_at: None, gate_status: None,
426        };
427        store.upsert_session(&s).unwrap();
428        s.repo = "OwnerB/repoB".into();
429        store.upsert_session(&s).unwrap();
430        let updated = store.get_session("s1").unwrap().unwrap();
431        assert_eq!(updated.repo, "OwnerB/repoB");
432    }
433
434    #[test]
435    fn get_session_by_id() {
436        let store = test_store();
437        let s = Session {
438            id: "s1".into(), orchestrator_id: None, name: "w".into(),
439            repo: "r".into(), status: SessionStatus::Working,
440            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
441            pr_number: None, pr_id: None, workspace_path: None, pid: None,
442            model: None, context_tokens: None, catalogue_path: None,
443            context_used_pct: None, context_total_tokens: None, context_window_size: None,
444            claude_session_id: None,
445            summary: None,
446            terminal_at: None, gate_status: None,
447        };
448        store.upsert_session(&s).unwrap();
449        let found = store.get_session("s1").unwrap();
450        assert!(found.is_some());
451        assert_eq!(found.unwrap().name, "w");
452        assert!(store.get_session("missing").unwrap().is_none());
453    }
454
455    #[test]
456    fn model_and_context_tokens_round_trip() {
457        let store = test_store();
458        let s = Session {
459            id: "s1".into(), orchestrator_id: None, name: "w".into(),
460            repo: "r".into(), status: SessionStatus::Working,
461            agent_type: "claude-code".into(), cost_usd: 1.5, started_at: 0,
462            pr_number: None, pr_id: None, workspace_path: None, pid: None,
463            model: Some("claude-fable-5".into()), context_tokens: Some(214_000), catalogue_path: None,
464            context_used_pct: None, context_total_tokens: None, context_window_size: None,
465            claude_session_id: None,
466            summary: None,
467            terminal_at: None, gate_status: None,
468        };
469        store.upsert_session(&s).unwrap();
470        let found = store.get_session("s1").unwrap().unwrap();
471        assert_eq!(found.model.as_deref(), Some("claude-fable-5"));
472        assert_eq!(found.context_tokens, Some(214_000));
473    }
474
475    #[test]
476    fn catalogue_path_round_trips() {
477        let store = test_store();
478        let s = Session {
479            id: "s2".into(), orchestrator_id: None, name: "w".into(),
480            repo: "r".into(), status: SessionStatus::Working,
481            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
482            pr_number: None, pr_id: None, workspace_path: None, pid: None,
483            model: None, context_tokens: None,
484            catalogue_path: Some("/brains/x".into()),
485            context_used_pct: None, context_total_tokens: None, context_window_size: None,
486            claude_session_id: None,
487            summary: None,
488            terminal_at: None, gate_status: None,
489        };
490        store.upsert_session(&s).unwrap();
491        let found = store.get_session("s2").unwrap().unwrap();
492        assert_eq!(found.catalogue_path.as_deref(), Some("/brains/x"));
493        // list path decodes it too
494        assert_eq!(store.list_sessions().unwrap()[0].catalogue_path.as_deref(), Some("/brains/x"));
495    }
496
497    #[test]
498    fn summary_round_trips() {
499        let store = test_store();
500        let s = Session {
501            id: "s2b".into(), orchestrator_id: None, name: "w".into(),
502            repo: "r".into(), status: SessionStatus::Working,
503            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
504            pr_number: None, pr_id: None, workspace_path: None, pid: None,
505            model: None, context_tokens: None, catalogue_path: None,
506            context_used_pct: None, context_total_tokens: None, context_window_size: None,
507            claude_session_id: None,
508            summary: Some("Fix flaky CI on the auth suite".into()),
509            terminal_at: None, gate_status: None,
510        };
511        store.upsert_session(&s).unwrap();
512        let found = store.get_session("s2b").unwrap().unwrap();
513        assert_eq!(found.summary.as_deref(), Some("Fix flaky CI on the auth suite"));
514        // list path decodes it too
515        assert_eq!(
516            store.list_sessions().unwrap().iter().find(|x| x.id == "s2b").unwrap().summary.as_deref(),
517            Some("Fix flaky CI on the auth suite"),
518        );
519
520        // None round-trips as None, not "" or "none"
521        let mut s2 = s.clone();
522        s2.id = "s2c".into();
523        s2.summary = None;
524        store.upsert_session(&s2).unwrap();
525        assert_eq!(store.get_session("s2c").unwrap().unwrap().summary, None);
526    }
527
528    #[test]
529    fn claude_session_id_round_trips() {
530        let store = test_store();
531        let s = Session {
532            id: "s3".into(), orchestrator_id: None, name: "w".into(),
533            repo: "r".into(), status: SessionStatus::Working,
534            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
535            pr_number: None, pr_id: None, workspace_path: None, pid: None,
536            model: None, context_tokens: None, catalogue_path: None,
537            context_used_pct: None, context_total_tokens: None, context_window_size: None,
538            claude_session_id: Some("b7e0b3a0-0000-4000-8000-000000000001".into()),
539            summary: None,
540            terminal_at: None, gate_status: None,
541        };
542        store.upsert_session(&s).unwrap();
543        let found = store.get_session("s3").unwrap().unwrap();
544        assert_eq!(found.claude_session_id.as_deref(), Some("b7e0b3a0-0000-4000-8000-000000000001"));
545        // list path decodes it too
546        assert_eq!(
547            store.list_sessions().unwrap().iter().find(|x| x.id == "s3").unwrap().claude_session_id.as_deref(),
548            Some("b7e0b3a0-0000-4000-8000-000000000001"),
549        );
550
551        // None round-trips as None, not "" or "none"
552        let mut s2 = s.clone();
553        s2.id = "s4".into();
554        s2.claude_session_id = None;
555        store.upsert_session(&s2).unwrap();
556        assert_eq!(store.get_session("s4").unwrap().unwrap().claude_session_id, None);
557    }
558
559    #[test]
560    fn terminal_at_round_trips() {
561        let store = test_store();
562        let s = Session {
563            id: "s5".into(), orchestrator_id: None, name: "w".into(),
564            repo: "r".into(), status: SessionStatus::Done,
565            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
566            pr_number: None, pr_id: None, workspace_path: None, pid: None,
567            model: None, context_tokens: None, catalogue_path: None,
568            context_used_pct: None, context_total_tokens: None, context_window_size: None,
569            claude_session_id: None,
570            summary: None,
571            terminal_at: Some(1_720_000_000_000), gate_status: None,
572        };
573        store.upsert_session(&s).unwrap();
574        let found = store.get_session("s5").unwrap().unwrap();
575        assert_eq!(found.terminal_at, Some(1_720_000_000_000));
576        // list path decodes it too
577        assert_eq!(
578            store.list_sessions().unwrap().iter().find(|x| x.id == "s5").unwrap().terminal_at,
579            Some(1_720_000_000_000),
580        );
581
582        // None round-trips as None — non-terminal / not-yet-retired sessions.
583        let mut s2 = s.clone();
584        s2.id = "s6".into();
585        s2.terminal_at = None;
586        store.upsert_session(&s2).unwrap();
587        assert_eq!(store.get_session("s6").unwrap().unwrap().terminal_at, None);
588    }
589
590    /// A Re-file respawns the same session id with a fresh `started_at` —
591    /// the conflict-update path must persist it, or the in-memory time
592    /// silently reverts to the original spawn time on app restart.
593    #[test]
594    fn upsert_conflict_updates_started_at() {
595        let store = test_store();
596        let mut s = Session {
597            id: "s3".into(), orchestrator_id: None, name: "w".into(),
598            repo: "r".into(), status: SessionStatus::Working,
599            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 100,
600            pr_number: None, pr_id: None, workspace_path: None, pid: None,
601            model: None, context_tokens: None, catalogue_path: None,
602            context_used_pct: None, context_total_tokens: None, context_window_size: None,
603            claude_session_id: None,
604            summary: None,
605            terminal_at: None, gate_status: None,
606        };
607        store.upsert_session(&s).unwrap();
608        s.started_at = 200;
609        store.upsert_session(&s).unwrap();
610        assert_eq!(store.get_session("s3").unwrap().unwrap().started_at, 200);
611    }
612
613    #[test]
614    fn get_pr_round_trips_and_misses_cleanly() {
615        let store = test_store();
616        assert!(store.get_pr(9).unwrap().is_none());
617        let pr = PR {
618            id: 9, number: 9, title: "t".into(),
619            url: "https://github.com/org/repo/pull/9".into(),
620            body: String::new(), session_id: "s1".into(),
621        };
622        store.upsert_pr(&pr).unwrap();
623        let found = store.get_pr(9).unwrap().unwrap();
624        assert_eq!(found.number, 9);
625        assert_eq!(found.session_id, "s1");
626        assert_eq!(found.url, pr.url);
627    }
628
629    #[test]
630    fn context_fields_round_trip() {
631        let store = test_store();
632        let s = Session {
633            id: "s1".into(), orchestrator_id: None, name: "w".into(),
634            repo: "r".into(), status: SessionStatus::Working,
635            agent_type: "claude-code".into(), cost_usd: 2.6, started_at: 0,
636            pr_number: None, pr_id: None, workspace_path: None, pid: None,
637            model: None, context_tokens: None, catalogue_path: None,
638            context_used_pct: Some(62.0),
639            context_total_tokens: Some(124_000),
640            context_window_size: Some(200_000),
641            claude_session_id: None,
642            summary: None,
643            terminal_at: None, gate_status: None,
644        };
645        store.upsert_session(&s).unwrap();
646        let found = store.get_session("s1").unwrap().unwrap();
647        assert_eq!(found.context_used_pct, Some(62.0));
648        assert_eq!(found.context_total_tokens, Some(124_000));
649        assert_eq!(found.context_window_size, Some(200_000));
650        // list path decodes it too
651        assert_eq!(store.list_sessions().unwrap()[0].context_used_pct, Some(62.0));
652    }
653
654    #[test]
655    fn context_fields_default_to_none() {
656        let store = test_store();
657        let s = Session {
658            id: "s2".into(), orchestrator_id: None, name: "w".into(),
659            repo: "r".into(), status: SessionStatus::Working,
660            agent_type: "claude-code".into(), cost_usd: 0.0, started_at: 0,
661            pr_number: None, pr_id: None, workspace_path: None, pid: None,
662            model: None, context_tokens: None, catalogue_path: None,
663            context_used_pct: None, context_total_tokens: None, context_window_size: None,
664            claude_session_id: None,
665            summary: None,
666            terminal_at: None, gate_status: None,
667        };
668        store.upsert_session(&s).unwrap();
669        let found = store.get_session("s2").unwrap().unwrap();
670        assert_eq!(found.context_used_pct, None);
671        assert_eq!(found.context_total_tokens, None);
672        assert_eq!(found.context_window_size, None);
673    }
674
675    #[test]
676    fn cost_samples_filters_by_agent_and_model_and_excludes_zero() {
677        let store = test_store();
678        for (id, agent_type, model, cost) in [
679            ("a", "claude-code", Some("claude-fable-5"), 3.0),
680            ("b", "claude-code", Some("claude-fable-5"), 5.0),
681            ("c", "claude-code", Some("claude-fable-5"), 0.0), // excluded: zero cost
682            ("d", "claude-code", Some("claude-opus-4-8"), 2.0), // excluded: different model
683            ("e", "codex",       Some("claude-fable-5"), 4.0),  // excluded: different harness
684        ] {
685            store.upsert_session(&Session {
686                id: id.into(), orchestrator_id: None, name: id.into(),
687                repo: "r".into(), status: SessionStatus::Working,
688                agent_type: agent_type.into(), cost_usd: cost, started_at: 0,
689                pr_number: None, pr_id: None, workspace_path: None, pid: None,
690                model: model.map(String::from), context_tokens: None, catalogue_path: None,
691                context_used_pct: None, context_total_tokens: None, context_window_size: None,
692                claude_session_id: None, summary: None,
693                terminal_at: None, gate_status: None,
694            }).unwrap();
695        }
696        let samples = store.cost_samples("claude-code", Some("claude-fable-5")).unwrap();
697        assert_eq!(samples.len(), 2);
698        assert!((samples.iter().sum::<f64>() - 8.0).abs() < f64::EPSILON);
699    }
700
701    #[test]
702    fn upsert_and_fetch_session_round_trips_gate_status() {
703        let store = Store::open(tempdir().unwrap().keep().join("t.db")).unwrap();
704        let mut session = crate::types::Session {
705            id: "s1".into(), orchestrator_id: None, name: "w".into(),
706            repo: "r".into(), status: crate::types::SessionStatus::PrOpen,
707            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
708            pr_number: Some(1), pr_id: Some(1), workspace_path: None, pid: None,
709            model: None, context_tokens: None, catalogue_path: None,
710            context_used_pct: None, context_total_tokens: None, context_window_size: None,
711            claude_session_id: None, summary: None, terminal_at: None,
712            gate_status: None,
713        };
714        session.gate_status = Some(crate::types::GateStatus {
715            ci: crate::types::GateCheck::Failing,
716            review: crate::types::GateCheck::Passing,
717            mergeable: crate::types::GateCheck::Unknown,
718            since: 12_345,
719        });
720        store.upsert_session(&session).unwrap();
721
722        let fetched = store.get_session("s1").unwrap().unwrap();
723        assert_eq!(fetched.gate_status, session.gate_status);
724
725        let listed = store.list_sessions().unwrap();
726        assert_eq!(listed[0].gate_status, session.gate_status);
727    }
728
729    #[test]
730    fn legacy_row_without_gate_status_column_defaults_to_none() {
731        // A row written before this migration has no gate_status column
732        // value — column_exists-gated ALTER TABLE means the column is added
733        // but existing rows get SQL NULL, which must deserialize to `None`.
734        let store = Store::open(tempdir().unwrap().keep().join("t.db")).unwrap();
735        let session = crate::types::Session {
736            id: "s2".into(), orchestrator_id: None, name: "w".into(),
737            repo: "r".into(), status: crate::types::SessionStatus::Working,
738            agent_type: "c".into(), cost_usd: 0.0, started_at: 0,
739            pr_number: None, pr_id: None, workspace_path: None, pid: None,
740            model: None, context_tokens: None, catalogue_path: None,
741            context_used_pct: None, context_total_tokens: None, context_window_size: None,
742            claude_session_id: None, summary: None, terminal_at: None,
743            gate_status: None,
744        };
745        store.upsert_session(&session).unwrap();
746        let fetched = store.get_session("s2").unwrap().unwrap();
747        assert_eq!(fetched.gate_status, None);
748    }
749}
ninox_core/store.rs

ninox_core/
store.rs
