chore: Commit

1 files changed, 6 insertions, 287 deletions

diff --git a/crates/daemon/src/components/history.rs b/crates/daemon/src/components/history.rs
index 7a0882a3..b476f627 100644
--- a/crates/daemon/src/components/history.rs
+++ b/crates/daemon/src/components/history.rs
@@ -2,84 +2,19 @@
 //!
 //! Handles command history lifecycle (start/end) and provides the History gRPC service.
 
-use std::{pin::Pin, sync::Arc};
-
-use crate::aclient::{
-    history::{History, HistoryId, store::HistoryStore},
-    settings::Settings,
+use crate::{
+    aclient::{history::store::HistoryStore, settings::Settings},
+    api::server::history::HistoryGrpcService,
 };
-use dashmap::DashMap;
 use eyre::Result;
-use time::OffsetDateTime;
-use tokio_stream::Stream;
-use tonic::{Request, Response, Status};
-use tracing::{Level, instrument};
 
 use crate::{
     daemon::{Component, DaemonHandle},
     events::DaemonEvent,
-    generated::history::{
-        EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, ShutdownReply,
-        ShutdownRequest, StartHistoryReply, StartHistoryRequest, StatusReply, StatusRequest,
-        TailHistoryReply, TailHistoryRequest,
-        history_server::{History as HistorySvc, HistoryServer},
-    },
 };
 
-const DAEMON_PROTOCOL_VERSION: u32 = 1;
-
-/// History component - manages command history lifecycle.
-///
-/// This component:
-/// - Tracks currently running commands (stored in memory)
-/// - Saves completed commands to the database and record store
-/// - Emits history events for other components (e.g., search indexing)
-/// - Provides the History gRPC service
-pub(crate) struct HistoryComponent {
-    inner: Arc<HistoryComponentInner>,
-}
-
-struct HistoryComponentInner {
-    /// Commands currently running (not yet completed).
-    running: DashMap<HistoryId, History>,
-
-    /// Handle to the daemon (set during start).
-    handle: tokio::sync::RwLock<Option<DaemonHandle>>,
-
-    /// History store for pushing records (set during start).
-    history_store: tokio::sync::RwLock<Option<HistoryStore>>,
-}
-
-impl HistoryComponent {
-    /// Create a new history component.
-    pub(crate) fn new() -> Self {
-        Self {
-            inner: Arc::new(HistoryComponentInner {
-                running: DashMap::new(),
-                handle: tokio::sync::RwLock::new(None),
-                history_store: tokio::sync::RwLock::new(None),
-            }),
-        }
-    }
-
-    /// Get the gRPC service for this component.
-    ///
-    /// This returns a tonic service that can be added to a gRPC server.
-    pub(crate) fn grpc_service(&self) -> HistoryServer<HistoryGrpcService> {
-        HistoryServer::new(HistoryGrpcService {
-            inner: self.inner.clone(),
-        })
-    }
-}
-
-impl Default for HistoryComponent {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[tonic::async_trait]
-impl Component for HistoryComponent {
+impl Component for HistoryGrpcService {
     fn name(&self) -> &'static str {
         "history"
     }
@@ -90,8 +25,8 @@ impl Component for HistoryComponent {
         let history_store =
             HistoryStore::new(handle.store().clone(), host_id, *handle.encryption_key());
 
-        *self.inner.history_store.write().await = Some(history_store);
-        *self.inner.handle.write().await = Some(handle);
+        *self.history_store.write().await = Some(history_store);
+        *self.handle.write().await = Some(handle);
 
         tracing::info!("history component started");
         Ok(())
@@ -107,219 +42,3 @@ impl Component for HistoryComponent {
         Ok(())
     }
 }
-
-/// The gRPC service implementation.
-///
-/// This is a thin wrapper that delegates to the component's shared state.
-pub(crate) struct HistoryGrpcService {
-    inner: Arc<HistoryComponentInner>,
-}
-
-fn history_to_tail_reply(kind: HistoryEventKind, history: History) -> TailHistoryReply {
-    TailHistoryReply {
-        kind: kind as i32,
-        history: Some(HistoryEntry {
-            timestamp: history.timestamp.unix_timestamp_nanos() as u64,
-            id: history.id.0,
-            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: history.duration,
-        }),
-    }
-}
-
-#[tonic::async_trait]
-impl HistorySvc for HistoryGrpcService {
-    type TailHistoryStream = Pin<Box<dyn Stream<Item = Result<TailHistoryReply, Status>> + Send>>;
-
-    #[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 = OffsetDateTime::from_unix_timestamp_nanos(i128::from(req.timestamp))
-            .map_err(|_| {
-                Status::invalid_argument(
-                    "failed to parse timestamp as unix time (expected nanos since epoch)",
-                )
-            })?;
-
-        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)
-            .build()
-            .into();
-
-        // Emit the event
-        if let Some(handle) = self.inner.handle.read().await.as_ref() {
-            handle.emit(DaemonEvent::HistoryStarted(h.clone()));
-        }
-
-        let id = h.id.clone();
-        tracing::info!(id = id.to_string(), "start history");
-        self.inner.running.insert(id.clone(), 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)]
-    #[expect(clippy::significant_drop_tightening, reason = "Would be a logic-bug")]
-    async fn end_history(
-        &self,
-        request: Request<EndHistoryRequest>,
-    ) -> Result<Response<EndHistoryReply>, Status> {
-        let req = request.into_inner();
-        let id = HistoryId(req.id);
-
-        if let Some((_, mut history)) = self.inner.running.remove(&id) {
-            history.exit = req.exit;
-            history.duration = match req.duration {
-                0 => i64::try_from(
-                    (OffsetDateTime::now_utc() - history.timestamp).whole_nanoseconds(),
-                )
-                .expect("failed to convert calculated duration to i64"),
-                value => i64::try_from(value).expect("failed to get i64 duration"),
-            };
-
-            // Get the handle and store to save the history
-            let handle_guard = self.inner.handle.read().await;
-            let handle = handle_guard
-                .as_ref()
-                .ok_or_else(|| Status::internal("component not initialized"))?;
-
-            let store_guard = self.inner.history_store.read().await;
-            let history_store = store_guard
-                .as_ref()
-                .ok_or_else(|| Status::internal("component not initialized"))?;
-
-            // Save to database
-            handle
-                .history_db()
-                .save(&history)
-                .await
-                .map_err(|e| Status::internal(format!("failed to write to db: {e:?}")))?;
-
-            tracing::info!(id = id.0, duration = history.duration, "end history");
-
-            // Push to record store
-            let (record_id, idx) = history_store
-                .push(history.clone())
-                .await
-                .map_err(|e| Status::internal(format!("failed to push record to store: {e:?}")))?;
-
-            // Emit the event
-            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)]
-    #[expect(clippy::significant_drop_tightening, reason = "Would be a logic-bug")]
-    async fn tail_history(
-        &self,
-        _request: Request<TailHistoryRequest>,
-    ) -> Result<Response<Self::TailHistoryStream>, Status> {
-        let handle_guard = self.inner.handle.read().await;
-        let handle = handle_guard
-            .as_ref()
-            .cloned()
-            .ok_or_else(|| Status::internal("component not initialized"))?;
-
-        let mut rx = 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(history_to_tail_reply(HistoryEventKind::Started, history))
-                    }
-                    DaemonEvent::HistoryEnded(history) => {
-                        Some(history_to_tail_reply(HistoryEventKind::Ended, 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)))
-    }
-
-    #[instrument(skip_all, level = Level::INFO)]
-    async fn status(
-        &self,
-        _request: Request<StatusRequest>,
-    ) -> Result<Response<StatusReply>, Status> {
-        let reply = StatusReply {
-            healthy: true,
-            version: env!("CARGO_PKG_VERSION").to_string(),
-            pid: std::process::id(),
-            protocol: DAEMON_PROTOCOL_VERSION,
-        };
-
-        Ok(Response::new(reply))
-    }
-
-    #[instrument(skip_all, level = Level::INFO)]
-    async fn shutdown(
-        &self,
-        _request: Request<ShutdownRequest>,
-    ) -> Result<Response<ShutdownReply>, Status> {
-        // Use the daemon handle to request shutdown
-        if let Some(handle) = self.inner.handle.read().await.as_ref() {
-            handle.shutdown();
-        }
-        Ok(Response::new(ShutdownReply { accepted: true }))
-    }
-}