1use anyhow::{Context, Result};
7use rusqlite::{Connection, OptionalExtension, params};
8
9use super::*;
10
11pub struct TasksRepo<'a> {
23 pub(crate) conn: &'a Connection,
24}
25
26pub struct SessionsRepo<'a> {
27 pub(crate) conn: &'a Connection,
28}
29
30impl SessionsRepo<'_> {
31 pub fn upsert(&self, new: NewSession) -> Result<SessionRecord> {
36 let now = now_rfc3339();
37 let id = new.id.unwrap_or_else(|| fresh_id("session"));
38 self.conn.execute(
39 "INSERT INTO sessions
40 (id, project_path, model_id, title, conversation_path, created_at, updated_at, total_tokens)
41 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
42 ON CONFLICT(id) DO UPDATE SET
43 project_path = excluded.project_path,
44 model_id = excluded.model_id,
45 title = excluded.title,
46 conversation_path = excluded.conversation_path,
47 updated_at = excluded.updated_at,
48 total_tokens = excluded.total_tokens",
49 params![
50 id,
51 new.project_path,
52 new.model_id,
53 new.title,
54 new.conversation_path,
55 now,
56 now,
57 new.total_tokens,
58 ],
59 )?;
60 self.get(&id)?
61 .context("session was upserted but could not be reloaded")
62 }
63
64 pub fn get(&self, id: &str) -> Result<Option<SessionRecord>> {
69 self.conn
70 .query_row(
71 "SELECT id, project_path, model_id, title, conversation_path,
72 created_at, updated_at, total_tokens
73 FROM sessions WHERE id = ?1",
74 [id],
75 session_from_row,
76 )
77 .optional()
78 .map_err(Into::into)
79 }
80
81 pub fn list(&self, limit: usize) -> Result<Vec<SessionRecord>> {
86 let mut stmt = self.conn.prepare(
87 "SELECT id, project_path, model_id, title, conversation_path,
88 created_at, updated_at, total_tokens
89 FROM sessions ORDER BY updated_at DESC LIMIT ?1",
90 )?;
91 let rows = stmt.query_map([clamp_limit(limit)], session_from_row)?;
92 rows.collect::<rusqlite::Result<Vec<_>>>()
93 .map_err(Into::into)
94 }
95}
96
97pub struct MessagesRepo<'a> {
98 pub(crate) conn: &'a Connection,
99}
100
101impl MessagesRepo<'_> {
102 pub fn add(&self, new: NewMessage) -> Result<MessageRecord> {
107 self.conn.execute(
108 "INSERT INTO messages (session_id, role, content_json, created_at)
109 VALUES (?1, ?2, ?3, ?4)",
110 params![new.session_id, new.role, new.content_json, now_rfc3339()],
111 )?;
112 let id = self.conn.last_insert_rowid();
113 self.get(id)?
114 .context("message was inserted but could not be reloaded")
115 }
116
117 pub fn get(&self, id: i64) -> Result<Option<MessageRecord>> {
122 self.conn
123 .query_row(
124 "SELECT id, session_id, role, content_json, created_at
125 FROM messages WHERE id = ?1",
126 [id],
127 message_from_row,
128 )
129 .optional()
130 .map_err(Into::into)
131 }
132
133 pub fn list_for_session(&self, session_id: &str) -> Result<Vec<MessageRecord>> {
147 let mut stmt = self.conn.prepare(
148 "SELECT id, session_id, role, content_json, created_at FROM (
149 SELECT id, session_id, role, content_json, created_at
150 FROM messages WHERE session_id = ?1
151 ORDER BY id DESC LIMIT ?2
152 ) ORDER BY id ASC",
153 )?;
154 let rows = stmt.query_map(params![session_id, MAX_SESSION_MESSAGES], message_from_row)?;
155 rows.collect::<rusqlite::Result<Vec<_>>>()
156 .map_err(Into::into)
157 }
158}
159
160impl TasksRepo<'_> {
161 pub fn create(&self, new: NewTask) -> Result<TaskRecord> {
166 let now = now_rfc3339();
167 let owner_kind = new.owner_kind;
170 let record = TaskRecord {
171 id: fresh_id("task"),
172 title: new.title,
173 status: TaskStatus::Queued,
174 priority: new.priority,
175 project_path: new.project_path,
176 model_id: new.model_id,
177 conversation_id: new.conversation_id,
178 created_at: now.clone(),
179 updated_at: now.clone(),
180 final_report: None,
181 prompt: new.prompt,
182 };
183 let tx = self.conn.unchecked_transaction()?;
186 tx.execute(
187 "INSERT INTO tasks
188 (id, title, status, priority, project_path, model_id, conversation_id, created_at, updated_at, final_report, owner_kind, prompt)
189 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
190 params![
191 record.id,
192 record.title,
193 record.status.as_str(),
194 record.priority.as_str(),
195 record.project_path,
196 record.model_id,
197 record.conversation_id,
198 record.created_at,
199 record.updated_at,
200 record.final_report,
201 owner_kind,
202 record.prompt,
203 ],
204 )?;
205 tx.execute(
206 "INSERT INTO task_events (task_id, kind, message, created_at)
207 VALUES (?1, ?2, ?3, ?4)",
208 params![record.id, "task_created", "task created", now],
209 )?;
210 tx.commit()?;
211 self.get(&record.id)?
212 .context("task was inserted but could not be reloaded")
213 }
214
215 pub fn get(&self, id: &str) -> Result<Option<TaskRecord>> {
220 self.conn
221 .query_row(
222 "SELECT id, title, status, priority, project_path, model_id, conversation_id,
223 created_at, updated_at, final_report, prompt
224 FROM tasks WHERE id = ?1",
225 [id],
226 task_from_row,
227 )
228 .optional()
229 .map_err(Into::into)
230 }
231
232 pub fn list(&self, limit: usize) -> Result<Vec<TaskRecord>> {
237 let mut stmt = self.conn.prepare(
238 "SELECT id, title, status, priority, project_path, model_id, conversation_id,
239 created_at, updated_at, final_report, prompt
240 FROM tasks
241 ORDER BY updated_at DESC
242 LIMIT ?1",
243 )?;
244 let rows = stmt.query_map([clamp_limit(limit)], task_from_row_opt)?;
248 collect_tolerant(rows)
249 }
250
251 pub fn update_status(
256 &self,
257 id: &str,
258 status: TaskStatus,
259 final_report: Option<&str>,
260 ) -> Result<()> {
261 let now = now_rfc3339();
262 let tx = self.conn.unchecked_transaction()?;
264 tx.execute(
265 "UPDATE tasks
266 SET status = ?2, updated_at = ?3, final_report = COALESCE(?4, final_report)
267 WHERE id = ?1",
268 params![id, status.as_str(), now, final_report],
269 )?;
270 tx.execute(
271 "INSERT INTO task_events (task_id, kind, message, created_at)
272 VALUES (?1, ?2, ?3, ?4)",
273 params![
274 id,
275 "status_changed",
276 format!("status changed to {status}"),
277 now
278 ],
279 )?;
280 tx.commit()?;
281 Ok(())
282 }
283
284 pub fn claim_next_queued(&self) -> Result<Option<TaskRecord>> {
301 let tx = self.conn.unchecked_transaction()?;
302 let claimed = tx
303 .query_row(
304 "UPDATE tasks SET status = 'running', updated_at = ?1
305 WHERE id = (
306 SELECT id FROM tasks
307 WHERE status = 'queued' AND owner_kind = ?2 AND prompt IS NOT NULL
308 ORDER BY CASE priority
309 WHEN 'high' THEN 0
310 WHEN 'normal' THEN 1
311 WHEN 'low' THEN 2
312 ELSE 1
313 END,
314 created_at ASC, id ASC
315 LIMIT 1
316 )
317 RETURNING id, title, status, priority, project_path, model_id,
318 conversation_id, created_at, updated_at, final_report, prompt",
319 params![now_rfc3339(), OWNER_KIND_DAEMON],
320 task_from_row,
321 )
322 .optional()?;
323 if let Some(task) = &claimed {
324 tx.execute(
325 "INSERT INTO task_events (task_id, kind, message, created_at)
326 VALUES (?1, ?2, ?3, ?4)",
327 params![
328 task.id,
329 "status_changed",
330 "status changed to running (claimed by scheduler)",
331 now_rfc3339(),
332 ],
333 )?;
334 }
335 tx.commit()?;
336 Ok(claimed)
337 }
338
339 pub fn add_event(&self, task_id: &str, kind: &str, message: &str) -> Result<()> {
343 self.conn.execute(
344 "INSERT INTO task_events (task_id, kind, message, created_at)
345 VALUES (?1, ?2, ?3, ?4)",
346 params![task_id, kind, message, now_rfc3339()],
347 )?;
348 Ok(())
349 }
350
351 pub fn events(&self, task_id: &str) -> Result<Vec<TaskTimelineEvent>> {
356 let mut stmt = self.conn.prepare(
357 "SELECT id, task_id, kind, message, created_at
358 FROM task_events
359 WHERE task_id = ?1
360 ORDER BY id ASC",
361 )?;
362 let rows = stmt.query_map([task_id], task_event_from_row_opt)?;
364 collect_tolerant(rows)
365 }
366}
367
368pub struct ToolRunsRepo<'a> {
369 pub(crate) conn: &'a Connection,
370}
371
372impl ToolRunsRepo<'_> {
373 pub fn start(&self, mut new: NewToolRun) -> Result<ToolRunRecord> {
378 new.args_json = new
382 .args_json
383 .as_deref()
384 .map(crate::redact::redact_json_text);
385 let id = new.id.unwrap_or_else(|| fresh_id("toolrun"));
386 self.conn.execute(
387 "INSERT INTO tool_runs
388 (id, task_id, turn_id, call_id, tool_name, status, args_json, output_json, started_at, finished_at)
389 VALUES (?1, ?2, ?3, ?4, ?5, 'running', ?6, NULL, ?7, NULL)",
390 params![
391 id,
392 new.task_id,
393 new.turn_id,
394 new.call_id,
395 new.tool_name,
396 new.args_json,
397 now_rfc3339(),
398 ],
399 )?;
400 self.get(&id)?
401 .context("tool run was inserted but could not be reloaded")
402 }
403
404 pub fn finish(&self, id: &str, status: &str, output_json: Option<&str>) -> Result<()> {
408 let output_json = output_json.map(crate::redact::redact_json_text);
409 let changed = self.conn.execute(
410 "UPDATE tool_runs
411 SET status = ?2, output_json = ?3, finished_at = ?4
412 WHERE id = ?1",
413 params![id, status, output_json, now_rfc3339()],
414 )?;
415 anyhow::ensure!(changed > 0, "tool run not found: {id}");
416 Ok(())
417 }
418
419 pub fn get(&self, id: &str) -> Result<Option<ToolRunRecord>> {
424 self.conn
425 .query_row(
426 "SELECT id, task_id, turn_id, call_id, tool_name, status, args_json,
427 output_json, started_at, finished_at
428 FROM tool_runs WHERE id = ?1",
429 [id],
430 tool_run_from_row,
431 )
432 .optional()
433 .map_err(Into::into)
434 }
435
436 pub fn list(&self, limit: usize) -> Result<Vec<ToolRunRecord>> {
441 let mut stmt = self.conn.prepare(
442 "SELECT id, task_id, turn_id, call_id, tool_name, status, args_json,
443 output_json, started_at, finished_at
444 FROM tool_runs ORDER BY started_at DESC LIMIT ?1",
445 )?;
446 let rows = stmt.query_map([clamp_limit(limit)], tool_run_from_row)?;
447 rows.collect::<rusqlite::Result<Vec<_>>>()
448 .map_err(Into::into)
449 }
450}
451
452pub struct OutcomesRepo<'a> {
453 pub(crate) conn: &'a Connection,
454}
455
456impl OutcomesRepo<'_> {
457 pub fn record(&self, new: NewOutcome) -> Result<OutcomeRecord> {
465 let id = new.id.unwrap_or_else(|| fresh_id("outcome"));
466 self.conn.execute(
467 "INSERT INTO outcomes
468 (id, task_id, tool_run_id, kind, label, reward, source, detail_json, created_at)
469 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
470 params![
471 id,
472 new.task_id,
473 new.tool_run_id,
474 new.kind,
475 new.label,
476 new.reward,
477 new.source,
478 new.detail_json,
479 now_rfc3339(),
480 ],
481 )?;
482 self.get(&id)?
483 .context("outcome was inserted but could not be reloaded")
484 }
485
486 pub fn get(&self, id: &str) -> Result<Option<OutcomeRecord>> {
491 self.conn
492 .query_row(
493 "SELECT id, task_id, tool_run_id, kind, label, reward, source,
494 detail_json, created_at
495 FROM outcomes WHERE id = ?1",
496 [id],
497 outcome_from_row,
498 )
499 .optional()
500 .map_err(Into::into)
501 }
502
503 pub fn list_for_task(&self, task_id: &str) -> Result<Vec<OutcomeRecord>> {
511 let mut stmt = self.conn.prepare(
512 "SELECT id, task_id, tool_run_id, kind, label, reward, source,
513 detail_json, created_at
514 FROM outcomes WHERE task_id = ?1 ORDER BY created_at ASC",
515 )?;
516 let rows = stmt.query_map([task_id], outcome_from_row)?;
517 rows.collect::<rusqlite::Result<Vec<_>>>()
518 .map_err(Into::into)
519 }
520
521 pub fn list(&self, limit: usize) -> Result<Vec<OutcomeRecord>> {
526 let mut stmt = self.conn.prepare(
527 "SELECT id, task_id, tool_run_id, kind, label, reward, source,
528 detail_json, created_at
529 FROM outcomes ORDER BY created_at DESC LIMIT ?1",
530 )?;
531 let rows = stmt.query_map([clamp_limit(limit)], outcome_from_row)?;
532 rows.collect::<rusqlite::Result<Vec<_>>>()
533 .map_err(Into::into)
534 }
535}
536
537pub(crate) fn outcome_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<OutcomeRecord> {
538 Ok(OutcomeRecord {
539 id: row.get(0)?,
540 task_id: row.get(1)?,
541 tool_run_id: row.get(2)?,
542 kind: row.get(3)?,
543 label: row.get(4)?,
544 reward: row.get(5)?,
545 source: row.get(6)?,
546 detail_json: row.get(7)?,
547 created_at: row.get(8)?,
548 })
549}
550
551pub struct ApprovalsRepo<'a> {
552 pub(crate) conn: &'a Connection,
553}
554
555impl ApprovalsRepo<'_> {
556 pub fn create(&self, new: NewApproval) -> Result<ApprovalRecord> {
561 let record = ApprovalRecord {
562 id: fresh_id("approval"),
563 task_id: new.task_id,
564 proposed_action: new.proposed_action,
565 risk_classification: new.risk_classification,
566 policy_decision: new.policy_decision,
567 user_decision: None,
568 args_summary: new.args_summary,
569 checkpoint_id: new.checkpoint_id,
570 pending_action_json: new.pending_action_json,
571 created_at: now_rfc3339(),
572 decided_at: None,
573 archived_at: None,
574 archive_reason: None,
575 };
576 self.conn.execute(
577 "INSERT INTO approvals
578 (id, task_id, proposed_action, risk_classification, policy_decision, user_decision,
579 args_summary, checkpoint_id, pending_action_json, created_at, decided_at)
580 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
581 params![
582 record.id,
583 record.task_id,
584 record.proposed_action,
585 record.risk_classification,
586 record.policy_decision,
587 record.user_decision,
588 record.args_summary,
589 record.checkpoint_id,
590 record.pending_action_json,
591 record.created_at,
592 record.decided_at,
593 ],
594 )?;
595 self.get(&record.id)?
596 .context("approval was inserted but could not be reloaded")
597 }
598
599 pub fn get(&self, id: &str) -> Result<Option<ApprovalRecord>> {
604 self.conn
605 .query_row(
606 "SELECT id, task_id, proposed_action, risk_classification, policy_decision,
607 user_decision, args_summary, checkpoint_id, pending_action_json,
608 created_at, decided_at, archived_at, archive_reason
609 FROM approvals WHERE id = ?1",
610 [id],
611 approval_from_row,
612 )
613 .optional()
614 .map_err(Into::into)
615 }
616
617 pub fn decide(&self, id: &str, user_decision: &str) -> Result<()> {
621 let changed = self.conn.execute(
629 "UPDATE approvals
630 SET user_decision = ?2, decided_at = ?3
631 WHERE id = ?1 AND user_decision IS NULL AND archived_at IS NULL",
632 params![id, user_decision, now_rfc3339()],
633 )?;
634 anyhow::ensure!(
635 changed > 0,
636 "approval {id} cannot be decided (already decided, archived, or not found)"
637 );
638 Ok(())
639 }
640
641 pub fn claim(&self, id: &str) -> Result<bool> {
654 let changed = self.conn.execute(
655 "UPDATE approvals
656 SET user_decision = 'approving'
657 WHERE id = ?1 AND user_decision IS NULL AND archived_at IS NULL",
658 params![id],
659 )?;
660 Ok(changed == 1)
661 }
662
663 pub fn release_claim(&self, id: &str) -> Result<()> {
670 self.conn.execute(
671 "UPDATE approvals SET user_decision = NULL
672 WHERE id = ?1 AND user_decision = 'approving'",
673 params![id],
674 )?;
675 Ok(())
676 }
677
678 pub fn finalize_claimed(&self, id: &str, user_decision: &str) -> Result<()> {
685 let changed = self.conn.execute(
686 "UPDATE approvals
687 SET user_decision = ?2, decided_at = ?3
688 WHERE id = ?1 AND user_decision = 'approving'",
689 params![id, user_decision, now_rfc3339()],
690 )?;
691 anyhow::ensure!(changed > 0, "approval {id} was not in the claimed state");
692 Ok(())
693 }
694
695 pub fn list_pending(&self) -> Result<Vec<ApprovalRecord>> {
699 self.list_pending_with_archived(false)
700 }
701
702 pub fn list_pending_all(&self) -> Result<Vec<ApprovalRecord>> {
706 self.list_pending_with_archived(true)
707 }
708
709 pub fn list_all(&self, limit: usize) -> Result<Vec<ApprovalRecord>> {
714 let mut stmt = self.conn.prepare(
715 "SELECT id, task_id, proposed_action, risk_classification, policy_decision,
716 user_decision, args_summary, checkpoint_id, pending_action_json,
717 created_at, decided_at, archived_at, archive_reason
718 FROM approvals
719 ORDER BY created_at DESC
720 LIMIT ?1",
721 )?;
722 let rows = stmt.query_map([clamp_limit(limit)], approval_from_row)?;
723 rows.collect::<rusqlite::Result<Vec<_>>>()
724 .map_err(Into::into)
725 }
726
727 pub(crate) fn list_pending_with_archived(
728 &self,
729 include_archived: bool,
730 ) -> Result<Vec<ApprovalRecord>> {
731 let archived_filter = if include_archived {
732 ""
733 } else {
734 " AND archived_at IS NULL"
735 };
736 let mut stmt = self.conn.prepare(&format!(
737 "SELECT id, task_id, proposed_action, risk_classification, policy_decision,
738 user_decision, args_summary, checkpoint_id, pending_action_json,
739 created_at, decided_at, archived_at, archive_reason
740 FROM approvals
741 WHERE user_decision IS NULL{archived_filter}
742 ORDER BY created_at DESC"
743 ))?;
744 let rows = stmt.query_map([], approval_from_row)?;
745 rows.collect::<rusqlite::Result<Vec<_>>>()
746 .map_err(Into::into)
747 }
748
749 pub fn archive(&self, ids: &[String], reason: &str) -> Result<usize> {
756 let archived_at = now_rfc3339();
757 let mut changed = 0;
758 for id in ids {
759 changed += self.conn.execute(
760 "UPDATE approvals
761 SET archived_at = COALESCE(archived_at, ?2),
762 archive_reason = COALESCE(archive_reason, ?3)
763 WHERE id = ?1 AND archived_at IS NULL",
764 params![id, archived_at, reason],
765 )?;
766 }
767 Ok(changed)
768 }
769
770 pub fn count_archived(&self) -> Result<usize> {
774 self.conn
775 .query_row(
776 "SELECT COUNT(*) FROM approvals WHERE archived_at IS NOT NULL",
777 [],
778 |row| row.get::<_, i64>(0),
779 )
780 .map(|count| count as usize)
781 .map_err(Into::into)
782 }
783}
784
785pub struct ProcessesRepo<'a> {
786 pub(crate) conn: &'a Connection,
787}
788
789impl ProcessesRepo<'_> {
790 pub fn upsert(&self, new: NewProcess) -> Result<ProcessRecord> {
795 let now = now_rfc3339();
796 let id = new.id.unwrap_or_else(|| fresh_id("process"));
797 self.conn.execute(
798 "INSERT INTO processes
799 (id, task_id, pid, command, cwd, log_path, detected_url, status, health, created_at, updated_at)
800 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
801 ON CONFLICT(id) DO UPDATE SET
802 task_id = excluded.task_id,
803 pid = excluded.pid,
804 command = excluded.command,
805 cwd = excluded.cwd,
806 log_path = excluded.log_path,
807 detected_url = excluded.detected_url,
808 status = excluded.status,
809 health = excluded.health,
810 updated_at = excluded.updated_at",
811 params![
812 id,
813 new.task_id,
814 new.pid,
815 new.command,
816 new.cwd,
817 new.log_path,
818 new.detected_url,
819 new.status.as_str(),
820 new.health,
821 now,
822 now,
823 ],
824 )?;
825 self.get(&id)?
826 .context("process was upserted but could not be reloaded")
827 }
828
829 pub fn get(&self, id: &str) -> Result<Option<ProcessRecord>> {
834 self.conn
835 .query_row(
836 "SELECT id, task_id, pid, command, cwd, log_path, detected_url, status, health,
837 created_at, updated_at
838 FROM processes WHERE id = ?1",
839 [id],
840 process_from_row,
841 )
842 .optional()
843 .map_err(Into::into)
844 }
845
846 pub fn list(&self, limit: usize) -> Result<Vec<ProcessRecord>> {
851 let mut stmt = self.conn.prepare(
852 "SELECT id, task_id, pid, command, cwd, log_path, detected_url, status, health,
853 created_at, updated_at
854 FROM processes
855 ORDER BY updated_at DESC
856 LIMIT ?1",
857 )?;
858 let rows = stmt.query_map([clamp_limit(limit)], process_from_row_opt)?;
861 collect_tolerant(rows)
862 }
863}
864
865pub struct CheckpointsRepo<'a> {
866 pub(crate) conn: &'a Connection,
867}
868
869impl CheckpointsRepo<'_> {
870 pub fn create(&self, new: NewCheckpoint) -> Result<CheckpointRecord> {
875 let id = new.id.unwrap_or_else(|| fresh_id("checkpoint"));
876 self.conn.execute(
877 "INSERT INTO checkpoints
878 (id, task_id, project_path, snapshot_path, changed_files_json,
879 pending_action_json, approval_id, created_at, session_id, message_index)
880 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
881 params![
882 id,
883 new.task_id,
884 new.project_path,
885 new.snapshot_path,
886 new.changed_files_json,
887 new.pending_action_json,
888 new.approval_id,
889 now_rfc3339(),
890 new.session_id,
891 new.message_index,
892 ],
893 )?;
894 self.get(&id)?
895 .context("checkpoint was inserted but could not be reloaded")
896 }
897
898 pub fn get(&self, id: &str) -> Result<Option<CheckpointRecord>> {
903 self.conn
904 .query_row(
905 "SELECT id, task_id, project_path, snapshot_path, changed_files_json,
906 pending_action_json, approval_id, created_at, archived_at, archive_reason,
907 session_id, message_index
908 FROM checkpoints WHERE id = ?1",
909 [id],
910 checkpoint_from_row,
911 )
912 .optional()
913 .map_err(Into::into)
914 }
915
916 pub fn set_approval(&self, id: &str, approval_id: &str) -> Result<()> {
920 let changed = self.conn.execute(
921 "UPDATE checkpoints SET approval_id = ?2 WHERE id = ?1",
922 params![id, approval_id],
923 )?;
924 anyhow::ensure!(changed > 0, "checkpoint not found: {id}");
925 Ok(())
926 }
927
928 pub fn delete(&self, id: &str) -> Result<bool> {
943 let changed = self
944 .conn
945 .execute("DELETE FROM checkpoints WHERE id = ?1", params![id])?;
946 Ok(changed > 0)
947 }
948
949 pub fn list(&self, limit: usize) -> Result<Vec<CheckpointRecord>> {
953 self.list_with_archived(limit, false)
954 }
955
956 pub fn list_all(&self, limit: usize) -> Result<Vec<CheckpointRecord>> {
960 self.list_with_archived(limit, true)
961 }
962
963 pub(crate) fn list_with_archived(
964 &self,
965 limit: usize,
966 include_archived: bool,
967 ) -> Result<Vec<CheckpointRecord>> {
968 let archived_filter = if include_archived {
969 ""
970 } else {
971 "WHERE archived_at IS NULL"
972 };
973 let mut stmt = self.conn.prepare(&format!(
974 "SELECT id, task_id, project_path, snapshot_path, changed_files_json,
975 pending_action_json, approval_id, created_at, archived_at, archive_reason,
976 session_id, message_index
977 FROM checkpoints {archived_filter} ORDER BY created_at DESC LIMIT ?1"
978 ))?;
979 let rows = stmt.query_map([clamp_limit(limit)], checkpoint_from_row)?;
980 rows.collect::<rusqlite::Result<Vec<_>>>()
981 .map_err(Into::into)
982 }
983
984 pub fn list_for_session(
998 &self,
999 session_id: &str,
1000 after_message_index: i64,
1001 ) -> Result<Vec<CheckpointRecord>> {
1002 let mut stmt = self.conn.prepare(
1003 "SELECT id, task_id, project_path, snapshot_path, changed_files_json,
1004 pending_action_json, approval_id, created_at, archived_at, archive_reason,
1005 session_id, message_index
1006 FROM checkpoints
1007 WHERE session_id = ?1 AND message_index > ?2 AND archived_at IS NULL
1008 ORDER BY created_at ASC",
1009 )?;
1010 let rows = stmt.query_map(
1011 params![session_id, after_message_index],
1012 checkpoint_from_row,
1013 )?;
1014 rows.collect::<rusqlite::Result<Vec<_>>>()
1015 .map_err(Into::into)
1016 }
1017
1018 pub fn archive(&self, ids: &[String], reason: &str) -> Result<usize> {
1025 let archived_at = now_rfc3339();
1026 let mut changed = 0;
1027 for id in ids {
1028 changed += self.conn.execute(
1029 "UPDATE checkpoints
1030 SET archived_at = COALESCE(archived_at, ?2),
1031 archive_reason = COALESCE(archive_reason, ?3)
1032 WHERE id = ?1 AND archived_at IS NULL",
1033 params![id, archived_at, reason],
1034 )?;
1035 }
1036 Ok(changed)
1037 }
1038
1039 pub fn count_archived(&self) -> Result<usize> {
1043 self.conn
1044 .query_row(
1045 "SELECT COUNT(*) FROM checkpoints WHERE archived_at IS NOT NULL",
1046 [],
1047 |row| row.get::<_, i64>(0),
1048 )
1049 .map(|count| count as usize)
1050 .map_err(Into::into)
1051 }
1052}
1053
1054pub struct CompactionsRepo<'a> {
1055 pub(crate) conn: &'a Connection,
1056}
1057
1058impl CompactionsRepo<'_> {
1059 pub fn create(&self, new: NewCompaction) -> Result<CompactionRecord> {
1064 let id = new.id.unwrap_or_else(|| fresh_id("compaction"));
1065 self.conn.execute(
1066 "INSERT INTO compactions
1067 (id, task_id, session_id, source_token_estimate, summary_token_count,
1068 preserved_turns, archive_path, verification_status, created_at)
1069 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
1070 ON CONFLICT(id) DO UPDATE SET
1071 task_id = excluded.task_id,
1072 session_id = excluded.session_id,
1073 source_token_estimate = excluded.source_token_estimate,
1074 summary_token_count = excluded.summary_token_count,
1075 preserved_turns = excluded.preserved_turns,
1076 archive_path = excluded.archive_path,
1077 verification_status = excluded.verification_status",
1078 params![
1079 id,
1080 new.task_id,
1081 new.session_id,
1082 new.source_token_estimate,
1083 new.summary_token_count,
1084 new.preserved_turns,
1085 new.archive_path,
1086 new.verification_status,
1087 now_rfc3339(),
1088 ],
1089 )?;
1090 self.get(&id)?
1091 .context("compaction was inserted but could not be reloaded")
1092 }
1093
1094 pub fn get(&self, id: &str) -> Result<Option<CompactionRecord>> {
1099 self.conn
1100 .query_row(
1101 "SELECT id, task_id, session_id, source_token_estimate, summary_token_count,
1102 preserved_turns, archive_path, verification_status, created_at
1103 FROM compactions WHERE id = ?1",
1104 [id],
1105 compaction_from_row,
1106 )
1107 .optional()
1108 .map_err(Into::into)
1109 }
1110
1111 pub fn list(&self, limit: usize) -> Result<Vec<CompactionRecord>> {
1116 let mut stmt = self.conn.prepare(
1117 "SELECT id, task_id, session_id, source_token_estimate, summary_token_count,
1118 preserved_turns, archive_path, verification_status, created_at
1119 FROM compactions ORDER BY created_at DESC LIMIT ?1",
1120 )?;
1121 let rows = stmt.query_map([clamp_limit(limit)], compaction_from_row)?;
1122 rows.collect::<rusqlite::Result<Vec<_>>>()
1123 .map_err(Into::into)
1124 }
1125}
1126
1127pub struct PluginsRepo<'a> {
1128 pub(crate) conn: &'a Connection,
1129}
1130
1131impl PluginsRepo<'_> {
1132 pub fn install(&self, new: NewPluginInstall) -> Result<PluginInstallRecord> {
1137 let now = now_rfc3339();
1138 let id = new.id.unwrap_or_else(|| fresh_id("plugin"));
1139 self.conn.execute(
1140 "INSERT INTO plugin_installs
1141 (id, name, source, version, enabled, manifest_json, installed_at, updated_at)
1142 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
1143 ON CONFLICT(id) DO UPDATE SET
1144 name = excluded.name,
1145 source = excluded.source,
1146 version = excluded.version,
1147 enabled = excluded.enabled,
1148 manifest_json = excluded.manifest_json,
1149 updated_at = excluded.updated_at",
1150 params![
1151 id,
1152 new.name,
1153 new.source,
1154 new.version,
1155 if new.enabled { 1 } else { 0 },
1156 new.manifest_json,
1157 now,
1158 now,
1159 ],
1160 )?;
1161 self.get(&id)?
1162 .context("plugin install was inserted but could not be reloaded")
1163 }
1164
1165 pub fn get(&self, id: &str) -> Result<Option<PluginInstallRecord>> {
1170 self.conn
1171 .query_row(
1172 "SELECT id, name, source, version, enabled, manifest_json, installed_at, updated_at
1173 FROM plugin_installs WHERE id = ?1",
1174 [id],
1175 plugin_from_row,
1176 )
1177 .optional()
1178 .map_err(Into::into)
1179 }
1180
1181 pub fn list(&self) -> Result<Vec<PluginInstallRecord>> {
1186 let mut stmt = self.conn.prepare(
1187 "SELECT id, name, source, version, enabled, manifest_json, installed_at, updated_at
1188 FROM plugin_installs ORDER BY name ASC",
1189 )?;
1190 let rows = stmt.query_map([], plugin_from_row)?;
1191 rows.collect::<rusqlite::Result<Vec<_>>>()
1192 .map_err(Into::into)
1193 }
1194
1195 pub fn set_enabled(&self, id: &str, enabled: bool) -> Result<()> {
1199 self.conn.execute(
1200 "UPDATE plugin_installs SET enabled = ?2, updated_at = ?3 WHERE id = ?1",
1201 params![id, if enabled { 1 } else { 0 }, now_rfc3339()],
1202 )?;
1203 Ok(())
1204 }
1205}
1206
1207pub struct ProviderProbesRepo<'a> {
1208 pub(crate) conn: &'a Connection,
1209}
1210
1211impl ProviderProbesRepo<'_> {
1212 pub fn upsert(&self, new: NewProviderProbe) -> Result<ProviderProbeRecord> {
1217 let now = now_rfc3339();
1218 let provider = new.provider;
1219 let model_id = new.model_id;
1220 let capability_key = new.capability_key;
1221 self.conn.execute(
1222 "INSERT INTO provider_probes
1223 (provider, model_id, capability_key, capability_value, confidence, error, probed_at)
1224 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
1225 ON CONFLICT(provider, model_id, capability_key) DO UPDATE SET
1226 capability_value = excluded.capability_value,
1227 confidence = excluded.confidence,
1228 error = excluded.error,
1229 probed_at = excluded.probed_at",
1230 params![
1231 &provider,
1232 &model_id,
1233 &capability_key,
1234 new.capability_value,
1235 new.confidence,
1236 new.error,
1237 now,
1238 ],
1239 )?;
1240 self.get(&provider, &model_id, &capability_key)?
1241 .context("provider probe was inserted but could not be reloaded")
1242 }
1243
1244 pub fn get(
1249 &self,
1250 provider: &str,
1251 model_id: &str,
1252 capability_key: &str,
1253 ) -> Result<Option<ProviderProbeRecord>> {
1254 self.conn
1255 .query_row(
1256 "SELECT provider, model_id, capability_key, capability_value, confidence, error, probed_at
1257 FROM provider_probes
1258 WHERE provider = ?1 AND model_id = ?2 AND capability_key = ?3",
1259 params![provider, model_id, capability_key],
1260 provider_probe_from_row,
1261 )
1262 .optional()
1263 .map_err(Into::into)
1264 }
1265
1266 pub fn list(
1271 &self,
1272 provider: Option<&str>,
1273 model_id: Option<&str>,
1274 ) -> Result<Vec<ProviderProbeRecord>> {
1275 let mut stmt = self.conn.prepare(
1276 "SELECT provider, model_id, capability_key, capability_value, confidence, error, probed_at
1277 FROM provider_probes ORDER BY provider ASC, model_id ASC, capability_key ASC",
1278 )?;
1279 let rows = stmt.query_map([], provider_probe_from_row)?;
1280 let mut out = Vec::new();
1281 for row in rows {
1282 let probe = row?;
1283 if provider.is_some_and(|p| probe.provider != p) {
1284 continue;
1285 }
1286 if model_id.is_some_and(|m| probe.model_id != m) {
1287 continue;
1288 }
1289 out.push(probe);
1290 }
1291 Ok(out)
1292 }
1293}
1294
1295pub struct PairingTokensRepo<'a> {
1296 pub(crate) conn: &'a Connection,
1297}
1298
1299impl PairingTokensRepo<'_> {
1300 pub fn create(
1305 &self,
1306 token_hash: &str,
1307 label: Option<&str>,
1308 expires_at: Option<&str>,
1309 ) -> Result<PairingTokenRecord> {
1310 let id = fresh_id("pairing");
1311 self.conn.execute(
1312 "INSERT INTO pairing_tokens
1313 (id, token_hash, label, enabled, created_at, last_used_at, expires_at)
1314 VALUES (?1, ?2, ?3, 1, ?4, NULL, ?5)",
1315 params![id, token_hash, label, now_rfc3339(), expires_at],
1316 )?;
1317 self.get(&id)?
1318 .context("pairing token was inserted but could not be reloaded")
1319 }
1320
1321 pub fn get(&self, id: &str) -> Result<Option<PairingTokenRecord>> {
1326 self.conn
1327 .query_row(
1328 "SELECT id, token_hash, label, enabled, created_at, last_used_at, expires_at
1329 FROM pairing_tokens WHERE id = ?1",
1330 [id],
1331 pairing_from_row,
1332 )
1333 .optional()
1334 .map_err(Into::into)
1335 }
1336
1337 pub fn verify_token(&self, token_hash: &str) -> Result<Option<PairingTokenRecord>> {
1351 let now = chrono::Utc::now();
1356 let mut stmt = self.conn.prepare(
1357 "SELECT id, token_hash, label, enabled, created_at, last_used_at, expires_at
1358 FROM pairing_tokens
1359 WHERE enabled = 1",
1360 )?;
1361 let candidates = stmt
1362 .query_map([], pairing_from_row)?
1363 .collect::<rusqlite::Result<Vec<_>>>()?;
1364 let mut found = None;
1365 for record in candidates {
1366 if is_expired(record.expires_at.as_deref(), now) {
1367 continue;
1368 }
1369 if ct_eq(record.token_hash.as_bytes(), token_hash.as_bytes()) {
1370 found = Some(record);
1371 }
1372 }
1373 Ok(found)
1374 }
1375
1376 pub fn list(&self) -> Result<Vec<PairingTokenRecord>> {
1381 let mut stmt = self.conn.prepare(
1382 "SELECT id, token_hash, label, enabled, created_at, last_used_at, expires_at
1383 FROM pairing_tokens ORDER BY created_at DESC",
1384 )?;
1385 let rows = stmt.query_map([], pairing_from_row)?;
1386 rows.collect::<rusqlite::Result<Vec<_>>>()
1387 .map_err(Into::into)
1388 }
1389
1390 pub fn list_redacted(&self) -> Result<Vec<PairingTokenRecord>> {
1400 Ok(self
1401 .list()?
1402 .into_iter()
1403 .map(|mut record| {
1404 record.token_hash = String::new();
1405 record
1406 })
1407 .collect())
1408 }
1409
1410 pub fn mark_used(&self, id: &str) -> Result<()> {
1414 self.conn.execute(
1415 "UPDATE pairing_tokens SET last_used_at = ?2 WHERE id = ?1 AND enabled = 1",
1416 params![id, now_rfc3339()],
1417 )?;
1418 Ok(())
1419 }
1420
1421 pub fn revoke(&self, id: &str) -> Result<bool> {
1429 let changed = self.conn.execute(
1430 "UPDATE pairing_tokens SET enabled = 0 WHERE id = ?1 AND enabled = 1",
1431 params![id],
1432 )?;
1433 Ok(changed > 0)
1434 }
1435}