Skip to main content

mermaid_runtime/storage/
repos.rs

1//! One repository type per table. Each is a thin `&Connection` wrapper; they
2//! share nothing else, which is what made this the most mechanical seam in the
3//! tree — and why `TasksRepo`'s declaration had drifted 100 lines from its own
4//! `impl`, with two unrelated repos in between.
5
6use anyhow::{Context, Result};
7use rusqlite::{Connection, OptionalExtension, params};
8
9use super::*;
10
11// Bumped to 5 for the additive `tasks.prompt` column (the daemon scheduler
12// executes queued tasks later, so the full prompt must be persisted at enqueue
13// time — `title` is truncated at 80 chars). Additive, but the bump lets a DB
14// already at v4 re-run the migration once to pick it up. The bump is
15// load-bearing alongside the F17 early-return in `init_schema`: a DB at an
16// older version still runs the migration (the idempotent baseline plus any
17// per-version step dispatched by `migrate_within_txn`) exactly once, while an
18// already-current DB skips the write lock entirely.
19//
20// History: v2 added the additive `tasks.owner_kind` column (F18/RC-E); v3 added
21// the F75 covering indexes; v4 added the `outcomes` table.
22pub 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    /// # Errors
32    ///
33    /// Errors if the write statement fails, or if the row cannot be read back
34    /// afterwards -- the reload is what produces the returned record.
35    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    /// # Errors
65    ///
66    /// Errors if the query fails or the stored row does not decode. A row that is
67    /// not there is `Ok(None)`, not an error.
68    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    /// # Errors
82    ///
83    /// Errors if the statement fails to prepare or run, or if any row does not
84    /// decode -- one undecodable row fails the whole call.
85    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    /// # Errors
103    ///
104    /// Errors if the write statement fails, or if the row cannot be read back
105    /// afterwards -- the reload is what produces the returned record.
106    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    /// # Errors
118    ///
119    /// Errors if the query fails or the stored row does not decode. A row that is
120    /// not there is `Ok(None)`, not an error.
121    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    /// Load a session's messages in chronological order, capped at
134    /// [`MAX_SESSION_MESSAGES`] (F24/RC-F).
135    ///
136    /// A session transcript is otherwise unbounded, and the daemon
137    /// `session_messages` path loads it whole into RAM — a pathological session
138    /// could OOM the daemon. We return the **most recent** `MAX_SESSION_MESSAGES`
139    /// (newest activity is what a viewer wants) but still in ascending `id`
140    /// order, by taking the tail in a subquery and re-sorting it ascending.
141    ///
142    /// # Errors
143    ///
144    /// Errors if the statement fails to prepare or run, or if any row does not
145    /// decode -- one undecodable row fails the whole call.
146    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    /// # Errors
162    ///
163    /// Errors if the write transaction fails, or if the row cannot be read back
164    /// afterwards -- the reload is what produces the returned record.
165    pub fn create(&self, new: NewTask) -> Result<TaskRecord> {
166        let now = now_rfc3339();
167        // Owner tag isn't part of the public `TaskRecord`; move it out before the
168        // record consumes the rest of `new`, then persist it on its own column.
169        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        // The task row and its initial event are one logical write — commit
184        // them atomically so a crash between can't leave an event-less task.
185        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    /// # Errors
216    ///
217    /// Errors if the query fails or the stored row does not decode. A row that is
218    /// not there is `Ok(None)`, not an error.
219    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    /// # Errors
233    ///
234    /// Errors if the statement fails to prepare or run, or if any row does not
235    /// decode -- one undecodable row fails the whole call.
236    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        // F19 (RC-E): skip-and-warn a single undecodable row (e.g. a status enum
245        // a different binary wrote) instead of `collect`ing a `Result` that would
246        // blank the WHOLE tasks panel on one poison row.
247        let rows = stmt.query_map([clamp_limit(limit)], task_from_row_opt)?;
248        collect_tolerant(rows)
249    }
250
251    /// # Errors
252    ///
253    /// Errors if the statement fails. The work runs in a transaction, so a failure
254    /// leaves the table unchanged.
255    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        // Status update + its event are one logical write.
263        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    /// Atomically claim the next runnable queued task for the daemon scheduler:
285    /// flip it to `Running` and return it, or `None` when the queue is empty.
286    ///
287    /// Only daemon-owned tasks WITH a persisted prompt are claimable —
288    /// metadata-only tasks (interactive CLI runs, external `create_task`
289    /// callers) are never executed by the scheduler. Order: priority
290    /// (high → normal → low), then FIFO by `created_at` (id as tiebreaker,
291    /// since two enqueues can share a coarse-clock timestamp). The claim is a
292    /// single `UPDATE … RETURNING`, so concurrent claimers can never run the
293    /// same task twice.
294    ///
295    /// # Errors
296    ///
297    /// Errors if the query fails or the stored row does not decode. A row that is
298    /// not there is `Ok(None)`, not an error. The work runs in a transaction, so a
299    /// failure leaves the table unchanged.
300    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    /// # Errors
340    ///
341    /// Errors if the statement fails.
342    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    /// # Errors
352    ///
353    /// Errors if the statement fails to prepare or run, or if any row does not
354    /// decode -- one undecodable row fails the whole call.
355    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        // F19 (RC-E): one undecodable event row must not blank the whole timeline.
363        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    /// # Errors
374    ///
375    /// Errors if the write statement fails, or if the row cannot be read back
376    /// afterwards -- the reload is what produces the returned record.
377    pub fn start(&self, mut new: NewToolRun) -> Result<ToolRunRecord> {
378        // The repository is the mandatory persistence choke point. Callers may
379        // pass executable arguments unchanged; only this cloned serialized
380        // representation is scrubbed before SQLite sees it.
381        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    /// # Errors
405    ///
406    /// Errors if the statement fails.
407    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    /// # Errors
420    ///
421    /// Errors if the query fails or the stored row does not decode. A row that is
422    /// not there is `Ok(None)`, not an error.
423    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    /// # Errors
437    ///
438    /// Errors if the statement fails to prepare or run, or if any row does not
439    /// decode -- one undecodable row fails the whole call.
440    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    /// Record a verifiable outcome / reward signal for a trajectory. Append-only
458    /// — the loop reads these; nothing mutates them after the fact.
459    ///
460    /// # Errors
461    ///
462    /// Errors if the write statement fails, or if the row cannot be read back
463    /// afterwards -- the reload is what produces the returned record.
464    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    /// # Errors
487    ///
488    /// Errors if the query fails or the stored row does not decode. A row that is
489    /// not there is `Ok(None)`, not an error.
490    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    /// Every outcome recorded against one task, oldest first (the order the
504    /// trajectory earned them).
505    ///
506    /// # Errors
507    ///
508    /// Errors if the statement fails to prepare or run, or if any row does not
509    /// decode -- one undecodable row fails the whole call.
510    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    /// # Errors
522    ///
523    /// Errors if the statement fails to prepare or run, or if any row does not
524    /// decode -- one undecodable row fails the whole call.
525    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    /// # Errors
557    ///
558    /// Errors if the write statement fails, or if the row cannot be read back
559    /// afterwards -- the reload is what produces the returned record.
560    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    /// # Errors
600    ///
601    /// Errors if the query fails or the stored row does not decode. A row that is
602    /// not there is `Ok(None)`, not an error.
603    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    /// # Errors
618    ///
619    /// Errors if the statement fails.
620    pub fn decide(&self, id: &str, user_decision: &str) -> Result<()> {
621        // Single-shot decision: only an undecided, un-archived approval can be
622        // decided, so a denied approval cannot be resurrected as "approved".
623        // `approval::approve_and_replay` runs the (un-rollback-able) replay
624        // effect *before* calling `decide`, so the "approved" mark lands only
625        // after the action ran: a crash mid-replay leaves the row undecided and
626        // safely re-runnable, never "approved but never applied" (#62). Mirrors
627        // the `archive` `WHERE archived_at IS NULL` idempotency pattern below.
628        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    /// Atomically claim an undecided approval for replay (#118). Sets
642    /// `user_decision='approving'` only when it is currently NULL and
643    /// un-archived, and reports whether THIS caller won the claim. Two concurrent
644    /// `approve <id>` calls race this single UPDATE; exactly one sees
645    /// `rows_affected == 1` and runs the un-rollback-able effect, the other sees
646    /// `false` and bails — so the effect can't fire twice. A claim that crashes
647    /// before finalizing is reset to NULL by the daemon's startup reconcile.
648    ///
649    /// # Errors
650    ///
651    /// Errors if the UPDATE fails. Losing the race is `Ok(false)`, not an error:
652    /// the row was already decided, already claimed, or archived.
653    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    /// Release a claim taken by [`Self::claim`] back to undecided, so the action
664    /// stays re-runnable after the replay effect failed.
665    ///
666    /// # Errors
667    ///
668    /// Errors if the statement fails.
669    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    /// Finalize a claimed approval's decision (the `approving` → terminal-value
679    /// transition that [`Self::decide`]'s `WHERE user_decision IS NULL` can't make).
680    ///
681    /// # Errors
682    ///
683    /// Errors if the statement fails.
684    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    /// # Errors
696    ///
697    /// Errors if the underlying query fails or any row does not decode.
698    pub fn list_pending(&self) -> Result<Vec<ApprovalRecord>> {
699        self.list_pending_with_archived(false)
700    }
701
702    /// # Errors
703    ///
704    /// Errors if the underlying query fails or any row does not decode.
705    pub fn list_pending_all(&self) -> Result<Vec<ApprovalRecord>> {
706        self.list_pending_with_archived(true)
707    }
708
709    /// # Errors
710    ///
711    /// Errors if the statement fails to prepare or run, or if any row does not
712    /// decode -- one undecodable row fails the whole call.
713    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    /// # Errors
750    ///
751    /// Errors if any one of the per-id updates fails, and stops there -- the ids
752    /// before it stay archived, because the loop runs outside a transaction. The
753    /// count is of rows actually changed, so ids already archived add nothing and
754    /// are not an error.
755    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    /// # Errors
771    ///
772    /// Errors if the count query fails.
773    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    /// # Errors
791    ///
792    /// Errors if the write statement fails, or if the row cannot be read back
793    /// afterwards -- the reload is what produces the returned record.
794    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    /// # Errors
830    ///
831    /// Errors if the query fails or the stored row does not decode. A row that is
832    /// not there is `Ok(None)`, not an error.
833    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    /// # Errors
847    ///
848    /// Errors if the statement fails to prepare or run, or if any row does not
849    /// decode -- one undecodable row fails the whole call.
850    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        // F19 (RC-E): skip-and-warn an undecodable row (e.g. a status enum a
859        // different binary wrote) rather than blanking the whole processes panel.
860        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    /// # Errors
871    ///
872    /// Errors if the write statement fails, or if the row cannot be read back
873    /// afterwards -- the reload is what produces the returned record.
874    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    /// # Errors
899    ///
900    /// Errors if the query fails or the stored row does not decode. A row that is
901    /// not there is `Ok(None)`, not an error.
902    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    /// # Errors
917    ///
918    /// Errors if the statement fails.
919    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    /// Delete a checkpoint row outright. Returns whether a row was removed.
929    ///
930    /// F23 (RC-F): coordinates the on-disk checkpoint-dir GC
931    /// ([`crate::checkpoint::gc_old_checkpoint_dirs`]) with the DB. The dir GC
932    /// prunes by mtime regardless of archive state, while storage [`Self`] /
933    /// `gc()` only removes ARCHIVED checkpoint rows — so a never-archived old
934    /// checkpoint would lose its directory while its row survived, and a later
935    /// `restore_checkpoint` would fail on the missing manifest. The dir GC now
936    /// calls this so `list()` and the on-disk dirs stay in agreement.
937    ///
938    /// # Errors
939    ///
940    /// Errors if the DELETE fails. A checkpoint that was not there is `Ok(false)`,
941    /// not an error.
942    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    /// # Errors
950    ///
951    /// Errors if the statement fails.
952    pub fn list(&self, limit: usize) -> Result<Vec<CheckpointRecord>> {
953        self.list_with_archived(limit, false)
954    }
955
956    /// # Errors
957    ///
958    /// Errors if the statement fails.
959    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    /// Unarchived checkpoints of `session_id` anchored STRICTLY past
985    /// `after_message_index`, oldest first. Strict `>` is the fork-boundary
986    /// invariant: a fork at user-message index `k` keeps `messages[..k]`, and
987    /// a checkpoint stamped `message_index == k` snapshotted state from
988    /// BEFORE that user message existed — it belongs to the kept prefix, not
989    /// the discarded timeline. Oldest-first because each checkpoint is a
990    /// PRE-mutation snapshot: the oldest one past the cut holds the file
991    /// state closest to the fork point.
992    ///
993    /// # Errors
994    ///
995    /// Errors if the statement fails to prepare or run, or if any row does not
996    /// decode -- one undecodable row fails the whole call.
997    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    /// # Errors
1019    ///
1020    /// Errors if any one of the per-id updates fails, and stops there -- the ids
1021    /// before it stay archived, because the loop runs outside a transaction. The
1022    /// count is of rows actually changed, so ids already archived add nothing and
1023    /// are not an error.
1024    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    /// # Errors
1040    ///
1041    /// Errors if the count query fails.
1042    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    /// # Errors
1060    ///
1061    /// Errors if the write statement fails, or if the row cannot be read back
1062    /// afterwards -- the reload is what produces the returned record.
1063    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    /// # Errors
1095    ///
1096    /// Errors if the query fails or the stored row does not decode. A row that is
1097    /// not there is `Ok(None)`, not an error.
1098    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    /// # Errors
1112    ///
1113    /// Errors if the statement fails to prepare or run, or if any row does not
1114    /// decode -- one undecodable row fails the whole call.
1115    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    /// # Errors
1133    ///
1134    /// Errors if the write statement fails, or if the row cannot be read back
1135    /// afterwards -- the reload is what produces the returned record.
1136    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    /// # Errors
1166    ///
1167    /// Errors if the query fails or the stored row does not decode. A row that is
1168    /// not there is `Ok(None)`, not an error.
1169    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    /// # Errors
1182    ///
1183    /// Errors if the statement fails to prepare or run, or if any row does not
1184    /// decode -- one undecodable row fails the whole call.
1185    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    /// # Errors
1196    ///
1197    /// Errors if the statement fails.
1198    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    /// # Errors
1213    ///
1214    /// Errors if the write statement fails, or if the row cannot be read back
1215    /// afterwards -- the reload is what produces the returned record.
1216    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    /// # Errors
1245    ///
1246    /// Errors if the query fails or the stored row does not decode. A row that is
1247    /// not there is `Ok(None)`, not an error.
1248    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    /// # Errors
1267    ///
1268    /// Errors if the statement fails to prepare or run, or if any row does not
1269    /// decode -- one undecodable row fails the whole call.
1270    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    /// # Errors
1301    ///
1302    /// Errors if the write statement fails, or if the row cannot be read back
1303    /// afterwards -- the reload is what produces the returned record.
1304    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    /// # Errors
1322    ///
1323    /// Errors if the query fails or the stored row does not decode. A row that is
1324    /// not there is `Ok(None)`, not an error.
1325    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    /// Look up an enabled, unexpired pairing token by hash.
1338    ///
1339    /// The hash is **not** matched in SQL (`WHERE token_hash = ?`) — that is a
1340    /// DB-level equality on the secret and a theoretical timing channel.
1341    /// Instead we fetch the enabled, unexpired candidates (neither predicate is
1342    /// secret) and compare each hash in constant time. The candidate count is
1343    /// tiny and not secret. All candidates are scanned without early exit so the
1344    /// timing doesn't reveal which (if any) token matched.
1345    ///
1346    /// # Errors
1347    ///
1348    /// Errors if the query fails or a row does not decode. No match is `Ok(None)`,
1349    /// not an error, and so is a token that matches but has expired.
1350    pub fn verify_token(&self, token_hash: &str) -> Result<Option<PairingTokenRecord>> {
1351        // Expiry is evaluated in Rust as a parsed instant (see `is_expired`),
1352        // not via a SQL `expires_at > ?` string compare. The skipped-because-
1353        // expired branch is on non-secret data; the hash itself is still matched
1354        // in constant time over every non-expired candidate with no early exit.
1355        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    /// # Errors
1377    ///
1378    /// Errors if the statement fails to prepare or run, or if any row does not
1379    /// decode -- one undecodable row fails the whole call.
1380    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    /// Like [`list`](Self::list), but with `token_hash` blanked. Use for any
1391    /// surface that crosses a trust boundary — e.g. the daemon snapshot served
1392    /// over the local socket to same-UID processes. The hash is
1393    /// secret-equivalent (it's all `verify_token` compares against) and must
1394    /// not leave the store.
1395    ///
1396    /// # Errors
1397    ///
1398    /// Errors if the underlying query fails or any row does not decode.
1399    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    /// # Errors
1411    ///
1412    /// Errors if the statement fails.
1413    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    /// Revoke a token by disabling it. Returns `true` if a live token was
1422    /// revoked, `false` if it was already disabled or unknown.
1423    ///
1424    /// # Errors
1425    ///
1426    /// Errors if the UPDATE fails. A token that was already revoked, or absent, is
1427    /// `Ok(false)`, not an error.
1428    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}