doover_core/
journal.rs

1//! Append-only action journal (SQLite, WAL).
2//!
3//! Semantics grounded in the captured hook contract (fixtures README):
4//! - success is implied by a PostToolUse arriving; there is no exit code —
5//!   a `pending` action with no post event is closed as `abandoned` when the
6//!   next action starts in its session, or at session end;
7//! - `tool_use_id` correlates pre/post pairs;
8//! - undo never deletes rows: it is itself a journaled action referencing its
9//!   target, so undoing an undo is redo, and history stays append-only
10//!   (YoloFS-style travel, not destructive rollback).
11//!
12//! Concurrency: hook invocations are separate short-lived processes. WAL mode
13//! plus BEGIN IMMEDIATE transactions and a busy timeout give unique,
14//! contiguous per-session sequence numbers across racing writers.
15
16use crate::snapshot::{EntryKind, Manifest};
17use std::collections::BTreeSet;
18use std::path::Path;
19use std::time::{Duration, SystemTime, UNIX_EPOCH};
20
21const SCHEMA_VERSION: i64 = 2;
22const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
23
24/// Which side of an action a manifest captures: state before the command
25/// (restored by undo) or after (restored by redo, and the conflict oracle).
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum ManifestRole {
28    Pre,
29    Post,
30}
31
32impl ManifestRole {
33    fn as_str(self) -> &'static str {
34        match self {
35            Self::Pre => "pre",
36            Self::Post => "post",
37        }
38    }
39}
40
41#[derive(Debug, thiserror::Error)]
42pub enum JournalError {
43    #[error("sqlite error: {0}")]
44    Sqlite(#[from] rusqlite::Error),
45    #[error("journal io: {0}")]
46    Io(#[from] std::io::Error),
47    #[error("manifest serialization: {0}")]
48    Serde(#[from] serde_json::Error),
49    #[error("action {0} not found")]
50    ActionNotFound(i64),
51    #[error("no completable action with tool_use_id {tool_use_id} in session {session_id}")]
52    NoPendingForToolUse {
53        session_id: String,
54        tool_use_id: String,
55    },
56    #[error("cannot undo action {id}: status is {status:?} (must be completed or abandoned)")]
57    NotUndoable { id: i64, status: ActionStatus },
58    #[error(
59        "cannot undo action {id}: it reverses another undo; run undo on the original action {original} instead"
60    )]
61    UndoTooDeep { id: i64, original: i64 },
62    #[error(
63        "manifest was written by a newer doover (schema {found}, this build understands {})",
64        crate::snapshot::MANIFEST_SCHEMA
65    )]
66    ManifestTooNew { found: u32 },
67    #[error(
68        "journal schema version {found} is newer than this doover understands ({SCHEMA_VERSION})"
69    )]
70    SchemaTooNew { found: i64 },
71}
72
73pub type ActionId = i64;
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub enum ActionKind {
77    Command,
78    Undo,
79}
80
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub enum ActionStatus {
83    Pending,
84    Completed,
85    Abandoned,
86    Undone,
87}
88
89impl ActionKind {
90    fn parse(s: &str) -> Self {
91        match s {
92            "undo" => Self::Undo,
93            _ => Self::Command,
94        }
95    }
96}
97
98impl ActionStatus {
99    fn as_str(self) -> &'static str {
100        match self {
101            Self::Pending => "pending",
102            Self::Completed => "completed",
103            Self::Abandoned => "abandoned",
104            Self::Undone => "undone",
105        }
106    }
107    fn parse(s: &str) -> Self {
108        match s {
109            "completed" => Self::Completed,
110            "abandoned" => Self::Abandoned,
111            "undone" => Self::Undone,
112            _ => Self::Pending,
113        }
114    }
115}
116
117pub struct NewAction<'a> {
118    pub session_id: &'a str,
119    pub tool_use_id: Option<&'a str>,
120    pub raw_command: &'a str,
121    /// Severity string from the resolver ("safe" … "irreversible", "unknown").
122    pub effect: &'a str,
123    pub rule_id: Option<&'a str>,
124    pub has_unknown: bool,
125}
126
127#[derive(Debug, Clone)]
128pub struct ActionRecord {
129    pub id: ActionId,
130    pub session_id: String,
131    pub seq: i64,
132    pub kind: ActionKind,
133    pub tool_use_id: Option<String>,
134    pub raw_command: String,
135    pub effect: String,
136    pub rule_id: Option<String>,
137    pub has_unknown: bool,
138    pub status: ActionStatus,
139    pub target_action_id: Option<ActionId>,
140    /// For undo actions: the target's status before it was flipped to
141    /// `undone` — what redo must restore (never a fabricated `completed`).
142    pub target_prior_status: Option<ActionStatus>,
143    pub pinned: bool,
144    pub started_at_ms: i64,
145    pub duration_ms: Option<i64>,
146    pub note: Option<String>,
147}
148
149/// True for SQLITE_BUSY / SQLITE_LOCKED — the transient contention class the
150/// open() retry loop absorbs.
151fn is_busy(e: &JournalError) -> bool {
152    matches!(
153        e,
154        JournalError::Sqlite(rusqlite::Error::SqliteFailure(f, _))
155            if matches!(
156                f.code,
157                rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked
158            )
159    )
160}
161
162pub fn now_ms() -> i64 {
163    SystemTime::now()
164        .duration_since(UNIX_EPOCH)
165        .map(|d| d.as_millis() as i64)
166        .unwrap_or(0)
167}
168
169pub struct Journal {
170    conn: rusqlite::Connection,
171}
172
173impl Journal {
174    pub fn open(path: &Path) -> Result<Self, JournalError> {
175        // Bounded retry on BUSY/LOCKED: the busy handler cannot wait in every
176        // path (SQLite returns immediately where waiting could deadlock, e.g.
177        // around the fresh-file WAL switch under concurrent first opens —
178        // round 20). Total worst-case wait ~1s, far under the hook timeout,
179        // and hook callers fail open anyway.
180        let mut last = None;
181        for _ in 0..40 {
182            match Self::open_attempt(path) {
183                Err(e) if is_busy(&e) => {
184                    last = Some(e);
185                    std::thread::sleep(Duration::from_millis(25));
186                }
187                other => return other,
188            }
189        }
190        Err(last.expect("loop ran at least once"))
191    }
192
193    fn open_attempt(path: &Path) -> Result<Self, JournalError> {
194        let conn = rusqlite::Connection::open(path)?;
195        // D4: raw commands may embed secrets — owner-only, umask-proof. The
196        // WAL/SHM sidecars inherit the db's mode on creation, but tighten any
197        // pre-existing ones from an older install too (best effort).
198        #[cfg(unix)]
199        {
200            use std::os::unix::fs::PermissionsExt;
201            let private = std::fs::Permissions::from_mode(0o600);
202            std::fs::set_permissions(path, private.clone())?;
203            for ext in ["-wal", "-shm"] {
204                let mut side = path.as_os_str().to_owned();
205                side.push(ext);
206                let _ = std::fs::set_permissions(std::path::Path::new(&side), private.clone());
207            }
208        }
209        conn.busy_timeout(BUSY_TIMEOUT)?;
210        conn.pragma_update(None, "journal_mode", "wal")?;
211        conn.pragma_update(None, "synchronous", "NORMAL")?;
212        conn.pragma_update(None, "foreign_keys", "ON")?;
213
214        // Double-checked migration under an EXCLUSIVE transaction (round 20,
215        // found by the S9 concurrency stress test): concurrent FIRST opens of
216        // a fresh journal each read user_version before anyone migrated, and
217        // the losers re-ran `ALTER TABLE` into "duplicate column name" — every
218        // hook failed open, so a brand-new install with parallel agents had
219        // ZERO protection. BEGIN EXCLUSIVE serializes racers (busy_timeout
220        // queues them) and the version is re-read INSIDE the lock, so late
221        // arrivals see the migrated version and skip.
222        conn.execute_batch("BEGIN EXCLUSIVE;")?;
223        let migrate = (|| -> Result<(), JournalError> {
224            let mut version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
225            if version > SCHEMA_VERSION {
226                return Err(JournalError::SchemaTooNew { found: version });
227            }
228            // stepwise migrations: each block moves exactly one version
229            // forward, so any past journal upgrades along the same audited path
230            if version == 0 {
231                conn.execute_batch(
232                    "CREATE TABLE IF NOT EXISTS sessions(
233                    id TEXT PRIMARY KEY,
234                    harness TEXT NOT NULL,
235                    cwd TEXT NOT NULL,
236                    started_at_ms INTEGER NOT NULL,
237                    ended_at_ms INTEGER
238                 );
239                 CREATE TABLE IF NOT EXISTS actions(
240                    id INTEGER PRIMARY KEY,
241                    session_id TEXT NOT NULL REFERENCES sessions(id),
242                    seq INTEGER NOT NULL,
243                    kind TEXT NOT NULL CHECK(kind IN ('command','undo')),
244                    tool_use_id TEXT,
245                    raw_command TEXT NOT NULL,
246                    effect TEXT NOT NULL,
247                    rule_id TEXT,
248                    has_unknown INTEGER NOT NULL DEFAULT 0,
249                    status TEXT NOT NULL
250                        CHECK(status IN ('pending','completed','abandoned','undone')),
251                    target_action_id INTEGER REFERENCES actions(id),
252                    target_prior_status TEXT,
253                    pinned INTEGER NOT NULL DEFAULT 0,
254                    started_at_ms INTEGER NOT NULL,
255                    duration_ms INTEGER,
256                    note TEXT,
257                    UNIQUE(session_id, seq)
258                 );
259                 CREATE INDEX IF NOT EXISTS idx_actions_session ON actions(session_id);
260                 CREATE INDEX IF NOT EXISTS idx_actions_tool_use ON actions(session_id, tool_use_id);
261                 CREATE TABLE IF NOT EXISTS manifests(
262                    id INTEGER PRIMARY KEY,
263                    action_id INTEGER NOT NULL REFERENCES actions(id),
264                    path TEXT NOT NULL,
265                    manifest_json TEXT NOT NULL,
266                    hashes TEXT NOT NULL,
267                    truncated INTEGER NOT NULL
268                 );
269                 CREATE INDEX IF NOT EXISTS idx_manifests_action ON manifests(action_id);
270                 PRAGMA user_version = 1;",
271                )?;
272                version = 1;
273            }
274            if version == 1 {
275                // v2: manifests gain a role — pre (undo restores it) or post
276                // (redo restores it; conflict oracle). Existing rows are all pre.
277                conn.execute_batch(
278                    "ALTER TABLE manifests ADD COLUMN role TEXT NOT NULL DEFAULT 'pre'
279                        CHECK(role IN ('pre','post'));
280                     PRAGMA user_version = 2;",
281                )?;
282            }
283            Ok(())
284        })();
285        match &migrate {
286            Ok(()) => conn.execute_batch("COMMIT;")?,
287            Err(_) => conn.execute_batch("ROLLBACK;").unwrap_or(()),
288        }
289        migrate?;
290        Ok(Self { conn })
291    }
292
293    pub fn begin_session(&self, id: &str, harness: &str, cwd: &str) -> Result<(), JournalError> {
294        self.conn.execute(
295            "INSERT INTO sessions(id, harness, cwd, started_at_ms) VALUES (?1, ?2, ?3, ?4)
296             ON CONFLICT(id) DO UPDATE SET cwd = excluded.cwd",
297            rusqlite::params![id, harness, cwd, now_ms()],
298        )?;
299        Ok(())
300    }
301
302    /// Record a new action as `pending`. Any older still-pending action in the
303    /// same session is closed as `abandoned` — its post event will never come
304    /// (captured contract: failures emit no PostToolUse).
305    pub fn start_action(&self, new: &NewAction) -> Result<ActionId, JournalError> {
306        self.conn.execute_batch("BEGIN IMMEDIATE;")?;
307        let result = (|| -> Result<ActionId, JournalError> {
308            self.conn.execute(
309                "UPDATE actions SET status = 'abandoned',
310                        note = COALESCE(note, 'no post event received')
311                 WHERE session_id = ?1 AND status = 'pending'",
312                [new.session_id],
313            )?;
314            let seq: i64 = self.conn.query_row(
315                "SELECT COALESCE(MAX(seq), 0) + 1 FROM actions WHERE session_id = ?1",
316                [new.session_id],
317                |r| r.get(0),
318            )?;
319            self.conn.execute(
320                "INSERT INTO actions(session_id, seq, kind, tool_use_id, raw_command,
321                                     effect, rule_id, has_unknown, status, started_at_ms)
322                 VALUES (?1, ?2, 'command', ?3, ?4, ?5, ?6, ?7, 'pending', ?8)",
323                rusqlite::params![
324                    new.session_id,
325                    seq,
326                    new.tool_use_id,
327                    new.raw_command,
328                    new.effect,
329                    new.rule_id,
330                    new.has_unknown,
331                    now_ms(),
332                ],
333            )?;
334            Ok(self.conn.last_insert_rowid())
335        })();
336        match &result {
337            Ok(_) => self.conn.execute_batch("COMMIT;")?,
338            Err(_) => self.conn.execute_batch("ROLLBACK;").unwrap_or(()),
339        }
340        result
341    }
342
343    /// Close a pending action via its pre/post correlation key. Also accepts
344    /// an `abandoned` action: with interleaved or background tool calls a post
345    /// can arrive after the next action's start already abandoned its pre —
346    /// the late post is ground truth and wins over our guess.
347    pub fn complete_by_tool_use(
348        &self,
349        session_id: &str,
350        tool_use_id: &str,
351        duration_ms: i64,
352    ) -> Result<ActionId, JournalError> {
353        let updated: Option<i64> = self
354            .conn
355            .query_row(
356                // newest matching row only: a duplicated correlation id (e.g.
357                // session replay) must never complete multiple actions
358                "UPDATE actions SET status = 'completed', duration_ms = ?3
359                 WHERE id = (SELECT id FROM actions
360                             WHERE session_id = ?1 AND tool_use_id = ?2
361                               AND status IN ('pending', 'abandoned')
362                             ORDER BY seq DESC LIMIT 1)
363                 RETURNING id",
364                rusqlite::params![session_id, tool_use_id, duration_ms],
365                |r| r.get(0),
366            )
367            .map(Some)
368            .or_else(|e| match e {
369                rusqlite::Error::QueryReturnedNoRows => Ok(None),
370                other => Err(other),
371            })?;
372        updated.ok_or_else(|| JournalError::NoPendingForToolUse {
373            session_id: session_id.to_string(),
374            tool_use_id: tool_use_id.to_string(),
375        })
376    }
377
378    pub fn end_session(&self, id: &str) -> Result<(), JournalError> {
379        self.conn.execute(
380            "UPDATE actions SET status = 'abandoned',
381                    note = COALESCE(note, 'session ended without post event')
382             WHERE session_id = ?1 AND status = 'pending'",
383            [id],
384        )?;
385        self.conn.execute(
386            "UPDATE sessions SET ended_at_ms = ?2 WHERE id = ?1",
387            rusqlite::params![id, now_ms()],
388        )?;
389        Ok(())
390    }
391
392    pub fn attach_manifest(
393        &self,
394        action: ActionId,
395        manifest: &Manifest,
396        role: ManifestRole,
397    ) -> Result<(), JournalError> {
398        let json = serde_json::to_string(manifest)?;
399        let hashes: Vec<&str> = manifest
400            .entries
401            .iter()
402            .filter_map(|e| match &e.kind {
403                EntryKind::File { hash, .. } => Some(hash.as_str()),
404                _ => None,
405            })
406            .collect();
407        self.conn.execute(
408            "INSERT INTO manifests(action_id, path, manifest_json, hashes, truncated, role)
409             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
410            rusqlite::params![
411                action,
412                manifest.path.to_string_lossy(),
413                json,
414                serde_json::to_string(&hashes)?,
415                manifest.truncated,
416                role.as_str(),
417            ],
418        )?;
419        Ok(())
420    }
421
422    /// All manifests of an action, regardless of role.
423    pub fn manifests(&self, action: ActionId) -> Result<Vec<Manifest>, JournalError> {
424        self.manifests_query(
425            "SELECT manifest_json FROM manifests WHERE action_id = ?1 ORDER BY id",
426            action,
427        )
428    }
429
430    /// Manifests of an action filtered by role (pre = undo side, post = redo
431    /// side / conflict oracle).
432    pub fn manifests_by_role(
433        &self,
434        action: ActionId,
435        role: ManifestRole,
436    ) -> Result<Vec<Manifest>, JournalError> {
437        match role {
438            ManifestRole::Pre => self.manifests_query(
439                "SELECT manifest_json FROM manifests
440                 WHERE action_id = ?1 AND role = 'pre' ORDER BY id",
441                action,
442            ),
443            ManifestRole::Post => self.manifests_query(
444                "SELECT manifest_json FROM manifests
445                 WHERE action_id = ?1 AND role = 'post' ORDER BY id",
446                action,
447            ),
448        }
449    }
450
451    fn manifests_query(&self, sql: &str, action: ActionId) -> Result<Vec<Manifest>, JournalError> {
452        let mut stmt = self.conn.prepare(sql)?;
453        let rows = stmt.query_map([action], |r| r.get::<_, String>(0))?;
454        let mut out: Vec<Manifest> = Vec::new();
455        for json in rows {
456            let m: Manifest = serde_json::from_str(&json?)?;
457            if m.schema > crate::snapshot::MANIFEST_SCHEMA {
458                return Err(JournalError::ManifestTooNew { found: m.schema });
459            }
460            out.push(m);
461        }
462        Ok(out)
463    }
464
465    /// Newest command-kind action that is plausibly undoable: completed or
466    /// abandoned, with at least one pre-manifest. Searches across sessions.
467    pub fn latest_undoable(&self) -> Result<Option<ActionRecord>, JournalError> {
468        let mut stmt = self.conn.prepare(&format!(
469            "{SELECT_ACTION} WHERE kind = 'command'
470               AND status IN ('completed','abandoned')
471               AND EXISTS (SELECT 1 FROM manifests m
472                           WHERE m.action_id = actions.id AND m.role = 'pre')
473             ORDER BY id DESC LIMIT 1"
474        ))?;
475        let mut rows = stmt.query_map([], row_to_action)?;
476        Ok(rows.next().transpose()?)
477    }
478
479    /// Most recent actions across all sessions, newest first (for `log`).
480    pub fn recent_actions(&self, limit: i64) -> Result<Vec<ActionRecord>, JournalError> {
481        let mut stmt = self
482            .conn
483            .prepare(&format!("{SELECT_ACTION} ORDER BY id DESC LIMIT ?1"))?;
484        let rows = stmt.query_map([limit], row_to_action)?;
485        Ok(rows.collect::<Result<Vec<_>, _>>()?)
486    }
487
488    /// Newest live undo action (redo target): kind undo, still completed.
489    pub fn latest_redoable(&self) -> Result<Option<ActionRecord>, JournalError> {
490        let mut stmt = self.conn.prepare(&format!(
491            "{SELECT_ACTION} WHERE kind = 'undo' AND status = 'completed'
492             ORDER BY id DESC LIMIT 1"
493        ))?;
494        let mut rows = stmt.query_map([], row_to_action)?;
495        Ok(rows.next().transpose()?)
496    }
497
498    /// Record an undo of `target` as a new journaled action. The target flips
499    /// to `undone`, remembering its prior status on the undo row; undoing an
500    /// *undo* restores that recorded status to the original — redo without
501    /// fabricating history. The status check runs INSIDE the transaction, so
502    /// racing double-undos admit exactly one winner, and an already-undone
503    /// target is refused (redo targets the undo action, not the original).
504    ///
505    /// Chains are bounded by design: undoable targets are command actions and
506    /// first-level undos (redo). Undoing a *redo* is refused with a pointer to
507    /// the original — the same capability with a trivially-consistent state
508    /// machine instead of recursive status cascades (audit round 4). The
509    /// exhaustive small-model test in T4 checks every sequence to depth 4
510    /// against a reference implementation of these rules.
511    pub fn record_undo(
512        &self,
513        session_id: &str,
514        target: ActionId,
515    ) -> Result<ActionId, JournalError> {
516        self.conn.execute_batch("BEGIN IMMEDIATE;")?;
517        let result = (|| -> Result<ActionId, JournalError> {
518            let target_rec = self.action(target)?;
519            if matches!(
520                target_rec.status,
521                ActionStatus::Pending | ActionStatus::Undone
522            ) {
523                return Err(JournalError::NotUndoable {
524                    id: target,
525                    status: target_rec.status,
526                });
527            }
528            if target_rec.kind == ActionKind::Undo {
529                let inner = self.action(
530                    target_rec
531                        .target_action_id
532                        .ok_or(JournalError::ActionNotFound(target))?,
533                )?;
534                if inner.kind == ActionKind::Undo {
535                    // walk to the ultimate command action for the error message
536                    let mut original = inner;
537                    let mut hops = 0;
538                    while original.kind == ActionKind::Undo && hops < 64 {
539                        match original.target_action_id {
540                            Some(t) => original = self.action(t)?,
541                            None => break,
542                        }
543                        hops += 1;
544                    }
545                    return Err(JournalError::UndoTooDeep {
546                        id: target,
547                        original: original.id,
548                    });
549                }
550            }
551            let seq: i64 = self.conn.query_row(
552                "SELECT COALESCE(MAX(seq), 0) + 1 FROM actions WHERE session_id = ?1",
553                [session_id],
554                |r| r.get(0),
555            )?;
556            self.conn.execute(
557                "INSERT INTO actions(session_id, seq, kind, raw_command, effect,
558                                     status, target_action_id, target_prior_status, started_at_ms)
559                 VALUES (?1, ?2, 'undo', ?3, 'destructive', 'completed', ?4, ?5, ?6)",
560                rusqlite::params![
561                    session_id,
562                    seq,
563                    format!("undo of action {target}"),
564                    target,
565                    target_rec.status.as_str(),
566                    now_ms(),
567                ],
568            )?;
569            let undo_id = self.conn.last_insert_rowid();
570            self.conn.execute(
571                "UPDATE actions SET status = 'undone' WHERE id = ?1",
572                [target],
573            )?;
574            // redo semantics: undoing an undo restores ITS target to the
575            // status the undo recorded — completed stays completed, abandoned
576            // stays abandoned
577            if target_rec.kind == ActionKind::Undo {
578                if let (Some(original), Some(prior)) =
579                    (target_rec.target_action_id, target_rec.target_prior_status)
580                {
581                    self.conn.execute(
582                        "UPDATE actions SET status = ?2 WHERE id = ?1",
583                        rusqlite::params![original, prior.as_str()],
584                    )?;
585                }
586            }
587            Ok(undo_id)
588        })();
589        match &result {
590            Ok(_) => self.conn.execute_batch("COMMIT;")?,
591            Err(_) => self.conn.execute_batch("ROLLBACK;").unwrap_or(()),
592        }
593        result
594    }
595
596    /// Append a note line to an action (used for loud protection-gap records:
597    /// truncated snapshots, per-path failures).
598    pub fn add_note(&self, action: ActionId, note: &str) -> Result<(), JournalError> {
599        let n = self.conn.execute(
600            "UPDATE actions SET note = COALESCE(note || char(10), '') || ?2 WHERE id = ?1",
601            rusqlite::params![action, note],
602        )?;
603        if n == 0 {
604            return Err(JournalError::ActionNotFound(action));
605        }
606        Ok(())
607    }
608
609    pub fn set_pinned(&self, action: ActionId, pinned: bool) -> Result<(), JournalError> {
610        let n = self.conn.execute(
611            "UPDATE actions SET pinned = ?2 WHERE id = ?1",
612            rusqlite::params![action, pinned],
613        )?;
614        if n == 0 {
615            return Err(JournalError::ActionNotFound(action));
616        }
617        Ok(())
618    }
619
620    /// Store hashes that GC must keep: referenced by a pinned action, by any
621    /// action started at/after `retain_after_ms`, or by an action that a live
622    /// (pinned/recent) undo targets — a kept chain row must keep its objects,
623    /// or redo would fail on a journal entry that still exists.
624    pub fn live_hashes(&self, retain_after_ms: i64) -> Result<BTreeSet<String>, JournalError> {
625        let mut stmt = self.conn.prepare(
626            // `pending` is live regardless of age: a long-running command's
627            // pre-snapshot must survive until its post event settles it —
628            // evicting those objects would strand the action doover is
629            // actively protecting (D2 review)
630            "SELECT m.hashes FROM manifests m
631             JOIN actions a ON a.id = m.action_id
632             WHERE a.pinned = 1 OR a.started_at_ms >= ?1 OR a.status = 'pending'
633                OR EXISTS (SELECT 1 FROM actions r
634                           WHERE r.target_action_id = a.id
635                             AND (r.pinned = 1 OR r.started_at_ms >= ?1))",
636        )?;
637        let rows = stmt.query_map([retain_after_ms], |r| r.get::<_, String>(0))?;
638        let mut out = BTreeSet::new();
639        for json in rows {
640            let hashes: Vec<String> = serde_json::from_str(&json?)?;
641            out.extend(hashes);
642        }
643        Ok(out)
644    }
645
646    pub fn action(&self, id: ActionId) -> Result<ActionRecord, JournalError> {
647        self.conn
648            .query_row(
649                &format!("{SELECT_ACTION} WHERE id = ?1"),
650                [id],
651                row_to_action,
652            )
653            .map_err(|e| match e {
654                rusqlite::Error::QueryReturnedNoRows => JournalError::ActionNotFound(id),
655                other => other.into(),
656            })
657    }
658
659    pub fn session_actions(&self, session_id: &str) -> Result<Vec<ActionRecord>, JournalError> {
660        let mut stmt = self.conn.prepare(&format!(
661            "{SELECT_ACTION} WHERE session_id = ?1 ORDER BY seq"
662        ))?;
663        let rows = stmt.query_map([session_id], row_to_action)?;
664        Ok(rows.collect::<Result<Vec<_>, _>>()?)
665    }
666
667    /// Newest action timestamp in the journal — the reference point for
668    /// retention cutoffs. NEVER use the wall clock for that (CLAUDE.md: a
669    /// backward NTP jump would make recent snapshots look collectable).
670    pub fn max_started_at(&self) -> Result<Option<i64>, JournalError> {
671        Ok(self
672            .conn
673            .query_row("SELECT MAX(started_at_ms) FROM actions", [], |r| r.get(0))?)
674    }
675
676    /// Prune journal rows older than `cutoff_ms`. Never touches pinned rows,
677    /// pending rows, or rows referenced as an undo target by a surviving row
678    /// (chain integrity; the referencing row's own pruning frees them for the
679    /// NEXT pass — eventual cleanup). Old `raw_command` strings may embed
680    /// secrets, which is why pruning rows (not just store objects) matters.
681    /// Returns (actions_pruned, sessions_pruned); `dry_run` only counts.
682    pub fn prune_before(&self, cutoff_ms: i64, dry_run: bool) -> Result<(u64, u64), JournalError> {
683        const CANDIDATES: &str = "FROM actions a
684             WHERE a.started_at_ms < ?1 AND a.pinned = 0 AND a.status != 'pending'
685               AND NOT EXISTS (SELECT 1 FROM actions r WHERE r.target_action_id = a.id)";
686        if dry_run {
687            let n: i64 = self.conn.query_row(
688                &format!("SELECT COUNT(*) {CANDIDATES}"),
689                [cutoff_ms],
690                |r| r.get(0),
691            )?;
692            // honest estimate, not a hardcoded zero: old sessions that are
693            // already empty or whose every action is itself a candidate
694            let s: i64 = self.conn.query_row(
695                "SELECT COUNT(*) FROM sessions s
696                 WHERE s.started_at_ms < ?1
697                   AND NOT EXISTS (
698                     SELECT 1 FROM actions a WHERE a.session_id = s.id
699                       AND NOT (a.started_at_ms < ?1 AND a.pinned = 0
700                                AND a.status != 'pending'
701                                AND NOT EXISTS (SELECT 1 FROM actions r
702                                                WHERE r.target_action_id = a.id)))",
703                [cutoff_ms],
704                |r| r.get(0),
705            )?;
706            return Ok((n as u64, s as u64));
707        }
708        self.conn.execute_batch("BEGIN IMMEDIATE;")?;
709        let result = (|| -> Result<(u64, u64), JournalError> {
710            self.conn.execute(
711                &format!("DELETE FROM manifests WHERE action_id IN (SELECT a.id {CANDIDATES})"),
712                [cutoff_ms],
713            )?;
714            let actions = self.conn.execute(
715                &format!("DELETE FROM actions WHERE id IN (SELECT a.id {CANDIDATES})"),
716                [cutoff_ms],
717            )? as u64;
718            // empty AND old (journal-relative), never merely empty: a session
719            // between begin_session and its first start_action is empty but
720            // live — deleting it would break the in-flight hook's FK insert
721            // and silently drop that action's protection
722            let sessions = self.conn.execute(
723                "DELETE FROM sessions
724                 WHERE started_at_ms < ?1
725                   AND id NOT IN (SELECT DISTINCT session_id FROM actions)",
726                [cutoff_ms],
727            )? as u64;
728            Ok((actions, sessions))
729        })();
730        match &result {
731            Ok(_) => self.conn.execute_batch("COMMIT;")?,
732            Err(_) => self.conn.execute_batch("ROLLBACK;").unwrap_or(()),
733        }
734        result
735    }
736
737    /// The `started_at_ms` of the last row in the next oldest-first batch of
738    /// evictable actions strictly before `before_ms` (size-cap eviction, D2).
739    /// Evictable = the same condition `prune_before` deletes by: unpinned,
740    /// non-pending, not referenced by an undo. `None` = nothing left to evict
741    /// below the ceiling — the caller must stop and report, never force.
742    pub fn oldest_evictable_batch_end(
743        &self,
744        batch: u32,
745        before_ms: i64,
746    ) -> Result<Option<i64>, JournalError> {
747        let mut stmt = self.conn.prepare(
748            "SELECT a.started_at_ms FROM actions a
749             WHERE a.started_at_ms < ?1 AND a.pinned = 0 AND a.status != 'pending'
750               AND NOT EXISTS (SELECT 1 FROM actions r WHERE r.target_action_id = a.id)
751             ORDER BY a.started_at_ms ASC
752             LIMIT ?2",
753        )?;
754        let mut last = None;
755        for row in stmt.query_map(rusqlite::params![before_ms, batch], |r| r.get::<_, i64>(0))? {
756            last = Some(row?);
757        }
758        Ok(last)
759    }
760
761    /// Test support: rewrite an action's timestamp so retention tests can
762    /// construct explicit timelines. Not part of the product surface.
763    pub fn set_started_at_for_test(
764        &self,
765        action: ActionId,
766        at_ms: i64,
767    ) -> Result<(), JournalError> {
768        let n = self.conn.execute(
769            "UPDATE actions SET started_at_ms = ?2 WHERE id = ?1",
770            rusqlite::params![action, at_ms],
771        )?;
772        if n == 0 {
773            return Err(JournalError::ActionNotFound(action));
774        }
775        Ok(())
776    }
777
778    /// Test support: backdate a session's start so retention tests can build
779    /// explicit timelines. Not part of the product surface.
780    pub fn set_session_started_at_for_test(
781        &self,
782        session_id: &str,
783        at_ms: i64,
784    ) -> Result<(), JournalError> {
785        self.conn.execute(
786            "UPDATE sessions SET started_at_ms = ?2 WHERE id = ?1",
787            rusqlite::params![session_id, at_ms],
788        )?;
789        Ok(())
790    }
791
792    /// (sessions, per-status action counts) for `status`/`doctor`.
793    pub fn stats(&self) -> Result<(u64, Vec<(String, u64)>), JournalError> {
794        let sessions: i64 = self
795            .conn
796            .query_row("SELECT COUNT(*) FROM sessions", [], |r| r.get(0))?;
797        let mut stmt = self
798            .conn
799            .prepare("SELECT status, COUNT(*) FROM actions GROUP BY status ORDER BY status")?;
800        let rows = stmt.query_map([], |r| {
801            Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)? as u64))
802        })?;
803        Ok((sessions as u64, rows.collect::<Result<Vec<_>, _>>()?))
804    }
805
806    pub fn integrity_check(&self) -> Result<bool, JournalError> {
807        let verdict: String = self
808            .conn
809            .query_row("PRAGMA integrity_check", [], |r| r.get(0))?;
810        Ok(verdict == "ok")
811    }
812}
813
814const SELECT_ACTION: &str = "SELECT id, session_id, seq, kind, tool_use_id, raw_command, effect,
815        rule_id, has_unknown, status, target_action_id, target_prior_status, pinned,
816        started_at_ms, duration_ms, note
817 FROM actions";
818
819fn row_to_action(r: &rusqlite::Row) -> Result<ActionRecord, rusqlite::Error> {
820    Ok(ActionRecord {
821        id: r.get(0)?,
822        session_id: r.get(1)?,
823        seq: r.get(2)?,
824        kind: ActionKind::parse(&r.get::<_, String>(3)?),
825        tool_use_id: r.get(4)?,
826        raw_command: r.get(5)?,
827        effect: r.get(6)?,
828        rule_id: r.get(7)?,
829        has_unknown: r.get(8)?,
830        status: ActionStatus::parse(&r.get::<_, String>(9)?),
831        target_action_id: r.get(10)?,
832        target_prior_status: r
833            .get::<_, Option<String>>(11)?
834            .map(|s| ActionStatus::parse(&s)),
835        pinned: r.get(12)?,
836        started_at_ms: r.get(13)?,
837        duration_ms: r.get(14)?,
838        note: r.get(15)?,
839    })
840}
doover_core/journal.rs

doover_core/
journal.rs
