db: Add dual-mode TursoDb (local/sync) and wire sce sync push|pull

Extend TursoDb<M> to support Turso Cloud sync mode via
turso::sync::Builder::new_remote() when both SCE_SYNC_URL and
SCE_SYNC_TOKEN are set. Sync-mode methods (push/pull/checkpoint/stats)
are explicit — no auto-push from execute().

Wire a user-invocable sce sync push|pull CLI command behind the new
sync module (cli/src/services/sync/), connected through clap schema,
command registry, and parse runtime dispatch.

Remove the unused SYNC_PUSH_THRESHOLD_ENV_KEY constant and write-counter
auto-push approach from the earlier design.

Co-authored-by: SCE <sce@crocoder.dev>

Author: davidabram

cli/src/cli_schema.rs

@@ -35,6 +35,10 @@ pub const VERSION_CLAP_ABOUT: &str = "Print deterministic runtime version metada
 pub const VERSION_TOP_LEVEL_PURPOSE: &str = "Print deterministic runtime version metadata";
 pub const VERSION_SHOW_IN_TOP_LEVEL_HELP: bool = true;
 
+pub const SYNC_CLAP_ABOUT: &str = "Push or pull changes to/from Turso Cloud";
+pub const SYNC_TOP_LEVEL_PURPOSE: &str = "Sync local database with Turso Cloud (push/pull)";
+pub const SYNC_SHOW_IN_TOP_LEVEL_HELP: bool = true;
+
 pub const COMPLETION_CLAP_ABOUT: &str = "Generate deterministic shell completion scripts";
 pub const COMPLETION_TOP_LEVEL_PURPOSE: &str = "Generate deterministic shell completion scripts";
 pub const COMPLETION_SHOW_IN_TOP_LEVEL_HELP: bool = true;
@@ -75,6 +79,11 @@ pub const TOP_LEVEL_COMMANDS: &[TopLevelCommandMetadata] = &[
         purpose: COMPLETION_TOP_LEVEL_PURPOSE,
         show_in_top_level_help: COMPLETION_SHOW_IN_TOP_LEVEL_HELP,
     },
+    TopLevelCommandMetadata {
+        name: crate::services::sync::NAME,
+        purpose: SYNC_TOP_LEVEL_PURPOSE,
+        show_in_top_level_help: SYNC_SHOW_IN_TOP_LEVEL_HELP,
+    },
 ];
 
 #[derive(Parser, Debug)]
@@ -190,6 +199,12 @@ pub enum Commands {
         #[arg(long, value_enum)]
         shell: CompletionShell,
     },
+
+    #[command(about = SYNC_CLAP_ABOUT, hide = !SYNC_SHOW_IN_TOP_LEVEL_HELP)]
+    Sync {
+        #[command(subcommand)]
+        subcommand: SyncSubcommand,
+    },
 }
 
 #[derive(Subcommand, Debug, Clone, PartialEq, Eq)]
@@ -273,6 +288,15 @@ pub enum HooksSubcommand {
     DiffTrace,
 }
 
+#[derive(Subcommand, Debug, Clone, PartialEq, Eq)]
+pub enum SyncSubcommand {
+    #[command(about = "Push local changes to Turso Cloud")]
+    Push,
+
+    #[command(about = "Pull remote changes from Turso Cloud")]
+    Pull,
+}
+
 #[derive(ValueEnum, Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub enum OutputFormat {
     #[default]

cli/src/services/command_registry.rs

@@ -104,6 +104,7 @@ pub fn build_default_registry() -> CommandRegistry {
         "completion",
         crate::services::completion::command::make_completion_command,
     );
+    registry.register("sync", crate::services::sync::command::make_sync_command);
     registry
 }
 

cli/src/services/config/mod.rs

@@ -44,14 +44,10 @@ pub(crate) const ENV_ATTRIBUTION_HOOKS_ENABLED: &str = "SCE_ATTRIBUTION_HOOKS_EN
 // Sync (Turso Cloud) env var keys.
 // When SCE_SYNC_URL and SCE_SYNC_TOKEN are both set, TursoDb<M> opens in sync
 // (remote-backed) mode. When either is absent, local-only mode is used.
-// SCE_SYNC_PUSH_THRESHOLD controls how many writes accumulate before an
-// automatic push (defaults to 10 when unset).
 #[allow(dead_code)]
 pub(crate) const SYNC_URL_ENV_KEY: &str = "SCE_SYNC_URL";
 #[allow(dead_code)]
 pub(crate) const SYNC_TOKEN_ENV_KEY: &str = "SCE_SYNC_TOKEN";
-#[allow(dead_code)]
-pub(crate) const SYNC_PUSH_THRESHOLD_ENV_KEY: &str = "SCE_SYNC_PUSH_THRESHOLD";
 
 const WORKOS_CLIENT_ID_ENV: &str = "WORKOS_CLIENT_ID";
 const WORKOS_CLIENT_ID_BAKED_DEFAULT: &str = "client_sce_default";

cli/src/services/db/mod.rs

@@ -139,10 +139,20 @@ fn sentence_case(value: &str) -> String {
 /// Wraps a Turso connection with a tokio current-thread runtime so callers can
 /// use synchronous `execute`/`query` methods while the underlying Turso API
 /// remains async.
+///
+/// Supports two modes:
+/// - **Local mode** (default): opens a plain Turso file database.
+/// - **Sync mode**: opens a Turso Cloud synced database when both
+///   `SCE_SYNC_URL` and `SCE_SYNC_TOKEN` environment variables are set.
+///   Sync operations (`push`, `pull`, `checkpoint`, `stats`) are available
+///   through explicit methods or the `sce sync` CLI command.
 #[allow(dead_code)]
 pub struct TursoDb<M: DbSpec> {
     conn: turso::Connection,
     runtime: tokio::runtime::Runtime,
+    /// Optional sync database handle for Turso Cloud operations (push, pull,
+    /// checkpoint, stats). `None` when the database is in local-only mode.
+    sync_db: Option<turso::sync::Database>,
     spec: PhantomData<fn() -> M>,
 }
 
@@ -152,6 +162,9 @@ impl<M: DbSpec> TursoDb<M> {
     ///
     /// Parent directories are created automatically. Migrations are run after
     /// the database connection is established.
+    ///
+    /// When both `SCE_SYNC_URL` and `SCE_SYNC_TOKEN` are set, the database opens
+    /// in sync (Turso Cloud) mode. Otherwise, local-only mode is used.
     pub fn new() -> Result<Self> {
         let db_name = M::db_name();
         let db_path = M::db_path().with_context(|| format!("failed to resolve {db_name} path"))?;
@@ -173,33 +186,70 @@ impl<M: DbSpec> TursoDb<M> {
                 format!("failed to create {db_name} tokio runtime. Try: rerun the command; if the issue persists, verify the local Tokio runtime environment.")
             })?;
 
-        let conn = runtime.block_on(async {
-            let path_str = db_path.to_str().ok_or_else(|| {
-                anyhow::anyhow!("invalid UTF-8 in database path: {}", db_path.display())
-            })?;
-            let db = turso::Builder::new_local(path_str)
-                .build()
-                .await
-                .map_err(|e| {
-                    anyhow::anyhow!(
-                        "failed to open {db_name} database at {}: {e}",
-                        db_path.display()
-                    )
-                })?;
-            db.connect()
-                .map_err(|e| anyhow::anyhow!("failed to connect to {db_name} database: {e}"))
+        let sync_url = std::env::var(crate::services::config::SYNC_URL_ENV_KEY).ok();
+        let sync_token = std::env::var(crate::services::config::SYNC_TOKEN_ENV_KEY).ok();
+
+        let path_str = db_path.to_str().ok_or_else(|| {
+            anyhow::anyhow!("invalid UTF-8 in database path: {}", db_path.display())
         })?;
 
-        let db = Self {
-            conn,
-            runtime,
-            spec: PhantomData,
-        };
+        if let (Some(url), Some(token)) = (sync_url, sync_token) {
+            let (conn, sync_db) = runtime.block_on(async {
+                let sync_db = turso::sync::Builder::new_remote(path_str)
+                    .with_remote_url(url)
+                    .with_auth_token(token)
+                    .build()
+                    .await
+                    .map_err(|e| {
+                        anyhow::anyhow!(
+                            "failed to open {db_name} synced database at {}: {e}",
+                            db_path.display()
+                        )
+                    })?;
+                let conn = sync_db.connect().await.map_err(|e| {
+                    anyhow::anyhow!("failed to connect to {db_name} synced database: {e}")
+                })?;
+                Ok::<_, anyhow::Error>((conn, sync_db))
+            })?;
+
+            let db = Self {
+                conn,
+                runtime,
+                sync_db: Some(sync_db),
+                spec: PhantomData,
+            };
+
+            db.run_migrations()
+                .with_context(|| format!("failed to run {db_name} migrations"))?;
+
+            Ok(db)
+        } else {
+            let conn = runtime.block_on(async {
+                let db = turso::Builder::new_local(path_str)
+                    .build()
+                    .await
+                    .map_err(|e| {
+                        anyhow::anyhow!(
+                            "failed to open {db_name} database at {}: {e}",
+                            db_path.display()
+                        )
+                    })?;
+                db.connect()
+                    .map_err(|e| anyhow::anyhow!("failed to connect to {db_name} database: {e}"))
+            })?;
 
-        db.run_migrations()
-            .with_context(|| format!("failed to run {db_name} migrations"))?;
+            let db = Self {
+                conn,
+                runtime,
+                sync_db: None,
+                spec: PhantomData,
+            };
 
-        Ok(db)
+            db.run_migrations()
+                .with_context(|| format!("failed to run {db_name} migrations"))?;
+
+            Ok(db)
+        }
     }
 
     /// Execute a SQL statement that does not return rows.
@@ -210,6 +260,10 @@ impl<M: DbSpec> TursoDb<M> {
     ///
     /// # Returns
     /// Number of rows affected.
+    ///
+    /// Sync is not triggered automatically — callers use the explicit `push()`
+    /// or `pull()` methods (or the `sce sync` CLI command) to sync with the
+    /// remote.
     pub fn execute(&self, sql: &str, params: impl turso::params::IntoParams) -> Result<u64> {
         self.runtime.block_on(async {
             self.conn
@@ -219,6 +273,65 @@ impl<M: DbSpec> TursoDb<M> {
         })
     }
 
+    /// Push local changes to the remote (sync mode only).
+    ///
+    /// In local mode this is a no-op that returns `Ok(())`.
+    pub fn push(&self) -> Result<()> {
+        match self.sync_db {
+            Some(ref sync_db) => self
+                .runtime
+                .block_on(sync_db.push())
+                .map_err(|e| anyhow::anyhow!("{} sync push failed: {e}", M::db_name())),
+            None => Ok(()),
+        }
+    }
+
+    /// Pull remote changes (sync mode only).
+    ///
+    /// Returns `true` if any changes were applied to the local database.
+    /// In local mode this is a no-op that returns `Ok(false)`.
+    pub fn pull(&self) -> Result<bool> {
+        match self.sync_db {
+            Some(ref sync_db) => self
+                .runtime
+                .block_on(sync_db.pull())
+                .map_err(|e| anyhow::anyhow!("{} sync pull failed: {e}", M::db_name())),
+            None => Ok(false),
+        }
+    }
+
+    /// Force a WAL checkpoint (sync mode only).
+    ///
+    /// In local mode this is a no-op that returns `Ok(())`.
+    pub fn checkpoint(&self) -> Result<()> {
+        match self.sync_db {
+            Some(ref sync_db) => self
+                .runtime
+                .block_on(sync_db.checkpoint())
+                .map_err(|e| anyhow::anyhow!("{} sync checkpoint failed: {e}", M::db_name())),
+            None => Ok(()),
+        }
+    }
+
+    /// Retrieve sync statistics (sync mode only).
+    ///
+    /// Returns `None` in local mode.
+    pub fn stats(&self) -> Result<Option<turso::sync::DatabaseSyncStats>> {
+        match self.sync_db {
+            Some(ref sync_db) => self
+                .runtime
+                .block_on(sync_db.stats())
+                .map(Some)
+                .map_err(|e| anyhow::anyhow!("{} sync stats failed: {e}", M::db_name())),
+            None => Ok(None),
+        }
+    }
+
+    /// Returns `true` if the database is in sync (Turso Cloud) mode.
+    pub fn is_sync_mode(&self) -> bool {
+        self.sync_db.is_some()
+    }
+
     /// Execute a SQL query that returns rows.
     ///
     /// # Arguments

cli/src/services/mod.rs

@@ -23,5 +23,6 @@ pub mod resilience;
 pub mod security;
 pub mod setup;
 pub mod style;
+pub mod sync;
 pub mod token_storage;
 pub mod version;

cli/src/services/parse/command_runtime.rs

@@ -139,6 +139,10 @@ fn render_missing_subcommand_help(args: &[String]) -> Option<RuntimeCommandHandl
             name: services::config::NAME.to_string(),
             text: cli_schema::render_help_for_path(&[services::config::NAME])?,
         })),
+        services::sync::NAME => Some(Box::new(services::help::command::HelpTextCommand {
+            name: services::sync::NAME.to_string(),
+            text: cli_schema::render_help_for_path(&[services::sync::NAME])?,
+        })),
         _ => None,
     }
 }
@@ -244,9 +248,24 @@ fn convert_clap_command(
                 },
             }))
         }
+        cli_schema::Commands::Sync { ref subcommand } => convert_sync_subcommand(subcommand),
     }
 }
 
+#[allow(clippy::unnecessary_wraps)]
+fn convert_sync_subcommand(
+    subcommand: &cli_schema::SyncSubcommand,
+) -> Result<RuntimeCommandHandle, ClassifiedError> {
+    let subcommand = match subcommand {
+        cli_schema::SyncSubcommand::Push => services::sync::SyncSubcommand::Push,
+        cli_schema::SyncSubcommand::Pull => services::sync::SyncSubcommand::Pull,
+    };
+
+    Ok(Box::new(services::sync::command::SyncCommand {
+        subcommand,
+    }))
+}
+
 #[allow(clippy::unnecessary_wraps, clippy::needless_pass_by_value)]
 fn convert_auth_subcommand(
     subcommand: cli_schema::AuthSubcommand,

cli/src/services/sync/command.rs

@@ -0,0 +1,28 @@
+use std::borrow::Cow;
+
+use crate::app::AppContext;
+use crate::services::command_registry::{RuntimeCommand, RuntimeCommandHandle};
+use crate::services::error::ClassifiedError;
+use crate::services::sync;
+
+pub struct SyncCommand {
+    pub subcommand: sync::SyncSubcommand,
+}
+
+impl RuntimeCommand for SyncCommand {
+    fn name(&self) -> Cow<'_, str> {
+        Cow::Borrowed(sync::NAME)
+    }
+
+    fn execute(&self, _context: &AppContext) -> Result<String, ClassifiedError> {
+        sync::run_sync(self.subcommand).map_err(|error| ClassifiedError::runtime(error.to_string()))
+    }
+}
+
+/// Construct a `SyncCommand` with the Push subcommand (used by the registry).
+#[allow(dead_code)]
+pub fn make_sync_command() -> RuntimeCommandHandle {
+    Box::new(SyncCommand {
+        subcommand: sync::SyncSubcommand::Push,
+    })
+}

cli/src/services/sync/mod.rs

@@ -0,0 +1,43 @@
+pub mod command;
+
+use anyhow::{Context, Result};
+
+use crate::services::agent_trace_db::AgentTraceDb;
+
+pub const NAME: &str = "sync";
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum SyncSubcommand {
+    Push,
+    Pull,
+}
+
+/// Perform a push or pull sync operation on the Agent Trace database.
+///
+/// Opens the Agent Trace DB (which auto-detects sync mode from env vars).
+/// If the database is in sync mode, executes the requested operation.
+/// If in local mode, returns an error indicating sync is not configured.
+pub fn run_sync(subcommand: SyncSubcommand) -> Result<String> {
+    let db = AgentTraceDb::new().context("failed to open Agent Trace DB for sync")?;
+
+    if !db.is_sync_mode() {
+        return Err(anyhow::anyhow!(
+            "Sync is not configured. Set SCE_SYNC_URL and SCE_SYNC_TOKEN to enable sync mode."
+        ));
+    }
+
+    match subcommand {
+        SyncSubcommand::Push => {
+            db.push().with_context(|| "sync push failed")?;
+            Ok("Pushed local changes to remote.".to_string())
+        }
+        SyncSubcommand::Pull => {
+            let has_changes = db.pull().with_context(|| "sync pull failed")?;
+            if has_changes {
+                Ok("Pulled remote changes.".to_string())
+            } else {
+                Ok("No remote changes to pull.".to_string())
+            }
+        }
+    }
+}