agent_tui_app/app/daemon/
server.rs

1//! Daemon server runtime.
2
3use crate::adapters::rpc::RpcResponse;
4use crate::common::DaemonError;
5use crate::common::join_thread_with_timeout_or_reap;
6use crate::common::telemetry;
7use libc::POLLIN;
8use libc::poll;
9use libc::pollfd;
10use std::collections::HashSet;
11use std::io::Read;
12use std::io::Write;
13use std::os::unix::io::AsRawFd;
14use std::os::unix::io::RawFd;
15use std::os::unix::net::UnixStream;
16use tracing::debug;
17use tracing::error;
18use tracing::info;
19use tracing::warn;
20
21use crate::app::daemon::rpc_core::RpcCore;
22use crate::app::daemon::rpc_core::RpcCoreError;
23use crate::app::daemon::rpc_core::RpcResponseWriter;
24use crate::app::daemon::transport::TransportConnection;
25use crate::app::daemon::transport::TransportError;
26use crate::app::daemon::transport::TransportListener;
27use crate::app::daemon::transport::UnixSocketConnection;
28use crate::app::daemon::transport::UnixSocketListener;
29use crate::app::daemon::ws_server::WsConfig;
30use crate::app::daemon::ws_server::WsServerError;
31use crate::app::daemon::ws_server::WsServerHandle;
32use crate::app::daemon::ws_server::start_ws_server;
33use crate::infra::daemon::DaemonConfig;
34use crate::infra::daemon::LockFile;
35use crate::infra::daemon::SignalHandler;
36use crate::infra::daemon::remove_lock_file;
37use crate::infra::ipc::socket_path;
38use std::sync::Arc;
39use std::sync::atomic::AtomicBool;
40use std::sync::atomic::AtomicU64;
41use std::sync::atomic::AtomicUsize;
42use std::sync::atomic::Ordering;
43use std::sync::mpsc;
44use std::sync::mpsc::SyncSender;
45use std::thread;
46use std::time::Duration;
47use std::time::Instant;
48
49const CHANNEL_CAPACITY: usize = 128;
50static CONNECTION_ID: AtomicU64 = AtomicU64::new(1);
51
52struct ShutdownWaker {
53    reader: UnixStream,
54    writer: Arc<std::sync::Mutex<UnixStream>>,
55}
56
57impl ShutdownWaker {
58    fn new() -> std::io::Result<Self> {
59        let (reader, writer) = UnixStream::pair()?;
60        reader.set_nonblocking(true)?;
61        Ok(Self {
62            reader,
63            writer: Arc::new(std::sync::Mutex::new(writer)),
64        })
65    }
66
67    fn notifier(&self) -> crate::usecases::ports::ShutdownNotifierHandle {
68        Arc::new(ShutdownNotify {
69            writer: Arc::clone(&self.writer),
70        })
71    }
72
73    fn reader_fd(&self) -> RawFd {
74        self.reader.as_raw_fd()
75    }
76
77    fn drain(&mut self) {
78        let mut buf = [0u8; 64];
79        loop {
80            match self.reader.read(&mut buf) {
81                Ok(0) => break,
82                Ok(_) => continue,
83                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => break,
84                Err(_) => break,
85            }
86        }
87    }
88}
89
90struct ShutdownNotify {
91    writer: Arc<std::sync::Mutex<UnixStream>>,
92}
93
94impl crate::usecases::ports::ShutdownNotifier for ShutdownNotify {
95    fn notify(&self) -> Result<(), std::io::Error> {
96        let mut writer = self
97            .writer
98            .lock()
99            .map_err(|_| std::io::Error::other("shutdown notifier mutex poisoned"))?;
100        writer.write_all(&[1])
101    }
102}
103
104struct UnixRpcWriter<'a> {
105    conn: &'a mut UnixSocketConnection,
106}
107
108impl RpcResponseWriter for UnixRpcWriter<'_> {
109    fn write_response(&mut self, response: &RpcResponse) -> Result<(), RpcCoreError> {
110        self.conn.write_response(response).map_err(|err| match err {
111            TransportError::ConnectionClosed => RpcCoreError::ConnectionClosed,
112            other => RpcCoreError::Other(other.to_string()),
113        })
114    }
115}
116
117fn session_selector_for_log(request: &crate::adapters::rpc::RpcRequest) -> &'static str {
118    match request.param_str("session") {
119        None => "implicit-active",
120        Some(raw) => {
121            let trimmed = raw.trim();
122            if trimmed.is_empty() {
123                "blank"
124            } else if trimmed == "active" {
125                "explicit-active"
126            } else {
127                "explicit"
128            }
129        }
130    }
131}
132
133pub struct DaemonServer {
134    core: Arc<RpcCore>,
135    active_connections: Arc<AtomicUsize>,
136    active_fds: Arc<std::sync::Mutex<HashSet<RawFd>>>,
137    active_fds_cv: Arc<std::sync::Condvar>,
138    connection_wait_lock: Arc<std::sync::Mutex<()>>,
139    connection_cv: Arc<std::sync::Condvar>,
140    stream_threads: std::sync::Mutex<Vec<thread::JoinHandle<()>>>,
141}
142
143struct ThreadPool {
144    workers: Vec<thread::JoinHandle<()>>,
145    sender: SyncSender<UnixSocketConnection>,
146}
147
148impl ThreadPool {
149    fn new(size: usize, server: Arc<DaemonServer>) -> std::io::Result<Self> {
150        let (sender, receiver) = mpsc::sync_channel::<UnixSocketConnection>(CHANNEL_CAPACITY);
151        let receiver = Arc::new(std::sync::Mutex::new(receiver));
152
153        let mut workers = Vec::with_capacity(size);
154        for id in 0..size {
155            let receiver = Arc::clone(&receiver);
156            let server = Arc::clone(&server);
157
158            let handle = match thread::Builder::new()
159                .name(format!("worker-{id}"))
160                .spawn(move || {
161                    let worker_span = tracing::info_span!("daemon_worker", worker_id = id);
162                    let _worker_guard = worker_span.enter();
163                    debug!(worker_id = id, "Worker thread started");
164                    loop {
165                        let conn = {
166                            let lock = match receiver.lock() {
167                                Ok(l) => l,
168                                Err(e) => {
169                                    error!(worker_id = id, error = %e, "Worker receiver lock poisoned");
170                                    break;
171                                }
172                            };
173                            match lock.recv() {
174                                Ok(conn) => conn,
175                                Err(mpsc::RecvError) => break,
176                            }
177                        };
178
179                        server.active_connections.fetch_add(1, Ordering::Relaxed);
180                        Arc::clone(&server).handle_client(conn);
181                        let remaining = server.active_connections.fetch_sub(1, Ordering::Relaxed) - 1;
182                        if remaining == 0 {
183                            server.connection_cv.notify_all();
184                        }
185                    }
186                    debug!(worker_id = id, "Worker thread stopped");
187                }) {
188                Ok(h) => h,
189                Err(e) => {
190                    error!(worker_id = id, error = %e, "Failed to spawn worker");
191                    continue;
192                }
193            };
194
195            workers.push(handle);
196        }
197
198        if workers.is_empty() {
199            return Err(std::io::Error::other("Failed to spawn any worker threads"));
200        }
201
202        if workers.len() < size {
203            warn!(
204                spawned = workers.len(),
205                requested = size,
206                "Only spawned partial worker threads"
207            );
208        }
209
210        Ok(Self { workers, sender })
211    }
212
213    fn execute(&self, conn: UnixSocketConnection) -> Result<(), UnixSocketConnection> {
214        self.sender.try_send(conn).map_err(|e| match e {
215            mpsc::TrySendError::Full(c) | mpsc::TrySendError::Disconnected(c) => c,
216        })
217    }
218
219    fn shutdown(self) {
220        drop(self.sender);
221        for worker in self.workers {
222            let _ = worker.join();
223        }
224    }
225}
226
227impl DaemonServer {
228    pub fn with_config(
229        config: DaemonConfig,
230        shutdown_flag: Arc<AtomicBool>,
231        shutdown_notifier: crate::usecases::ports::ShutdownNotifierHandle,
232    ) -> Result<Self, crate::infra::daemon::SessionError> {
233        let core = Arc::new(RpcCore::with_config(
234            config,
235            shutdown_flag,
236            shutdown_notifier,
237        )?);
238        Ok(Self {
239            core,
240            active_connections: Arc::new(AtomicUsize::new(0)),
241            active_fds: Arc::new(std::sync::Mutex::new(HashSet::new())),
242            active_fds_cv: Arc::new(std::sync::Condvar::new()),
243            connection_wait_lock: Arc::new(std::sync::Mutex::new(())),
244            connection_cv: Arc::new(std::sync::Condvar::new()),
245            stream_threads: std::sync::Mutex::new(Vec::new()),
246        })
247    }
248
249    fn session_repository_handle(&self) -> Arc<dyn crate::usecases::ports::SessionRepository> {
250        self.core.session_repository_handle()
251    }
252
253    pub fn shutdown_all_sessions(&self) {
254        self.core.shutdown_all_sessions();
255    }
256
257    fn register_connection(&self, fd: RawFd) {
258        let mut fds = self
259            .active_fds
260            .lock()
261            .unwrap_or_else(std::sync::PoisonError::into_inner);
262        fds.insert(fd);
263        self.active_fds_cv.notify_all();
264    }
265
266    fn unregister_connection(&self, fd: RawFd) {
267        let mut fds = self
268            .active_fds
269            .lock()
270            .unwrap_or_else(std::sync::PoisonError::into_inner);
271        fds.remove(&fd);
272        self.active_fds_cv.notify_all();
273    }
274
275    #[cfg(test)]
276    fn wait_for_registered_connection(&self, timeout: Duration) -> bool {
277        let fds = self
278            .active_fds
279            .lock()
280            .unwrap_or_else(std::sync::PoisonError::into_inner);
281        let result = self
282            .active_fds_cv
283            .wait_timeout_while(fds, timeout, |fds| fds.is_empty())
284            .unwrap_or_else(std::sync::PoisonError::into_inner);
285        !result.0.is_empty()
286    }
287
288    fn shutdown_connections(&self) {
289        let fds = {
290            let guard = self
291                .active_fds
292                .lock()
293                .unwrap_or_else(std::sync::PoisonError::into_inner);
294            guard.iter().copied().collect::<Vec<_>>()
295        };
296        for fd in fds {
297            // SAFETY: shutting down a socket fd is safe and idempotent for active connections.
298            unsafe {
299                libc::shutdown(fd, libc::SHUT_RDWR);
300            }
301        }
302    }
303
304    fn register_stream_thread(&self, handle: thread::JoinHandle<()>) {
305        let mut guard = self
306            .stream_threads
307            .lock()
308            .unwrap_or_else(std::sync::PoisonError::into_inner);
309        guard.push(handle);
310    }
311
312    fn join_stream_threads(&self, timeout: Duration) {
313        let handles = {
314            let mut guard = self
315                .stream_threads
316                .lock()
317                .unwrap_or_else(std::sync::PoisonError::into_inner);
318            guard.drain(..).collect::<Vec<_>>()
319        };
320
321        for handle in handles {
322            let _ = join_thread_with_timeout_or_reap(
323                handle,
324                timeout,
325                "stream thread",
326                "daemon-stream-reaper",
327            );
328        }
329    }
330
331    fn handle_client(self: Arc<Self>, mut conn: UnixSocketConnection) {
332        let idle_timeout = DaemonConfig::from_env().idle_timeout();
333        let conn_id = CONNECTION_ID.fetch_add(1, Ordering::Relaxed);
334        let conn_fd = conn.raw_fd();
335        self.register_connection(conn_fd);
336        let conn_span = tracing::info_span!("daemon_connection", conn_id);
337        let _conn_guard = conn_span.enter();
338        debug!(conn_id, "Client connected");
339
340        if let Err(err) = conn.set_read_timeout(Some(idle_timeout)) {
341            warn!(error = %err, "Failed to set read timeout");
342        }
343        if let Err(err) = conn.set_write_timeout(Some(Duration::from_secs(30))) {
344            warn!(error = %err, "Failed to set write timeout");
345        }
346
347        loop {
348            let request = match conn.read_request() {
349                Ok(req) => req,
350                Err(TransportError::ConnectionClosed) | Err(TransportError::Timeout) => break,
351                Err(TransportError::SizeLimit { max_bytes }) => {
352                    warn!(max_bytes, "Request size limit exceeded");
353                    let error_response = RpcResponse::error_without_id(
354                        -32700,
355                        &format!(
356                            "Parse error: request size limit exceeded ({}MB max)",
357                            max_bytes / 1024 / 1024
358                        ),
359                    );
360                    let _ = conn.write_response(&error_response);
361                    break;
362                }
363                Err(TransportError::Parse { source, request_id }) => {
364                    debug!(error = %source, "Request parse error");
365                    let message = format!("Parse error: {source}");
366                    let error_response = request_id.map_or_else(
367                        || RpcResponse::error_without_id(-32700, &message),
368                        |id| RpcResponse::error(id, -32700, &message),
369                    );
370                    let _ = conn.write_response(&error_response);
371                    continue;
372                }
373                Err(TransportError::Serialize(err)) => {
374                    error!(error = %err, "Response serialize error");
375                    break;
376                }
377                Err(TransportError::Io(err)) => {
378                    error!(error = %err, "Client connection error");
379                    break;
380                }
381            };
382
383            let request_id = request.id.clone();
384            let method = request.method.clone();
385            let request_span = tracing::debug_span!(
386                "rpc_request",
387                request_id = %request_id,
388                method = %method,
389                session_selector = session_selector_for_log(&request)
390            );
391            let _request_guard = request_span.enter();
392            let start = Instant::now();
393
394            if let Some(kind) = RpcCore::stream_kind_for_method(&method) {
395                self.spawn_stream_thread(conn, request, kind, conn_id, conn_fd);
396                return;
397            }
398
399            let response = self.core.route(request);
400            debug!(
401                request_id = %request_id,
402                method = %method,
403                elapsed_ms = start.elapsed().as_millis(),
404                "RPC request handled"
405            );
406
407            if let Err(err) = conn.write_response(&response) {
408                match err {
409                    TransportError::ConnectionClosed => break,
410                    _ => {
411                        error!(error = %err, "Client write error");
412                        break;
413                    }
414                }
415            }
416        }
417
418        self.unregister_connection(conn_fd);
419        debug!(conn_id, "Client disconnected");
420    }
421
422    fn spawn_stream_thread(
423        self: &Arc<Self>,
424        conn: UnixSocketConnection,
425        request: crate::adapters::rpc::RpcRequest,
426        kind: crate::app::daemon::rpc_core::StreamKind,
427        conn_id: u64,
428        conn_fd: RawFd,
429    ) {
430        let server = Arc::clone(self);
431        let method = match kind {
432            crate::app::daemon::rpc_core::StreamKind::Attach => "attach_stream",
433            crate::app::daemon::rpc_core::StreamKind::LivePreview => "live_preview_stream",
434            crate::app::daemon::rpc_core::StreamKind::Flightdeck => "flightdeck_stream",
435        };
436
437        server.active_connections.fetch_add(1, Ordering::Relaxed);
438        let payload = Arc::new(std::sync::Mutex::new(Some((conn, request))));
439        let payload_for_thread = Arc::clone(&payload);
440        let server_for_thread = Arc::clone(&server);
441        let span = tracing::info_span!("daemon_stream", conn_id, method = %method);
442        let spawn_result = thread::Builder::new()
443            .name(format!("stream-{method}-{conn_id}"))
444            .spawn(move || {
445                let _guard = span.enter();
446                let start = Instant::now();
447
448                let Some((mut conn, request)) = payload_for_thread
449                    .lock()
450                    .unwrap_or_else(std::sync::PoisonError::into_inner)
451                    .take()
452                else {
453                    warn!("Stream payload missing; dropping connection");
454                    let remaining = server_for_thread
455                        .active_connections
456                        .fetch_sub(1, Ordering::Relaxed)
457                        - 1;
458                    if remaining == 0 {
459                        server_for_thread.connection_cv.notify_all();
460                    }
461                    server_for_thread.unregister_connection(conn_fd);
462                    return;
463                };
464
465                let stream_result = {
466                    let mut writer = UnixRpcWriter { conn: &mut conn };
467                    server_for_thread
468                        .core
469                        .handle_stream(&mut writer, request, kind, None)
470                };
471
472                debug!(
473                    conn_id,
474                    method = %method,
475                    elapsed_ms = start.elapsed().as_millis(),
476                    "RPC stream handled"
477                );
478                if let Err(err) = stream_result {
479                    match err {
480                        RpcCoreError::ConnectionClosed => {}
481                        other => error!(error = %other, "RPC stream error"),
482                    }
483                }
484                debug!(conn_id, method = %method, "Stream connection closed");
485
486                let remaining = server_for_thread
487                    .active_connections
488                    .fetch_sub(1, Ordering::Relaxed)
489                    - 1;
490                if remaining == 0 {
491                    server_for_thread.connection_cv.notify_all();
492                }
493                server_for_thread.unregister_connection(conn_fd);
494            });
495
496        match spawn_result {
497            Ok(handle) => server.register_stream_thread(handle),
498            Err(_) => {
499                let Some((mut conn, request)) = payload
500                    .lock()
501                    .unwrap_or_else(std::sync::PoisonError::into_inner)
502                    .take()
503                else {
504                    warn!("Stream payload missing; dropping connection");
505                    let remaining = server.active_connections.fetch_sub(1, Ordering::Relaxed) - 1;
506                    if remaining == 0 {
507                        server.connection_cv.notify_all();
508                    }
509                    return;
510                };
511
512                warn!("Failed to spawn stream thread, handling on worker thread");
513                let mut writer = UnixRpcWriter { conn: &mut conn };
514                let _ = server.core.handle_stream(&mut writer, request, kind, None);
515
516                let remaining = server.active_connections.fetch_sub(1, Ordering::Relaxed) - 1;
517                if remaining == 0 {
518                    server.connection_cv.notify_all();
519                }
520                server.unregister_connection(conn_fd);
521            }
522        }
523    }
524}
525
526fn init_logging() -> telemetry::TelemetryGuard {
527    telemetry::init_tracing("info")
528}
529
530fn bind_socket(
531    socket_path: &std::path::Path,
532    max_request_bytes: usize,
533) -> Result<UnixSocketListener, DaemonError> {
534    if socket_path.exists() {
535        std::fs::remove_file(socket_path).map_err(|e| DaemonError::SocketBind {
536            operation: "remove stale socket",
537            source: Box::new(e),
538        })?;
539    }
540
541    let listener = UnixSocketListener::bind(socket_path, max_request_bytes).map_err(|e| {
542        DaemonError::SocketBind {
543            operation: "bind socket",
544            source: Box::new(e),
545        }
546    })?;
547    listener
548        .set_nonblocking(true)
549        .map_err(|e| DaemonError::SocketBind {
550            operation: "set non-blocking",
551            source: Box::new(e),
552        })?;
553
554    Ok(listener)
555}
556
557fn run_accept_loop(
558    listener: &UnixSocketListener,
559    pool: &ThreadPool,
560    shutdown: &AtomicBool,
561    waker: &mut ShutdownWaker,
562) {
563    let listener_fd = listener.as_raw_fd();
564    let wake_fd = waker.reader_fd();
565    let mut fds = [
566        pollfd {
567            fd: listener_fd,
568            events: POLLIN,
569            revents: 0,
570        },
571        pollfd {
572            fd: wake_fd,
573            events: POLLIN,
574            revents: 0,
575        },
576    ];
577
578    while !shutdown.load(Ordering::Relaxed) {
579        // SAFETY: `fds` is stack-allocated and length matches call argument.
580        let poll_result = unsafe { poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, -1) };
581        if poll_result < 0 {
582            let err = std::io::Error::last_os_error();
583            if err.kind() == std::io::ErrorKind::Interrupted {
584                continue;
585            }
586            error!(error = %err, "poll failed");
587            continue;
588        }
589
590        if fds[1].revents & POLLIN != 0 {
591            waker.drain();
592            if shutdown.load(Ordering::Relaxed) {
593                break;
594            }
595        }
596
597        if fds[0].revents & POLLIN != 0 {
598            loop {
599                match listener.accept() {
600                    Ok(conn) => {
601                        if let Err(conn) = pool.execute(conn) {
602                            warn!("Thread pool channel closed, dropping connection");
603                            drop(conn);
604                        }
605                    }
606                    Err(TransportError::Timeout) => break,
607                    Err(err) => {
608                        if !shutdown.load(Ordering::Relaxed) {
609                            error!(error = %err, "Error accepting connection");
610                        }
611                        break;
612                    }
613                }
614            }
615        }
616    }
617}
618
619fn wait_for_connections(server: &DaemonServer, timeout_secs: u64) {
620    info!(
621        active_connections = server.active_connections.load(Ordering::Relaxed),
622        "Waiting for active connections to complete"
623    );
624    let shutdown_deadline = Instant::now() + Duration::from_secs(timeout_secs);
625    let mut guard = server
626        .connection_wait_lock
627        .lock()
628        .unwrap_or_else(std::sync::PoisonError::into_inner);
629    while server.active_connections.load(Ordering::Relaxed) > 0 {
630        let now = Instant::now();
631        if now >= shutdown_deadline {
632            warn!("Shutdown timeout, forcing close");
633            break;
634        }
635        let remaining = shutdown_deadline.saturating_duration_since(now);
636        let (_guard, result) = server
637            .connection_cv
638            .wait_timeout(guard, remaining)
639            .unwrap_or_else(std::sync::PoisonError::into_inner);
640        guard = _guard;
641        if result.timed_out() {
642            warn!("Shutdown timeout, forcing close");
643            break;
644        }
645    }
646}
647
648fn cleanup(
649    socket_path: &std::path::Path,
650    lock_path: &std::path::Path,
651    server: &DaemonServer,
652    pool: ThreadPool,
653    ws_handle: Option<WsServerHandle>,
654) {
655    if let Some(handle) = ws_handle {
656        handle.shutdown();
657    }
658
659    info!("Cleaning up sessions...");
660    server.shutdown_all_sessions();
661
662    info!("Stopping thread pool...");
663    pool.shutdown();
664
665    if socket_path.exists() {
666        let _ = std::fs::remove_file(socket_path);
667    }
668
669    remove_lock_file(lock_path);
670
671    info!("Daemon shutdown complete");
672}
673
674pub fn start_daemon() -> Result<(), DaemonError> {
675    let _telemetry = init_logging();
676
677    let socket_path = socket_path();
678    let lock_path = socket_path.with_extension("lock");
679
680    let _lock = LockFile::acquire(&lock_path)?;
681
682    let config = DaemonConfig::from_env();
683    let max_connections = config.max_connections();
684    let max_request_bytes = config.max_request_bytes();
685    let listener = bind_socket(&socket_path, max_request_bytes)?;
686    info!(socket = %socket_path.display(), pid = std::process::id(), "Daemon started");
687
688    let shutdown = Arc::new(AtomicBool::new(false));
689    let waker = ShutdownWaker::new()
690        .map_err(|e| DaemonError::SignalSetup(format!("failed to create shutdown waker: {e}")))?;
691    let shutdown_notifier = waker.notifier();
692
693    let server = Arc::new(
694        DaemonServer::with_config(
695            config,
696            Arc::clone(&shutdown),
697            Arc::clone(&shutdown_notifier),
698        )
699        .map_err(|source| DaemonError::Startup {
700            operation: "initialize_sessions",
701            source: Box::new(source),
702        })?,
703    );
704
705    let ws_handle = match start_ws_server(
706        Arc::clone(&server.core),
707        Arc::clone(&shutdown),
708        WsConfig::from_env(),
709    ) {
710        Ok(handle) => Some(handle),
711        Err(WsServerError::Disabled) => None,
712        Err(WsServerError::Io { operation, source }) => {
713            warn!(operation = %operation, error = %source, "Failed to start WS server");
714            None
715        }
716        Err(WsServerError::InvalidListen { message }) => {
717            warn!(reason = %message, "Invalid WS listen address");
718            None
719        }
720    };
721
722    let _signal_handler = SignalHandler::setup(Arc::clone(&shutdown), Some(shutdown_notifier))?;
723
724    let pool = ThreadPool::new(max_connections, Arc::clone(&server))
725        .map_err(|e| DaemonError::ThreadPool(e.to_string()))?;
726
727    let mut waker = waker;
728    run_accept_loop(&listener, &pool, &shutdown, &mut waker);
729
730    info!("Shutting down daemon...");
731    server.shutdown_connections();
732    wait_for_connections(&server, 5);
733    server.join_stream_threads(Duration::from_secs(2));
734    cleanup(&socket_path, &lock_path, &server, pool, ws_handle);
735
736    Ok(())
737}
738
739#[cfg(test)]
740#[path = "server_tests.rs"]
741mod tests;
agent_tui_app/app/daemon/server.rs

agent_tui_app/app/daemon/
server.rs
