feat(octavo): move the hub protos into atuin-octavo and generate its client

.codespellrc

@@ -1,6 +1,6 @@
 [codespell]
 # Ref: https://github.com/codespell-project/codespell#using-a-config-file
-skip = .git*,*.lock,.codespellrc,CODE_OF_CONDUCT.md,CONTRIBUTORS,./vendor/axoasset,./crates/atuin-api-client/proto/buf
+skip = .git*,*.lock,.codespellrc,CODE_OF_CONDUCT.md,CONTRIBUTORS,./vendor/axoasset,./crates/atuin-octavo/proto/buf
 check-hidden = true
 # ignore-regex = 
 ignore-words-list = crate,ratatui,inbetween,iterm,fo,brunch,ser,writet

Cargo.lock

@@ -277,6 +277,7 @@ dependencies = [
  "atuin-dotfiles",
  "atuin-history",
  "atuin-kv",
+ "atuin-octavo",
  "atuin-pty-proxy",
  "atuin-scripts",
  "atuin-vt100",
@@ -406,10 +407,6 @@ dependencies = [
  "yaml-rust2",
 ]
 
-[[package]]
-name = "atuin-api-client"
-version = "18.23.0"
-
 [[package]]
 name = "atuin-client"
 version = "18.23.0"
@@ -553,6 +550,7 @@ dependencies = [
  "atuin-common",
  "atuin-domain",
  "atuin-history",
+ "atuin-octavo",
  "dashmap",
  "derive_more",
  "divan",
@@ -682,6 +680,32 @@ dependencies = [
  "typed-builder",
 ]
 
+[[package]]
+name = "atuin-octavo"
+version = "18.23.0"
+dependencies = [
+ "atuin-client",
+ "atuin-common",
+ "atuin-domain",
+ "eyre",
+ "futures",
+ "hyper-tls",
+ "hyper-util",
+ "native-tls",
+ "prost",
+ "prost-types",
+ "protox",
+ "secrecy",
+ "sqlx",
+ "thiserror 2.0.21",
+ "tonic",
+ "tonic-prost",
+ "tonic-prost-build",
+ "url",
+ "uuid",
+ "xxhash-rust",
+]
+
 [[package]]
 name = "atuin-pty-proxy"
 version = "18.23.0"

Cargo.toml

@@ -21,6 +21,7 @@ atuin-domain = { path = "crates/atuin-domain", version = "18.23.0" }
 atuin-dotfiles = { path = "crates/atuin-dotfiles", version = "18.23.0" }
 atuin-history = { path = "crates/atuin-history", version = "18.23.0" }
 atuin-kv = { path = "crates/atuin-kv", version = "18.23.0" }
+atuin-octavo = { path = "crates/atuin-octavo", version = "18.23.0" }
 atuin-pty-proxy = { path = "crates/atuin-pty-proxy", version = "18.23.0", default-features = false }
 atuin-scripts = { path = "crates/atuin-scripts", version = "18.23.0" }
 atuin-server = { path = "crates/atuin-server", version = "18.23.0" }
@@ -73,6 +74,7 @@ getrandom = "0.4"
 glob-match = "0.2"
 http = "1"
 humantime = "2"
+hyper-tls = "0.6"
 hyper-util = "0.1"
 i18n-embed = { version = "0.16", features = ["fluent-system"] }
 i18n-embed-fl = "0.10"
@@ -92,6 +94,7 @@ metrics = "0.24"
 metrics-exporter-prometheus = { version = "0.18", default-features = false }
 minijinja = "2"
 minspan = "0.1"
+native-tls = "0.2"
 norm = { version = "0.1", features = ["fzf-v2"] }
 notify = "8"
 notify-debouncer-full = "0.7"

crates/atuin-api-client/Cargo.toml

@@ -1,15 +0,0 @@
-[package]
-name = "atuin-api-client"
-edition = "2024"
-description = "Low-level clients for the Atuin APIs"
-
-authors.workspace = true
-homepage.workspace = true
-license.workspace = true
-readme.workspace = true
-repository.workspace = true
-rust-version.workspace = true
-version.workspace = true
-
-[lints]
-workspace = true

crates/atuin-api-client/src/lib.rs

@@ -1 +0,0 @@
-#![deny(unsafe_code)]

crates/atuin-client/src/database.rs

@@ -501,6 +501,13 @@ impl Sqlite {
         Ok(res)
     }
 
+    #[instrument(level = "trace", skip_all, err)]
+    pub async fn active_ids(&self) -> Result<Vec<HistoryId>> {
+        db::query_scalar::<_, HistoryId>("select id from history where deleted_at is null")
+            .fetch_all(self.sqlite.pool())
+            .await
+    }
+
     #[instrument(level = "trace", skip_all, err)]
     pub async fn load_active(
         &self,

crates/atuin-client/src/meta.rs

@@ -25,6 +25,7 @@ const KEY_LAST_VERSION_CHECK: &str = "last_version_check_time";
 const KEY_LATEST_VERSION: &str = "latest_version";
 const KEY_SESSION: &str = "session";
 const KEY_HUB_SESSION: &str = "hub_session";
+const KEY_HUB_USER_ID: &str = "hub_user_id";
 const KEY_FILES_MIGRATED: &str = "files_migrated";
 
 const HUB_TOKEN_PREFIX: &str = "atapi_";
@@ -188,11 +189,68 @@ impl MetaStore {
     }
 
     pub async fn save_hub_session(&self, token: &SecretString) -> Result<()> {
-        self.set(KEY_HUB_SESSION, token.expose_secret()).await
+        let mut tx = self.sqlite.pool().begin().await?;
+        db::query(
+            "INSERT INTO meta (key, value, updated_at) VALUES (?1, ?2, strftime('%s', 'now'))
+             ON CONFLICT(key) DO UPDATE SET value = ?2, updated_at = strftime('%s', 'now')",
+        )
+        .bind(KEY_HUB_SESSION)
+        .bind(token.expose_secret())
+        .execute(&mut *tx)
+        .await?;
+        db::query("DELETE FROM meta WHERE key = ?1")
+            .bind(KEY_HUB_USER_ID)
+            .execute(&mut *tx)
+            .await?;
+        tx.commit().await?;
+
+        Ok(())
     }
 
     pub async fn delete_hub_session(&self) -> Result<()> {
-        self.delete(KEY_HUB_SESSION).await
+        db::query("DELETE FROM meta WHERE key IN (?1, ?2)")
+            .bind(KEY_HUB_SESSION)
+            .bind(KEY_HUB_USER_ID)
+            .execute(self.sqlite.pool())
+            .await?;
+
+        Ok(())
+    }
+
+    /// The id of the hub user `token` logs in, if `token` is still the hub session and
+    /// [`Self::save_hub_user_id`] recorded it.
+    pub async fn hub_user_id(&self, token: &SecretString) -> Result<Option<String>> {
+        let row: Option<(String,)> = db::query_as(
+            "SELECT user_id.value FROM meta AS user_id
+             JOIN meta AS session ON session.key = ?2 AND session.value = ?3
+             WHERE user_id.key = ?1",
+        )
+        .bind(KEY_HUB_USER_ID)
+        .bind(KEY_HUB_SESSION)
+        .bind(token.expose_secret())
+        .fetch_optional(self.sqlite.pool())
+        .await?;
+
+        Ok(row.map(|r| r.0))
+    }
+
+    /// Records that `token` logs in the hub user `user_id`, unless a login or logout replaced
+    /// `token` as the hub session since it was read.
+    pub async fn save_hub_user_id(&self, token: &SecretString, user_id: &str) -> Result<()> {
+        db::query(
+            "INSERT INTO meta (key, value, updated_at)
+             SELECT ?1, ?2, strftime('%s', 'now') WHERE EXISTS
+               (SELECT 1 FROM meta WHERE key = ?3 AND value = ?4)
+             ON CONFLICT(key) DO UPDATE SET value = ?2, updated_at = strftime('%s', 'now')",
+        )
+        .bind(KEY_HUB_USER_ID)
+        .bind(user_id)
+        .bind(KEY_HUB_SESSION)
+        .bind(token.expose_secret())
+        .execute(self.sqlite.pool())
+        .await?;
+
+        Ok(())
     }
 
     pub async fn hub_logged_in(&self) -> Result<bool> {

crates/atuin-client/src/settings.rs

@@ -540,6 +540,12 @@ pub struct PtyProxy {
     pub enabled: bool,
 }
 
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+pub struct Octavo {
+    pub enabled: bool,
+    pub endpoint: Option<Url>,
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct Search {
     /// The list of enabled filter modes, in order of priority.
@@ -1134,6 +1140,9 @@ pub struct Settings {
     #[serde(default)]
     pub output: OutputCapture,
 
+    #[serde(default)]
+    pub octavo: Octavo,
+
     #[serde(default)]
     pub search: Search,
 
@@ -1223,6 +1232,11 @@ impl Settings {
         Self::effective_data_dir().join("ai_harness_sessions.db")
     }
 
+    #[must_use]
+    pub fn octavo_queue_path() -> PathBuf {
+        Self::effective_data_dir().join("octavo.db")
+    }
+
     // -- Meta store: lazily initialized on first access --
 
     pub async fn meta_store() -> Result<&'static crate::meta::MetaStore> {
@@ -1628,6 +1642,8 @@ impl Settings {
             )?
             .set_default("no_mouse", false)?
             .set_default("pty_proxy.enabled", false)?
+            .set_default("octavo.enabled", false)?
+            .set_default("octavo.endpoint", None::<String>)?
             .add_source(Environment::with_prefix("atuin").prefix_separator("_").separator("__")))
     }
 

crates/atuin-daemon/Cargo.toml

@@ -11,12 +11,16 @@ repository.workspace = true
 rust-version.workspace = true
 version.workspace = true
 
+[features]
+octavo = ["dep:atuin-octavo"]
+
 [dependencies]
 async-stream = { workspace = true }
 atuin-client = { workspace = true, features = ["proto"] }
 atuin-common = { workspace = true, features = ["ai", "os", "proto", "sqlite"] }
 atuin-domain = { workspace = true }
 atuin-history = { workspace = true }
+atuin-octavo = { workspace = true, optional = true }
 dashmap = { workspace = true }
 derive_more = { workspace = true }
 easy-cast = { workspace = true }

crates/atuin-daemon/src/history_journal.rs

@@ -98,6 +98,7 @@ use tokio_stream::wrappers::BroadcastStream;
 use tracing::field::Empty;
 use tracing::{Instrument, Span};
 
+use crate::octavo::Octavo;
 use crate::output_capture::{CaptureError, GetOutputError, OutputCaptureEngine};
 use crate::search::SearchIndex;
 
@@ -212,6 +213,8 @@ pub struct HistoryJournal {
     /// Durable store for captured command output.
     output_capture: OutputCaptureEngine,
 
+    octavo: Octavo,
+
     /// Ids a [`Self::delete`] is currently tearing down, reference-counted across concurrent
     /// deletes. [`Self::register_command_output`] refuses these, so no capture can land between a
     /// delete's output removal and its record removal.
@@ -284,6 +287,7 @@ impl HistoryJournal {
         history_db: HistoryDatabase,
         search_index: Arc<tokio::sync::RwLock<SearchIndex>>,
         output_capture: OutputCaptureEngine,
+        octavo: Octavo,
     ) -> Self {
         const DEFAULT_LIFECYCLE_SHARDS: NonZeroUsize = NonZeroUsize::new(64).unwrap();
 
@@ -296,6 +300,7 @@ impl HistoryJournal {
             search_index,
             broadcast,
             output_capture,
+            octavo,
             deleting: DashMap::new(),
             lifecycle_mutex: AsyncShardedMutex::new(DEFAULT_LIFECYCLE_SHARDS),
         }
@@ -406,6 +411,7 @@ impl HistoryJournal {
         //              .read() and the subsequent .write() are completely discarded from the new
         //              index.
         self.search_index.read().await.add_history(&history);
+        self.octavo.push_history(&history, history_record_id).await;
 
         if self.broadcast.receiver_count() > 0 {
             let _ = self.broadcast.send(CmdEvent::Finished(history));
@@ -534,6 +540,9 @@ impl HistoryJournal {
             return Ok(deleted);
         }
 
+        // The tombstones commit the deletion, so Octavo hears of it even if a chunk below fails.
+        self.octavo.delete_history(&to_delete).await;
+
         // The tombstones were just written above, so replaying them through
         // `HistoryStore::build_all` would only read and decrypt them back into these same ids.
         // Chunked like `build_all` is: one transaction over the whole set would hold the history
@@ -699,8 +708,14 @@ mod tests {
         let history_store = HistoryStore::new(store, HostId(uuid_v7()), paseto_v4::Key::generate());
         let search_index = Arc::new(RwLock::new(SearchIndex::new(OrFilter::all())));
         let caps = CapClient::new("http://127.0.0.1:1".parse().unwrap(), reqwest::Client::new());
-        let journal =
-            HistoryJournal::new(caps, history_store, history_db, search_index, output_capture);
+        let journal = HistoryJournal::new(
+            caps,
+            history_store,
+            history_db,
+            search_index,
+            output_capture,
+            Octavo::nop(),
+        );
         (journal, tmp)
     }
 

crates/atuin-daemon/src/lib.rs

@@ -20,6 +20,7 @@ pub mod daemon;
 pub mod events;
 pub mod grpc;
 pub(crate) mod history_journal;
+mod octavo;
 mod output_capture;
 pub mod pidfile;
 pub mod search;
@@ -38,6 +39,7 @@ pub use history_journal::{
     CmdCancelError, CmdDeleteError, CmdEvent, CmdFinishError, CmdRebuildError, FinishedCmd,
     GetCmdInFlightError, HistoryJournal, RegisterOutputError,
 };
+pub use octavo::Octavo;
 pub use output_capture::{
     CaptureError, DeleteOutputError, GetOutputError, OutputCaptureEngine, OutputLine, OutputMatch,
 };
@@ -343,6 +345,7 @@ pub async fn boot(
         handle.history_db().clone(),
         search_index,
         output_capture,
+        Octavo::open(&settings, handle.clone()).await,
     ));
     let history_service = HistoryServer::new(grpc::HistoryService::new(journal, handle.clone()));
     let ai_session_service = AiSessionServer::new(grpc::AiSessionService::new(ai_session_capture));

crates/atuin-daemon/src/octavo/client/active/engine.rs

@@ -0,0 +1,203 @@
+use std::num::NonZeroU32;
+use std::ops::ControlFlow;
+use std::sync::Arc;
+use std::time::Duration;
+
+use atuin_client::history::HistoryId;
+use atuin_common::futures::Backoff;
+use atuin_octavo::hub::{HubCallError, HubClient, HubLogin};
+use atuin_octavo::queue::{PendingHistoryUpload, UploadQueue};
+use futures::StreamExt;
+use tokio::sync::Notify;
+use tokio::task::JoinHandle;
+
+use crate::daemon::DaemonHandle;
+
+const POLL_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const BACKOFF: Backoff = Backoff::Exponential {
+    initial: Duration::from_secs(5),
+    max: Duration::from_secs(30 * 60),
+    factor: NonZeroU32::new(2).unwrap(),
+};
+
+#[derive(Debug)]
+pub struct UploadEngine {
+    task: JoinHandle<()>,
+}
+
+impl UploadEngine {
+    pub fn spawn(
+        queue: UploadQueue,
+        hub: HubClient,
+        handle: DaemonHandle,
+        wake: Arc<Notify>,
+    ) -> Self {
+        let inner = EngineInner {
+            queue,
+            hub,
+            handle,
+            wake,
+        };
+
+        Self {
+            task: tokio::spawn(inner.run()),
+        }
+    }
+}
+
+impl Drop for UploadEngine {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+struct RetryLater;
+
+struct EngineInner {
+    queue: UploadQueue,
+    hub: HubClient,
+    handle: DaemonHandle,
+    wake: Arc<Notify>,
+}
+
+impl EngineInner {
+    async fn run(self) {
+        loop {
+            BACKOFF
+                .retry_forever(|| async {
+                    self.send_pending().await.map_or(ControlFlow::Continue(()), ControlFlow::Break)
+                })
+                .await;
+
+            tokio::select! {
+                () = self.wake.notified() => {}
+                () = tokio::time::sleep(POLL_INTERVAL) => {}
+            }
+        }
+    }
+
+    /// Sends the logged-in user's queued deletions, then, while Octavo is on, their queued
+    /// uploads. What other users queued waits for them to log in again.
+    async fn send_pending(&self) -> Result<(), RetryLater> {
+        let settings = self.handle.settings().await.clone();
+        let login = match self.hub.login(&settings).await {
+            Ok(login) => login,
+            Err(HubCallError::NotLoggedIn) => return Ok(()),
+            Err(err) => {
+                tracing::warn!(?err, "failed to log in to octavo; retrying later");
+                return Err(RetryLater);
+            }
+        };
+
+        self.delete_pending(&login).await?;
+        if !settings.octavo.enabled {
+            return Ok(());
+        }
+
+        let mut pending = self.queue.pending_history(login.user_id()).items();
+        while let Some(entry) = pending.next().await {
+            let entry = entry.map_err(|err| {
+                tracing::warn!(?err, "failed to read the octavo upload queue; retrying later");
+                RetryLater
+            })?;
+            self.upload(&login, entry).await?;
+        }
+
+        Ok(())
+    }
+
+    async fn upload(
+        &self,
+        login: &HubLogin,
+        PendingHistoryUpload {
+            history_id: id,
+            record_id,
+        }: PendingHistoryUpload,
+    ) -> Result<(), RetryLater> {
+        let history = self.handle.history_db().load(id).await.map_err(|err| {
+            tracing::warn!(?err, %id, "failed to load a queued history entry; retrying later");
+            RetryLater
+        })?;
+        let Some(history) = history.filter(|history| history.deleted_at.is_none()) else {
+            return self.dequeue(login, id).await;
+        };
+
+        match login.upload(history, record_id).await {
+            Ok(()) => {}
+            Err(err) if err.is_permanent() => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "octavo will never accept this history entry; dropping it"
+                );
+            }
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "failed to upload a history entry to octavo; retrying later"
+                );
+                return Err(RetryLater);
+            }
+        }
+
+        self.dequeue(login, id).await
+    }
+
+    async fn delete_pending(&self, login: &HubLogin) -> Result<(), RetryLater> {
+        loop {
+            let ids = self.queue.pending_deletions(login.user_id()).await.map_err(|err| {
+                tracing::warn!(?err, "failed to read the octavo deletion queue; retrying later");
+                RetryLater
+            })?;
+            if ids.is_empty() {
+                return Ok(());
+            }
+
+            for id in ids {
+                self.delete(login, id).await?;
+            }
+        }
+    }
+
+    async fn delete(&self, login: &HubLogin, id: HistoryId) -> Result<(), RetryLater> {
+        match login.delete(id).await {
+            Ok(()) => {}
+            Err(err) if err.is_permanent() => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "octavo refused to delete this history entry; dropping the deletion"
+                );
+            }
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "failed to delete a history entry from octavo; retrying later"
+                );
+                return Err(RetryLater);
+            }
+        }
+
+        self.queue.remove_deletion(login.user_id(), id).await.map_err(|err| {
+            tracing::warn!(
+                ?err,
+                %id,
+                "failed to remove a deletion from the octavo queue; retrying later"
+            );
+            RetryLater
+        })
+    }
+
+    async fn dequeue(&self, login: &HubLogin, id: HistoryId) -> Result<(), RetryLater> {
+        self.queue.remove_history(login.user_id(), id).await.map_err(|err| {
+            tracing::warn!(
+                ?err,
+                %id,
+                "failed to remove a history entry from the octavo upload queue; retrying later"
+            );
+            RetryLater
+        })
+    }
+}

crates/atuin-daemon/src/octavo/client/active/mod.rs

@@ -0,0 +1,213 @@
+mod engine;
+
+use std::sync::Arc;
+
+use atuin_client::history::{History, HistoryId};
+use atuin_client::settings::Settings;
+use atuin_domain::record::RecordId;
+use atuin_octavo::hub::{ConnectError, HubCallError, HubClient, HubToken, UserId};
+use atuin_octavo::pb;
+use atuin_octavo::queue::{QueueError, UploadQueue};
+use engine::UploadEngine;
+use thiserror::Error;
+use tokio::sync::Notify;
+
+use super::OctavoClient;
+use crate::daemon::DaemonHandle;
+
+#[derive(Debug, Error)]
+pub enum OpenActiveOctavoClientError {
+    #[error("failed to open the upload queue")]
+    Queue(#[from] QueueError),
+    #[error("failed to set up the hub client")]
+    Hub(#[from] ConnectError),
+}
+
+#[derive(Debug)]
+pub struct ActiveOctavoClient {
+    uploader: Arc<DirectUploader>,
+    _engine: UploadEngine,
+}
+
+impl ActiveOctavoClient {
+    pub async fn open(
+        settings: &Settings,
+        handle: DaemonHandle,
+    ) -> Result<Self, OpenActiveOctavoClientError> {
+        let queue = UploadQueue::open(Settings::octavo_queue_path()).await?;
+        let hub = HubClient::new(settings)?;
+        let wake = Arc::new(Notify::new());
+        let engine = UploadEngine::spawn(queue.clone(), hub.clone(), handle.clone(), wake.clone());
+
+        Ok(Self {
+            uploader: Arc::new(DirectUploader {
+                hub,
+                handle,
+                queue,
+                wake,
+            }),
+            _engine: engine,
+        })
+    }
+}
+
+impl OctavoClient for ActiveOctavoClient {
+    async fn push_history(&self, history: &History, record_id: RecordId) {
+        let settings = self.uploader.handle.settings().await.clone();
+        if !settings.octavo.enabled {
+            return;
+        }
+        // Read now, so the command goes to whoever was logged in as it finished.
+        let Some(token) = self.uploader.token(&settings).await else {
+            return;
+        };
+
+        // Detached, so finishing a command never waits on the network. A daemon that exits while
+        // the upload is in flight loses only that entry's upload.
+        tokio::spawn(Arc::clone(&self.uploader).upload_or_queue(
+            settings,
+            token,
+            history.clone(),
+            record_id,
+        ));
+    }
+
+    async fn delete_history(&self, ids: &[HistoryId]) {
+        let settings = self.uploader.handle.settings().await.clone();
+        // Read now, so the deletions go to whoever was logged in when the entries were deleted.
+        let Some(token) = self.uploader.token(&settings).await else {
+            return;
+        };
+
+        // Queued before the local delete returns, so the deletion outlives the daemon. That needs
+        // the token's user, known once the daemon has reached the hub since the login; until
+        // then a detached task asks the hub first. The engine sends them once they're queued.
+        match self.uploader.hub.user_of(&token).await {
+            Ok(Some(user)) => self.uploader.queue_deletions(&user, ids).await,
+            Ok(None) => {
+                tokio::spawn(Arc::clone(&self.uploader).learn_user_and_queue_deletions(
+                    settings,
+                    token,
+                    ids.to_vec(),
+                ));
+            }
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    count = ids.len(),
+                    "failed to read the hub login; these deletions will not reach octavo"
+                );
+            }
+        }
+    }
+}
+
+/// Uploads a finished command as soon as it finishes, leaving the queue to the ones that can't go
+/// right away.
+#[derive(Debug)]
+struct DirectUploader {
+    hub: HubClient,
+    handle: DaemonHandle,
+    queue: UploadQueue,
+    wake: Arc<Notify>,
+}
+
+impl DirectUploader {
+    /// Uploads `history`, or queues it for the engine to retry when the upload fails for a reason
+    /// that may pass, such as no network.
+    async fn upload_or_queue(
+        self: Arc<Self>,
+        settings: Settings,
+        token: HubToken,
+        history: History,
+        record_id: RecordId,
+    ) {
+        let id = history.id;
+        let login = match self.hub.login_as(token, &settings).await {
+            Ok(login) => login,
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "failed to learn whose hub token this is; not uploading this history entry"
+                );
+                return;
+            }
+        };
+
+        match login.upload(history, Some(record_id)).await {
+            Ok(()) => {}
+            Err(err) if err.is_permanent() => {
+                tracing::warn!(
+                    ?err,
+                    %id,
+                    "octavo will never accept this history entry; dropping it"
+                );
+            }
+            Err(err) => {
+                tracing::debug!(
+                    ?err,
+                    %id,
+                    "failed to upload a history entry to octavo; queueing it"
+                );
+                if let Err(err) = self.queue.push_history(login.user_id(), id, record_id).await {
+                    tracing::warn!(
+                        ?err,
+                        %id,
+                        "failed to queue a history entry for octavo; it will not be uploaded"
+                    );
+                    return;
+                }
+                self.wake.notify_one();
+            }
+        }
+    }
+
+    /// Learns whose `token` is from the hub, then queues the deletion of `ids` for them.
+    async fn learn_user_and_queue_deletions(
+        self: Arc<Self>,
+        settings: Settings,
+        token: HubToken,
+        ids: Vec<HistoryId>,
+    ) {
+        match self.hub.login_as(token, &settings).await {
+            Ok(login) => self.queue_deletions(login.user_id(), &ids).await,
+            Err(err) => {
+                tracing::warn!(
+                    ?err,
+                    count = ids.len(),
+                    "failed to learn whose hub token this is; these deletions will not reach \
+                     octavo"
+                );
+            }
+        }
+    }
+
+    /// Queues the deletion of `user`'s entries `ids` for the engine to send.
+    async fn queue_deletions(&self, user: &UserId, ids: &[HistoryId]) {
+        // Octavo never stored an entry whose id isn't a UUIDv7.
+        let stored = ids.iter().copied().filter(|&id| pb::UuidV7::try_from(id).is_ok());
+        if let Err(err) = self.queue.push_deletions(user, stored).await {
+            tracing::warn!(
+                ?err,
+                count = ids.len(),
+                "failed to queue history deletions for octavo; they will not reach it"
+            );
+            return;
+        }
+
+        self.wake.notify_one();
+    }
+
+    /// The hub token logged in now, or `None` with Octavo left alone when there's none to use.
+    async fn token(&self, settings: &Settings) -> Option<HubToken> {
+        match self.hub.token(settings).await {
+            Ok(token) => Some(token),
+            Err(HubCallError::NotLoggedIn) => None,
+            Err(err) => {
+                tracing::warn!(?err, "failed to read the hub session; leaving octavo alone");
+                None
+            }
+        }
+    }
+}

crates/atuin-daemon/src/octavo/client/mod.rs

@@ -0,0 +1,25 @@
+#[cfg(feature = "octavo")]
+mod active;
+mod nop;
+
+#[cfg(feature = "octavo")]
+pub use active::ActiveOctavoClient;
+use atuin_client::history::{History, HistoryId};
+use atuin_domain::record::RecordId;
+use enum_dispatch::enum_dispatch;
+pub use nop::NopOctavoClient;
+
+#[enum_dispatch]
+#[allow(async_fn_in_trait)]
+pub trait OctavoClient {
+    async fn push_history(&self, history: &History, record_id: RecordId);
+    async fn delete_history(&self, ids: &[HistoryId]);
+}
+
+#[enum_dispatch(OctavoClient)]
+#[derive(Debug)]
+pub enum AnyOctavoClient {
+    #[cfg(feature = "octavo")]
+    Active(ActiveOctavoClient),
+    Nop(NopOctavoClient),
+}

crates/atuin-daemon/src/octavo/client/nop.rs

@@ -0,0 +1,13 @@
+use atuin_client::history::{History, HistoryId};
+use atuin_domain::record::RecordId;
+
+use super::OctavoClient;
+
+#[derive(Debug, Clone, Copy)]
+pub struct NopOctavoClient;
+
+impl OctavoClient for NopOctavoClient {
+    async fn push_history(&self, _history: &History, _record_id: RecordId) {}
+
+    async fn delete_history(&self, _ids: &[HistoryId]) {}
+}

crates/atuin-daemon/src/octavo/mod.rs

@@ -0,0 +1,55 @@
+mod client;
+
+use atuin_client::history::{History, HistoryId};
+use atuin_client::settings::Settings;
+use atuin_domain::record::RecordId;
+use client::{AnyOctavoClient, NopOctavoClient, OctavoClient};
+
+use crate::daemon::DaemonHandle;
+
+#[derive(Debug)]
+pub struct Octavo {
+    client: AnyOctavoClient,
+}
+
+impl Octavo {
+    #[cfg(feature = "octavo")]
+    pub async fn open(settings: &Settings, handle: DaemonHandle) -> Self {
+        // Once Octavo has been on here, deletions keep reaching it with uploads off, so what the
+        // user deletes locally leaves Octavo too.
+        if !settings.octavo.enabled && !Settings::octavo_queue_path().exists() {
+            return Self::nop();
+        }
+
+        match client::ActiveOctavoClient::open(settings, handle).await {
+            Ok(client) => Self {
+                client: client.into(),
+            },
+            Err(err) => {
+                tracing::error!(?err, "failed to open octavo; octavo uploads are disabled");
+                Self::nop()
+            }
+        }
+    }
+
+    #[cfg(not(feature = "octavo"))]
+    #[allow(clippy::unused_async)]
+    pub async fn open(_settings: &Settings, _handle: DaemonHandle) -> Self {
+        Self::nop()
+    }
+
+    #[must_use]
+    pub fn nop() -> Self {
+        Self {
+            client: NopOctavoClient.into(),
+        }
+    }
+
+    pub async fn push_history(&self, history: &History, record_id: RecordId) {
+        self.client.push_history(history, record_id).await;
+    }
+
+    pub async fn delete_history(&self, ids: &[HistoryId]) {
+        self.client.delete_history(ids).await;
+    }
+}

crates/atuin-daemon/tests/common/mod.rs

@@ -30,7 +30,7 @@ use atuin_daemon::grpc::history::pb;
 use atuin_daemon::grpc::history::pb::history_server::HistoryServer;
 use atuin_daemon::search::{IndexFilterMode, SearchIndex};
 use atuin_daemon::{
-    Daemon, DaemonEvent, DaemonHandle, HistoryJournal, OutputCaptureEngine, SearchComponent,
+    Daemon, DaemonEvent, DaemonHandle, HistoryJournal, Octavo, OutputCaptureEngine, SearchComponent,
 };
 use atuin_domain::record::{CmdOrigin, HostId, RecordTag};
 use corpus::{HistoryGen, Seeded};
@@ -188,6 +188,7 @@ impl TestEnvBuilder {
             history_db.clone(),
             index.clone(),
             output_capture,
+            Octavo::nop(),
         ));
         let history_service =
             HistoryServer::new(HistoryService::new(journal.clone(), handle.clone()));

crates/atuin-domain/src/caps/all.rs

@@ -50,6 +50,25 @@ impl Capability for PackfileCap {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OctavoCap {
+    pub version: u32,
+}
+
+impl Capability for OctavoCap {
+    fn static_name() -> &'static str {
+        "sh.atuin.server/octavo"
+    }
+
+    fn name(&self) -> &'static str {
+        Self::static_name()
+    }
+
+    fn json(&self) -> Result<serde_json::Value, serde_json::Error> {
+        serde_json::to_value(self)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct PageSizeCap {
     pub version: u32,

crates/atuin-domain/src/caps/mod.rs

@@ -57,7 +57,7 @@ mod client;
 mod middleware;
 mod server;
 
-pub use all::{CapabilitiesCap, PackfileCap, PageSizeCap};
+pub use all::{CapabilitiesCap, OctavoCap, PackfileCap, PageSizeCap};
 pub use client::{AuthHeaderProvider, CapClient, ServerSupportError};
 pub use middleware::{CapMiddleware, CapMismatch, CapabilitiesExt};
 pub use server::{CapServer, Negotiation};

crates/atuin-octavo/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "atuin-octavo"
+edition = "2024"
+description = "Client for the Atuin hub's gRPC API"
+
+authors.workspace = true
+homepage.workspace = true
+license.workspace = true
+readme.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+atuin-client = { workspace = true }
+atuin-common = { workspace = true, features = ["sqlite"] }
+atuin-domain = { workspace = true }
+eyre = { workspace = true }
+futures = { workspace = true }
+hyper-tls = { workspace = true }
+hyper-util = { workspace = true, features = ["client-legacy", "tokio"] }
+native-tls = { workspace = true, features = ["alpn"] }
+prost = { workspace = true }
+prost-types = { workspace = true }
+secrecy = { workspace = true }
+sqlx = { workspace = true }
+thiserror = { workspace = true }
+tonic = { workspace = true }
+tonic-prost = { workspace = true }
+url = { workspace = true }
+uuid = { workspace = true }
+xxhash-rust = { workspace = true }
+
+[build-dependencies]
+protox = { workspace = true }
+tonic-prost-build = { workspace = true }
+
+[lints]
+workspace = true

crates/atuin-octavo/build.rs

@@ -0,0 +1,11 @@
+use std::error::Error;
+
+fn main() -> Result<(), Box<dyn Error>> {
+    println!("cargo::rerun-if-changed=proto");
+    println!("cargo::rerun-if-changed=migrations");
+
+    let descriptors = protox::compile(["atuin/api/v1/hub_service.proto"], ["proto"])?;
+    tonic_prost_build::configure().build_server(false).compile_fds(descriptors)?;
+
+    Ok(())
+}

crates/atuin-octavo/migrations/0001_init.sql

@@ -0,0 +1,4 @@
+CREATE TABLE history_uploads (
+    history_id TEXT PRIMARY KEY NOT NULL,
+    record_id BLOB
+) WITHOUT ROWID;

crates/atuin-octavo/migrations/0002_history_deletions.sql

@@ -0,0 +1,3 @@
+CREATE TABLE history_deletions (
+    history_id TEXT PRIMARY KEY NOT NULL
+) WITHOUT ROWID;

crates/atuin-octavo/migrations/0003_queue_owners.sql

@@ -0,0 +1,14 @@
+DROP TABLE history_uploads;
+CREATE TABLE history_uploads (
+    user_id TEXT NOT NULL,
+    history_id TEXT NOT NULL,
+    record_id BLOB,
+    PRIMARY KEY (user_id, history_id)
+) WITHOUT ROWID;
+
+DROP TABLE history_deletions;
+CREATE TABLE history_deletions (
+    user_id TEXT NOT NULL,
+    history_id TEXT NOT NULL,
+    PRIMARY KEY (user_id, history_id)
+) WITHOUT ROWID;

crates/atuin-octavo/proto/atuin/api/v1/common.proto

@@ -20,7 +20,7 @@ message CmdOrigin {
 // Who decided to run a command: a person or an agent. Not a hub user.
 message Author {
   // e.g. `alice` or `claude-code`.
-  string name = 1;
+  string name = 1 [(buf.validate.field).string.max_bytes = 255];
   // Logically required; the hub rejects `AUTHOR_KIND_UNSPECIFIED`.
   AuthorKind kind = 2 [
     (buf.validate.field).required = true,

crates/atuin-octavo/proto/atuin/api/v1/history.proto

@@ -23,8 +23,9 @@ message History {
   google.protobuf.Timestamp start_time = 6 [(buf.validate.field).required = true];
   // How long the command ran and never negative.
   google.protobuf.Duration duration = 7 [(buf.validate.field).duration.gte = {}];
-  // The terminal session the command ran in. Client-generated, unique per user.
-  UuidV7 session = 8 [(buf.validate.field).required = true];
+  // The terminal session the command ran in, as the client names it. Unique per user, and empty
+  // when the client doesn't know it.
+  string session = 8 [(buf.validate.field).string.max_bytes = 128];
   // The machine and account the command ran on.
   CmdOrigin cmd_origin = 9 [(buf.validate.field).required = true];
   // Who decided to run the command.

crates/atuin-octavo/proto/atuin/api/v1/hub_service.proto

@@ -2,6 +2,7 @@ syntax = "proto3";
 
 package atuin.api.v1;
 
+import "atuin/api/v1/common.proto";
 import "atuin/api/v1/history.proto";
 import "buf/validate/validate.proto";
 
@@ -16,6 +17,15 @@ service HubService {
   //
   // Entries stored before the call, or while no call is open, are not sent.
   rpc WatchHistory(WatchHistoryRequest) returns (stream WatchHistoryResponse);
+
+  // Delete the calling user's history entry with this id, and keep it from being stored again.
+  //
+  // An id that names no stored entry is still remembered as deleted, so a client can repeat a
+  // deletion, or send it before the upload it races.
+  rpc DeleteHistory(DeleteHistoryRequest) returns (DeleteHistoryResponse);
+
+  // Identify the user the call's token logs in.
+  rpc WhoAmI(WhoAmIRequest) returns (WhoAmIResponse);
 }
 
 message InsertHistoryRequest {
@@ -31,3 +41,17 @@ message WatchHistoryResponse {
   // An entry the user stored.
   History history = 1 [(buf.validate.field).required = true];
 }
+
+message DeleteHistoryRequest {
+  // The id of the entry to delete, as `History.id` gave it.
+  UuidV7 id = 1 [(buf.validate.field).required = true];
+}
+
+message DeleteHistoryResponse {}
+
+message WhoAmIRequest {}
+
+message WhoAmIResponse {
+  // The calling user's id. Opaque, and stable for the life of the account.
+  string user_id = 1 [(buf.validate.field).string.min_len = 1];
+}

crates/atuin-octavo/src/hub.rs

@@ -0,0 +1,263 @@
+use std::time::Duration;
+
+use atuin_client::history::{History, HistoryId};
+use atuin_client::settings::{Settings, SyncAuth};
+use atuin_domain::record::RecordId;
+use hyper_tls::HttpsConnector;
+use hyper_util::client::legacy::connect::HttpConnector;
+use secrecy::{ExposeSecret, SecretString};
+use thiserror::Error;
+use tonic::metadata::errors::InvalidMetadataValue;
+use tonic::metadata::{Ascii, MetadataValue};
+use tonic::transport::{Channel, Endpoint};
+use tonic::{Code, Request};
+use url::Url;
+
+use crate::pb;
+use crate::pb::hub_service_client::HubServiceClient;
+use crate::pb::{HistoryConversionError, NotUuidV7};
+
+#[derive(Debug, Error)]
+pub enum ConnectError {
+    #[error("the Octavo endpoint {address} is not a valid URI")]
+    Endpoint {
+        address: Url,
+        #[source]
+        source: tonic::transport::Error,
+    },
+    #[error("failed to set up TLS")]
+    Tls(#[source] native_tls::Error),
+}
+
+#[derive(Debug, Error)]
+pub enum HubCallError {
+    #[error(transparent)]
+    Convert(#[from] HistoryConversionError),
+    #[error(transparent)]
+    Id(#[from] NotUuidV7),
+    #[error("Octavo needs a hub login; run `atuin login`")]
+    NotLoggedIn,
+    #[error("failed to read the hub login: {0:#}")]
+    Meta(eyre::Report),
+    #[error(transparent)]
+    Connect(#[from] ConnectError),
+    #[error("the Octavo endpoint is now {0}; restart the daemon to use it")]
+    EndpointChanged(Url),
+    #[error("the hub token is not a valid header value")]
+    InvalidToken(#[source] InvalidMetadataValue),
+    #[error("the hub named no user for the token")]
+    NoUserId,
+    #[error("the hub refused the call")]
+    Hub(#[source] tonic::Status),
+}
+
+impl HubCallError {
+    /// Whether the call would fail the same way however often it's retried.
+    #[must_use]
+    pub fn is_permanent(&self) -> bool {
+        match self {
+            Self::Convert(_) | Self::Id(_) => true,
+            Self::Hub(status) => status.code() == Code::InvalidArgument,
+            Self::NotLoggedIn
+            | Self::Meta(_)
+            | Self::Connect(_)
+            | Self::EndpointChanged(_)
+            | Self::InvalidToken(_)
+            | Self::NoUserId => false,
+        }
+    }
+}
+
+/// A hub user's id, as `WhoAmI` names it. Opaque.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UserId(String);
+
+impl UserId {
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+/// The hub Octavo's settings pointed at when the client was made.
+#[derive(Debug, Clone)]
+pub struct HubClient {
+    address: Url,
+    service: HubServiceClient<Channel>,
+}
+
+impl HubClient {
+    pub fn new(settings: &Settings) -> Result<Self, ConnectError> {
+        let address = octavo_address(settings);
+        let service = HubServiceClient::new(connect(&address, settings.network_connect_timeout)?);
+        Ok(Self { address, service })
+    }
+
+    /// The hub token logged in now, so work done for this moment's user later still goes to them
+    /// after another login.
+    pub async fn token(&self, settings: &Settings) -> Result<HubToken, HubCallError> {
+        // A token for one hub must never reach another, so a client keeps to the hub it was made
+        // for.
+        let configured = octavo_address(settings);
+        if configured != self.address {
+            return Err(HubCallError::EndpointChanged(configured));
+        }
+
+        let SyncAuth::Hub { token } = settings.resolve_sync_auth().await else {
+            return Err(HubCallError::NotLoggedIn);
+        };
+        Ok(HubToken(token))
+    }
+
+    /// Logs in with the hub token logged in now.
+    pub async fn login(&self, settings: &Settings) -> Result<HubLogin, HubCallError> {
+        let token = self.token(settings).await?;
+        self.login_as(token, settings).await
+    }
+
+    /// Logs in with `token`, asking the hub whose it is the first time it's used, and remembering
+    /// the answer for as long as it's the hub session.
+    pub async fn login_as(
+        &self,
+        HubToken(token): HubToken,
+        settings: &Settings,
+    ) -> Result<HubLogin, HubCallError> {
+        let mut authorization =
+            MetadataValue::try_from(format!("Bearer {}", token.expose_secret()))
+                .map_err(HubCallError::InvalidToken)?;
+        authorization.set_sensitive(true);
+        let caller = Caller {
+            service: self.service.clone(),
+            authorization,
+            timeout: settings.network_timeout,
+        };
+
+        let user_id = match Self::known_user(&token).await? {
+            Some(id) => id,
+            None => {
+                let id = caller.who_am_i().await?;
+                Settings::meta_store()
+                    .await
+                    .map_err(HubCallError::Meta)?
+                    .save_hub_user_id(&token, id.as_str())
+                    .await
+                    .map_err(HubCallError::Meta)?;
+                id
+            }
+        };
+
+        Ok(HubLogin { caller, user_id })
+    }
+
+    /// The user `token` belongs to, if a login with it already learned that while it's been the
+    /// hub session. Never calls the hub.
+    pub async fn user_of(
+        &self,
+        HubToken(token): &HubToken,
+    ) -> Result<Option<UserId>, HubCallError> {
+        Self::known_user(token).await
+    }
+
+    async fn known_user(token: &SecretString) -> Result<Option<UserId>, HubCallError> {
+        let meta = Settings::meta_store().await.map_err(HubCallError::Meta)?;
+        let id = meta.hub_user_id(token).await.map_err(HubCallError::Meta)?;
+        Ok(id.map(UserId))
+    }
+}
+
+/// A hub token, as the hub session held it when [`HubClient::token`] read it.
+#[derive(Debug)]
+pub struct HubToken(SecretString);
+
+/// The hub, as the user a hub token belongs to.
+#[derive(Debug)]
+pub struct HubLogin {
+    caller: Caller,
+    user_id: UserId,
+}
+
+impl HubLogin {
+    pub fn user_id(&self) -> &UserId {
+        &self.user_id
+    }
+
+    pub async fn upload(
+        &self,
+        history: History,
+        record_id: Option<RecordId>,
+    ) -> Result<(), HubCallError> {
+        let mut history = pb::History::try_from(history)?;
+        history.record_id = record_id.map(pb::UuidV7::from);
+        let request = self.caller.request(pb::InsertHistoryRequest {
+            history: Some(history),
+        });
+
+        match self.caller.service.clone().insert_history(request).await {
+            Ok(_) => Ok(()),
+            Err(status) if status.code() == Code::AlreadyExists => Ok(()),
+            Err(status) => Err(HubCallError::Hub(status)),
+        }
+    }
+
+    /// Deletes the entry with `id`, and keeps it from being stored again.
+    pub async fn delete(&self, id: HistoryId) -> Result<(), HubCallError> {
+        let request = self.caller.request(pb::DeleteHistoryRequest {
+            id: Some(id.try_into()?),
+        });
+
+        self.caller.service.clone().delete_history(request).await.map_err(HubCallError::Hub)?;
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+struct Caller {
+    service: HubServiceClient<Channel>,
+    authorization: MetadataValue<Ascii>,
+    timeout: Duration,
+}
+
+impl Caller {
+    fn request<T>(&self, message: T) -> Request<T> {
+        let mut request = Request::new(message);
+        request.set_timeout(self.timeout);
+        request.metadata_mut().insert("authorization", self.authorization.clone());
+        request
+    }
+
+    async fn who_am_i(&self) -> Result<UserId, HubCallError> {
+        let request = self.request(pb::WhoAmIRequest {});
+        let user_id =
+            self.service.clone().who_am_i(request).await.map_err(HubCallError::Hub)?.into_inner();
+
+        if user_id.user_id.is_empty() {
+            return Err(HubCallError::NoUserId);
+        }
+        Ok(UserId(user_id.user_id))
+    }
+}
+
+fn octavo_address(settings: &Settings) -> Url {
+    settings.octavo.endpoint.clone().unwrap_or_else(|| settings.hub_endpoint())
+}
+
+fn connect(address: &Url, connect_timeout: Duration) -> Result<Channel, ConnectError> {
+    let endpoint = Endpoint::from_shared(address.to_string())
+        .map_err(|source| ConnectError::Endpoint {
+            address: address.clone(),
+            source,
+        })?
+        .connect_timeout(connect_timeout);
+
+    Ok(endpoint.connect_with_connector_lazy(connector()?))
+}
+
+fn connector() -> Result<HttpsConnector<HttpConnector>, ConnectError> {
+    let mut http = HttpConnector::new();
+    http.enforce_http(false);
+    let tls = native_tls::TlsConnector::builder()
+        .request_alpns(&["h2"])
+        .build()
+        .map_err(ConnectError::Tls)?;
+    Ok(HttpsConnector::from((http, tls.into())))
+}

crates/atuin-octavo/src/lib.rs

@@ -0,0 +1,5 @@
+#![deny(unsafe_code)]
+
+pub mod hub;
+pub mod pb;
+pub mod queue;

crates/atuin-octavo/src/pb.rs

@@ -0,0 +1,128 @@
+mod codegen {
+    #![allow(clippy::must_use_candidate, clippy::derive_partial_eq_without_eq)]
+    tonic::include_proto!("atuin.api.v1");
+}
+
+use std::time::{Duration, SystemTime};
+
+use atuin_client::history::{History as DomainHistory, HistoryId};
+use atuin_domain::record::{CmdOrigin as DomainCmdOrigin, RecordId};
+pub use codegen::*;
+use prost::Message;
+use thiserror::Error;
+use uuid::{Uuid, Version};
+use xxhash_rust::xxh3::xxh3_128;
+
+#[derive(Debug, Error)]
+#[error("history id {0} is not a UUIDv7, so Octavo can't store it")]
+pub struct NotUuidV7(pub HistoryId);
+
+impl TryFrom<HistoryId> for UuidV7 {
+    type Error = NotUuidV7;
+
+    fn try_from(id: HistoryId) -> Result<Self, Self::Error> {
+        let bytes = id.into_bytes();
+        if Uuid::from_bytes(bytes).get_version() != Some(Version::SortRand) {
+            return Err(NotUuidV7(id));
+        }
+
+        Ok(Self {
+            value: bytes.to_vec(),
+        })
+    }
+}
+
+impl From<RecordId> for UuidV7 {
+    fn from(id: RecordId) -> Self {
+        id.0.into()
+    }
+}
+
+impl From<Uuid> for UuidV7 {
+    fn from(uuid: Uuid) -> Self {
+        Self {
+            value: uuid.into_bytes().to_vec(),
+        }
+    }
+}
+
+impl From<&DomainCmdOrigin> for CmdOrigin {
+    fn from(origin: &DomainCmdOrigin) -> Self {
+        Self {
+            host: origin.host().into_inner().to_owned(),
+            user: origin.user().into_inner().to_owned(),
+        }
+    }
+}
+
+const MAX_AUTHOR_NAME_BYTES: usize = 255;
+const MAX_SESSION_BYTES: usize = 128;
+const MAX_HISTORY_BYTES: usize = 1 << 20;
+
+#[derive(Debug, Error)]
+pub enum HistoryConversionError {
+    #[error(transparent)]
+    Id(#[from] NotUuidV7),
+    #[error("exit code {0} does not fit in 32 bits")]
+    ExitOutOfRange(i64),
+    #[error("invalid duration: {0}")]
+    InvalidDuration(#[from] prost_types::DurationError),
+    #[error("the author name is {0} bytes, over the {MAX_AUTHOR_NAME_BYTES} Octavo allows")]
+    AuthorTooLong(usize),
+    #[error("the entry is {0} bytes, over the {MAX_HISTORY_BYTES} Octavo takes")]
+    TooLarge(usize),
+}
+
+impl TryFrom<DomainHistory> for History {
+    type Error = HistoryConversionError;
+
+    fn try_from(history: DomainHistory) -> Result<Self, Self::Error> {
+        let exit = i32::try_from(history.exit)
+            .map_err(|_| HistoryConversionError::ExitOutOfRange(history.exit))?;
+        let duration = u64::try_from(history.duration)
+            .ok()
+            .map(|nanos| prost_types::Duration::try_from(Duration::from_nanos(nanos)))
+            .transpose()?;
+        if history.author.len() > MAX_AUTHOR_NAME_BYTES {
+            return Err(HistoryConversionError::AuthorTooLong(history.author.len()));
+        }
+        let kind = if history.is_agent() {
+            AuthorKind::Agent
+        } else {
+            AuthorKind::User
+        };
+
+        let converted = Self {
+            id: Some(history.id.try_into()?),
+            record_id: None,
+            cmd_origin: Some(CmdOrigin::from(&history.cmd_origin)),
+            start_time: Some(SystemTime::from(history.timestamp).into()),
+            duration,
+            session: session_key(history.session),
+            author: Some(Author {
+                name: history.author,
+                kind: kind.into(),
+            }),
+            command: history.command,
+            cwd: history.cwd,
+            exit,
+            intent: history.intent,
+            shell: history.shell,
+        };
+
+        match converted.encoded_len() {
+            len if len > MAX_HISTORY_BYTES => Err(HistoryConversionError::TooLarge(len)),
+            _ => Ok(converted),
+        }
+    }
+}
+
+/// `session` as `History.session` can carry it: as is when it fits the bytes the field allows, or
+/// else as the hex of its 128-bit xxh3 hash, which keeps two long sessions apart.
+fn session_key(session: String) -> String {
+    if session.len() <= MAX_SESSION_BYTES {
+        return session;
+    }
+
+    format!("{:032x}", xxh3_128(session.as_bytes()))
+}

crates/atuin-octavo/src/queue.rs

@@ -0,0 +1,170 @@
+use std::num::NonZeroU32;
+use std::path::Path;
+
+use atuin_client::history::HistoryId;
+use atuin_common::db;
+use atuin_common::db::sqlite::{Sqlite, SqliteOpenOrCreateError};
+use atuin_common::futures::stream::ChunkedStream;
+use atuin_domain::record::RecordId;
+use futures::stream;
+use sqlx::migrate::MigrateError;
+use thiserror::Error;
+
+use crate::hub::UserId;
+
+const PAGE_SIZE: NonZeroU32 = NonZeroU32::new(100).unwrap();
+
+#[derive(Debug, Error)]
+pub enum QueueError {
+    #[error("failed to open the upload queue")]
+    Open(#[from] SqliteOpenOrCreateError),
+    #[error("failed to migrate the upload queue")]
+    Migrate(#[from] MigrateError),
+    #[error("an upload queue query failed")]
+    Query(#[from] sqlx::Error),
+}
+
+#[derive(Debug, Clone, Copy, sqlx::FromRow)]
+pub struct PendingHistoryUpload {
+    pub history_id: HistoryId,
+    pub record_id: Option<RecordId>,
+}
+
+#[derive(Debug, Clone)]
+pub struct UploadQueue {
+    db: Sqlite,
+}
+
+impl UploadQueue {
+    pub async fn open(path: impl AsRef<Path>) -> Result<Self, QueueError> {
+        let db = Sqlite::builder(path.as_ref().as_os_str()).restrict_permissions().open().await?;
+        db::migrate!(db.pool(), "./migrations").await?;
+        Ok(Self { db })
+    }
+
+    /// Queues `user`'s entry `id` for upload.
+    pub async fn push_history(
+        &self,
+        user: &UserId,
+        id: HistoryId,
+        record_id: RecordId,
+    ) -> Result<(), QueueError> {
+        db::query(
+            "INSERT OR IGNORE INTO history_uploads (user_id, history_id, record_id) VALUES (?1, \
+             ?2, ?3)",
+        )
+        .bind(user.as_str())
+        .bind(id)
+        .bind(record_id)
+        .execute(self.db.pool())
+        .await?;
+        Ok(())
+    }
+
+    pub async fn push_many_history(
+        &self,
+        user: &UserId,
+        ids: impl IntoIterator<Item = HistoryId>,
+    ) -> Result<u64, QueueError> {
+        self.insert_ids("history_uploads", user, ids).await
+    }
+
+    /// Queues the deletion of `user`'s entries `ids` from Octavo, returning how many were not
+    /// queued already.
+    pub async fn push_deletions(
+        &self,
+        user: &UserId,
+        ids: impl IntoIterator<Item = HistoryId>,
+    ) -> Result<u64, QueueError> {
+        self.insert_ids("history_deletions", user, ids).await
+    }
+
+    /// The first page of `user`'s queued deletions, in id order.
+    pub async fn pending_deletions(&self, user: &UserId) -> Result<Vec<HistoryId>, QueueError> {
+        let ids = db::query_scalar(
+            "SELECT history_id FROM history_deletions WHERE user_id = ?1 ORDER BY history_id \
+             LIMIT ?2",
+        )
+        .bind(user.as_str())
+        .bind(PAGE_SIZE.get())
+        .fetch_all(self.db.pool())
+        .await?;
+        Ok(ids)
+    }
+
+    pub async fn remove_deletion(&self, user: &UserId, id: HistoryId) -> Result<(), QueueError> {
+        db::query("DELETE FROM history_deletions WHERE user_id = ?1 AND history_id = ?2")
+            .bind(user.as_str())
+            .bind(id)
+            .execute(self.db.pool())
+            .await?;
+        Ok(())
+    }
+
+    async fn insert_ids(
+        &self,
+        table: &'static str,
+        user: &UserId,
+        ids: impl IntoIterator<Item = HistoryId>,
+    ) -> Result<u64, QueueError> {
+        let rows_per_insert = (self.db.info().await.variable_number_limit() / 2).max(1);
+        let mut ids = ids.into_iter().peekable();
+        let mut tx = self.db.pool().begin().await?;
+        let mut queued = 0;
+
+        while ids.peek().is_some() {
+            let mut insert = sqlx::QueryBuilder::new(format!(
+                "INSERT OR IGNORE INTO {table} (user_id, history_id) "
+            ));
+            insert.push_values(ids.by_ref().take(rows_per_insert), |mut row, id| {
+                row.push_bind(user.as_str()).push_bind(id);
+            });
+            queued += insert.build().execute(&mut *tx).await?.rows_affected();
+        }
+
+        tx.commit().await?;
+        Ok(queued)
+    }
+
+    /// `user`'s queued uploads, in id order.
+    pub fn pending_history(
+        &self,
+        user: &UserId,
+    ) -> ChunkedStream<Result<PendingHistoryUpload, QueueError>> {
+        let pool = self.db.pool().clone();
+        let user = user.clone();
+
+        ChunkedStream::new(stream::unfold(Some(None), move |after: Option<Option<HistoryId>>| {
+            let pool = pool.clone();
+            let user = user.clone();
+            async move {
+                let after = after?;
+                let page = match db::query_as::<_, PendingHistoryUpload>(
+                    "SELECT history_id, record_id FROM history_uploads WHERE user_id = ?1 AND (?2 \
+                     IS NULL OR history_id > ?2) ORDER BY history_id LIMIT ?3",
+                )
+                .bind(user.as_str())
+                .bind(after)
+                .bind(PAGE_SIZE.get())
+                .fetch_all(&pool)
+                .await
+                {
+                    Ok(page) => page,
+                    Err(err) => return Some((vec![Err(err.into())], None)),
+                };
+
+                let last = page.last()?.history_id;
+                Some((page.into_iter().map(Ok).collect(), Some(Some(last))))
+            }
+        }))
+    }
+
+    pub async fn remove_history(&self, user: &UserId, id: HistoryId) -> Result<(), QueueError> {
+        db::query("DELETE FROM history_uploads WHERE user_id = ?1 AND history_id = ?2")
+            .bind(user.as_str())
+            .bind(id)
+            .execute(self.db.pool())
+            .await?;
+        Ok(())
+    }
+}

crates/atuin/Cargo.toml

@@ -48,12 +48,14 @@ default = [
     "daemon",
     "ai",
     "pty-proxy",
+    "octavo",
 ]
 client = ["atuin-client"]
 sync = ["atuin-client/sync"]
 daemon = ["dep:async-stream", "atuin-client/daemon", "atuin-daemon", "atuin-ai?/daemon"]
 ai = ["atuin-ai"]
 pty-proxy = ["dep:atuin-pty-proxy"]
+octavo = ["sync", "daemon", "atuin-daemon/octavo", "dep:atuin-octavo"]
 hex = ["pty-proxy"]
 clipboard = ["dep:arboard", "atuin-ai?/clipboard"]
 check-update = ["atuin-client/check-update"]
@@ -85,6 +87,7 @@ atuin-dotfiles = { workspace = true }
 atuin-history = { workspace = true }
 atuin-vt100 = { workspace = true }
 atuin-kv = { workspace = true }
+atuin-octavo = { workspace = true, optional = true }
 atuin-pty-proxy = { workspace = true, optional = true }
 atuin-scripts = { workspace = true }
 

crates/atuin/i18n/en-US/atuin.ftl

@@ -135,6 +135,15 @@ value-config-set-the-type-float = Store the value as a float
 arg-config-enable-feature = The feature to enable
 value-config-enable-feature-daemon = Run the daemon, autostart it, and use it for search
 value-config-enable-feature-output-capture = Capture command output, via the daemon and the pty proxy
+value-config-enable-feature-octavo = Use Octavo, once it is enabled for your account on the hub
+config-enable-octavo-not-logged-in = You are not logged in to the hub, so it cannot say whether Octavo is enabled for your account. Run 'atuin login' first.
+config-enable-octavo-not-enabled = Octavo is not enabled for your account on { $address }.
+config-enable-octavo-description = Octavo is Atuin's sync for sharing and searching your history.
+config-enable-octavo-disclaimer = Unlike Atuin Sync, Octavo is not end-to-end encrypted, but your data is encrypted in transit and at rest. It's yours, and you can delete it at any time.
+config-enable-octavo-declined = Octavo was not enabled.
+config-enable-octavo-backfill = Also upload the { $count } entries already in your history, not just new commands.
+config-enable-octavo-backfilled = Queued { $count } history entries for upload.
+config-enable-octavo-backfill-old = { $count } entries recorded before Atuin v14 were left out: Octavo only takes the history ids Atuin has used since.
 arg-config-print-key = Print the value of a specific key and all its children
 arg-daemon-daemonize = Internal flag for daemonization
 arg-daemon-show-logs = Also write daemon logs to the console (useful for debugging)

crates/atuin/src/command/client/config.rs

@@ -1,6 +1,18 @@
+#[cfg(feature = "octavo")]
+use atuin_client::history::HistoryId;
 use atuin_client::settings::Settings;
+#[cfg(feature = "octavo")]
+use atuin_domain::caps::OctavoCap;
+#[cfg(feature = "octavo")]
+use atuin_octavo::hub::{HubCallError, HubClient, UserId};
+#[cfg(feature = "octavo")]
+use atuin_octavo::pb;
+#[cfg(feature = "octavo")]
+use atuin_octavo::queue::UploadQueue;
 use clap::{Args, Subcommand, ValueEnum};
 use eyre::Result;
+#[cfg(feature = "octavo")]
+use eyre::WrapErr;
 use toml_edit::{Document, DocumentMut, Item, Table, TableLike, Value};
 use tracing::instrument;
 
@@ -239,10 +251,25 @@ pub enum Feature {
     Daemon,
     #[value(help = fl!("value-config-enable-feature-output-capture"))]
     OutputCapture,
+    #[cfg(feature = "octavo")]
+    #[value(help = fl!("value-config-enable-feature-octavo"))]
+    Octavo,
 }
 
 impl EnableCmd {
     pub async fn run(self, settings: &Settings) -> Result<()> {
+        #[cfg(feature = "octavo")]
+        let octavo = match self.feature {
+            Feature::Octavo => {
+                let Some(onboarding) = onboard_octavo(settings).await? else {
+                    println!("{}", fl!("config-enable-octavo-declined"));
+                    return Ok(());
+                };
+                Some(onboarding)
+            }
+            Feature::Daemon | Feature::OutputCapture => None,
+        };
+
         let config_file = Settings::get_config_path()?;
         let config_str = tokio::fs::read_to_string(&config_file).await?;
 
@@ -251,13 +278,23 @@ impl EnableCmd {
 
         println!("Enabled.");
 
+        // Queued before the restart below so the daemon starts on it, and reported after it so a
+        // failure still leaves the daemon on the new config.
+        #[cfg(feature = "octavo")]
+        let backfilled = match octavo {
+            Some(onboarding) => onboarding.backfill().await,
+            None => Ok(()),
+        };
+        #[cfg(not(feature = "octavo"))]
+        let backfilled = Ok(());
+
         // The daemon reads these settings only at startup, so it needs a restart to pick them
         // up. A running daemon with autostart off may be externally managed (systemd, launchd),
         // so leave that one alone and tell the user instead.
         #[cfg(feature = "daemon")]
         if settings.daemon.enabled && !settings.daemon.autostart {
             println!("Restart the Atuin daemon and your shell for the change to take effect.");
-            return Ok(());
+            return backfilled;
         } else if let Err(e) = daemon::restart_cmd(settings).await {
             eprintln!(
                 "Could not restart the Atuin daemon: {e}\nRun `atuin daemon restart` manually."
@@ -266,7 +303,7 @@ impl EnableCmd {
 
         println!("Restart your shell for the change to take effect.");
 
-        Ok(())
+        backfilled
     }
 
     fn get_updated_config(&self, config_str: &str, daemon_enabled: bool) -> Result<String> {
@@ -285,6 +322,8 @@ impl EnableCmd {
                 set_deep_key(&mut doc, "pty_proxy.enabled", Value::from(true))?;
                 set_deep_key(&mut doc, "output.enabled", Value::from(true))?;
             }
+            #[cfg(feature = "octavo")]
+            Feature::Octavo => set_deep_key(&mut doc, "octavo.enabled", Value::from(true))?,
         }
 
         let updated = doc.to_string();
@@ -295,6 +334,93 @@ impl EnableCmd {
     }
 }
 
+/// What enabling Octavo queues once the config says it's on.
+#[cfg(feature = "octavo")]
+struct OctavoOnboarding {
+    queue: UploadQueue,
+    user: UserId,
+    backfill: Vec<HistoryId>,
+    old: usize,
+}
+
+#[cfg(feature = "octavo")]
+impl OctavoOnboarding {
+    async fn backfill(self) -> Result<()> {
+        if !self.backfill.is_empty() {
+            let queued = self.queue.push_many_history(&self.user, self.backfill).await.wrap_err(
+                "failed to queue your history for upload; run `atuin config enable octavo` again \
+                 to retry",
+            )?;
+            println!("{}", fl!("config-enable-octavo-backfilled", count = queued));
+        }
+        if self.old > 0 {
+            println!("{}", fl!("config-enable-octavo-backfill-old", count = self.old));
+        }
+
+        Ok(())
+    }
+}
+
+/// Asks the user to enable Octavo, and whether to upload the history they already have; `None`
+/// if they decline.
+#[cfg(feature = "octavo")]
+async fn onboard_octavo(settings: &Settings) -> Result<Option<OctavoOnboarding>> {
+    let login = match HubClient::new(settings)?.login(settings).await {
+        Ok(login) => login,
+        Err(HubCallError::NotLoggedIn) => eyre::bail!(fl!("config-enable-octavo-not-logged-in")),
+        Err(err) => return Err(err).wrap_err("failed to ask the hub who you are logged in as"),
+    };
+
+    let caps = atuin_client::api_client::caps_client(settings)?;
+    caps.refresh().await?;
+    if caps.get_server::<OctavoCap>().await?.is_none() {
+        eyre::bail!(fl!(
+            "config-enable-octavo-not-enabled",
+            address = settings.sync_address.to_string()
+        ));
+    }
+
+    let accepted = super::setup::prompt(
+        "Octavo",
+        &fl!("config-enable-octavo-description"),
+        Some(&fl!("config-enable-octavo-disclaimer")),
+        super::setup::DefaultAnswer::No,
+    )?;
+    if !accepted {
+        return Ok(None);
+    }
+
+    let history =
+        atuin_client::database::Sqlite::new(&settings.db_path, settings.local_timeout).await?;
+    let (ids, old): (Vec<_>, Vec<_>) =
+        history.active_ids().await?.into_iter().partition(|&id| pb::UuidV7::try_from(id).is_ok());
+
+    let backfill = super::setup::prompt(
+        "history backfill",
+        &fl!("config-enable-octavo-backfill", count = ids.len()),
+        None,
+        super::setup::DefaultAnswer::No,
+    )?;
+
+    let (backfill, old) = if backfill {
+        (ids, old.len())
+    } else {
+        (Vec::new(), 0)
+    };
+
+    // Opened before the config says Octavo is on, so a queue that can't be opened leaves it off.
+    let queue = UploadQueue::open(Settings::octavo_queue_path())
+        .await
+        .wrap_err("failed to open Octavo's upload queue")?;
+
+    Ok(Some(OctavoOnboarding {
+        queue,
+        user: login.user_id().clone(),
+        backfill,
+        old,
+    }))
+}
+
 #[derive(Args, Debug)]
 pub struct PrintCmd {
     #[arg(help = fl!("arg-config-print-key"))]

crates/atuin/src/command/client/history.rs

@@ -21,6 +21,12 @@ use atuin_daemon::grpc::history::pb::{
     TailHistoryReply, tail_history_reply::Event as TailEventProto,
 };
 use atuin_domain::record::CmdOrigin;
+#[cfg(feature = "octavo")]
+use atuin_octavo::hub::{HubCallError, HubClient};
+#[cfg(feature = "octavo")]
+use atuin_octavo::pb;
+#[cfg(feature = "octavo")]
+use atuin_octavo::queue::UploadQueue;
 use clap::Subcommand;
 #[cfg(feature = "daemon")]
 use colored::Colorize;
@@ -610,11 +616,49 @@ pub(super) async fn delete_history_entries(
         return Ok(());
     }
 
+    let entries: Vec<History> = entries.into_iter().collect();
+    #[cfg(feature = "octavo")]
+    let history_ids: Vec<HistoryId> = entries.iter().map(|h| h.id).collect();
+
     let ids = history_store.delete_entries(entries).await?;
     history_store.build_all(db, &ids).await?;
+
+    #[cfg(feature = "octavo")]
+    queue_octavo_deletions(settings, history_ids).await;
     Ok(())
 }
 
+/// Queues the deletion of `ids` from Octavo, for the daemon to send once it runs, if Octavo was
+/// ever on here.
+#[cfg(feature = "octavo")]
+async fn queue_octavo_deletions(settings: &Settings, ids: Vec<HistoryId>) {
+    let path = Settings::octavo_queue_path();
+    if !path.exists() {
+        return;
+    }
+
+    let login = async { HubClient::new(settings)?.login(settings).await };
+    let login = match login.await {
+        Ok(login) => login,
+        Err(HubCallError::NotLoggedIn) => return,
+        Err(err) => {
+            warn!(
+                ?err,
+                "failed to learn which hub user is logged in; deletions won't reach octavo"
+            );
+            return;
+        }
+    };
+
+    // Octavo never stored an entry whose id isn't a UUIDv7.
+    let stored = ids.into_iter().filter(|&id| pb::UuidV7::try_from(id).is_ok());
+    let queued =
+        async { UploadQueue::open(path).await?.push_deletions(login.user_id(), stored).await };
+    if let Err(err) = queued.await {
+        warn!(?err, "failed to queue history deletions for octavo; they will not reach it");
+    }
+}
+
 #[cfg(feature = "daemon")]
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum TailKind {

crates/atuin/src/command/client/setup.rs

@@ -15,12 +15,14 @@ pub async fn run(_settings: &Settings) -> Result<()> {
             "By default, Atuin AI only has access to the name and version of your operating \
              system and shell - your shell history is not sent to the AI.",
         ),
+        DefaultAnswer::Yes,
     )?;
 
     let enable_daemon = prompt(
         "Atuin Daemon",
         "This will enable improved search and history sync using a persistent background process",
         None,
+        DefaultAnswer::Yes,
     )?;
 
     let config_file = Settings::get_config_path()?;
@@ -57,19 +59,39 @@ pub async fn run(_settings: &Settings) -> Result<()> {
     Ok(())
 }
 
-pub fn prompt(feature: &str, description: &str, note: Option<&str>) -> Result<bool> {
+#[derive(Clone, Copy, Debug)]
+pub enum DefaultAnswer {
+    Yes,
+    #[cfg_attr(not(feature = "octavo"), expect(dead_code))]
+    No,
+}
+
+pub fn prompt(
+    feature: &str,
+    description: &str,
+    note: Option<&str>,
+    default: DefaultAnswer,
+) -> Result<bool> {
+    let q = match default {
+        DefaultAnswer::Yes => "[Y/n]",
+        DefaultAnswer::No => "[y/N]",
+    };
+
     println!("> Enable {feature}?", feature = feature.bold().bright_blue());
     if let Some(note) = note {
         println!("  {description}");
-        print!("  {note} {q} ", q = "[Y/n]".bold());
+        print!("  {note} {q} ", q = q.bold());
     } else {
-        print!("  {description} {q} ", q = "[Y/n]".bold());
+        print!("  {description} {q} ", q = q.bold());
     }
 
     io::stdout().flush().ok();
 
     let mut input = String::new();
     io::stdin().read_line(&mut input)?;
-    let answer = input.trim().to_lowercase();
-    Ok(answer.is_empty() || answer == "y" || answer == "yes")
+    Ok(match input.trim().to_lowercase().as_str() {
+        "" => matches!(default, DefaultAnswer::Yes),
+        "y" | "yes" => true,
+        _ => false,
+    })
 }

scripts/release.sh

@@ -447,10 +447,12 @@ publish_crates() {
 
     local crates=(
         atuin-common
+        atuin-domain
         atuin-client
         atuin-ai
         atuin-dotfiles
         atuin-history
+        atuin-octavo
         atuin-daemon
         atuin-kv
         atuin-scripts