huskies: merge 1249 story Checkpoint the CRDT snapshot periodically, not once by accident

server/src/config/mod.rs

@@ -155,6 +155,12 @@ pub struct ProjectConfig {
     /// (disabled) — the on-demand `gc` MCP tool remains available regardless.
     #[serde(default)]
     pub gc_min_free_gb: u64,
+    /// Number of CRDT ops applied (locally or received from sync peers)
+    /// between periodic snapshot checkpoints (story 1249). Keeps the
+    /// replayed tail on startup bounded instead of growing forever between
+    /// the one-time post-replay snapshot and the next restart. Default: 5000.
+    #[serde(default = "default_snapshot_interval_ops")]
+    pub snapshot_interval_ops: usize,
 }
 
 /// Configuration for the filesystem watcher's sweep behaviour.
@@ -255,6 +261,11 @@ fn default_max_mesh_peers() -> usize {
     3
 }
 
+/// Default number of ops between periodic CRDT snapshot checkpoints (story 1249).
+pub fn default_snapshot_interval_ops() -> usize {
+    5000
+}
+
 /// Configuration for the low-disk-space watchdog's free-space thresholds.
 ///
 /// Sleds check free space on the `/workspace` filesystem each tick and
@@ -472,6 +483,7 @@ impl Default for ProjectConfig {
             status_push_enabled: default_status_push_enabled(),
             merge_failure_block_threshold: default_merge_failure_block_threshold(),
             gc_min_free_gb: 0,
+            snapshot_interval_ops: default_snapshot_interval_ops(),
         }
     }
 }
@@ -564,6 +576,7 @@ impl ProjectConfig {
                         status_push_enabled: default_status_push_enabled(),
                         merge_failure_block_threshold: default_merge_failure_block_threshold(),
                         gc_min_free_gb: 0,
+                        snapshot_interval_ops: default_snapshot_interval_ops(),
                     };
                     validate_agents(&config.agent)?;
                     return Ok(config);
@@ -607,6 +620,7 @@ impl ProjectConfig {
                         status_push_enabled: default_status_push_enabled(),
                         merge_failure_block_threshold: default_merge_failure_block_threshold(),
                         gc_min_free_gb: 0,
+                        snapshot_interval_ops: default_snapshot_interval_ops(),
                     };
                     validate_agents(&config.agent)?;
                     Ok(config)
@@ -638,6 +652,7 @@ impl ProjectConfig {
                         status_push_enabled: default_status_push_enabled(),
                         merge_failure_block_threshold: default_merge_failure_block_threshold(),
                         gc_min_free_gb: 0,
+                        snapshot_interval_ops: default_snapshot_interval_ops(),
                     })
                 }
             }

server/src/config/tests.rs

@@ -11,6 +11,21 @@ fn default_config_when_missing() {
     assert!(config.component.is_empty());
 }
 
+#[test]
+fn snapshot_interval_ops_defaults_to_5000() {
+    let config = ProjectConfig::default();
+    assert_eq!(config.snapshot_interval_ops, 5000);
+}
+
+#[test]
+fn snapshot_interval_ops_overridable() {
+    let toml_str = r#"
+snapshot_interval_ops = 250
+"#;
+    let config = ProjectConfig::parse(toml_str).unwrap();
+    assert_eq!(config.snapshot_interval_ops, 250);
+}
+
 #[test]
 fn parse_multi_agent_toml() {
     let toml_str = r#"

server/src/crdt_state/mod.rs

@@ -49,7 +49,7 @@ pub use read::{
     read_all_items, read_item, tombstoned_ids,
 };
 pub(crate) use state::flush_persistence;
-pub use state::{init, subscribe};
+pub use state::{checkpoint_on_shutdown, init, subscribe};
 pub use types::{
     ActiveAgentCrdt, ActiveAgentView, AgentThrottleCrdt, AgentThrottleView, CrdtEvent, EpicId,
     EventLogEntryCrdt, GatewayConfigCrdt, GatewayProjectCrdt, GatewayProjectView, LlmSessionCrdt,

server/src/crdt_state/state/init.rs

@@ -24,7 +24,7 @@ use super::indices::{
     rebuild_index, rebuild_llm_session_index, rebuild_merge_job_index, rebuild_node_index,
     rebuild_test_job_index, rebuild_token_index,
 };
-use super::statics::{ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK};
+use super::statics::{self, ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK};
 use super::{CRDT_STATE, CrdtState};
 use crate::slog;
 
@@ -41,8 +41,11 @@ pub(crate) enum PersistMsg {
 /// Opens the SQLite database, loads or creates a node keypair, replays any
 /// persisted ops to reconstruct state, and spawns a background persistence
 /// task.  Safe to call only once; subsequent calls are no-ops.
+///
+/// `snapshot_interval_ops` (story 1249) is the number of ops applied between
+/// periodic snapshot checkpoints — see `ProjectConfig::snapshot_interval_ops`.
 #[allow(clippy::string_slice)] // op_id is hex::encode output (ASCII-only), &op_id[..12] is always valid
-pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
+pub async fn init(db_path: &Path, snapshot_interval_ops: usize) -> Result<(), sqlx::Error> {
     if CRDT_STATE.get().is_some() {
         return Ok(());
     }
@@ -52,6 +55,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
         .create_if_missing(true);
     let pool = SqlitePool::connect_with(options).await?;
     sqlx::migrate!("./migrations").run(&pool).await?;
+    let _ = statics::CRDT_POOL.set(pool.clone());
+    let _ = statics::SNAPSHOT_INTERVAL_OPS.set(snapshot_interval_ops);
 
     // Load or create the node keypair.
     let keypair = load_or_create_keypair(&pool).await?;
@@ -274,6 +279,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
                         slog!("[crdt] Failed to persist op {}: {e}", &op_id[..12]);
                     }
                     PERSIST_PENDING.fetch_sub(1, Ordering::Relaxed);
+
+                    maybe_periodic_checkpoint().await;
                 }
                 PersistMsg::Flush(reply) => {
                     // All ops queued before this message have already been processed.
@@ -379,21 +386,27 @@ async fn load_or_create_keypair(pool: &SqlitePool) -> Result<Ed25519KeyPair, sql
 /// and back up the database file.  Errors are logged but not propagated —
 /// a failed snapshot just means the next restart will do a full replay.
 async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_floor: u64) {
-    // Find the highest rowid currently in crdt_ops — ops with rowid <= this
-    // value are already captured in the snapshot.
-    let max_rowid: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(rowid), 0) FROM crdt_ops")
-        .fetch_one(pool)
-        .await
-        .unwrap_or(0);
-
-    let doc_ref = &crdt.doc;
-    let json = match serde_json::to_string(doc_ref) {
+    let json = match serde_json::to_string(&crdt.doc) {
         Ok(j) => j,
         Err(e) => {
             slog!("[crdt] Failed to serialize snapshot: {e}");
             return;
         }
     };
+    save_snapshot_json(pool, &json, lamport_floor).await;
+}
+
+/// Shared snapshot-writing body: back up the DB file and write `json` (an
+/// already-serialized `PipelineDoc`) into the `crdt_snapshot` table.  Used by
+/// both the one-time post-replay snapshot ([`save_snapshot`]) and periodic /
+/// shutdown checkpoints ([`checkpoint_now`]) added by story 1249.
+async fn save_snapshot_json(pool: &SqlitePool, json: &str, at_seq: u64) {
+    // Find the highest rowid currently in crdt_ops — ops with rowid <= this
+    // value are already captured in the snapshot.
+    let max_rowid: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(rowid), 0) FROM crdt_ops")
+        .fetch_one(pool)
+        .await
+        .unwrap_or(0);
 
     let json_len = json.len();
     let now = chrono::Utc::now().to_rfc3339();
@@ -412,9 +425,9 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
         "INSERT OR REPLACE INTO crdt_snapshot (id, at_seq, max_rowid, state_json, created_at) \
          VALUES (1, ?1, ?2, ?3, ?4)",
     )
-    .bind(lamport_floor as i64)
+    .bind(at_seq as i64)
     .bind(max_rowid)
-    .bind(&json)
+    .bind(json)
     .bind(&now)
     .execute(pool)
     .await;
@@ -422,7 +435,7 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
     match result {
         Ok(_) => slog!(
             "[crdt] Snapshot saved: at_seq={}, max_rowid={}, json={}B",
-            lamport_floor,
+            at_seq,
             max_rowid,
             json_len
         ),
@@ -430,6 +443,69 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
     }
 }
 
+/// Pure threshold decision for periodic checkpoints (story 1249 AC2/AC3).
+///
+/// Returns `true` only when at least one op has been applied since the last
+/// checkpoint AND that count has reached `threshold`. The `ops_since_last >
+/// 0` guard is what makes AC3 (skip when nothing changed) hold even for a
+/// `threshold` of 0 or 1.
+fn should_checkpoint(ops_since_last: usize, threshold: usize) -> bool {
+    ops_since_last > 0 && ops_since_last >= threshold
+}
+
+/// Called after every op the persistence task writes. Triggers a checkpoint
+/// once `OPS_SINCE_SNAPSHOT` reaches the configured `SNAPSHOT_INTERVAL_OPS`
+/// (story 1249 AC1/AC2).
+async fn maybe_periodic_checkpoint() {
+    let threshold = statics::SNAPSHOT_INTERVAL_OPS
+        .get()
+        .copied()
+        .unwrap_or(crate::config::default_snapshot_interval_ops());
+    let ops_since_last = statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed);
+    if should_checkpoint(ops_since_last, threshold) {
+        checkpoint_now("periodic").await;
+    }
+}
+
+/// Take a snapshot of the current CRDT state, unless no ops have been
+/// applied since the last checkpoint (story 1249 AC3). Resets
+/// `OPS_SINCE_SNAPSHOT` back to zero afterwards. `reason` is only used for
+/// the log line (e.g. `"periodic"` or `"shutdown"`).
+async fn checkpoint_now(reason: &str) {
+    if statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed) == 0 {
+        return;
+    }
+    let Some(pool) = statics::CRDT_POOL.get() else {
+        return;
+    };
+    let Some(state_mutex) = super::get_crdt() else {
+        return;
+    };
+    let json = {
+        let Ok(state) = state_mutex.lock() else {
+            return;
+        };
+        match serde_json::to_string(&state.crdt.doc) {
+            Ok(j) => j,
+            Err(e) => {
+                slog!("[crdt] Failed to serialize {reason} checkpoint: {e}");
+                return;
+            }
+        }
+    };
+    let at_seq = statics::LAST_SEQ.load(Ordering::Relaxed);
+    save_snapshot_json(pool, &json, at_seq).await;
+    statics::OPS_SINCE_SNAPSHOT.store(0, Ordering::Relaxed);
+    slog!("[crdt] Checkpoint ({reason}) complete");
+}
+
+/// Take a final checkpoint on clean shutdown (story 1249 AC1), if any ops
+/// have been applied since the last one. Safe to call even if the CRDT
+/// layer was never initialised — it's a no-op in that case.
+pub async fn checkpoint_on_shutdown() {
+    checkpoint_now("shutdown").await;
+}
+
 /// Build the abort message printed when a CRDT snapshot fails to deserialize
 /// into the current schema.  `deserialize_error` is the `Display` text of the
 /// `serde_json::Error`, which names the missing/mismatched field; it is
@@ -457,7 +533,25 @@ fn pool_path(pool: &SqlitePool) -> Option<String> {
 
 #[cfg(test)]
 mod tests {
-    use super::snapshot_load_abort_message;
+    use super::{should_checkpoint, snapshot_load_abort_message};
+
+    /// AC3: a checkpoint must never fire when nothing has changed, no matter
+    /// how low the configured threshold is.
+    #[test]
+    fn should_checkpoint_false_when_no_ops_applied() {
+        assert!(!should_checkpoint(0, 5000));
+        assert!(!should_checkpoint(0, 1));
+        assert!(!should_checkpoint(0, 0));
+    }
+
+    /// AC2: the threshold is whatever N is configured to, not a hardcoded value.
+    #[test]
+    fn should_checkpoint_respects_configured_threshold() {
+        assert!(!should_checkpoint(4999, 5000));
+        assert!(should_checkpoint(5000, 5000));
+        assert!(should_checkpoint(5001, 5000));
+        assert!(should_checkpoint(1, 1));
+    }
 
     /// The abort message must name the failing field (via the serde error
     /// text), explain that the ops log is intact, and give the exact

server/src/crdt_state/state/mod.rs

@@ -26,8 +26,8 @@ mod tests;
 
 // ── Re-exports for crdt_state siblings ──────────────────────────────
 
-pub use init::init;
 pub(crate) use init::{PersistMsg, flush_persistence};
+pub use init::{checkpoint_on_shutdown, init};
 
 /// Subscribe to CRDT state-transition events.
 ///

server/src/crdt_state/state/statics.rs

@@ -10,10 +10,11 @@
 //! tests do not share `ALL_OPS` — preventing one test's `apply_compaction`
 //! from pruning another test's freshly-written ops.
 
-use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::{AtomicU64, AtomicUsize};
 use std::sync::{Mutex, OnceLock};
 
 use bft_json_crdt::json_crdt::SignedOp;
+use sqlx::SqlitePool;
 use tokio::sync::broadcast;
 
 use super::super::VectorClock;
@@ -46,6 +47,31 @@ pub(crate) static ALL_OPS: OnceLock<Mutex<Vec<String>>> = OnceLock::new();
 /// re-parsing all ops when a peer requests `our_vector_clock()`.
 pub(crate) static VECTOR_CLOCK: OnceLock<Mutex<VectorClock>> = OnceLock::new();
 
+/// The CRDT SQLite pool, stashed here so periodic and shutdown checkpoints
+/// (story 1249) can reuse it without threading it through every call site
+/// that might trigger one. Set once, at the end of `init::init()`.
+pub(crate) static CRDT_POOL: OnceLock<SqlitePool> = OnceLock::new();
+
+/// Number of ops between periodic snapshot checkpoints (story 1249).
+/// Set once, from `ProjectConfig::snapshot_interval_ops`, at the end of
+/// `init::init()`. Falls back to `config::default_snapshot_interval_ops()`
+/// (5000) if `init()` hasn't set it yet.
+pub(crate) static SNAPSHOT_INTERVAL_OPS: OnceLock<usize> = OnceLock::new();
+
+/// Count of ops applied (locally created or received from sync peers) since
+/// the last snapshot checkpoint. Incremented in [`track_op`]; reset to zero
+/// after a checkpoint is taken. A checkpoint is skipped when this is zero
+/// (story 1249 AC3) so restarting the server or an idle period never writes
+/// a redundant snapshot.
+pub(crate) static OPS_SINCE_SNAPSHOT: AtomicUsize = AtomicUsize::new(0);
+
+/// Highest op sequence number seen since startup, updated in [`track_op`].
+/// Used as the informational `at_seq` value for periodic/shutdown
+/// checkpoints (the replay-boundary correctness depends only on
+/// `max_rowid`, computed fresh from SQLite at snapshot time — this is just
+/// for accurate logging/debugging).
+pub(crate) static LAST_SEQ: AtomicU64 = AtomicU64::new(0);
+
 #[cfg(test)]
 thread_local! {
     /// Per-thread op journal for test isolation.  Each test thread sees its
@@ -107,4 +133,6 @@ pub(in crate::crdt_state) fn track_op(signed: &SignedOp, json: String) {
         let author_hex = hex::encode(&signed.author());
         *clock.entry(author_hex).or_insert(0) += 1;
     }
+    OPS_SINCE_SNAPSHOT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    LAST_SEQ.fetch_max(signed.inner.seq, std::sync::atomic::Ordering::Relaxed);
 }

server/src/gateway/mod.rs

@@ -94,7 +94,9 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> {
 
     // Initialise the CRDT so gateway_config.active_project is persisted across restarts.
     let crdt_db = config_dir.join("gateway.db");
-    if let Err(e) = crate::crdt_state::init(&crdt_db).await {
+    if let Err(e) =
+        crate::crdt_state::init(&crdt_db, crate::config::default_snapshot_interval_ops()).await
+    {
         crate::slog!(
             "[gateway] Warning: CRDT init failed ({e}); active-project selection will not persist"
         );
@@ -146,6 +148,13 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> {
         .run(route)
         .await;
 
+    // Story 1249: take a final CRDT snapshot on clean shutdown (mirrors main.rs).
+    crate::crdt_state::checkpoint_on_shutdown().await;
+
+    // Story 1249: take a final CRDT snapshot on clean shutdown so the next
+    // restart's replayed tail stays bounded.
+    crate::crdt_state::checkpoint_on_shutdown().await;
+
     // Best-effort shutdown notification: signal the Matrix bot so it can post
     // "going offline" before the process exits.  Mirror of main.rs:346.
     {

server/src/main.rs

@@ -491,6 +491,11 @@ async fn main() -> Result<(), std::io::Error> {
 
     let result = Server::new(TcpListener::bind(&addr)).run(app).await;
 
+    // Story 1249: take a final CRDT snapshot on clean shutdown so the next
+    // restart's replayed tail stays bounded, even if fewer than
+    // `snapshot_interval_ops` ops have accumulated since the last periodic one.
+    crate::crdt_state::checkpoint_on_shutdown().await;
+
     // ── Shutdown notifications (best-effort) ──────────────────────────────────
     startup::bots::notify_shutdown(&bot_ctxs).await;
 

server/src/startup/project.rs

@@ -338,7 +338,15 @@ pub(crate) async fn init_subsystems(app_state: &Arc<SessionState>, cwd: &Path, i
             let huskies_dir = db_path.parent().unwrap_or(db_path);
             migrate_json_stores_to_sqlite(huskies_dir).await;
         }
-        if let Err(e) = crdt_state::init(db_path).await {
+        let snapshot_interval_ops = app_state
+            .project_root
+            .lock()
+            .unwrap()
+            .as_ref()
+            .and_then(|root| config::ProjectConfig::load(root).ok())
+            .map(|cfg| cfg.snapshot_interval_ops)
+            .unwrap_or_else(config::default_snapshot_interval_ops);
+        if let Err(e) = crdt_state::init(db_path, snapshot_interval_ops).await {
             crate::slog!("[crdt] Failed to initialise CRDT state layer: {e}");
         } else {
             crdt_state::migrate_names_from_slugs();

server/src/worktree/cleanup.rs

@@ -233,6 +233,7 @@ mod tests {
             status_push_enabled: true,
             merge_failure_block_threshold: 3,
             gc_min_free_gb: 0,
+            snapshot_interval_ops: 5000,
         }
     }
 

server/src/worktree/create.rs

@@ -272,6 +272,7 @@ mod tests {
             status_push_enabled: true,
             merge_failure_block_threshold: 3,
             gc_min_free_gb: 0,
+            snapshot_interval_ops: 5000,
         }
     }
 

server/src/worktree/remove.rs

@@ -130,6 +130,7 @@ mod tests {
             status_push_enabled: true,
             merge_failure_block_threshold: 3,
             gc_min_free_gb: 0,
+            snapshot_interval_ops: 5000,
         }
     }
 

server/src/worktree/sweep.rs

@@ -154,6 +154,7 @@ mod tests {
             status_push_enabled: true,
             merge_failure_block_threshold: 3,
             gc_min_free_gb: 0,
+            snapshot_interval_ops: 5000,
         }
     }