chore: Commit

19 files changed, 547 insertions, 1778 deletions

diff --git a/crates/client/src/command/client/history.rs b/crates/client/src/command/client/history.rs
index 9f383a08..4d633b7d 100644
--- a/crates/client/src/command/client/history.rs
+++ b/crates/client/src/command/client/history.rs
@@ -262,6 +262,111 @@ pub(crate) fn print_list(
     }
 }
 
+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),
+    }
+}
+
+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),
+    }
+}
+
 fn check_for_write_errors(write: Result<(), io::Error>) {
     if let Err(err) = write {
         // Ignore broken pipe (issue #626)
diff --git a/crates/daemon/build.rs b/crates/daemon/build.rs
index 5abf6b2a..62612968 100644
--- a/crates/daemon/build.rs
+++ b/crates/daemon/build.rs
@@ -7,7 +7,6 @@ fn main() -> Result<(), std::io::Error> {
     let proto_paths = [
         "proto/history.proto",
         "proto/control.proto",
-        "proto/semantic.proto",
     ];
     let proto_include_dirs = ["proto"];
 
diff --git a/crates/daemon/proto/control.proto b/crates/daemon/proto/control.proto
index 06347902..47a7b8b2 100644
--- a/crates/daemon/proto/control.proto
+++ b/crates/daemon/proto/control.proto
@@ -1,62 +1,23 @@
 syntax = "proto3";
 package control;
 
-// The Control service allows external processes (CLI commands, etc.)
-// to inject events into the running daemon.
 service Control {
-  // Send an event to the daemon's event bus
-  rpc SendEvent(SendEventRequest) returns (SendEventResponse);
-}
-
-message SendEventRequest {
-  oneof event {
-    // History was pruned - search index needs full rebuild
-    HistoryPrunedEvent history_pruned = 1;
-
-    // Specific history items were deleted
-    HistoryDeletedEvent history_deleted = 2;
-
-    // Request immediate sync
-    ForceSyncEvent force_sync = 3;
-
-    // Settings have changed, reload if needed
-    SettingsReloadedEvent settings_reloaded = 4;
-
-    // Request graceful shutdown
-    ShutdownEvent shutdown = 5;
-
-    // History was rebuilt - search index needs full rebuild
-    HistoryRebuiltEvent history_rebuilt = 6;
-  }
-}
-
-message SendEventResponse {
-  // Empty on success; errors come through gRPC status
-}
-
-// Individual event message types
-
-message HistoryPrunedEvent {
-  // No fields needed - just signals that pruning happened
-}
-
-message HistoryRebuiltEvent {
-  // No fields needed - just signals that rebuilding happened
-}
-
-message HistoryDeletedEvent {
-  // IDs of deleted history items (UUIDs as strings)
-  repeated string ids = 1;
-}
+  // Tell the daemon to perform a sync operation.
+  rpc ForceSync(ForceSyncRequest) returns (ForceSyncReply);
 
-message ForceSyncEvent {
-  // No fields needed - just triggers sync
+  // Query the daemon for it's status.
+  rpc Status(StatusRequest) returns (StatusReply);
 }
 
-message SettingsReloadedEvent {
-  // No fields needed - components should re-read settings
+message ForceSyncRequest {}
+message ForceSyncReply {
+  bool accepted = 1;
 }
 
-message ShutdownEvent {
-  // No fields needed - triggers graceful shutdown
+message StatusRequest {}
+message StatusReply {
+  bool healthy = 1;
+  string version = 2;
+  uint32 pid = 3;
+  uint32 protocol = 4;
 }
diff --git a/crates/daemon/proto/history.proto b/crates/daemon/proto/history.proto
index 59c12471..46a41a5e 100644
--- a/crates/daemon/proto/history.proto
+++ b/crates/daemon/proto/history.proto
@@ -1,8 +1,13 @@
 syntax = "proto3";
 package history;
 
+service History {
+  rpc StartHistory(StartHistoryRequest) returns (StartHistoryReply);
+  rpc EndHistory(EndHistoryRequest) returns (EndHistoryReply);
+  rpc TailHistory(TailHistoryRequest) returns (stream TailHistoryReply);
+}
+
 message StartHistoryRequest {
-  // If people are still using my software in ~530 years, they can figure out a u128 migration
   uint64 timestamp = 1; // nanosecond unix epoch
   string command = 2;
   string cwd = 3;
@@ -11,19 +16,17 @@ message StartHistoryRequest {
   string author = 6;
   string intent = 7;
 }
-
-message EndHistoryRequest {
-  string id = 1;
-  int64 exit = 2;
-  uint64 duration = 3;
-}
-
 message StartHistoryReply {
   string id = 1;
   string version = 2;
   uint32 protocol = 3;
 }
 
+message EndHistoryRequest {
+  string id = 1;
+  int64 exit = 2;
+  uint64 duration = 3;
+}
 message EndHistoryReply {
   string id = 1;
   uint64 idx = 2;
@@ -31,22 +34,12 @@ message EndHistoryReply {
   uint32 protocol = 4;
 }
 
-message StatusRequest {}
-
-message StatusReply {
-  bool healthy = 1;
-  string version = 2;
-  uint32 pid = 3;
-  uint32 protocol = 4;
-}
-
-message ShutdownRequest {}
-
-message ShutdownReply {
-  bool accepted = 1;
-}
 
 message TailHistoryRequest {}
+message TailHistoryReply {
+  HistoryEventKind kind = 1;
+  HistoryEntry history = 2;
+}
 
 enum HistoryEventKind {
   HISTORY_EVENT_KIND_UNSPECIFIED = 0;
@@ -66,16 +59,3 @@ message HistoryEntry {
   int64 exit = 9;
   int64 duration = 10;
 }
-
-message TailHistoryReply {
-  HistoryEventKind kind = 1;
-  HistoryEntry history = 2;
-}
-
-service History {
-  rpc StartHistory(StartHistoryRequest) returns (StartHistoryReply);
-  rpc EndHistory(EndHistoryRequest) returns (EndHistoryReply);
-  rpc TailHistory(TailHistoryRequest) returns (stream TailHistoryReply);
-  rpc Status(StatusRequest) returns (StatusReply);
-  rpc Shutdown(ShutdownRequest) returns (ShutdownReply);
-}
diff --git a/crates/daemon/proto/semantic.proto b/crates/daemon/proto/semantic.proto
deleted file mode 100644
index 07e550c8..00000000
--- a/crates/daemon/proto/semantic.proto
+++ /dev/null
@@ -1,47 +0,0 @@
-syntax = "proto3";
-package semantic;
-
-service Semantic {
-  rpc RecordCommands(stream CommandCapture) returns (RecordCommandsReply);
-  rpc CommandOutput(CommandOutputRequest) returns (CommandOutputReply);
-}
-
-message CommandCapture {
-  string prompt = 1;
-  string command = 2;
-  string output = 3;
-  optional int32 exit_code = 4;
-  optional string history_id = 5;
-  optional string session_id = 6;
-  bool output_truncated = 7;
-  uint64 output_observed_bytes = 8;
-}
-
-message RecordCommandsReply {
-  uint64 accepted = 1;
-}
-
-message CommandOutputRequest {
-  string history_id = 1;
-  repeated OutputRange ranges = 2;
-}
-
-message OutputRange {
-  int64 start = 1;
-  int64 end = 2;
-}
-
-message OutputLine {
-  uint64 line_number = 1;
-  string content = 2;
-}
-
-message CommandOutputReply {
-  bool found = 1;
-  string output = 2;
-  uint64 total_bytes = 3;
-  uint64 total_lines = 4;
-  repeated OutputLine lines = 5;
-  bool output_truncated = 6;
-  uint64 output_observed_bytes = 7;
-}
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