netget/server/socket_file/mod.rs
1//! Unix domain socket server implementation
2//!
3//! Platform: Unix/Linux only (uses Unix domain sockets)
4#![cfg(unix)]
5
6pub mod actions;
7
8use anyhow::{Context, Result};
9use bytes::Bytes;
10use std::collections::HashMap;
11use std::path::PathBuf;
12use std::sync::Arc;
13use std::time::Duration;
14use tokio::io::{AsyncReadExt, AsyncWriteExt};
15use tokio::net::UnixStream;
16use tokio::sync::{mpsc, Mutex};
17use tracing::{debug, info};
18
19use super::connection::ConnectionId;
20use crate::llm::action_helper::call_llm;
21use crate::llm::ollama_client::OllamaClient;
22use crate::llm::ActionResult;
23use crate::logging::emit::Log;
24use crate::protocol::Event;
25use crate::server::SocketFileProtocol;
26use crate::state::app_state::AppState;
27use actions::{SOCKET_FILE_CONNECTION_OPENED_EVENT, SOCKET_FILE_DATA_RECEIVED_EVENT};
28
29/// How long a connected peer may send nothing at all before the server closes it.
30///
31/// 300 seconds, the window a `manual` rule gives a human to answer one event
32/// (`src/state/intercepts.rs`), for the reason `tcp` gives for the same number: this is the
33/// generic byte stream over a filesystem socket, and the peer it most often has is NetGet's own
34/// `socket_file` client, which connects inside `connect()`, writes nothing, and waits for a
35/// person to use `[ send message ]` — **connected and silent**. A shorter bound drops that peer
36/// while the operator is still looking at it. A stranger is still bounded, and still capped at
37/// [`MAX_CONNECTIONS`]; the node is owner-only (0600), so the stranger is a local process of the
38/// same user. Declared as the `first_byte_timeout_secs` startup parameter.
39///
40/// A `send_first` server is not the exception it looks like: the banner task holds the
41/// connection's [`ConnectionActivity`](crate::server::accept_bounded::ConnectionActivity) busy
42/// for the whole of its model round-trip, so the clock does not run while the peer is waiting
43/// to be spoken to.
44const FIRST_BYTE_READ_TIMEOUT: Duration = Duration::from_secs(300);
45
46/// How long an established session may be silent between messages.
47///
48/// Fifteen minutes, `tcp`'s number: three times the window a `manual` rule gives a human, so a
49/// session whose last exchange was composed by hand still has minutes of think-time left. A
50/// message still being answered is never silence — the reader consults `ConnectionActivity`.
51/// Declared as the `idle_timeout_secs` startup parameter.
52const IDLE_BETWEEN_MESSAGES_TIMEOUT: Duration = Duration::from_secs(900);
53
54/// Concurrent connections this server admits.
55///
56/// [`crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS`]. Each connection may queue bytes
57/// while its LLM call is in flight, so this is the multiplier that turns the per-connection cost
58/// into a total one.
59const MAX_CONNECTIONS: usize = crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS;
60
61/// What a peer over [`MAX_CONNECTIONS`] is told before the socket closes: nothing.
62///
63/// A raw byte stream has no vocabulary — anything written would be read as payload — so the
64/// refusal is a clean EOF, logged by `accept_bounded_unix` with
65/// `decision=fail_closed_connection_cap`. Same reasoning as `tcp`.
66const CONNECTION_CAP_REFUSAL: &[u8] = b"";
67
68/// Read from `reader` with a deadline that a connection doing work can never trip.
69///
70/// Returns `None` when the peer has genuinely been silent for `bound`, and the read's own
71/// result otherwise. The read loop hands each message to a spawned task and goes straight back
72/// to `read()`, so the deadline and an answer in progress are live at the same moment;
73/// consulting [`ConnectionActivity::idle_for`](crate::server::accept_bounded::ConnectionActivity::idle_for)
74/// — which reports a connection with work in flight as not idle — keeps the deadline from
75/// closing the connection it is in the middle of answering. The same helper as `tcp`'s.
76async fn read_bounded<R>(
77 reader: &mut R,
78 buffer: &mut [u8],
79 activity: &crate::server::accept_bounded::ConnectionActivity,
80 bound: Duration,
81) -> Option<std::io::Result<usize>>
82where
83 R: tokio::io::AsyncRead + Unpin,
84{
85 loop {
86 match tokio::time::timeout(bound, reader.read(buffer)).await {
87 Ok(result) => return Some(result),
88 Err(_) => match activity.idle_for() {
89 // Work in flight: the peer is waiting on us, not the other way round.
90 None => continue,
91 // Something crossed the connection inside the window; wait again from there.
92 Some(idle) if idle < bound => continue,
93 Some(_) => return None,
94 },
95 }
96 }
97}
98
99/// Connection state for LLM processing
100#[derive(Debug, Clone, PartialEq)]
101enum ConnectionState {
102 Idle,
103 Processing,
104 Accumulating,
105}
106
107/// Per-connection data for LLM handling
108struct ConnectionData {
109 state: ConnectionState,
110 queued_data: Vec<u8>,
111 write_half: Arc<Mutex<tokio::io::WriteHalf<UnixStream>>>,
112}
113
114/// Describe a file type for the "refusing to unlink" error message.
115fn describe_file_type(ft: &std::fs::FileType) -> &'static str {
116 use std::os::unix::fs::FileTypeExt;
117 if ft.is_dir() {
118 "directory"
119 } else if ft.is_symlink() {
120 "symlink"
121 } else if ft.is_fifo() {
122 "FIFO"
123 } else if ft.is_char_device() {
124 "character device"
125 } else if ft.is_block_device() {
126 "block device"
127 } else {
128 "regular file"
129 }
130}
131
132/// Removes the socket node this server bound, when the accept loop ends — including when the
133/// task is aborted by `stop_server`, because aborting drops the task future and with it this
134/// guard.
135///
136/// Without it a stopped server left its socket node on disk advertising a service nothing was
137/// listening on: `connect(2)` gets ECONNREFUSED rather than ENOENT, which reads as "the service
138/// is down" instead of "there is no service". `named_pipe` and `pty` both clean up after
139/// themselves; this one did not. The node is always one we created — the bind is preceded by a
140/// guarded unlink of any stale socket at the same path.
141struct SocketCleanup(PathBuf);
142
143impl Drop for SocketCleanup {
144 fn drop(&mut self) {
145 let _ = std::fs::remove_file(&self.0);
146 }
147}
148
149/// Unix domain socket server that listens for incoming connections
150pub struct SocketFileServer;
151
152impl SocketFileServer {
153 /// Spawn the socket file server with integrated LLM actions
154 #[allow(clippy::too_many_arguments)]
155 pub async fn spawn_with_llm_actions(
156 socket_path: PathBuf,
157 llm_client: OllamaClient,
158 app_state: Arc<AppState>,
159 status_tx: mpsc::UnboundedSender<String>,
160 send_first: bool,
161 server_id: crate::state::ServerId,
162 first_byte_timeout_secs: Option<u64>,
163 idle_timeout_secs: Option<u64>,
164 ) -> Result<PathBuf> {
165 let first_byte_timeout = first_byte_timeout_secs
166 .map(Duration::from_secs)
167 .unwrap_or(FIRST_BYTE_READ_TIMEOUT);
168 let idle_timeout = idle_timeout_secs
169 .map(Duration::from_secs)
170 .unwrap_or(IDLE_BETWEEN_MESSAGES_TIMEOUT);
171 // Remove a stale socket file, but ONLY if the path really is a socket.
172 //
173 // `socket_path` comes from the LLM or an MCP caller, so an unconditional
174 // `remove_file` here is an arbitrary-file delete: "./netget.sock" typo'd as
175 // "./netget.rs", or a deliberately chosen "~/.ssh/id_ed25519", would be unlinked
176 // before the bind. `symlink_metadata` deliberately does not follow symlinks, so a
177 // symlink pointing at a regular file is refused rather than followed and deleted.
178 match std::fs::symlink_metadata(&socket_path) {
179 Ok(meta) => {
180 use std::os::unix::fs::FileTypeExt;
181 if !meta.file_type().is_socket() {
182 anyhow::bail!(
183 "Refusing to remove {:?}: it exists but is not a Unix domain socket \
184 (it is a {}). Delete it yourself if that is really what you want, or \
185 pass a different socket_path.",
186 socket_path,
187 describe_file_type(&meta.file_type())
188 );
189 }
190 std::fs::remove_file(&socket_path).with_context(|| {
191 format!("Failed to remove existing socket file: {:?}", socket_path)
192 })?;
193 }
194 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
195 // Nothing there - normal case.
196 }
197 Err(e) => {
198 return Err(e).with_context(|| format!("Failed to stat {:?}", socket_path));
199 }
200 }
201
202 // Create and bind Unix domain socket server
203 let listener = tokio::net::UnixListener::bind(&socket_path)
204 .with_context(|| format!("Failed to bind to socket path: {:?}", socket_path))?;
205
206 // Owner-only (0600). `bind` creates the node with 0777 & ~umask, which on a default
207 // umask of 022 is `srwxr-xr-x`: on both Linux and macOS the kernel checks write
208 // permission on the socket node at connect(2), so every local user could speak to a
209 // server the model was told to run for one process. The model or an MCP caller chooses
210 // `socket_path`, and a predictable path under a world-writable directory is the classic
211 // local-escalation shape, so the default has to be closed. There is a brief window
212 // between bind and chmod; closing it entirely means a private directory, which is the
213 // caller's choice of path, not ours.
214 std::fs::set_permissions(
215 &socket_path,
216 <std::fs::Permissions as std::os::unix::fs::PermissionsExt>::from_mode(0o600),
217 )
218 .with_context(|| {
219 format!(
220 "Failed to restrict permissions on socket file {:?} to owner-only",
221 socket_path
222 )
223 })?;
224
225 Log::new(Some(&status_tx))
226 .info(format!("Socket file server listening on {:?}", socket_path));
227
228 let connections = Arc::new(Mutex::new(HashMap::new()));
229 let protocol = Arc::new(SocketFileProtocol::new());
230
231 let socket_path_clone = socket_path.clone();
232
233 // Spawn accept loop
234 let task_registrar = app_state.clone();
235 let cleanup = SocketCleanup(socket_path.clone());
236 let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
237 let accept_handle = tokio::spawn(async move {
238 // Moved in so it drops (and unlinks) when the loop ends or the task is aborted.
239 let _cleanup = cleanup;
240 loop {
241 match crate::server::accept_bounded::accept_bounded_unix(
242 &listener,
243 &limiter,
244 CONNECTION_CAP_REFUSAL,
245 "Socket file",
246 Some(&status_tx),
247 )
248 .await
249 {
250 Ok((stream, permit)) => {
251 let connection_id =
252 ConnectionId::new(app_state.get_next_unified_id().await);
253 info!("Accepted socket file connection {}", connection_id);
254
255 // Split stream
256 let (read_half, write_half) = tokio::io::split(stream);
257 let write_half_arc = Arc::new(Mutex::new(write_half));
258
259 // Whether this connection is doing work the peer is waiting on. The
260 // read deadline consults it, so an LLM round-trip or a `manual` rule
261 // parked for a human is never mistaken for an idle peer.
262 let activity =
263 Arc::new(crate::server::accept_bounded::ConnectionActivity::new());
264
265 // Add connection to ServerInstance
266 use crate::state::server::{
267 ConnectionState as ServerConnectionState, ConnectionStatus,
268 ProtocolConnectionInfo,
269 };
270 let now = crate::utils::clock::Instant::now();
271 // Use a dummy SocketAddr since Unix sockets don't have IP addresses
272 let dummy_addr = "127.0.0.1:0".parse().unwrap();
273 let conn_state = ServerConnectionState {
274 id: connection_id,
275 remote_addr: dummy_addr,
276 local_addr: dummy_addr,
277 bytes_sent: 0,
278 bytes_received: 0,
279 packets_sent: 0,
280 packets_received: 0,
281 last_activity: now,
282 status: ConnectionStatus::Active,
283 status_changed_at: now,
284 protocol_info: ProtocolConnectionInfo::new(serde_json::json!({
285 "state": "Idle",
286 "socket_path": socket_path_clone.to_string_lossy()
287 })),
288 };
289 app_state
290 .add_connection_to_server(server_id, conn_state)
291 .await;
292 let _ = status_tx.send("__UPDATE_UI__".to_string());
293
294 // Register the connection HERE, before either task is spawned.
295 //
296 // This used to be the first thing the banner task did, racing the
297 // reader task: handle_data_with_actions returns silently when the
298 // connection is not in the map, so a client that wrote immediately
299 // after connecting (`printf ping | nc -U ...`) had its first payload
300 // dropped with no response and no log line. Inserting synchronously
301 // in the accept loop closes the window.
302 connections.lock().await.insert(
303 connection_id,
304 ConnectionData {
305 state: ConnectionState::Idle,
306 queued_data: Vec::new(),
307 write_half: write_half_arc.clone(),
308 },
309 );
310
311 // Send the greeting banner, if this server was asked for one.
312 if send_first {
313 let llm_client_clone = llm_client.clone();
314 let app_state_clone = app_state.clone();
315 let status_tx_clone = status_tx.clone();
316 let connections_clone = connections.clone();
317 let write_half_for_conn = write_half_arc.clone();
318 let protocol_clone = protocol.clone();
319 // The peer is waiting to be greeted, so the first-byte deadline
320 // must not run while the banner is being composed.
321 let banner_busy = activity.busy();
322 // Tracked, not detached: stop_server must abort this task too.
323 let task_owner = app_state.clone();
324 task_owner
325 .spawn_server_task(server_id, async move {
326 let _banner_busy = banner_busy;
327 Self::send_banner(
328 connection_id,
329 server_id,
330 llm_client_clone,
331 app_state_clone,
332 status_tx_clone,
333 connections_clone,
334 write_half_for_conn,
335 protocol_clone,
336 )
337 .await;
338 })
339 .await;
340 }
341
342 // Spawn reader task
343 let llm_client_clone = llm_client.clone();
344 let app_state_clone = app_state.clone();
345 let status_tx_clone = status_tx.clone();
346 let connections_clone = connections.clone();
347 let protocol_clone = protocol.clone();
348 let activity_clone = activity.clone();
349 // Tracked, not detached: stop_server must abort this task too.
350 let task_owner = app_state.clone();
351 task_owner
352 .spawn_server_task(server_id, async move {
353 // Held for the life of the connection, so the cap counts live
354 // connections rather than accepts.
355 let _permit = permit;
356 let mut buffer = vec![0u8; 8192];
357 let mut read_half = read_half;
358 let mut seen_bytes = false;
359
360 loop {
361 let bound = if seen_bytes {
362 idle_timeout
363 } else {
364 first_byte_timeout
365 };
366 // The deadline wraps this read and nothing else.
367 let read = match read_bounded(
368 &mut read_half,
369 &mut buffer,
370 &activity_clone,
371 bound,
372 )
373 .await
374 {
375 Some(result) => result,
376 None => {
377 connections_clone.lock().await.remove(&connection_id);
378 app_state_clone
379 .close_connection_on_server(
380 server_id,
381 connection_id,
382 )
383 .await;
384 Log::new(Some(&status_tx_clone)).info(format!(
385 "Socket file connection {connection_id} sent \
386 nothing for {}s; closing idle connection",
387 bound.as_secs()
388 ));
389 let _ =
390 status_tx_clone.send("__UPDATE_UI__".to_string());
391 break;
392 }
393 };
394 match read {
395 Ok(0) => {
396 // Connection closed
397 connections_clone.lock().await.remove(&connection_id);
398 app_state_clone
399 .close_connection_on_server(
400 server_id,
401 connection_id,
402 )
403 .await;
404 Log::new(Some(&status_tx_clone)).info(format!(
405 "Socket file connection {connection_id} closed"
406 ));
407 let _ =
408 status_tx_clone.send("__UPDATE_UI__".to_string());
409 break;
410 }
411 Ok(n) => {
412 seen_bytes = true;
413 activity_clone.touch();
414 let data = Bytes::copy_from_slice(&buffer[..n]);
415
416 // Feeds the dashboard rail's counters and refreshes
417 // `last_activity`; this protocol never called it, so a
418 // live connection read 0B/0B in the tree forever.
419 app_state_clone
420 .update_connection_stats(
421 server_id,
422 connection_id,
423 Some(n as u64),
424 None,
425 Some(1),
426 None,
427 )
428 .await;
429
430 // Data summary + full payload are FileOnly: the
431 // socket_file_data_received event template renders the
432 // equivalent lines to the TUI, so streaming the payload
433 // here too would duplicate it and load the unbounded
434 // status channel.
435 let log = Log::new(Some(&status_tx_clone));
436 if data.iter().all(|&b| {
437 b.is_ascii_graphic() || b.is_ascii_whitespace()
438 }) {
439 let data_str = String::from_utf8_lossy(&data);
440 let preview =
441 crate::utils::truncate_for_log(&data_str, 100);
442 log.debug(format!(
443 "Socket file received {} bytes on {}: {}",
444 n, connection_id, preview
445 ));
446 log.trace(format!(
447 "Socket file data (text): {:?}",
448 data_str
449 ));
450 } else {
451 log.debug(format!(
452 "Socket file received {} bytes on {} (binary data)",
453 n, connection_id
454 ));
455 log.trace(format!(
456 "Socket file data (hex): {}",
457 hex::encode(&data)
458 ));
459 }
460
461 // Handle data in separate task
462 let llm_clone = llm_client_clone.clone();
463 let state_clone = app_state_clone.clone();
464 let status_clone = status_tx_clone.clone();
465 let conns_clone = connections_clone.clone();
466 let protocol_clone = protocol_clone.clone();
467 // Busy for the whole of the answer — the LLM call,
468 // a script, or a `manual` rule parked for a human —
469 // so the read deadline cannot close the connection
470 // this is an answer for.
471 let busy = activity_clone.busy();
472 // Tracked, not detached: stop_server must abort this task too.
473 let task_owner = app_state_clone.clone();
474 task_owner
475 .spawn_server_task(server_id, async move {
476 let _busy = busy;
477 Self::handle_data_with_actions(
478 connection_id,
479 server_id,
480 data,
481 llm_clone,
482 state_clone,
483 status_clone,
484 conns_clone,
485 protocol_clone,
486 )
487 .await;
488 })
489 .await;
490 }
491 Err(e) => {
492 Log::new(Some(&status_tx_clone)).error(format!(
493 "Read error on socket file connection {}: {}",
494 connection_id, e
495 ));
496 connections_clone.lock().await.remove(&connection_id);
497 break;
498 }
499 }
500 }
501 })
502 .await;
503 }
504 Err(e) => {
505 Log::new(Some(&status_tx))
506 .error(format!("Accept error on socket file: {}", e));
507 break;
508 }
509 }
510 }
511 });
512
513 task_registrar
514 .register_server_task(server_id, accept_handle)
515 .await;
516
517 Ok(socket_path)
518 }
519
520 /// Send the greeting banner for a new connection (`send_first` servers only).
521 ///
522 /// The connection is already registered by the accept loop by the time this runs.
523 #[allow(clippy::too_many_arguments)]
524 async fn send_banner(
525 connection_id: ConnectionId,
526 server_id: crate::state::ServerId,
527 llm_client: OllamaClient,
528 app_state: Arc<AppState>,
529 status_tx: mpsc::UnboundedSender<String>,
530 connections: Arc<Mutex<HashMap<ConnectionId, ConnectionData>>>,
531 write_half: Arc<Mutex<tokio::io::WriteHalf<UnixStream>>>,
532 protocol: Arc<SocketFileProtocol>,
533 ) {
534 {
535 let log = Log::new(Some(&status_tx));
536 // Create connection opened event
537 let event = Event::new(&SOCKET_FILE_CONNECTION_OPENED_EVENT, serde_json::json!({}));
538
539 // Call LLM
540 match call_llm(
541 &llm_client,
542 &app_state,
543 server_id,
544 Some(connection_id),
545 &event,
546 protocol.as_ref(),
547 )
548 .await
549 {
550 Ok(execution_result) => {
551 debug!("LLM socket file banner response received");
552
553 // Display messages
554 for msg in execution_result.messages {
555 let _ = status_tx.send(msg);
556 }
557
558 // Handle protocol results (send banner)
559 let mut wrote_banner = false;
560 for protocol_result in execution_result.protocol_results {
561 match protocol_result {
562 ActionResult::Output(output_data) => {
563 let mut write = write_half.lock().await;
564 if let Err(e) = write.write_all(&output_data).await {
565 log.error(format!("Failed to send socket file banner: {}", e));
566 } else {
567 app_state
568 .update_connection_stats(
569 server_id,
570 connection_id,
571 None,
572 Some(output_data.len() as u64),
573 None,
574 Some(1),
575 )
576 .await;
577 // Sent-data summary + payload are FileOnly: the
578 // send_socket_data action template already reports the
579 // send to the TUI.
580 if output_data
581 .iter()
582 .all(|&b| b.is_ascii_graphic() || b.is_ascii_whitespace())
583 {
584 let data_str = String::from_utf8_lossy(&output_data);
585 let preview =
586 crate::utils::truncate_for_log(&data_str, 100);
587 log.debug(format!(
588 "Socket file sent {} bytes to {}: {}",
589 output_data.len(),
590 connection_id,
591 preview
592 ));
593 log.trace(format!(
594 "Socket file sent (text): {:?}",
595 data_str
596 ));
597 } else {
598 log.debug(format!(
599 "Socket file sent {} bytes to {} (binary data)",
600 output_data.len(),
601 connection_id
602 ));
603 log.trace(format!(
604 "Socket file sent (hex): {}",
605 hex::encode(&output_data)
606 ));
607 }
608 log.debug(format!(
609 "Sent banner to socket file connection {connection_id}"
610 ));
611 wrote_banner = true;
612 }
613 }
614 ActionResult::CloseConnection => {
615 connections.lock().await.remove(&connection_id);
616 log.info(format!(
617 "Closed socket file connection {connection_id} after banner: decision=model_close"
618 ));
619 }
620 _ => {}
621 }
622 }
623
624 // Greeting with nothing is a legitimate answer, not a failure —
625 // keep it distinguishable in the log from the two below.
626 if !wrote_banner {
627 log.debug(format!(
628 "No banner bytes for socket file connection {connection_id}: decision=model_no_actions"
629 ));
630 }
631 }
632 Err(e) => {
633 let failure = crate::utils::WireFailure::classify(&e);
634 let class = if failure.is_overloaded() {
635 "overloaded"
636 } else {
637 "unavailable"
638 };
639 // The full error goes to the log and the status stream, where an
640 // operator looks. Nothing derived from it reaches the peer.
641 log.warn(format!(
642 "Socket file banner failed for {connection_id}: decision=fail_closed_llm_error class={class} error={e}"
643 ));
644
645 // A send_first server owes this peer a greeting and now has none.
646 // A raw byte stream has no error frame, so the only honest signal is
647 // FIN: half-close so the peer's next read returns EOF immediately
648 // instead of blocking until its own timeout. Same shape as `tcp`.
649 {
650 let mut write = write_half.lock().await;
651 let _ = write.shutdown().await;
652 }
653 connections.lock().await.remove(&connection_id);
654 app_state
655 .close_connection_on_server(server_id, connection_id)
656 .await;
657 log.info(format!(
658 "Closed socket file connection {connection_id} after banner LLM error"
659 ));
660 }
661 }
662 }
663 }
664
665 /// Handle data received on a connection with LLM actions
666 async fn handle_data_with_actions(
667 connection_id: ConnectionId,
668 server_id: crate::state::ServerId,
669 data: Bytes,
670 llm_client: OllamaClient,
671 app_state: Arc<AppState>,
672 status_tx: mpsc::UnboundedSender<String>,
673 connections: Arc<Mutex<HashMap<ConnectionId, ConnectionData>>>,
674 protocol: Arc<SocketFileProtocol>,
675 ) {
676 let log = Log::new(Some(&status_tx));
677
678 // Check connection state
679 let current_state = {
680 let conns = connections.lock().await;
681 if let Some(conn_data) = conns.get(&connection_id) {
682 conn_data.state.clone()
683 } else {
684 // Never silent. A miss here means the connection was torn down between the
685 // read and this lookup (peer reset, or close_this_connection on another
686 // task) - legitimate, but indistinguishable from a registration race, which
687 // is exactly what made the bitcoin accept-order bug so hard to find: the
688 // read loop logged "received N bytes" and then nothing at all.
689 debug!(
690 "socket-file connection {} is no longer registered; dropping {} received bytes",
691 connection_id,
692 data.len()
693 );
694 return;
695 }
696 };
697
698 // If processing, queue the data
699 if current_state == ConnectionState::Processing {
700 connections
701 .lock()
702 .await
703 .entry(connection_id)
704 .and_modify(|conn| {
705 conn.queued_data.extend_from_slice(&data);
706 });
707 log.debug(format!(
708 "Queued {} bytes for socket file connection {}",
709 data.len(),
710 connection_id
711 ));
712 return;
713 }
714
715 // Merge any queued data with new data.
716 //
717 // Not `unwrap()`: the map lock was released after the state read above, so the
718 // connection can legitimately be gone by now - the peer reset, or another task ran
719 // `close_this_connection`. `unwrap()` panicked inside a `tokio::spawn`, which swallows
720 // the panic: the bytes vanished, the log said nothing, and the server stayed Running.
721 let mut all_data = {
722 let mut conns = connections.lock().await;
723 let Some(conn_data) = conns.get_mut(&connection_id) else {
724 debug!(
725 "socket-file connection {} went away before its {} received bytes could be \
726 processed",
727 connection_id,
728 data.len()
729 );
730 return;
731 };
732 conn_data.state = ConnectionState::Processing;
733 let mut merged = conn_data.queued_data.clone();
734 merged.extend_from_slice(&data);
735 conn_data.queued_data.clear();
736 Bytes::from(merged)
737 };
738
739 loop {
740 // Get write_half for context
741 let write_half = {
742 let conns = connections.lock().await;
743 conns.get(&connection_id).map(|c| c.write_half.clone())
744 };
745
746 let Some(write_half) = write_half else {
747 debug!(
748 "socket-file connection {} went away before its response could be written",
749 connection_id
750 );
751 return;
752 };
753
754 // Format data for event parameter. Printable ASCII is passed through as text,
755 // anything else is hex-encoded. `encoding` tells the LLM which one it got, so it
756 // can echo the payload back with a matching `encoding` on send_socket_data.
757 let (data_str, data_encoding) = if all_data
758 .iter()
759 .all(|&b| b.is_ascii_graphic() || b.is_ascii_whitespace())
760 {
761 (String::from_utf8_lossy(&all_data).to_string(), "utf8")
762 } else {
763 (hex::encode(&all_data), "hex")
764 };
765
766 // Create data received event
767 let event = Event::new(
768 &SOCKET_FILE_DATA_RECEIVED_EVENT,
769 serde_json::json!({
770 "data": data_str,
771 "encoding": data_encoding
772 }),
773 );
774
775 // Call LLM
776 match call_llm(
777 &llm_client,
778 &app_state,
779 server_id,
780 Some(connection_id),
781 &event,
782 protocol.as_ref(),
783 )
784 .await
785 {
786 Ok(execution_result) => {
787 debug!("LLM socket file response received");
788
789 // Display messages
790 for msg in execution_result.messages {
791 let _ = status_tx.send(msg);
792 }
793
794 // Handle protocol results
795 let mut should_close = false;
796 let mut should_wait = false;
797 let mut wrote_output = false;
798
799 for protocol_result in execution_result.protocol_results {
800 match protocol_result {
801 ActionResult::Output(output_data) => {
802 let mut write = write_half.lock().await;
803 if let Err(e) = write.write_all(&output_data).await {
804 log.error(format!(
805 "Failed to send socket file response: {}",
806 e
807 ));
808 } else {
809 app_state
810 .update_connection_stats(
811 server_id,
812 connection_id,
813 None,
814 Some(output_data.len() as u64),
815 None,
816 Some(1),
817 )
818 .await;
819 // Sent-data summary + payload are FileOnly: the
820 // send_socket_data action template already reports the
821 // send to the TUI.
822 if output_data
823 .iter()
824 .all(|&b| b.is_ascii_graphic() || b.is_ascii_whitespace())
825 {
826 let data_str = String::from_utf8_lossy(&output_data);
827 let preview =
828 crate::utils::truncate_for_log(&data_str, 100);
829 log.debug(format!(
830 "Socket file sent {} bytes to {}: {}",
831 output_data.len(),
832 connection_id,
833 preview
834 ));
835 log.trace(format!(
836 "Socket file sent (text): {:?}",
837 data_str
838 ));
839 } else {
840 log.debug(format!(
841 "Socket file sent {} bytes to {} (binary data)",
842 output_data.len(),
843 connection_id
844 ));
845 log.trace(format!(
846 "Socket file sent (hex): {}",
847 hex::encode(&output_data)
848 ));
849 }
850 log.debug(format!(
851 "Sent {} bytes to socket file connection {}",
852 output_data.len(),
853 connection_id
854 ));
855 wrote_output = true;
856 }
857 }
858 ActionResult::CloseConnection => {
859 should_close = true;
860 }
861 ActionResult::WaitForMore => {
862 should_wait = true;
863 }
864 _ => {}
865 }
866 }
867
868 // Handle wait_for_more
869 if should_wait {
870 connections
871 .lock()
872 .await
873 .entry(connection_id)
874 .and_modify(|conn| conn.state = ConnectionState::Accumulating);
875 log.debug(format!(
876 "Waiting for more data from socket file connection {connection_id}"
877 ));
878 return;
879 }
880
881 // Handle close_connection
882 if should_close {
883 connections.lock().await.remove(&connection_id);
884 // The model answered by hanging up — distinct in the log from
885 // "answered nothing" and from an LLM failure.
886 log.info(format!(
887 "Closed socket file connection {connection_id}: decision=model_close"
888 ));
889 return;
890 }
891
892 // The model answered, but with no bytes and no lifecycle action. On a
893 // raw byte stream that is a legitimate answer ("say nothing, keep
894 // listening"), so the connection stays open — but it must not be
895 // confused with the backend having failed.
896 if !wrote_output {
897 log.debug(format!(
898 "No response bytes for socket file connection {connection_id}: decision=model_no_actions"
899 ));
900 }
901
902 // Check for queued data
903 let has_queued = {
904 let conns = connections.lock().await;
905 conns
906 .get(&connection_id)
907 .map(|c| !c.queued_data.is_empty())
908 .unwrap_or(false)
909 };
910
911 if has_queued {
912 // Take the queue and make it the next iteration's payload.
913 // Leaving it in place re-sent the SAME bytes to the model on
914 // every pass and never emptied the queue: one response per
915 // iteration, forever, for a single line of input.
916 let queued = {
917 let mut conns = connections.lock().await;
918 match conns.get_mut(&connection_id) {
919 Some(conn) => std::mem::take(&mut conn.queued_data),
920 None => return,
921 }
922 };
923 if queued.is_empty() {
924 connections
925 .lock()
926 .await
927 .entry(connection_id)
928 .and_modify(|conn| conn.state = ConnectionState::Idle);
929 return;
930 }
931 all_data = Bytes::from(queued);
932 } else {
933 // Go to Idle state
934 connections
935 .lock()
936 .await
937 .entry(connection_id)
938 .and_modify(|conn| conn.state = ConnectionState::Idle);
939 return;
940 }
941 }
942 Err(e) => {
943 let failure = crate::utils::WireFailure::classify(&e);
944 let class = if failure.is_overloaded() {
945 "overloaded"
946 } else {
947 "unavailable"
948 };
949 // Full error to the log/status stream only — never to the peer.
950 log.warn(format!(
951 "LLM error for socket file data on {connection_id}: decision=fail_closed_llm_error class={class} error={e}"
952 ));
953 if failure.is_overloaded() {
954 log.warn(format!(
955 "Socket file connection {connection_id} closed: LLM capacity exhausted"
956 ));
957 }
958
959 // Say *something* on the wire. A raw byte stream has no error frame,
960 // so the only honest signal is FIN: half-close the connection so the
961 // peer's next read returns EOF immediately. This path used to reset
962 // to Idle and write nothing, leaving the peer blocked until its own
963 // timeout with no indication anything had gone wrong.
964 {
965 let mut write = write_half.lock().await;
966 let _ = write.shutdown().await;
967 }
968 connections.lock().await.remove(&connection_id);
969 app_state
970 .close_connection_on_server(server_id, connection_id)
971 .await;
972 log.info(format!(
973 "Closed socket file connection {connection_id} after LLM error"
974 ));
975 return;
976 }
977 }
978 }
979 }
980}