aboutsummaryrefslogtreecommitdiffstats
path: root/crates/daemon/src
diff options
context:
space:
mode:
authorBenedikt Peetz <benedikt.peetz@b-peetz.de>2026-07-20 14:18:36 +0200
committerBenedikt Peetz <benedikt.peetz@b-peetz.de>2026-07-20 14:18:36 +0200
commit63a0ec3901a863fb07d18f4b814a98a813644382 (patch)
treec94b7b501601b5df05cf654e9f9c9b5bc66a239e /crates/daemon/src
parentchore: Commit (diff)
downloadatuin-63a0ec3901a863fb07d18f4b814a98a813644382.zip
chore: Commit
Diffstat (limited to 'crates/daemon/src')
-rw-r--r--crates/daemon/src/api/client/mod.rs (renamed from crates/daemon/src/client.rs)140
-rw-r--r--crates/daemon/src/api/generated.rs (renamed from crates/daemon/src/generated.rs)10
-rw-r--r--crates/daemon/src/api/mod.rs6
-rw-r--r--crates/daemon/src/api/server/control.rs61
-rw-r--r--crates/daemon/src/api/server/history.rs229
-rw-r--r--crates/daemon/src/api/server/mod.rs2
-rw-r--r--crates/daemon/src/components/history.rs293
-rw-r--r--crates/daemon/src/components/mod.rs4
-rw-r--r--crates/daemon/src/components/semantic.rs912
-rw-r--r--crates/daemon/src/control/mod.rs74
-rw-r--r--crates/daemon/src/daemon.rs30
-rw-r--r--crates/daemon/src/lib.rs53
-rw-r--r--crates/daemon/src/main.rs170
-rw-r--r--crates/daemon/src/server.rs73
14 files changed, 414 insertions, 1643 deletions
diff --git a/crates/daemon/src/client.rs b/crates/daemon/src/api/client/mod.rs
index 5f8ea0f2..d6cbbe85 100644
--- a/crates/daemon/src/client.rs
+++ b/crates/daemon/src/api/client/mod.rs
@@ -8,30 +8,33 @@ use hyper_util::rt::TokioIo;
#[cfg(unix)]
use tokio::net::UnixStream;
-use crate::aclient::{history::History, settings::Settings};
-use crate::generated;
use crate::{
- events::DaemonEvent,
- generated::{
- control::{
- ForceSyncEvent, HistoryDeletedEvent, HistoryPrunedEvent, HistoryRebuiltEvent,
- SendEventRequest, SettingsReloadedEvent, ShutdownEvent,
- control_client::ControlClient as ControlServiceClient,
- },
- history::{
- EndHistoryReply, EndHistoryRequest, ShutdownRequest, StartHistoryReply,
- StartHistoryRequest, StatusReply, StatusRequest, TailHistoryReply, TailHistoryRequest,
- history_client::HistoryClient as HistoryServiceClient,
- },
- semantic::{
- CommandCapture, RecordCommandsReply,
- semantic_client::SemanticClient as SemanticServiceClient,
+ aclient::{history::History, settings::Settings},
+ api::{
+ DAEMON_PROTOCOL_VERSION, DAEMON_VERSION,
+ generated::{
+ control::{
+ StatusReply, StatusRequest, control_client::ControlClient as ControlServiceClient,
+ },
+ history::{
+ EndHistoryReply, EndHistoryRequest, StartHistoryReply, StartHistoryRequest,
+ TailHistoryReply, TailHistoryRequest,
+ history_client::HistoryClient as HistoryServiceClient,
+ },
},
},
};
-pub struct HistoryClient {
- client: HistoryServiceClient<Channel>,
+fn daemon_matches_expected(version: &str, protocol: u32) -> bool {
+ version == DAEMON_VERSION && protocol == DAEMON_PROTOCOL_VERSION
+}
+
+fn daemon_mismatch_message(version: &str, protocol: u32) -> String {
+ if protocol == DAEMON_PROTOCOL_VERSION {
+ format!("daemon is out of date: expected {DAEMON_VERSION}, got {version}")
+ } else {
+ format!("daemon protocol mismatch: expected {DAEMON_PROTOCOL_VERSION}, got {protocol}")
+ }
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -61,6 +64,14 @@ pub fn classify_error(error: &eyre::Report) -> DaemonClientErrorKind {
DaemonClientErrorKind::Other
}
+// ============================================================================
+// History Client
+// ============================================================================
+
+pub struct HistoryClient {
+ client: HistoryServiceClient<Channel>,
+}
+
// Wrap the grpc client
impl HistoryClient {
#[cfg(unix)]
@@ -114,10 +125,6 @@ impl HistoryClient {
Ok(self.client.end_history(req).await?.into_inner())
}
- pub async fn status(&mut self) -> Result<StatusReply> {
- Ok(self.client.status(StatusRequest {}).await?.into_inner())
- }
-
pub(crate) async fn tail_history(&mut self) -> Result<tonic::Streaming<TailHistoryReply>> {
Ok(self
.client
@@ -125,54 +132,6 @@ impl HistoryClient {
.await?
.into_inner())
}
-
- pub async fn shutdown(&mut self) -> Result<bool> {
- let resp = self.client.shutdown(ShutdownRequest {}).await?.into_inner();
- Ok(resp.accepted)
- }
-}
-
-pub(crate) struct SemanticClient {
- client: SemanticServiceClient<Channel>,
-}
-
-impl SemanticClient {
- #[cfg(unix)]
- pub(crate) async fn new(path: String) -> Result<Self> {
- let log_path = path.clone();
- let channel = Endpoint::try_from("http://atuin_local_daemon:0")?
- .connect_with_connector(service_fn(move |_: Uri| {
- let path = path.clone();
-
- async move {
- Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path.clone()).await?))
- }
- }))
- .await
- .wrap_err_with(|| {
- format!(
- "failed to connect to local atuin daemon at {}. Is it running?",
- &log_path
- )
- })?;
-
- let client = SemanticServiceClient::new(channel);
-
- Ok(Self { client })
- }
-
- #[cfg(unix)]
- pub(crate) async fn from_settings(settings: &Settings) -> Result<Self> {
- Self::new(settings.daemon.socket_path.clone()).await
- }
-
- pub(crate) async fn record_commands(
- &mut self,
- captures: Vec<CommandCapture>,
- ) -> Result<RecordCommandsReply> {
- let stream = tokio_stream::iter(captures);
- Ok(self.client.record_commands(stream).await?.into_inner())
- }
}
// ============================================================================
@@ -180,15 +139,12 @@ impl SemanticClient {
// ============================================================================
/// Client for the Control gRPC service.
-///
-/// Used to inject events into a running daemon from external processes.
pub struct ControlClient {
client: ControlServiceClient<Channel>,
}
impl ControlClient {
/// Connect to the daemon's control service.
- #[cfg(unix)]
pub(crate) async fn new(path: String) -> Result<Self> {
let log_path = path.clone();
let channel = Endpoint::try_from("http://atuin_local_daemon:0")?
@@ -218,39 +174,7 @@ impl ControlClient {
Self::new(settings.daemon.socket_path.clone()).await
}
- /// Send an event to the daemon.
- pub async fn send_event(&mut self, event: DaemonEvent) -> Result<()> {
- let proto_event = daemon_event_to_proto(event);
- let request = SendEventRequest {
- event: Some(proto_event),
- };
- self.client.send_event(request).await?;
- Ok(())
- }
-}
-
-/// Convert a daemon event to its proto representation.
-fn daemon_event_to_proto(event: DaemonEvent) -> generated::control::send_event_request::Event {
- use generated::control::send_event_request::Event;
-
- match event {
- DaemonEvent::HistoryPruned => Event::HistoryPruned(HistoryPrunedEvent {}),
- DaemonEvent::HistoryRebuilt => Event::HistoryRebuilt(HistoryRebuiltEvent {}),
- DaemonEvent::HistoryDeleted { ids } => Event::HistoryDeleted(HistoryDeletedEvent {
- ids: ids.into_iter().map(|id| id.0).collect(),
- }),
- DaemonEvent::ForceSync => Event::ForceSync(ForceSyncEvent {}),
- DaemonEvent::SettingsReloaded => Event::SettingsReloaded(SettingsReloadedEvent {}),
- DaemonEvent::ShutdownRequested => Event::Shutdown(ShutdownEvent {}),
- // These events are internal and not sent via the control service
- DaemonEvent::HistoryStarted(_)
- | DaemonEvent::HistoryEnded(_)
- | DaemonEvent::RecordsAdded(_)
- | DaemonEvent::SyncCompleted { .. }
- | DaemonEvent::SyncFailed { .. } => {
- // Use shutdown as a fallback, though this shouldn't happen
- tracing::warn!("attempted to send internal event via control service");
- Event::Shutdown(ShutdownEvent {})
- }
+ pub async fn status(&mut self) -> Result<StatusReply> {
+ Ok(self.client.status(StatusRequest {}).await?.into_inner())
}
}
diff --git a/crates/daemon/src/generated.rs b/crates/daemon/src/api/generated.rs
index 9deb4e0c..304edcd9 100644
--- a/crates/daemon/src/generated.rs
+++ b/crates/daemon/src/api/generated.rs
@@ -19,18 +19,10 @@ pub(crate) mod semantic {
///
/// This module contains the proto-generated types for the history gRPC service.
pub(crate) mod history {
- // Include the generated proto code
tonic::include_proto!("history");
}
-/// Control module for external event injection.
-///
-/// This module provides the gRPC service that allows external processes
-/// (like CLI commands) to inject events into the daemon's event bus.
+/// Control module for external control.
pub(crate) mod control {
- // Include the generated proto code
tonic::include_proto!("control");
-
- // Re-export the service
- pub(crate) use crate::control::ControlService;
}
diff --git a/crates/daemon/src/api/mod.rs b/crates/daemon/src/api/mod.rs
new file mode 100644
index 00000000..e6f8f1f0
--- /dev/null
+++ b/crates/daemon/src/api/mod.rs
@@ -0,0 +1,6 @@
+pub(crate) mod client;
+pub(crate) mod server;
+pub(crate) mod generated;
+
+const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
+const DAEMON_PROTOCOL_VERSION: u32 = 1;
diff --git a/crates/daemon/src/api/server/control.rs b/crates/daemon/src/api/server/control.rs
new file mode 100644
index 00000000..63fef340
--- /dev/null
+++ b/crates/daemon/src/api/server/control.rs
@@ -0,0 +1,61 @@
+use tonic::{Request, Response, Status};
+use tracing::{Level, instrument};
+
+use crate::{
+ api::{
+ DAEMON_PROTOCOL_VERSION, DAEMON_VERSION,
+ generated::control::{
+ ForceSyncReply, ForceSyncRequest, StatusReply, StatusRequest,
+ control_server::{Control, ControlServer},
+ },
+ },
+ daemon::DaemonHandle,
+};
+
+/// The Control gRPC service.
+///
+/// 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 ControlService {
+ /// Create a new control service with the given daemon handle.
+ pub(crate) fn new(handle: DaemonHandle) -> Self {
+ Self { handle }
+ }
+
+ /// Get a tonic server for this service.
+ pub(crate) fn into_server(self) -> ControlServer<Self> {
+ ControlServer::new(self)
+ }
+}
+
+#[tonic::async_trait]
+impl Control for ControlService {
+ #[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))
+ }
+
+ #[instrument(skip_all, level = Level::INFO)]
+ async fn force_sync(
+ &self,
+ _request: Request<ForceSyncRequest>,
+ ) -> Result<Response<ForceSyncReply>, Status> {
+ let reply = ForceSyncReply { accepted: false };
+
+ Ok(Response::new(reply))
+ }
+}
diff --git a/crates/daemon/src/api/server/history.rs b/crates/daemon/src/api/server/history.rs
new file mode 100644
index 00000000..032876a6
--- /dev/null
+++ b/crates/daemon/src/api/server/history.rs
@@ -0,0 +1,229 @@
+use std::pin::Pin;
+
+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::{History, HistoryId, store::HistoryStore},
+ api::{
+ DAEMON_PROTOCOL_VERSION,
+ generated::history::{
+ EndHistoryReply, EndHistoryRequest, HistoryEntry, HistoryEventKind, StartHistoryReply,
+ StartHistoryRequest, TailHistoryReply, TailHistoryRequest,
+ history_server::History as HistorySvc,
+ },
+ },
+ daemon::DaemonHandle,
+ events::DaemonEvent,
+};
+
+/// The gRPC service implementation.
+///
+/// This is a thin wrapper that delegates to the component's shared state.
+pub(crate) struct HistoryGrpcService {
+ /// Commands currently running (not yet completed).
+ running: DashMap<HistoryId, History>,
+
+ /// Handle to the daemon (set during start).
+ pub(crate) handle: tokio::sync::RwLock<Option<DaemonHandle>>,
+
+ /// History store for pushing records (set during start).
+ pub(crate) history_store: tokio::sync::RwLock<Option<HistoryStore>>,
+}
+
+impl HistoryGrpcService {
+ /// Create a new history component.
+ pub(crate) fn new() -> Self {
+ Self {
+ running: DashMap::new(),
+ handle: tokio::sync::RwLock::new(None),
+ history_store: tokio::sync::RwLock::new(None),
+ }
+ }
+}
+
+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.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.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.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.handle.read().await;
+ let handle = handle_guard
+ .as_ref()
+ .ok_or_else(|| Status::internal("component not initialized"))?;
+
+ let store_guard = self.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.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)))
+ }
+}
diff --git a/crates/daemon/src/api/server/mod.rs b/crates/daemon/src/api/server/mod.rs
new file mode 100644
index 00000000..8d475fe9
--- /dev/null
+++ b/crates/daemon/src/api/server/mod.rs
@@ -0,0 +1,2 @@
+pub(crate) mod control;
+pub(crate) mod history;
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 }))
- }
-}
diff --git a/crates/daemon/src/components/mod.rs b/crates/daemon/src/components/mod.rs
index 7c478efb..0b0319df 100644
--- a/crates/daemon/src/components/mod.rs
+++ b/crates/daemon/src/components/mod.rs
@@ -14,9 +14,7 @@
//! - [`sync::SyncComponent`]: Cloud sync
pub(crate) mod history;
-pub(crate) mod semantic;
pub(crate) mod sync;
-pub(crate) use history::HistoryComponent;
-pub(crate) use semantic::SemanticComponent;
pub(crate) use sync::SyncComponent;
+pub(crate) use history::HistoryComponent;
diff --git a/crates/daemon/src/components/semantic.rs b/crates/daemon/src/components/semantic.rs
deleted file mode 100644
index aec26887..00000000
--- a/crates/daemon/src/components/semantic.rs
+++ /dev/null
@@ -1,912 +0,0 @@
-//! Semantic command capture component.
-//!
-//! This is a prototype in-memory store for completed command captures emitted
-//! by atuin-pty-proxy. It keeps recent captures per Atuin session and indexes
-//! them by history ID for AI tool lookup.
-
-use std::collections::{HashMap, VecDeque};
-use std::fmt::{Display, Formatter};
-use std::sync::Arc;
-
-use crate::aclient::history::{History, HistoryId};
-use crate::generated::semantic;
-use eyre::Result;
-use tokio::sync::Mutex;
-use tonic::{Request, Response, Status, Streaming};
-use tracing::{Level, instrument};
-
-use crate::{
- daemon::{Component, DaemonHandle},
- events::DaemonEvent,
- generated::semantic::{
- CommandCapture, CommandOutputReply, CommandOutputRequest, OutputLine, RecordCommandsReply,
- semantic_server::{Semantic as SemanticSvc, SemanticServer},
- },
-};
-
-const MAX_SESSIONS: usize = 20;
-const MAX_COMMANDS_PER_SESSION: usize = 128;
-const MAX_BYTES_PER_SESSION: usize = 32 * 1024 * 1024;
-const MAX_PENDING_HISTORIES: usize = 128;
-
-/// Stores completed command captures and associates them with history events.
-pub(crate) struct SemanticComponent {
- inner: Arc<SemanticComponentInner>,
-}
-
-struct SemanticComponentInner {
- state: Mutex<SemanticState>,
-}
-
-#[derive(Default)]
-struct SemanticState {
- sessions: HashMap<SessionId, SessionCaptures>,
- session_lru: VecDeque<SessionId>,
- history_index: HashMap<HistoryId, CaptureRef>,
- pending_histories: VecDeque<History>,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-struct SessionId(String);
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-struct CaptureId(u64);
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct CaptureRef {
- session_id: SessionId,
- capture_id: CaptureId,
-}
-
-#[derive(Default)]
-struct SessionCaptures {
- next_id: u64,
- records: VecDeque<StoredCapture>,
- output_bytes: usize,
-}
-
-struct StoredCapture {
- id: CaptureId,
- history_id: HistoryId,
- output_bytes: usize,
- record: SemanticCommandRecord,
-}
-
-struct EvictedCapture {
- history_id: HistoryId,
- capture_id: CaptureId,
-}
-
-#[derive(Debug, Clone)]
-struct SemanticCommandRecord {
- capture: CommandCapture,
- history: Option<History>,
-}
-
-impl SemanticComponent {
- pub(crate) fn new() -> Self {
- Self {
- inner: Arc::new(SemanticComponentInner {
- state: Mutex::new(SemanticState::default()),
- }),
- }
- }
-
- pub(crate) fn grpc_service(&self) -> SemanticServer<SemanticGrpcService> {
- SemanticServer::new(SemanticGrpcService {
- inner: self.inner.clone(),
- })
- }
-}
-
-impl Default for SemanticComponent {
- fn default() -> Self {
- Self::new()
- }
-}
-
-#[tonic::async_trait]
-impl Component for SemanticComponent {
- fn name(&self) -> &'static str {
- "semantic"
- }
-
- async fn start(&mut self, _handle: DaemonHandle) -> Result<()> {
- tracing::info!("semantic component started");
- Ok(())
- }
-
- async fn handle_event(&mut self, event: &DaemonEvent) -> Result<()> {
- if let DaemonEvent::HistoryEnded(history) = event {
- self.inner.record_history(history.clone()).await;
- }
-
- Ok(())
- }
-
- async fn stop(&mut self) -> Result<()> {
- let state = self.inner.state.lock().await;
- tracing::info!(
- sessions = state.sessions.len(),
- records = state.record_count(),
- indexed_histories = state.history_index.len(),
- pending_histories = state.pending_histories.len(),
- "semantic component stopped"
- );
- Ok(())
- }
-}
-
-impl SemanticComponentInner {
- async fn record_capture(&self, capture: CommandCapture) -> bool {
- let mut state = self.state.lock().await;
- state.record_capture(capture)
- }
-
- async fn record_history(&self, history: History) {
- let mut state = self.state.lock().await;
- state.record_history(history);
- }
-
- async fn command_output(&self, request: &CommandOutputRequest) -> CommandOutputReply {
- let mut state = self.state.lock().await;
- state.command_output(request)
- }
-}
-
-impl SemanticState {
- fn record_capture(&mut self, mut capture: CommandCapture) -> bool {
- let Some(history_id) = history_id_from_str(capture.history_id.as_deref()) else {
- tracing::debug!(
- command_bytes = capture.command.len(),
- prompt_bytes = capture.prompt.len(),
- output_bytes = capture.output.len(),
- output_truncated = capture.output_truncated,
- "dropping semantic command capture without history id"
- );
- return false;
- };
-
- let history = take_pending_history(&mut self.pending_histories, &history_id);
- let Some(session_id) = capture
- .session_id
- .as_deref()
- .and_then(|session_id| SessionId::try_from(session_id).ok())
- .or_else(|| {
- history
- .as_ref()
- .and_then(|history| SessionId::try_from(history.session.as_str()).ok())
- })
- else {
- tracing::debug!(
- history_id = %history_id,
- command_bytes = capture.command.len(),
- prompt_bytes = capture.prompt.len(),
- output_bytes = capture.output.len(),
- output_truncated = capture.output_truncated,
- "dropping semantic command capture without session id"
- );
- return false;
- };
-
- capture.history_id = Some(history_id.to_string());
- capture.session_id = Some(session_id.to_string());
- if capture.output_observed_bytes == 0 {
- capture.output_observed_bytes = capture.output.len() as u64;
- }
-
- let record = SemanticCommandRecord { capture, history };
- log_record(&record, "recorded semantic command capture");
- self.push_record(&session_id, history_id, record);
- true
- }
-
- fn record_history(&mut self, history: History) {
- let history_id = history.id.clone();
-
- if let Some(capture_ref) = self.history_index.get(&history_id).cloned() {
- if let Some(stored) = self.stored_capture_mut(&capture_ref) {
- stored.record.history = Some(history);
- log_record(
- &stored.record,
- "associated semantic command capture with history",
- );
- return;
- }
-
- self.history_index.remove(&history_id);
- }
-
- tracing::debug!(
- id = %history.id,
- command_bytes = history.command.len(),
- "history ended before semantic capture arrived"
- );
- push_pending_history(&mut self.pending_histories, history);
- }
-
- fn command_output(&mut self, request: &CommandOutputRequest) -> CommandOutputReply {
- let Some(history_id) = history_id_from_str(Some(&request.history_id)) else {
- return command_output_not_found();
- };
- let Some(capture_ref) = self.history_index.get(&history_id).cloned() else {
- return command_output_not_found();
- };
-
- let Some(reply) = self.command_output_for_ref(&capture_ref, &request.ranges) else {
- self.history_index.remove(&history_id);
- return command_output_not_found();
- };
-
- self.touch_session(&capture_ref.session_id);
- reply
- }
-
- fn command_output_for_ref(
- &self,
- capture_ref: &CaptureRef,
- ranges: &[semantic::OutputRange],
- ) -> Option<CommandOutputReply> {
- let stored = self
- .sessions
- .get(&capture_ref.session_id)?
- .stored_capture(capture_ref.capture_id)?;
- let output = &stored.record.capture.output;
- let output_observed_bytes = stored
- .record
- .capture
- .output_observed_bytes
- .max(output.len() as u64);
-
- Some(CommandOutputReply {
- found: true,
- output: String::new(),
- total_bytes: output.len() as u64,
- total_lines: output.lines().count() as u64,
- lines: select_output_ranges(output, ranges),
- output_truncated: stored.record.capture.output_truncated,
- output_observed_bytes,
- })
- }
-
- fn push_record(
- &mut self,
- session_id: &SessionId,
- history_id: HistoryId,
- record: SemanticCommandRecord,
- ) {
- self.touch_session(session_id);
-
- let (capture_id, evicted) = {
- let session = self.sessions.entry(session_id.clone()).or_default();
- session.push(history_id.clone(), record)
- };
-
- let capture_ref = CaptureRef {
- session_id: session_id.clone(),
- capture_id,
- };
- self.history_index.insert(history_id, capture_ref);
-
- for evicted in evicted {
- self.remove_history_index_if_matches(
- session_id,
- &evicted.history_id,
- evicted.capture_id,
- );
- }
-
- self.expire_lru_sessions();
- }
-
- fn touch_session(&mut self, session_id: &SessionId) {
- if let Some(index) = self.session_lru.iter().position(|id| id == session_id) {
- self.session_lru.remove(index);
- }
- self.session_lru.push_back(session_id.clone());
- }
-
- fn expire_lru_sessions(&mut self) {
- while self.session_lru.len() > MAX_SESSIONS {
- let Some(session_id) = self.session_lru.pop_front() else {
- break;
- };
- let Some(session) = self.sessions.remove(&session_id) else {
- continue;
- };
-
- for stored in session.records {
- self.remove_history_index_if_matches(&session_id, &stored.history_id, stored.id);
- }
- }
- }
-
- fn remove_history_index_if_matches(
- &mut self,
- session_id: &SessionId,
- history_id: &HistoryId,
- capture_id: CaptureId,
- ) {
- if self
- .history_index
- .get(history_id)
- .is_some_and(|capture_ref| {
- &capture_ref.session_id == session_id && capture_ref.capture_id == capture_id
- })
- {
- self.history_index.remove(history_id);
- }
- }
-
- fn stored_capture_mut(&mut self, capture_ref: &CaptureRef) -> Option<&mut StoredCapture> {
- self.sessions
- .get_mut(&capture_ref.session_id)?
- .stored_capture_mut(capture_ref.capture_id)
- }
-
- fn record_count(&self) -> usize {
- self.sessions
- .values()
- .map(|session| session.records.len())
- .sum()
- }
-}
-
-impl SessionCaptures {
- fn push(
- &mut self,
- history_id: HistoryId,
- record: SemanticCommandRecord,
- ) -> (CaptureId, Vec<EvictedCapture>) {
- self.push_with_limits(
- history_id,
- record,
- MAX_COMMANDS_PER_SESSION,
- MAX_BYTES_PER_SESSION,
- )
- }
-
- fn push_with_limits(
- &mut self,
- history_id: HistoryId,
- record: SemanticCommandRecord,
- max_commands: usize,
- max_output_bytes: usize,
- ) -> (CaptureId, Vec<EvictedCapture>) {
- let capture_id = CaptureId(self.next_id);
- self.next_id = self.next_id.saturating_add(1);
- let output_bytes = record.capture.output.len();
- self.output_bytes = self.output_bytes.saturating_add(output_bytes);
- self.records.push_back(StoredCapture {
- id: capture_id,
- history_id,
- output_bytes,
- record,
- });
-
- (
- capture_id,
- self.evict_to_limits(max_commands, max_output_bytes),
- )
- }
-
- fn evict_to_limits(
- &mut self,
- max_commands: usize,
- max_output_bytes: usize,
- ) -> Vec<EvictedCapture> {
- let mut evicted = Vec::new();
- while self.records.len() > max_commands || self.output_bytes > max_output_bytes {
- let Some(record) = self.records.pop_front() else {
- break;
- };
- self.output_bytes = self.output_bytes.saturating_sub(record.output_bytes);
- evicted.push(EvictedCapture {
- history_id: record.history_id,
- capture_id: record.id,
- });
- }
- evicted
- }
-
- fn stored_capture(&self, capture_id: CaptureId) -> Option<&StoredCapture> {
- self.records.iter().find(|record| record.id == capture_id)
- }
-
- fn stored_capture_mut(&mut self, capture_id: CaptureId) -> Option<&mut StoredCapture> {
- self.records
- .iter_mut()
- .find(|record| record.id == capture_id)
- }
-}
-
-impl TryFrom<&str> for SessionId {
- type Error = ();
-
- fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
- let value = value.trim();
- if value.is_empty() {
- return Err(());
- }
-
- Ok(Self(value.to_string()))
- }
-}
-
-impl TryFrom<String> for SessionId {
- type Error = ();
-
- fn try_from(value: String) -> std::result::Result<Self, Self::Error> {
- Self::try_from(value.as_str())
- }
-}
-
-impl AsRef<str> for SessionId {
- fn as_ref(&self) -> &str {
- &self.0
- }
-}
-
-impl Display for SessionId {
- fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- f.write_str(&self.0)
- }
-}
-
-pub(crate) struct SemanticGrpcService {
- inner: Arc<SemanticComponentInner>,
-}
-
-#[tonic::async_trait]
-impl SemanticSvc for SemanticGrpcService {
- #[instrument(skip_all, level = Level::INFO)]
- async fn record_commands(
- &self,
- request: Request<Streaming<CommandCapture>>,
- ) -> Result<Response<RecordCommandsReply>, Status> {
- let mut stream = request.into_inner();
- let mut accepted = 0_u64;
-
- while let Some(capture) = stream.message().await? {
- if self.inner.record_capture(capture).await {
- accepted += 1;
- }
- }
-
- Ok(Response::new(RecordCommandsReply { accepted }))
- }
-
- #[instrument(skip_all, level = Level::INFO)]
- async fn command_output(
- &self,
- request: Request<CommandOutputRequest>,
- ) -> Result<Response<CommandOutputReply>, Status> {
- let request = request.into_inner();
- if request.history_id.trim().is_empty() {
- return Err(Status::invalid_argument("history_id is required"));
- }
-
- Ok(Response::new(self.inner.command_output(&request).await))
- }
-}
-
-fn history_id_from_str(value: Option<&str>) -> Option<HistoryId> {
- let value = value?.trim();
- (!value.is_empty()).then(|| HistoryId(value.to_string()))
-}
-
-fn take_pending_history(
- histories: &mut VecDeque<History>,
- history_id: &HistoryId,
-) -> Option<History> {
- let index = histories
- .iter()
- .position(|history| &history.id == history_id)?;
- histories.remove(index)
-}
-
-fn push_pending_history(histories: &mut VecDeque<History>, history: History) {
- if let Some(index) = histories
- .iter()
- .position(|pending| pending.id == history.id)
- {
- histories.remove(index);
- }
-
- histories.push_back(history);
- trim_front(histories, MAX_PENDING_HISTORIES);
-}
-
-fn trim_front<T>(records: &mut VecDeque<T>, max_len: usize) {
- while records.len() > max_len {
- records.pop_front();
- }
-}
-
-fn command_output_not_found() -> CommandOutputReply {
- CommandOutputReply {
- found: false,
- output: String::new(),
- total_bytes: 0,
- total_lines: 0,
- lines: Vec::new(),
- output_truncated: false,
- output_observed_bytes: 0,
- }
-}
-
-fn select_output_ranges(output: &str, ranges: &[semantic::OutputRange]) -> Vec<OutputLine> {
- let lines: Vec<&str> = output.lines().collect();
- if lines.is_empty() {
- return Vec::new();
- }
-
- let ranges = if ranges.is_empty() {
- vec![semantic::OutputRange { start: 0, end: 999 }]
- } else {
- ranges.to_vec()
- };
-
- let mut ranges = ranges
- .into_iter()
- .filter_map(|range| normalize_line_range(range.start, range.end, lines.len()))
- .collect::<Vec<_>>();
- ranges.sort_unstable_by_key(|(start, _)| *start);
-
- let mut merged: Vec<(usize, usize)> = Vec::new();
- for (start, end) in ranges {
- match merged.last_mut() {
- Some((_, merged_end)) if start <= merged_end.saturating_add(1) => {
- *merged_end = (*merged_end).max(end);
- }
- _ => merged.push((start, end)),
- }
- }
-
- merged
- .into_iter()
- .flat_map(|(start, end)| {
- lines[start..=end]
- .iter()
- .enumerate()
- .map(move |(offset, line)| OutputLine {
- line_number: (start + offset + 1) as u64,
- content: (*line).to_string(),
- })
- })
- .collect()
-}
-
-fn normalize_line_range(start: i64, end: i64, line_count: usize) -> Option<(usize, usize)> {
- let line_count = i64::try_from(line_count).ok()?;
- let start = if start < 0 { line_count + start } else { start };
- let end = if end < 0 { line_count + end } else { end };
-
- if end < 0 || start >= line_count {
- return None;
- }
-
- let start = start.max(0);
- let end = end.min(line_count - 1);
-
- (start <= end).then_some((start as usize, end as usize))
-}
-
-fn log_record(record: &SemanticCommandRecord, message: &'static str) {
- let history_id = record.capture.history_id.as_deref().unwrap_or("<missing>");
- let associated_history_id = record
- .history
- .as_ref()
- .map(|history| history.id.to_string());
- let exit = record.history.as_ref().map(|history| history.exit);
- let duration = record.history.as_ref().map(|history| history.duration);
- let author = record
- .history
- .as_ref()
- .map(|history| history.author.as_str());
- let session_id = record.capture.session_id.as_deref();
-
- tracing::debug!(
- history_id = %history_id,
- associated_history_id = ?associated_history_id,
- session_id = ?session_id,
- command_bytes = record.capture.command.len(),
- prompt_bytes = record.capture.prompt.len(),
- output_bytes = record.capture.output.len(),
- output_truncated = record.capture.output_truncated,
- output_observed_bytes = record.capture.output_observed_bytes,
- capture_exit_code = ?record.capture.exit_code,
- history_exit = ?exit,
- duration = ?duration,
- author = ?author,
- "{message}"
- );
-}
-
-#[cfg(test)]
-mod tests {
- use time::OffsetDateTime;
-
- use crate::{
- atuin_client::history::{History, HistoryId},
- atuin_daemon::{
- components::semantic::{
- MAX_COMMANDS_PER_SESSION, MAX_SESSIONS, SemanticCommandRecord, SemanticState,
- SessionCaptures, SessionId, select_output_ranges,
- },
- generated::semantic::{self, CommandOutputReply, CommandOutputRequest, OutputLine},
- },
- atuin_pty_proxy::CommandCapture,
- };
-
- fn history(id: &str, session: &str, command: &str) -> History {
- History {
- id: HistoryId(id.to_string()),
- timestamp: OffsetDateTime::UNIX_EPOCH,
- duration: 0,
- exit: 0,
- command: command.to_string(),
- cwd: String::new(),
- session: session.to_string(),
- hostname: String::new(),
- author: String::new(),
- intent: None,
- deleted_at: None,
- }
- }
-
- fn capture(history_id: Option<&str>, session_id: Option<&str>, output: &str) -> CommandCapture {
- CommandCapture {
- prompt: String::new(),
- command: String::new(),
- output: output.to_string(),
- exit_code: None,
- history_id: history_id.map(str::to_string),
- session_id: session_id.map(str::to_string),
- output_truncated: false,
- output_observed_bytes: output.len() as u64,
- }
- }
-
- fn command_output(state: &mut SemanticState, history_id: &str) -> CommandOutputReply {
- state.command_output(&CommandOutputRequest {
- history_id: history_id.to_string(),
- ranges: Vec::new(),
- })
- }
-
- fn output_line(line_number: u64, content: &str) -> OutputLine {
- OutputLine {
- line_number,
- content: content.to_string(),
- }
- }
-
- #[test]
- fn drops_capture_without_history_id() {
- let mut state = SemanticState::default();
-
- assert!(!state.record_capture(capture(None, Some("session-1"), "output")));
- assert!(!command_output(&mut state, "id-1").found);
- assert_eq!(state.record_count(), 0);
- }
-
- #[test]
- fn stores_capture_by_session_and_history_id() {
- let mut state = SemanticState::default();
-
- assert!(state.record_capture(capture(Some("id-1"), Some("session-1"), "output")));
-
- let reply = command_output(&mut state, "id-1");
- assert!(reply.found);
- assert_eq!(reply.total_bytes, 6);
- assert_eq!(reply.output_observed_bytes, 6);
- assert_eq!(reply.lines, vec![output_line(1, "output")]);
- }
-
- #[test]
- fn uses_pending_history_session_when_capture_session_is_missing() {
- let mut state = SemanticState::default();
-
- state.record_history(history("id-1", "session-from-history", "cargo test"));
- assert!(state.record_capture(capture(Some("id-1"), None, "output")));
-
- assert!(
- state
- .sessions
- .contains_key(&SessionId("session-from-history".to_string()))
- );
- assert!(command_output(&mut state, "id-1").found);
- }
-
- #[test]
- fn associates_history_by_id_after_capture_arrives() {
- let mut state = SemanticState::default();
-
- assert!(state.record_capture(capture(Some("id-1"), Some("session-1"), "output")));
- state.record_history(history("id-1", "session-1", "different command"));
-
- let capture_ref = state
- .history_index
- .get(&HistoryId("id-1".to_string()))
- .unwrap();
- let stored = state
- .sessions
- .get(&capture_ref.session_id)
- .unwrap()
- .stored_capture(capture_ref.capture_id)
- .unwrap();
- assert!(stored.record.history.is_some());
- }
-
- #[test]
- fn evicts_oldest_command_when_session_ring_is_full() {
- let mut state = SemanticState::default();
-
- for index in 0..=MAX_COMMANDS_PER_SESSION {
- assert!(state.record_capture(capture(
- Some(&format!("id-{index}")),
- Some("session-1"),
- "output",
- )));
- }
-
- assert!(!command_output(&mut state, "id-0").found);
- assert!(command_output(&mut state, &format!("id-{MAX_COMMANDS_PER_SESSION}")).found);
- assert_eq!(state.record_count(), MAX_COMMANDS_PER_SESSION);
- }
-
- #[test]
- fn evicts_oldest_session_after_lru_limit() {
- let mut state = SemanticState::default();
-
- for index in 0..MAX_SESSIONS {
- assert!(state.record_capture(capture(
- Some(&format!("id-{index}")),
- Some(&format!("session-{index}")),
- "output",
- )));
- }
- assert!(command_output(&mut state, "id-0").found);
-
- assert!(state.record_capture(capture(Some("new-id"), Some("new-session"), "output",)));
-
- assert!(command_output(&mut state, "id-0").found);
- assert!(!command_output(&mut state, "id-1").found);
- assert!(command_output(&mut state, "new-id").found);
- assert_eq!(state.sessions.len(), MAX_SESSIONS);
- }
-
- #[test]
- fn evicts_by_session_byte_limit() {
- let mut session = SessionCaptures::default();
- let first_output = "x".repeat(10);
- let second_output = "y";
- let (_, evicted_first) = session.push_with_limits(
- HistoryId("first".to_string()),
- SemanticCommandRecord {
- capture: capture(Some("first"), Some("session-1"), &first_output),
- history: None,
- },
- MAX_COMMANDS_PER_SESSION,
- 10,
- );
- assert!(evicted_first.is_empty());
-
- let (_, evicted_second) = session.push_with_limits(
- HistoryId("second".to_string()),
- SemanticCommandRecord {
- capture: capture(Some("second"), Some("session-1"), second_output),
- history: None,
- },
- MAX_COMMANDS_PER_SESSION,
- 10,
- );
-
- assert_eq!(evicted_second.len(), 1);
- assert_eq!(evicted_second[0].history_id, HistoryId("first".to_string()));
- assert_eq!(session.records.len(), 1);
- assert_eq!(session.output_bytes, 1);
- }
-
- #[test]
- fn command_output_reports_truncation_metadata() {
- let mut state = SemanticState::default();
- let mut capture = capture(Some("id-1"), Some("session-1"), "partial");
- capture.output_truncated = true;
- capture.output_observed_bytes = 1024;
-
- assert!(state.record_capture(capture));
-
- let reply = command_output(&mut state, "id-1");
- assert!(reply.output_truncated);
- assert_eq!(reply.total_bytes, 7);
- assert_eq!(reply.output_observed_bytes, 1024);
- }
-
- #[test]
- fn output_ranges_are_line_based_inclusive_and_support_negative_offsets() {
- let output = "zero\none\ntwo\nthree\nfour";
- let ranges = vec![
- semantic::OutputRange { start: 1, end: 2 },
- semantic::OutputRange { start: -2, end: -1 },
- ];
-
- assert_eq!(
- select_output_ranges(output, &ranges),
- vec![
- output_line(2, "one"),
- output_line(3, "two"),
- output_line(4, "three"),
- output_line(5, "four"),
- ]
- );
- }
-
- #[test]
- fn output_ranges_merge_overlaps_and_adjacent_ranges() {
- let output = (0..100)
- .map(|n| format!("line {n}"))
- .collect::<Vec<_>>()
- .join("\n");
- let ranges = vec![
- semantic::OutputRange { start: 0, end: 100 },
- semantic::OutputRange {
- start: -100,
- end: -1,
- },
- ];
-
- let selected = select_output_ranges(&output, &ranges);
-
- assert_eq!(selected.len(), 100);
- assert_eq!(selected.first(), Some(&output_line(1, "line 0")));
- assert_eq!(selected.last(), Some(&output_line(100, "line 99")));
- }
-
- #[test]
- fn output_ranges_can_leave_gaps_for_client_formatting() {
- let output = "zero\none\ntwo\nthree\nfour";
- let ranges = vec![
- semantic::OutputRange { start: 0, end: 1 },
- semantic::OutputRange { start: 4, end: 4 },
- ];
-
- assert_eq!(
- select_output_ranges(output, &ranges),
- vec![
- output_line(1, "zero"),
- output_line(2, "one"),
- output_line(5, "four"),
- ]
- );
- }
-
- #[test]
- fn empty_output_ranges_default_to_first_thousand_lines() {
- let output = (0..1001)
- .map(|n| format!("line {n}"))
- .collect::<Vec<_>>()
- .join("\n");
-
- let selected = select_output_ranges(&output, &[]);
-
- assert_eq!(selected.len(), 1000);
- assert_eq!(selected.first(), Some(&output_line(1, "line 0")));
- assert_eq!(selected.last(), Some(&output_line(1000, "line 999")));
- }
-
- #[test]
- fn output_ranges_skip_ranges_fully_outside_output() {
- let output = "zero\none\ntwo";
- let ranges = vec![
- semantic::OutputRange { start: 10, end: 20 },
- semantic::OutputRange {
- start: -20,
- end: -10,
- },
- ];
-
- assert_eq!(select_output_ranges(output, &ranges), Vec::new());
- }
-}
diff --git a/crates/daemon/src/control/mod.rs b/crates/daemon/src/control/mod.rs
deleted file mode 100644
index f727a3a2..00000000
--- a/crates/daemon/src/control/mod.rs
+++ /dev/null
@@ -1,74 +0,0 @@
-//! Control service implementation.
-//!
-//! This gRPC service allows external processes (like CLI commands) to inject
-//! events into the daemon's event bus.
-
-use tonic::{Request, Response, Status};
-use tracing::{Level, info, instrument};
-
-use crate::{
- aclient::history::HistoryId,
- daemon::DaemonHandle,
- events::DaemonEvent,
- generated::control::{
- SendEventRequest, SendEventResponse,
- control_server::{Control, ControlServer},
- send_event_request::Event,
- },
-};
-
-/// The Control gRPC service.
-///
-/// 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 ControlService {
- /// Create a new control service with the given daemon handle.
- pub(crate) fn new(handle: DaemonHandle) -> Self {
- Self { handle }
- }
-
- /// Get a tonic server for this service.
- pub(crate) fn into_server(self) -> ControlServer<Self> {
- ControlServer::new(self)
- }
-}
-
-#[tonic::async_trait]
-impl Control for ControlService {
- #[instrument(skip_all, level = Level::INFO, name = "control_send_event")]
- async fn send_event(
- &self,
- request: Request<SendEventRequest>,
- ) -> Result<Response<SendEventResponse>, Status> {
- let req = request.into_inner();
-
- let event = req
- .event
- .ok_or_else(|| Status::invalid_argument("event is required"))?;
-
- let daemon_event = proto_event_to_daemon_event(event);
-
- info!(?daemon_event, "received control event");
- self.handle.emit(daemon_event);
-
- Ok(Response::new(SendEventResponse {}))
- }
-}
-
-/// Convert a proto event to a daemon event.
-fn proto_event_to_daemon_event(event: Event) -> DaemonEvent {
- match event {
- Event::HistoryPruned(_) => DaemonEvent::HistoryPruned,
- Event::HistoryRebuilt(_) => DaemonEvent::HistoryRebuilt,
- Event::HistoryDeleted(e) => DaemonEvent::HistoryDeleted {
- ids: e.ids.into_iter().map(HistoryId).collect(),
- },
- Event::ForceSync(_) => DaemonEvent::ForceSync,
- Event::SettingsReloaded(_) => DaemonEvent::SettingsReloaded,
- Event::Shutdown(_) => DaemonEvent::ShutdownRequested,
- }
-}
diff --git a/crates/daemon/src/daemon.rs b/crates/daemon/src/daemon.rs
index 4e691be2..2e236f4e 100644
--- a/crates/daemon/src/daemon.rs
+++ b/crates/daemon/src/daemon.rs
@@ -248,7 +248,6 @@ pub(crate) trait Component: Send + Sync {
/// Events emitted during handling are queued and processed in subsequent
/// iterations, ensuring the loop eventually drains.
pub(crate) struct Daemon {
- components: Vec<Box<dyn Component>>,
handle: DaemonHandle,
}
@@ -269,14 +268,12 @@ impl Daemon {
///
/// This must be called before `run_event_loop()`. It initializes all
/// registered components with the daemon handle.
- pub(crate) async fn start_components(&mut self) -> Result<()> {
- for component in &mut self.components {
- tracing::info!(component = component.name(), "starting component");
- component
- .start(self.handle.clone())
- .await
- .with_context(|| format!("failed to start component: {}", component.name()))?;
- }
+ pub(crate) async fn start_component(&mut self, component: &mut impl Component) -> Result<()> {
+ tracing::info!(component = component.name(), "starting component");
+ component
+ .start(self.handle.clone())
+ .await
+ .with_context(|| format!("failed to start component: {}", component.name()))?;
Ok(())
}
@@ -365,7 +362,7 @@ pub(crate) struct DaemonBuilder {
settings: Settings,
store: Option<SqliteStore>,
history_db: Option<HistoryDatabase>,
- components: Vec<Box<dyn Component>>,
+ components: Vec<Arc<Box<dyn Component>>>,
}
impl DaemonBuilder {
@@ -391,14 +388,6 @@ impl DaemonBuilder {
self
}
- /// Register a component.
- ///
- /// Components are started in registration order and stopped in reverse order.
- pub(crate) fn component(mut self, component: impl Component + 'static) -> Self {
- self.components.push(Box::new(component));
- self
- }
-
/// Build the daemon.
///
/// This loads the encryption key and creates the daemon state.
@@ -428,9 +417,6 @@ impl DaemonBuilder {
// Create the handle (just a reference to the state)
let handle = DaemonHandle { state };
- Ok(Daemon {
- components: self.components,
- handle,
- })
+ Ok(Daemon { handle })
}
}
diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs
index a5d233c6..3be39a1c 100644
--- a/crates/daemon/src/lib.rs
+++ b/crates/daemon/src/lib.rs
@@ -1,25 +1,30 @@
-use crate::aclient::database::ClientSqlite as HistoryDatabase;
+#![expect(unused_crate_dependencies, reason = "Didn't remove them yet")]
+
+use std::sync::Arc;
+
use crate::aclient::record::sqlite_store::SqliteStore;
-use crate::aclient::settings::{Settings, watcher::global_settings_watcher};
+use crate::aclient::settings::Settings;
+use crate::api::server::control::ControlService;
+use crate::{
+ aclient::database::ClientSqlite as HistoryDatabase,
+ api::generated::history::history_server::HistoryServer,
+};
use eyre::Result;
pub mod aclient;
-pub mod client;
+pub(crate) mod api;
pub(crate) mod components;
-pub(crate) mod control;
pub(crate) mod daemon;
pub(crate) mod events;
pub(crate) mod server;
-pub(crate) mod generated;
-
// Re-export core daemon types for convenience
pub(crate) use daemon::Daemon;
pub use events::DaemonEvent;
// Re-export components
-pub(crate) use components::{HistoryComponent, SemanticComponent, SyncComponent};
+pub(crate) use components::{HistoryComponent, SyncComponent};
/// Boot the daemon using the new component-based architecture.
///
@@ -31,53 +36,28 @@ pub async fn boot(
history_db: HistoryDatabase,
) -> Result<()> {
// Create the components
- let history_component = HistoryComponent::new();
- let semantic_component = SemanticComponent::new();
- let sync_component = SyncComponent::new();
+ let history_component = Arc::new(HistoryComponent::new());
+ let sync_component = Arc::new(Box::new(&SyncComponent::new()));
// Get the gRPC services before moving components into the daemon
// (The services share state with the components via Arc)
- let history_service = history_component.grpc_service();
- let semantic_service = semantic_component.grpc_service();
+ let history_service = HistoryServer::from_arc(Arc::clone(&history_component));
// Build the daemon
let mut daemon = Daemon::builder(settings.clone())
.store(store)
.history_db(history_db)
.component(history_component)
- .component(semantic_component)
.component(sync_component)
.build()?;
- // Get a handle for the control service and gRPC server shutdown
let handle = daemon.handle();
- // Create the control service
- let control_service = control::ControlService::new(handle.clone());
+ let control_service = ControlService::new(handle.clone());
// Start all components first (so gRPC services can work)
daemon.start_components().await?;
- // Spawn config file watcher to reload settings on changes
- if let Ok(watcher) = global_settings_watcher() {
- let mut settings_rx = watcher.subscribe();
- let watcher_handle = handle.clone();
- tokio::spawn(async move {
- tracing::info!("config file watcher started");
- while settings_rx.changed().await.is_ok() {
- // Use the already-loaded settings from the watcher
- // (avoids parsing the config file twice)
- let new_settings = (*settings_rx.borrow()).clone();
- watcher_handle.apply_settings((*new_settings).clone()).await;
- }
- tracing::debug!("config file watcher stopped");
- });
- } else {
- tracing::warn!(
- "failed to start config file watcher; settings changes will require daemon restart"
- );
- }
-
// Spawn signal handler to emit ShutdownRequested on Ctrl+C/SIGTERM
let signal_handle = handle.clone();
tokio::spawn(async move {
@@ -89,7 +69,6 @@ pub async fn boot(
server::run_grpc_server(
&settings,
history_service,
- semantic_service,
control_service.into_server(),
handle,
)?;
diff --git a/crates/daemon/src/main.rs b/crates/daemon/src/main.rs
index 26a5cafd..174cf94b 100644
--- a/crates/daemon/src/main.rs
+++ b/crates/daemon/src/main.rs
@@ -1,22 +1,16 @@
-#[allow(unused_imports)]
+#![expect(unused_crate_dependencies, reason = "Didn't remove them yet")]
+
use clap::Parser;
use eyre::{Result, WrapErr, bail, eyre};
use fs4::fs_std::FileExt;
use std::fs::{self, File, OpenOptions};
-use std::io::{ErrorKind, Write};
-#[cfg(unix)]
-use std::os::unix::net::UnixStream as StdUnixStream;
+use std::io::Write;
use std::path::{Path, PathBuf};
-use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use tokio::time::sleep;
use turtle_daemon::{
- DaemonEvent,
- aclient::{
- database::ClientSqlite, history::History, record::sqlite_store::SqliteStore,
- settings::Settings,
- },
- client::{ControlClient, DaemonClientErrorKind, HistoryClient, classify_error},
+ aclient::{database::ClientSqlite, record::sqlite_store::SqliteStore, settings::Settings},
+ client::{DaemonClientErrorKind, HistoryClient, classify_error},
};
#[derive(Parser, Debug)]
@@ -24,9 +18,6 @@ use turtle_daemon::{
pub(crate) enum Cmd {
/// Start the daemon server
Start {
- #[arg(long, hide = true)]
- daemonize: bool,
-
/// Also write daemon logs to the console (useful for debugging)
#[arg(long)]
show_logs: bool,
@@ -39,21 +30,6 @@ pub(crate) enum Cmd {
Stop,
}
-impl Cmd {
- pub(crate) async fn run(
- self,
- settings: Settings,
- store: SqliteStore,
- history_db: ClientSqlite,
- ) -> Result<()> {
- match self {
- Cmd::Start { .. } => run(settings, store, history_db).await,
- Cmd::Status => status_cmd(&settings).await,
- Cmd::Stop => stop_cmd(&settings).await,
- }
- }
-}
-
#[tokio::main]
async fn main() -> Result<()> {
let settings = Settings::new().wrap_err("could not load client settings")?;
@@ -63,13 +39,14 @@ async fn main() -> Result<()> {
let db = ClientSqlite::new(db_path, settings.local_timeout).await?;
let sqlite_store = SqliteStore::new(record_store_path, settings.local_timeout).await?;
- Cmd::parse().run(settings, sqlite_store, db).await
+ match Cmd::parse() {
+ Cmd::Start { show_logs, .. } => start_cmd(settings, store, history_db, show_logs).await,
+ Cmd::Status => status_cmd(&settings).await,
+ Cmd::Stop => stop_cmd(&settings).await,
+ }
}
-const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");
-const DAEMON_PROTOCOL_VERSION: u32 = 1;
const STARTUP_POLL: Duration = Duration::from_millis(40);
-const LOCK_POLL: Duration = Duration::from_millis(20);
const LEGACY_DAEMON_RESTART_MESSAGE: &str = "legacy daemon detected; restart daemon manually";
struct PidfileGuard {
@@ -109,18 +86,6 @@ enum Probe {
Unreachable(eyre::Report),
}
-fn daemon_matches_expected(version: &str, protocol: u32) -> bool {
- version == DAEMON_VERSION && protocol == DAEMON_PROTOCOL_VERSION
-}
-
-fn daemon_mismatch_message(version: &str, protocol: u32) -> String {
- if protocol == DAEMON_PROTOCOL_VERSION {
- format!("daemon is out of date: expected {DAEMON_VERSION}, got {version}")
- } else {
- format!("daemon protocol mismatch: expected {DAEMON_PROTOCOL_VERSION}, got {protocol}")
- }
-}
-
fn is_legacy_daemon_error(err: &eyre::Report) -> bool {
matches!(classify_error(err), DaemonClientErrorKind::Unimplemented)
}
@@ -141,6 +106,8 @@ fn open_lock_file(path: &Path) -> Result<File> {
}
async fn wait_for_lock(path: &Path, timeout: Duration) -> Result<File> {
+ const LOCK_POLL: Duration = Duration::from_millis(20);
+
let file = open_lock_file(path)?;
let start = Instant::now();
@@ -168,32 +135,6 @@ async fn wait_for_pidfile_available(path: &Path, timeout: Duration) -> Result<()
Ok(())
}
-async fn connect_client(settings: &Settings) -> Result<HistoryClient> {
- HistoryClient::new(
- #[cfg(unix)]
- settings.daemon.socket_path.clone(),
- )
- .await
-}
-
-async fn probe(settings: &Settings) -> Probe {
- let mut client = match connect_client(settings).await {
- Ok(client) => client,
- Err(err) => return Probe::Unreachable(err),
- };
-
- match client.status().await {
- Ok(status) => {
- if daemon_matches_expected(&status.version, status.protocol) {
- Probe::Ready(client)
- } else {
- Probe::NeedsRestart(daemon_mismatch_message(&status.version, status.protocol))
- }
- }
- Err(err) => Probe::Unreachable(err),
- }
-}
-
async fn request_shutdown(settings: &Settings) {
if let Ok(mut client) = connect_client(settings).await {
drop(client.shutdown().await);
@@ -204,85 +145,6 @@ fn startup_timeout(settings: &Settings) -> Duration {
Duration::from_secs_f64(settings.local_timeout.max(0.5) + 2.0)
}
-pub(crate) async fn start_history(settings: &Settings, history: History) -> Result<String> {
- match async {
- connect_client(settings)
- .await?
- .start_history(history.clone())
- .await
- }
- .await
- {
- Ok(resp) => {
- if daemon_matches_expected(&resp.version, resp.protocol) {
- return Ok(resp.id);
- }
-
- Err(eyre!(
- "{}. Restart the daemon manually",
- daemon_mismatch_message(&resp.version, resp.protocol)
- ))
- }
- Err(err) => Err(err),
- }
-}
-
-pub(crate) async fn end_history(
- settings: &Settings,
- id: String,
- duration: u64,
- exit: i64,
-) -> Result<()> {
- match async {
- connect_client(settings)
- .await?
- .end_history(id.clone(), duration, exit)
- .await
- }
- .await
- {
- Ok(resp) => {
- if daemon_matches_expected(&resp.version, resp.protocol) {
- return Ok(());
- }
-
- Err(eyre!(
- "{}. Restart the daemon manually",
- daemon_mismatch_message(&resp.version, resp.protocol)
- ))
- }
- Err(err) => Err(err),
- }
-}
-
-/// Emit a daemon event.
-pub(crate) async fn emit_event(settings: &Settings, event: DaemonEvent) {
- // Try to connect and send
- match ControlClient::from_settings(settings).await {
- Ok(mut client) => {
- if let Err(e) = client.send_event(event).await {
- tracing::debug!(?e, "failed to send event to daemon");
- }
- }
- Err(e) => {
- tracing::debug!(?e, "daemon not available, skipping event emission");
- }
- }
-}
-
-pub(crate) async fn tail_client(settings: &Settings) -> Result<HistoryClient> {
- match probe(settings).await {
- Probe::Ready(client) => Ok(client),
- Probe::NeedsRestart(reason) => {
- bail!("{reason}. Restart the daemon manually");
- }
- Probe::Unreachable(err) if is_legacy_daemon_error(&err) => {
- Err(err.wrap_err(LEGACY_DAEMON_RESTART_MESSAGE))
- }
- Probe::Unreachable(err) => Err(err),
- }
-}
-
async fn status_cmd(settings: &Settings) -> Result<()> {
match probe(settings).await {
Probe::Ready(mut client) => {
@@ -292,7 +154,6 @@ async fn status_cmd(settings: &Settings) -> Result<()> {
println!(" Version: {}", status.version);
println!(" Protocol: {}", status.protocol);
println!(" Healthy: {}", status.healthy);
- #[cfg(unix)]
println!(" Socket: {}", settings.daemon.socket_path);
}
Probe::NeedsRestart(reason) => {
@@ -331,7 +192,12 @@ async fn stop_cmd(settings: &Settings) -> Result<()> {
}
}
-async fn run(settings: Settings, store: SqliteStore, history_db: ClientSqlite) -> Result<()> {
+async fn start_cmd(
+ settings: Settings,
+ store: SqliteStore,
+ history_db: ClientSqlite,
+ show_logs: bool,
+) -> Result<()> {
let pidfile_path = PathBuf::from(&settings.daemon.pidfile_path);
let _pidfile_guard = PidfileGuard::acquire(&pidfile_path)?;
diff --git a/crates/daemon/src/server.rs b/crates/daemon/src/server.rs
index 97c4fe48..7ba8312c 100644
--- a/crates/daemon/src/server.rs
+++ b/crates/daemon/src/server.rs
@@ -1,14 +1,17 @@
+use std::os::unix::net::SocketAddr;
+use std::path::PathBuf;
+
use eyre::Result;
+use eyre::{OptionExt, WrapErr};
+#[cfg(unix)]
+use crate::api::server::{control::ControlService, history::HistoryGrpcService};
use crate::{
aclient::settings::Settings,
- components::{history::HistoryGrpcService, semantic::SemanticGrpcService},
- daemon::DaemonHandle,
- generated::{
- control::{ControlService, control_server::ControlServer},
- history::history_server::HistoryServer,
- semantic::semantic_server::SemanticServer,
+ api::generated::{
+ control::control_server::ControlServer, history::history_server::HistoryServer,
},
+ daemon::DaemonHandle,
};
/// Run the gRPC server with the given services.
@@ -19,7 +22,6 @@ use crate::{
pub(crate) fn run_grpc_server(
settings: &Settings,
history_service: HistoryServer<HistoryGrpcService>,
- semantic_service: SemanticServer<SemanticGrpcService>,
control_service: ControlServer<ControlService>,
handle: DaemonHandle,
) -> Result<()> {
@@ -28,42 +30,36 @@ pub(crate) fn run_grpc_server(
let socket_path = settings.daemon.socket_path.clone();
- let (uds, cleanup) = if cfg!(target_os = "linux") && settings.daemon.systemd_socket {
- #[cfg(target_os = "linux")]
- {
- use eyre::{OptionExt, WrapErr};
- use std::os::unix::net::SocketAddr;
- use std::path::PathBuf;
- tracing::info!("getting systemd socket");
- let listener = listenfd::ListenFd::from_env()
- .take_unix_listener(0)?
- .ok_or_eyre("missing systemd socket")?;
- listener.set_nonblocking(true)?;
- let actual_path: Result<PathBuf, eyre::Report> = listener
- .local_addr()
- .context("getting systemd socket's path")
- .and_then(|addr: SocketAddr| {
- addr.as_pathname()
- .ok_or_eyre("systemd socket missing path")
- .map(|path: &std::path::Path| path.to_owned())
- });
- match actual_path {
- Ok(actual_path) => {
- tracing::info!("listening on systemd socket: {actual_path:?}");
- if actual_path != std::path::Path::new(&socket_path) {
- tracing::warn!(
- "systemd socket is not at configured client path: {socket_path:?}"
- );
- }
- }
- Err(err) => {
+ let (uds, cleanup) = if settings.daemon.systemd_socket {
+ tracing::info!("getting systemd socket");
+ let listener = listenfd::ListenFd::from_env()
+ .take_unix_listener(0)?
+ .ok_or_eyre("missing systemd socket")?;
+ listener.set_nonblocking(true)?;
+ let actual_path: Result<PathBuf, eyre::Report> = listener
+ .local_addr()
+ .context("getting systemd socket's path")
+ .and_then(|addr: SocketAddr| {
+ addr.as_pathname()
+ .ok_or_eyre("systemd socket missing path")
+ .map(|path: &std::path::Path| path.to_owned())
+ });
+ match actual_path {
+ Ok(actual_path) => {
+ tracing::info!("listening on systemd socket: {actual_path:?}");
+ if actual_path != std::path::Path::new(&socket_path) {
tracing::warn!(
- "could not detect systemd socket path, ensure that it's at the configured path: {socket_path:?}, error: {err:?}"
+ "systemd socket is not at configured client path: {socket_path:?}"
);
}
}
- (UnixListener::from_std(listener)?, false)
+ Err(err) => {
+ tracing::warn!(
+ "could not detect systemd socket path, ensure that it's at the configured path: {socket_path:?}, error: {err:?}"
+ );
+ }
}
+ (UnixListener::from_std(listener)?, false)
} else {
tracing::info!("listening on unix socket {socket_path:?}");
(UnixListener::bind(socket_path.clone())?, true)
@@ -101,7 +97,6 @@ pub(crate) fn run_grpc_server(
if let Err(e) = Server::builder()
.add_service(history_service)
- .add_service(semantic_service)
.add_service(control_service)
.serve_with_incoming_shutdown(uds_stream, shutdown_signal)
.await