3 files changed, 346 insertions, 85 deletions

diff --git a/crates/turtle/src/atuin_daemon/components/sync.rs b/crates/daemon/src/api/control.rs
index 20d49839..16b4bd94 100644
--- a/crates/turtle/src/atuin_daemon/components/sync.rs
+++ b/crates/daemon/src/api/control.rs
@@ -1,29 +1,26 @@
-//! Sync component.
-//!
-//! Handles periodic synchronization with the Atuin cloud server.
-
 use std::time::Duration;
 
 use eyre::Result;
-use rand::Rng;
-use tokio::sync::mpsc;
+use rand::RngExt;
 use tokio::time::{self, MissedTickBehavior};
+use tonic::{Request, Response, Status};
+use tracing::{Level, instrument};
 
-use crate::atuin_client::{history::store::HistoryStore, record::sync, settings::Settings};
+use turtle_api::generated::{
+    DAEMON_PROTOCOL_VERSION,
+    control::{
+        ForceSyncReply, ForceSyncRequest, PathsReply, PathsRequest, StatusReply, StatusRequest,
+        control_server::{Control, ControlServer},
+    },
+};
 
-use crate::atuin_daemon::{
-    daemon::{Component, DaemonHandle},
+use crate::{
+    DAEMON_VERSION,
+    aclient::{history::store::HistoryStore, record::sync, settings::Settings},
+    daemon::DaemonHandle,
     events::DaemonEvent,
 };
 
-/// Commands that can be sent to the sync task.
-enum SyncCommand {
-    /// Trigger an immediate sync.
-    ForceSync,
-    /// Stop the sync loop.
-    Stop,
-}
-
 /// Sync state - tracks whether we're in normal operation or retrying after failure.
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum SyncState {
@@ -34,77 +31,98 @@ enum SyncState {
     Retrying,
 }
 
-/// Sync component - handles periodic cloud synchronization.
+/// The Control gRPC service.
 ///
-/// This component:
-/// - Runs a background sync loop on a configurable interval
-/// - Implements exponential backoff on sync failures
-/// - Responds to [`ForceSync`] events for immediate sync
-/// - Emits SyncCompleted/SyncFailed events
-pub(crate) struct SyncComponent {
-    task_handle: Option<tokio::task::JoinHandle<()>>,
-    command_tx: Option<mpsc::Sender<SyncCommand>>,
+/// This service is used by external processes to inject events into the daemon.
+/// It's not a component - it's part of the daemon's core infrastructure.
+pub(crate) struct ControlService {
+    handle: DaemonHandle,
 }
 
-impl SyncComponent {
-    /// Create a new sync component.
-    pub(crate) fn new() -> Self {
-        Self {
-            task_handle: None,
-            command_tx: None,
-        }
+impl ControlService {
+    /// Create a new control service with the given daemon handle.
+    pub(crate) fn new(handle: DaemonHandle) -> Self {
+        tokio::spawn(sync_loop(handle.clone()));
+
+        Self { handle }
     }
-}
 
-impl Default for SyncComponent {
-    fn default() -> Self {
-        Self::new()
+    /// Get a tonic server for this service.
+    pub(crate) fn into_server(self) -> ControlServer<Self> {
+        ControlServer::new(self)
     }
 }
 
 #[tonic::async_trait]
-impl Component for SyncComponent {
-    fn name(&self) -> &'static str {
-        "sync"
-    }
+impl Control for ControlService {
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn paths(&self, _request: Request<PathsRequest>) -> Result<Response<PathsReply>, Status> {
+        let settings = self.handle.settings().await;
 
-    async fn start(&mut self, handle: DaemonHandle) -> Result<()> {
-        let (cmd_tx, cmd_rx) = mpsc::channel(16);
-        self.command_tx = Some(cmd_tx);
+        let config = Settings::get_config_path()
+            .map_err(|e| Status::internal(format!("failed to get settings path: {e:?}")))?;
 
-        // Spawn the sync loop with its own copy of the handle
-        self.task_handle = Some(tokio::spawn(sync_loop(handle, cmd_rx)));
+        let reply = PathsReply {
+            config: config.to_string_lossy().to_string(),
+            db: settings.db_path.clone(),
+            socket: settings.daemon.socket_path.clone(),
+        };
 
-        tracing::info!("sync component started");
-        Ok(())
+        Ok(Response::new(reply))
     }
 
-    async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> {
-        match event {
-            DaemonEvent::ForceSync => {
-                tracing::info!("force sync requested");
-                if let Some(tx) = &self.command_tx {
-                    drop(tx.send(SyncCommand::ForceSync).await);
-                }
-            }
-            DaemonEvent::SyncFailed { error } => {
-                tracing::error!(?error, "Sync failed.");
-            }
-            _ => (),
-        }
-        Ok(())
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn status(
+        &self,
+        _request: Request<StatusRequest>,
+    ) -> Result<Response<StatusReply>, Status> {
+        let reply = StatusReply {
+            healthy: true,
+            version: DAEMON_VERSION.to_owned(),
+            pid: std::process::id(),
+            protocol: DAEMON_PROTOCOL_VERSION,
+        };
+
+        Ok(Response::new(reply))
     }
 
-    async fn stop(&mut self) -> Result<()> {
-        if let Some(tx) = &self.command_tx {
-            drop(tx.send(SyncCommand::Stop).await);
-        }
-        if let Some(handle) = self.task_handle.take() {
-            // Give the task a moment to shut down gracefully
-            drop(time::timeout(Duration::from_secs(5), handle).await);
-        }
-        tracing::info!("sync component stopped");
-        Ok(())
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn force_sync(
+        &self,
+        _request: Request<ForceSyncRequest>,
+    ) -> Result<Response<ForceSyncReply>, Status> {
+        self.handle.emit(DaemonEvent::ForceSync);
+        let event = self
+            .handle
+            .wait_for(|e| {
+                matches!(
+                    e,
+                    DaemonEvent::SyncFailed { .. } | DaemonEvent::SyncCompleted { .. }
+                )
+            })
+            .await
+            .map_err(|e| {
+                Status::internal(format!("failed to wait for sync response event: {e:?}"))
+            })?;
+
+        let reply = match event {
+            DaemonEvent::SyncCompleted {
+                uploaded,
+                downloaded,
+            } => ForceSyncReply {
+                error: None,
+                uploaded: uploaded as u32,
+                downloaded: downloaded as u32,
+            },
+            DaemonEvent::SyncFailed { error } => ForceSyncReply {
+                error: Some(error),
+                uploaded: 0,
+                downloaded: 0,
+            },
+            _ => unreachable!(),
+        };
+
+        Ok(Response::new(reply))
     }
 }
 
@@ -113,7 +131,7 @@ impl Component for SyncComponent {
 /// This runs in a spawned task and handles periodic sync as well as
 /// force sync requests.
 #[expect(clippy::significant_drop_tightening, reason = "false positive")]
-async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>) {
+async fn sync_loop(handle: DaemonHandle) {
     tracing::info!("sync loop starting");
 
     // Clone settings since we need them across await points
@@ -131,7 +149,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>
     let history_store = HistoryStore::new(handle.store().clone(), host_id, encryption_key);
 
     // Don't backoff by more than 30 mins (with a random jitter of up to 1 min)
-    let max_interval: f64 = 60.0f64.mul_add(30.0, rand::thread_rng().gen_range(0.0..60.0));
+    let max_interval: f64 = 60.0f64.mul_add(30.0, rand::rng().random_range(0.0..60.0));
 
     let mut ticker = time::interval(Duration::from_secs(settings.daemon.sync_frequency));
 
@@ -141,6 +159,7 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>
 
     let mut sync_state = SyncState::Idle;
 
+    let mut daemon_rx = handle.subscribe();
     loop {
         tokio::select! {
             _ = ticker.tick() => {
@@ -161,9 +180,9 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>
                     &settings,
                 ).await;
             }
-            cmd = cmd_rx.recv() => {
+            cmd = daemon_rx.recv() => {
                 match cmd {
-                    Some(SyncCommand::ForceSync) => {
+                    Ok(DaemonEvent::ForceSync) => {
                         tracing::info!("executing force sync");
                         let settings = handle.settings().await;
                         sync_state = do_sync_tick(
@@ -173,11 +192,12 @@ async fn sync_loop(handle: DaemonHandle, mut cmd_rx: mpsc::Receiver<SyncCommand>
                             max_interval,
                             &settings,
                         ).await;
-                    }
-                    Some(SyncCommand::Stop) | None => {
+                    },
+                    Ok(DaemonEvent::ShutdownRequested) | Err(_) => {
                         tracing::info!("sync loop stopping");
                         break;
-                    }
+                    },
+                    _ => ()
                 }
             }
         }
@@ -217,14 +237,13 @@ async fn do_sync_tick(
         Err(e) => {
             tracing::error!("sync tick failed with {e}");
 
-            // Emit failure event
             handle.emit(DaemonEvent::SyncFailed {
                 error: e.to_string(),
             });
 
             // Exponential backoff
-            let mut rng = rand::thread_rng();
-            let mut new_interval = ticker.period().as_secs_f64() * rng.gen_range(2.0..2.2);
+            let mut rng = rand::rng();
+            let mut new_interval = ticker.period().as_secs_f64() * rng.random_range(2.0..2.2);
 
             if new_interval > max_interval {
                 new_interval = max_interval;
@@ -256,9 +275,6 @@ async fn do_sync_tick(
                 tracing::error!("failed to build history from downloaded records: {e}");
             }
 
-            // Emit the records added event (for search indexing)
-            handle.emit(DaemonEvent::RecordsAdded(downloaded_records.clone()));
-
             // Emit sync completed event
             handle.emit(DaemonEvent::SyncCompleted {
                 uploaded: uploaded_count as usize,
diff --git a/crates/daemon/src/api/history.rs b/crates/daemon/src/api/history.rs
new file mode 100644
index 00000000..078e8c64
--- /dev/null
+++ b/crates/daemon/src/api/history.rs
@@ -0,0 +1,243 @@
+use std::{pin::Pin, time::Duration};
+
+use dashmap::DashMap;
+use eyre::Result;
+use time::OffsetDateTime;
+use tokio_stream::Stream;
+use tonic::{Request, Response, Status};
+use tracing::{Level, instrument};
+
+use crate::{
+    aclient::{history::store::HistoryStore, settings::Settings},
+    daemon::DaemonHandle,
+    events::DaemonEvent,
+};
+use turtle_api::{
+    client::{
+        proto_duration_to_std, proto_timestamp_to_time, std_to_proto_duration,
+        time_to_proto_timestamp,
+    },
+    generated::{
+        DAEMON_PROTOCOL_VERSION,
+        history::{
+            EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, HistoryReply,
+            HistoryRequest, StartHistoryReply, StartHistoryRequest, TailHistoryReply,
+            TailHistoryRequest,
+            history_server::{History as HistorySvc, HistoryServer},
+        },
+    },
+    history::{History, HistoryId},
+};
+
+/// The gRPC service implementation.
+///
+/// This is a thin wrapper that delegates to the component's shared state.
+pub(crate) struct HistoryService {
+    /// Commands currently running (not yet completed).
+    running: DashMap<HistoryId, History>,
+
+    /// Handle to the daemon (set during start).
+    handle: DaemonHandle,
+
+    /// History store for pushing records
+    history_store: HistoryStore,
+}
+
+impl HistoryService {
+    pub(crate) async fn new(handle: DaemonHandle) -> Result<Self> {
+        let host_id = Settings::host_id().await?;
+        let history_store =
+            HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key());
+
+        Ok(Self {
+            running: DashMap::new(),
+            handle,
+            history_store,
+        })
+    }
+
+    /// Get a tonic server for this service.
+    pub(crate) fn into_server(self) -> HistoryServer<Self> {
+        HistoryServer::new(self)
+    }
+}
+
+fn history_to_reply(history: History) -> HistoryEntry {
+    HistoryEntry {
+        timestamp: time_to_proto_timestamp(history.timestamp),
+        id: history.id.to_string(),
+        command: history.command,
+        cwd: history.cwd,
+        session: history.session,
+        hostname: history.hostname,
+        author: history.author,
+        intent: history.intent.unwrap_or_default(),
+        exit: history.exit,
+        duration: std_to_proto_duration(history.duration),
+    }
+}
+
+#[tonic::async_trait]
+impl HistorySvc for HistoryService {
+    type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>;
+
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn history(
+        &self,
+        request: Request<HistoryRequest>,
+    ) -> Result<Response<HistoryReply>, Status> {
+        let req = request.into_inner();
+
+        let entries = if let Some(range) = req.range {
+            let from = proto_timestamp_to_time(range.start);
+            let to = proto_timestamp_to_time(range.end);
+
+            self.handle.history_db().range(from, to).await
+        } else {
+            self.handle.history_db().list(None, false, false).await
+        }
+        .map_err(|e| Status::internal(format!("failed to read db: {e:?}")))?
+        .into_iter()
+        .map(history_to_reply)
+        .collect();
+
+        Ok(Response::new(HistoryReply { entries }))
+    }
+
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn start_history(
+        &self,
+        request: Request<StartHistoryRequest>,
+    ) -> Result<Response<StartHistoryReply>, Status> {
+        let req = request.into_inner();
+
+        let timestamp = proto_timestamp_to_time(req.timestamp);
+
+        let h: History = History::daemon()
+            .timestamp(timestamp)
+            .command(req.command)
+            .cwd(req.cwd)
+            .session(req.session)
+            .hostname(req.hostname)
+            .author(req.author)
+            .intent(req.intent.unwrap_or_default())
+            .build()
+            .into();
+
+        self.handle.emit(DaemonEvent::HistoryStarted(h.clone()));
+
+        let id = h.id;
+        tracing::info!(id = id.to_string(), "start history called");
+        self.running.insert(id, h);
+
+        let reply = StartHistoryReply {
+            id: id.to_string(),
+            version: env!("CARGO_PKG_VERSION").to_string(),
+            protocol: DAEMON_PROTOCOL_VERSION,
+        };
+
+        Ok(Response::new(reply))
+    }
+
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn end_history(
+        &self,
+        request: Request<EndHistoryRequest>,
+    ) -> Result<Response<EndHistoryReply>, Status> {
+        let req = request.into_inner();
+        let id = HistoryId::from(req.id);
+
+        tracing::info!(id = id.to_string(), "end history called");
+
+        if let Some((_, mut history)) = self.running.remove(&id) {
+            history.exit = req.exit;
+            history.duration = match proto_duration_to_std(req.duration) {
+                Duration::ZERO => Duration::from_nanos_u128(
+                    (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds() as u128,
+                ),
+                value => value,
+            };
+
+            self.handle
+                .history_db()
+                .save(&history)
+                .await
+                .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?;
+
+            tracing::info!(
+                id = id.to_string(),
+                duration = history.duration.as_nanos(),
+                "end history"
+            );
+
+            let (record_id, idx) = self
+                .history_store
+                .push(history.clone())
+                .await
+                .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?;
+
+            self.handle.emit(DaemonEvent::HistoryEnded(history));
+
+            let reply = EndHistoryReply {
+                id: record_id.0.to_string(),
+                idx,
+                version: env!("CARGO_PKG_VERSION").to_string(),
+                protocol: DAEMON_PROTOCOL_VERSION,
+            };
+
+            return Ok(Response::new(reply));
+        }
+
+        Err(Status::not_found(format!(
+            "could not find history with id: {id}"
+        )))
+    }
+
+    #[instrument(skip_all, level = Level::INFO)]
+    async fn tail_history(
+        &self,
+        _request: Request<TailHistoryRequest>,
+    ) -> Result<Response<Self::TailHistoryStream>, Status> {
+        let mut rx = self.handle.subscribe();
+        let (tx, out_rx) = tokio::sync::mpsc::channel::<Result<TailHistoryReply, Status>>(128);
+
+        tokio::spawn(async move {
+            loop {
+                let event = match rx.recv().await {
+                    Ok(event) => event,
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
+                        drop(
+                            tx.send(Err(Status::resource_exhausted(format!(
+                                "tail stream lagged behind and dropped {skipped} events"
+                            ))))
+                            .await,
+                        );
+                        break;
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                };
+
+                let reply = match event {
+                    DaemonEvent::HistoryStarted(history) => Some(TailHistoryReply {
+                        kind: HistoryEventKind::Started.into(),
+                        history: Some(history_to_reply(history)),
+                    }),
+                    DaemonEvent::HistoryEnded(history) => Some(TailHistoryReply {
+                        kind: HistoryEventKind::Ended.into(),
+                        history: Some(history_to_reply(history)),
+                    }),
+                    _ => None,
+                };
+
+                if let Some(reply) = reply
+                    && tx.send(Ok(reply)).await.is_err()
+                {
+                    break;
+                }
+            }
+        });
+
+        let stream = tokio_stream::wrappers::ReceiverStream::new(out_rx);
+        Ok(Response::new(Box::pin(stream)))
+    }
+}
diff --git a/crates/daemon/src/api/mod.rs b/crates/daemon/src/api/mod.rs
new file mode 100644
index 00000000..8d475fe9
--- /dev/null
+++ b/crates/daemon/src/api/mod.rs
@@ -0,0 +1,2 @@
+pub(crate) mod control;
+pub(crate) mod history;