netget/server/telnet/mod.rs
1//! Telnet server implementation
2pub mod actions;
3
4use crate::server::connection::ConnectionId;
5use anyhow::Result;
6use std::net::SocketAddr;
7use std::sync::Arc;
8use std::time::Duration;
9use tokio::sync::mpsc;
10
11use crate::llm::action_helper::call_llm;
12use crate::llm::actions::protocol_trait::ActionResult;
13use crate::llm::ollama_client::OllamaClient;
14use crate::logging::emit::Log;
15use crate::protocol::Event;
16use crate::server::TelnetProtocol;
17use crate::state::app_state::AppState;
18use actions::{TELNET_CONNECTION_OPENED_EVENT, TELNET_MESSAGE_RECEIVED_EVENT};
19
20/// Truncate `text` to at most `max` bytes without splitting a UTF-8 character.
21///
22/// Slicing `&text[..max]` directly panics when the boundary lands inside a multi-byte
23/// character, and both the received line and the handler's response are attacker- or
24/// model-controlled, so that panic is reachable from the network.
25fn preview(text: &str, max: usize) -> String {
26 if text.len() <= max {
27 return text.to_string();
28 }
29 let mut end = max;
30 while end > 0 && !text.is_char_boundary(end) {
31 end -= 1;
32 }
33 format!("{}...", &text[..end])
34}
35
36/// A telnet line longer than this is not a line anyone typed.
37///
38/// `BufReader::read_line` grows its `String` until a newline arrives, so before this cap
39/// existed an unauthenticated peer could hold the connection open and stream gigabytes with
40/// no `\n` among them, and the server buffered every byte. Nothing authenticates before this
41/// loop, and telnet is the protocol an operator is most likely to point at a real device.
42pub const MAX_LINE_BYTES: usize = 8192;
43
44/// How long to wait for the first line from a peer that has only connected.
45///
46/// **This was 120 seconds, and the argument for it was right about the peer and wrong about
47/// the number.** It read: two minutes rather than the thirty seconds a machine protocol gets,
48/// because the other end of a telnet session is usually a *person* — they connect, read
49/// whatever banner `send_first` produced, and then start typing. That reasoning already knew
50/// this bound waits on a human. It just picked a figure this product contradicts elsewhere.
51///
52/// NetGet's own telnet client is the strongest case of it. `src/client/telnet/mod.rs` opens
53/// the socket, registers its command channel, raises `telnet_connected` and then reads: it
54/// writes nothing of its own, every byte comes from an action, and a client created from the
55/// dashboard's `[ + telnet client ]` is routed `telnet_connected` →
56/// static-with-no-actions and then `*` → manual (`src/tui/modal/form.rs`). So it connects,
57/// is answered with nothing, and waits for a person at `[ send message ]`. That person is
58/// given **300** seconds to answer one event (`src/state/intercepts.rs`); at 120 the server
59/// hung up on them less than halfway through their own window. Telnet was also the only one of
60/// the five protocols raised in the first pass of this sweep that had no startup parameter to
61/// raise it with, so an operator could not even work around it.
62///
63/// 300 seconds is that window, and it is the number this product already uses for how long
64/// someone might take.
65///
66/// What it costs: a stranger holding a socket, a task, an `AppState` row and one of
67/// [`MAX_CONNECTIONS`] slots while saying nothing now gets 300 seconds rather than 120 — two
68/// and a half times longer for one idle slot, not a removal of the bound. The cap still holds
69/// and a peer over it is still answered [`CONNECTION_CAP_REFUSAL`]. A listener genuinely
70/// exposed to strangers should set `first_byte_timeout_secs` low; 120 is the old value and
71/// remains a sound choice for one.
72///
73/// The banner itself sits outside the deadline by construction — it is generated and written
74/// before this loop starts, so however long the model takes, the clock below has not started.
75const FIRST_LINE_READ_TIMEOUT: Duration = Duration::from_secs(300);
76
77/// The two read deadlines for one connection, resolved from the server's startup parameters.
78///
79/// Carried as one value so threading them from `spawn_with_llm_actions` into the connection
80/// task costs one argument rather than two.
81#[derive(Clone, Copy)]
82struct ReadDeadlines {
83 /// [`FIRST_LINE_READ_TIMEOUT`], or this server's `first_byte_timeout_secs`.
84 first_byte: Duration,
85 /// [`IDLE_BETWEEN_LINES_TIMEOUT`], or this server's `idle_timeout_secs`.
86 idle: Duration,
87}
88
89impl ReadDeadlines {
90 fn resolve(first_byte_secs: Option<u64>, idle_secs: Option<u64>) -> Self {
91 Self {
92 first_byte: first_byte_secs
93 .map(Duration::from_secs)
94 .unwrap_or(FIRST_LINE_READ_TIMEOUT),
95 idle: idle_secs
96 .map(Duration::from_secs)
97 .unwrap_or(IDLE_BETWEEN_LINES_TIMEOUT),
98 }
99 }
100}
101
102/// How long to wait for a *further* line once the peer has sent one.
103///
104/// Ten minutes, which is Cisco IOS's `exec-timeout 10 0` default for vty lines — the canonical
105/// idle bound for exactly the kind of device an operator points this server at, and therefore
106/// the number every telnet user already expects. A session here is long-lived by nature: a
107/// person thinks, reads output, and types again, and cutting that off at a machine protocol's
108/// timescale would break the protocol's own use. Everything that can legitimately take minutes
109/// — the LLM round-trip, and a `manual` rule parking a line for a human to answer
110/// (`src/state/intercepts.rs`, 300s by default) — happens after a read has already returned, so
111/// a slow answer can never be timed out from under itself.
112///
113/// Overridable per server with `idle_timeout_secs`, for the same reason the first bound is:
114/// the right value is a property of who is on the other end, and only the operator knows that.
115const IDLE_BETWEEN_LINES_TIMEOUT: Duration = Duration::from_secs(600);
116
117/// Concurrent connections this server admits.
118///
119/// [`crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS`]. Each connection may buffer up to
120/// [`MAX_LINE_BYTES`], so this is the multiplier that turns that per-connection bound into a
121/// total one.
122const MAX_CONNECTIONS: usize = crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS;
123
124/// What a peer over [`MAX_CONNECTIONS`] is told before the socket closes.
125///
126/// Telnet has no error frame — it is a byte stream with a human on the other end — so the
127/// protocol-appropriate refusal is the same plain notice line this file already writes when a
128/// line runs past [`MAX_LINE_BYTES`]: fixed text, on its own line, with nothing interpolated
129/// into it. A terminal prints it; no client can mistake it for a prompt or a login success.
130const CONNECTION_CAP_REFUSAL: &[u8] = b"\r\n[netget] too many connections\r\n";
131
132mod iac {
133 pub const IAC: u8 = 255;
134 pub const SE: u8 = 240;
135 pub const SB: u8 = 250;
136 pub const WILL: u8 = 251;
137 pub const DONT: u8 = 254;
138}
139
140/// Where the IAC state machine is between reads.
141///
142/// The whole machine is this one enum plus a buffer capped at [`MAX_LINE_BYTES`], which is
143/// what makes "can option negotiation be driven into unbounded state?" answerable: it cannot.
144/// A subnegotiation that never sends `IAC SE` discards bytes as they arrive rather than
145/// accumulating them, and no arm can panic.
146#[derive(Clone, Copy, PartialEq, Eq)]
147enum IacState {
148 /// Ordinary data.
149 Data,
150 /// `IAC` seen; the next byte is the command.
151 Command,
152 /// `IAC WILL/WONT/DO/DONT` seen; the next byte is the option, and is consumed.
153 Option,
154 /// Inside `IAC SB …`; everything is consumed until `IAC SE`.
155 Subnegotiation,
156 /// `IAC` inside a subnegotiation: `SE` ends it, `IAC` is an escaped 0xFF.
157 SubnegotiationIac,
158}
159
160/// What one attempt to read a line produced.
161enum LineRead {
162 /// A line with its terminator stripped, and the wire bytes it consumed.
163 Line(Vec<u8>, usize),
164 /// The peer hung up with nothing pending.
165 Eof,
166 /// [`MAX_LINE_BYTES`] of data arrived with no newline in it.
167 TooLong,
168 /// Nothing arrived within the caller's deadline.
169 TimedOut,
170 /// The socket errored.
171 Failed(std::io::Error),
172}
173
174/// Reads `\n`-terminated lines with IAC sequences removed and a hard size cap.
175///
176/// Two defects this replaced, both reachable from any peer:
177///
178/// * **Unbounded buffering.** See [`MAX_LINE_BYTES`].
179/// * **A real `telnet(1)` client was dropped on its first line.** `read_line` validates UTF-8
180/// across the whole line and returns `InvalidData` when that fails. A real client opens by
181/// sending `IAC DO/WILL …` negotiation, whose bytes are not valid UTF-8, so the first line
182/// the user typed arrived with that junk in front of it, failed to decode, and the read loop
183/// treated the error as a reason to close. `actions.rs` told the model those bytes "arrive
184/// as part of the first message", which was never what happened — the connection died
185/// instead. Sequences are now removed from the stream and what is left is decoded lossily,
186/// so the client survives and the model sees what was typed.
187///
188/// Stripping has to happen on the byte stream *before* lines are split out, not on a line
189/// already cut at a newline: an option byte can itself be `0x0A` (`IAC DO NAOCRD` is
190/// `FF FD 0A`), and a sequence may straddle two reads.
191///
192/// Negotiation is still **not answered**: options are recognised only well enough to be
193/// skipped, so a real client gets no reply to its offers and stays in its default line mode,
194/// which is the mode this server can serve.
195struct TelnetLineReader<R> {
196 reader: R,
197 /// Data bytes received and not yet returned as a line.
198 pending: Vec<u8>,
199 chunk: Vec<u8>,
200 state: IacState,
201}
202
203impl<R: tokio::io::AsyncRead + Unpin> TelnetLineReader<R> {
204 fn new(reader: R) -> Self {
205 Self {
206 reader,
207 pending: Vec::new(),
208 chunk: vec![0u8; 4096],
209 state: IacState::Data,
210 }
211 }
212
213 /// Feed raw bytes through the IAC machine, appending only data bytes to `pending`.
214 fn absorb(&mut self, raw: &[u8]) {
215 for &byte in raw {
216 self.state = match self.state {
217 IacState::Data => {
218 if byte == iac::IAC {
219 IacState::Command
220 } else {
221 self.pending.push(byte);
222 IacState::Data
223 }
224 }
225 IacState::Command => match byte {
226 // `IAC IAC` is a literal 0xFF in the data stream.
227 iac::IAC => {
228 self.pending.push(iac::IAC);
229 IacState::Data
230 }
231 iac::SB => IacState::Subnegotiation,
232 iac::WILL..=iac::DONT => IacState::Option,
233 // Every other command (NOP, DM, BRK, AYT, GA …) is two bytes in total.
234 _ => IacState::Data,
235 },
236 IacState::Option => IacState::Data,
237 IacState::Subnegotiation => {
238 if byte == iac::IAC {
239 IacState::SubnegotiationIac
240 } else {
241 IacState::Subnegotiation
242 }
243 }
244 IacState::SubnegotiationIac => match byte {
245 iac::SE => IacState::Data,
246 // Anything else, including an escaped 0xFF, is still payload we discard.
247 _ => IacState::Subnegotiation,
248 },
249 };
250 }
251 }
252
253 /// The `bound` limits the wait for *more bytes*, not the whole line: a peer that is still
254 /// typing keeps the connection alive, which is what a person on a slow link needs.
255 async fn next_line(&mut self, bound: Duration) -> LineRead {
256 use tokio::io::AsyncReadExt;
257 loop {
258 if let Some(idx) = self.pending.iter().position(|b| *b == b'\n') {
259 let mut line: Vec<u8> = self.pending.drain(..=idx).collect();
260 let consumed = line.len();
261 line.pop();
262 if line.last() == Some(&b'\r') {
263 line.pop();
264 }
265 return LineRead::Line(line, consumed);
266 }
267 if self.pending.len() > MAX_LINE_BYTES {
268 return LineRead::TooLong;
269 }
270
271 let read = match tokio::time::timeout(bound, self.reader.read(&mut self.chunk)).await {
272 Err(_) => return LineRead::TimedOut,
273 Ok(read) => read,
274 };
275 let n = match read {
276 Ok(0) => {
277 if self.pending.is_empty() {
278 return LineRead::Eof;
279 }
280 // A final line with no terminator: answer it rather than discarding what
281 // the peer said on its way out.
282 let mut line = std::mem::take(&mut self.pending);
283 let consumed = line.len();
284 if line.last() == Some(&b'\r') {
285 line.pop();
286 }
287 return LineRead::Line(line, consumed);
288 }
289 Ok(n) => n,
290 Err(e) => return LineRead::Failed(e),
291 };
292 let raw = self.chunk[..n].to_vec();
293 self.absorb(&raw);
294 }
295 }
296}
297
298/// Telnet server that forwards messages to LLM
299pub struct TelnetServer;
300
301impl TelnetServer {
302 /// Spawn Telnet server with integrated LLM actions
303 #[allow(clippy::too_many_arguments)]
304 pub async fn spawn_with_llm_actions(
305 listen_addr: SocketAddr,
306 llm_client: OllamaClient,
307 app_state: Arc<AppState>,
308 status_tx: mpsc::UnboundedSender<String>,
309 server_id: crate::state::ServerId,
310 send_first: bool,
311 first_byte_timeout_secs: Option<u64>,
312 idle_timeout_secs: Option<u64>,
313 ) -> Result<SocketAddr> {
314 // Both bounds are tunable because their right value is a property of who is on the
315 // other end, which only the operator knows. The defaults serve a person at a keyboard
316 // — including NetGet's own telnet client parked at the dashboard waiting for one; a
317 // listener exposed to strangers wants the first one much lower.
318 let deadlines = ReadDeadlines::resolve(first_byte_timeout_secs, idle_timeout_secs);
319 let listener =
320 crate::server::socket_helpers::create_reusable_tcp_listener(listen_addr).await?;
321 let local_addr = listener.local_addr()?;
322 Log::new(Some(&status_tx)).info(format!("Telnet server listening on {}", local_addr));
323
324 let protocol = Arc::new(TelnetProtocol::new());
325
326 let task_registrar = app_state.clone();
327 let limiter = crate::server::accept_bounded::ConnectionLimiter::new(MAX_CONNECTIONS);
328 let accept_handle = tokio::spawn(async move {
329 loop {
330 match crate::server::accept_bounded::accept_bounded(
331 &listener,
332 &limiter,
333 CONNECTION_CAP_REFUSAL,
334 "Telnet",
335 Some(&status_tx),
336 )
337 .await
338 {
339 Ok((stream, remote_addr, permit)) => {
340 let connection_id =
341 ConnectionId::new(app_state.get_next_unified_id().await);
342 let local_addr_conn = stream.local_addr().unwrap_or(local_addr);
343 let llm_clone = llm_client.clone();
344 let state_clone = app_state.clone();
345 let status_clone = status_tx.clone();
346 let protocol_clone = protocol.clone();
347
348 // Tracked, not detached: stop_server must abort this task too.
349 let task_owner = app_state.clone();
350 task_owner.spawn_server_task(server_id, async move {
351 // Held for the life of the connection: releasing it here would cap
352 // the accept rate rather than the number of live connections.
353 let _permit = permit;
354 let (read_half, write_half) = tokio::io::split(stream);
355 let write_half_arc = Arc::new(tokio::sync::Mutex::new(write_half));
356
357 // Add connection to ServerInstance
358 use crate::state::server::{
359 ConnectionState as ServerConnectionState, ConnectionStatus,
360 ProtocolConnectionInfo,
361 };
362 let now = crate::utils::clock::Instant::now();
363 let conn_state = ServerConnectionState {
364 id: connection_id,
365 remote_addr,
366 local_addr: local_addr_conn,
367 bytes_sent: 0,
368 bytes_received: 0,
369 packets_sent: 0,
370 packets_received: 0,
371 last_activity: now,
372 status: ConnectionStatus::Active,
373 status_changed_at: now,
374 protocol_info: ProtocolConnectionInfo::empty(),
375 };
376 state_clone
377 .add_connection_to_server(server_id, conn_state)
378 .await;
379 let _ = status_clone.send("__UPDATE_UI__".to_string());
380 let log = Log::new(Some(&status_clone));
381
382 // Peer messaging: the dashboard's "message this peer" and
383 // "disconnect" inject actions into THIS connection through
384 // the same executor the handlers use. The handle is removed
385 // on every exit path below; the command task ends with it.
386 let peer_rx = crate::server::peer_support::register_peer_channel(
387 &state_clone,
388 server_id,
389 connection_id.as_u32(),
390 )
391 .await;
392 crate::server::peer_support::spawn_peer_command_task(
393 peer_rx,
394 protocol_clone.clone(),
395 state_clone.clone(),
396 server_id,
397 connection_id.as_u32(),
398 write_half_arc.clone(),
399 status_clone.clone(),
400 );
401
402 // Raise telnet_connection_opened for EVERY connection, before the
403 // first line is read, so the handler can greet before the client
404 // says anything. It used to be raised only with send_first, which a
405 // server started from a one-line instruction cannot set - so
406 // "greet everyone who connects" was unservable, and the real-model
407 // eval scored every banner case 0. With nothing to say, the model
408 // answers with no actions; a dashboard-created instance routes the
409 // event to a zero-action static rule and pays nothing.
410 //
411 // The line reader starts only after this returns, so the first read
412 // is always held until the connect event is answered. send_first
413 // now says the peer is OWED a greeting: silence is then a warning,
414 // and a backend failure prints the failure notice.
415 {
416 let event = Event::new(
417 &TELNET_CONNECTION_OPENED_EVENT,
418 crate::protocol::event_type::connect_event_data(),
419 );
420 match call_llm(
421 &llm_clone,
422 &state_clone,
423 server_id,
424 Some(connection_id),
425 &event,
426 protocol_clone.as_ref(),
427 )
428 .await
429 {
430 Ok(execution_result) => {
431 let mut greeted = 0usize;
432 for protocol_result in execution_result.protocol_results {
433 if let ActionResult::Output(data) = protocol_result {
434 greeted += 1;
435 use tokio::io::AsyncWriteExt;
436 let mut write = write_half_arc.lock().await;
437 let _ = write.write_all(&data).await;
438 let _ = write.flush().await;
439 drop(write);
440 state_clone
441 .update_connection_stats(
442 server_id,
443 connection_id,
444 None,
445 Some(data.len() as u64),
446 None,
447 Some(1),
448 )
449 .await;
450 log.debug(format!(
451 "Telnet sent greeting ({} bytes) on connection {}",
452 data.len(),
453 connection_id
454 ));
455 }
456 }
457 if greeted > 0 {
458 log.info(format!(
459 "Telnet greeting for {} (connection {}) \
460 decision=model_answer ({} write(s))",
461 remote_addr, connection_id, greeted
462 ));
463 } else if !send_first {
464 // Nothing to say on connect is the ordinary answer
465 // for a server that waits for the client.
466 log.debug(format!(
467 "Telnet connect for {} (connection {}) \
468 decision=model_no_actions",
469 remote_addr, connection_id
470 ));
471 } else {
472 // `send_first` was asked for and nothing came back,
473 // so the peer stares at a blank screen with no
474 // notice. Distinct from the backend failing below.
475 log.warn(format!(
476 "Telnet greeting for {} (connection {}) \
477 decision=model_silent: send_first was requested \
478 but the model produced no banner",
479 remote_addr, connection_id
480 ));
481 }
482 }
483 Err(e) if !send_first => {
484 // Nobody asked this server to speak first, so the peer
485 // has been failed by nothing yet: its first line gets
486 // the per-line path's answer, or that path's notice if
487 // the backend is still down.
488 log.warn(format!(
489 "Telnet connect event for {} (connection {}) \
490 decision=connect_event_failed: {}",
491 remote_addr, connection_id, e
492 ));
493 }
494 Err(e) => {
495 // The client asked for a banner and is sitting at a
496 // blank screen. Telnet has no error frame - it is a
497 // byte stream with a human on the other end - so the
498 // protocol-appropriate answer is a plain notice line.
499 //
500 // Tagged like the per-line path: the notice is a
501 // category, not the banner that was asked for.
502 let decision = match crate::utils::WireFailure::classify(&e)
503 {
504 crate::utils::WireFailure::Overloaded => {
505 "fail_closed_llm_overloaded"
506 }
507 crate::utils::WireFailure::Unavailable => {
508 "fail_closed_llm_error"
509 }
510 };
511 log.error(format!(
512 "Telnet greeting for {} (connection {}) decision={}: {}",
513 remote_addr, connection_id, decision, e
514 ));
515 let notice = telnet_failure_notice(&e);
516 use tokio::io::AsyncWriteExt;
517 let mut write = write_half_arc.lock().await;
518 let _ = write.write_all(notice.as_bytes()).await;
519 let _ = write.flush().await;
520 }
521 }
522 }
523
524 // Bounded, IAC-stripping line reading. Option negotiation is still
525 // not *answered* — sequences are recognised only well enough to be
526 // removed, so a real telnet client survives instead of being
527 // dropped when its negotiation fails to decode as UTF-8.
528 let mut reader = TelnetLineReader::new(read_half);
529 let mut close_requested = false;
530 let mut answered_one = false;
531
532 loop {
533 // The deadline wraps the read and nothing else: the LLM call
534 // and a `manual` park both happen further down this loop, after
535 // a line has already been read.
536 let read_timeout = if answered_one {
537 deadlines.idle
538 } else {
539 deadlines.first_byte
540 };
541 let (line_bytes, n) = match reader.next_line(read_timeout).await {
542 LineRead::Line(bytes, n) => (bytes, n),
543 LineRead::Eof => break,
544 LineRead::TimedOut => {
545 log.info(format!(
546 "Telnet client {} (connection {}) sent nothing for \
547 {}s; closing idle connection",
548 remote_addr,
549 connection_id,
550 read_timeout.as_secs()
551 ));
552 break;
553 }
554 LineRead::TooLong => {
555 // No error frame exists in telnet, so say it in words
556 // on their own line and hang up. A category, never an
557 // error string: see `telnet_failure_notice`.
558 // The protocol refused this, not the model: no LLM call
559 // was made and none will be. Its own token, so it is
560 // never confused with a backend failure.
561 log.warn(format!(
562 "Telnet line from {} (connection {}) \
563 decision=refused_line_too_long: more than {} bytes \
564 with no newline; closing",
565 remote_addr, connection_id, MAX_LINE_BYTES
566 ));
567 use tokio::io::AsyncWriteExt;
568 let mut write = write_half_arc.lock().await;
569 let _ = write
570 .write_all(b"\r\n[netget] line too long\r\n")
571 .await;
572 let _ = write.flush().await;
573 let _ = write.shutdown().await;
574 break;
575 }
576 LineRead::Failed(e) => {
577 log.debug(format!(
578 "Telnet read error on connection {}: {}",
579 connection_id, e
580 ));
581 break;
582 }
583 };
584 // Decoded lossily: after IAC stripping whatever is left is
585 // what the peer typed, and a stray non-UTF-8 byte must not
586 // cost them the connection.
587 let line = String::from_utf8_lossy(&line_bytes).into_owned();
588 answered_one = true;
589
590 // Counters and last_activity: the rail shows ↓/↑ per peer,
591 // and until this was added every telnet peer read 0/0.
592 state_clone
593 .update_connection_stats(
594 server_id,
595 connection_id,
596 Some(n as u64),
597 None,
598 Some(1),
599 None,
600 )
601 .await;
602
603 // Summary + full payload are FileOnly: the
604 // telnet_message_received event template renders the
605 // equivalent line to the TUI, so streaming it here too
606 // would duplicate it and load the unbounded status channel.
607 let line_preview = preview(&line, 100);
608 log.debug(format!(
609 "Telnet received {} bytes on connection {}: {}",
610 n,
611 connection_id,
612 line_preview.trim()
613 ));
614 log.trace(format!("Telnet data (text): {:?}", line.trim()));
615
616 let event = Event::new(
617 &TELNET_MESSAGE_RECEIVED_EVENT,
618 serde_json::json!({
619 "message": line.trim()
620 }),
621 );
622
623 log.debug(format!(
624 "Telnet calling LLM for connection {}",
625 connection_id
626 ));
627
628 match call_llm(
629 &llm_clone,
630 &state_clone,
631 server_id,
632 Some(connection_id),
633 &event,
634 protocol_clone.as_ref(),
635 )
636 .await
637 {
638 Ok(execution_result) => {
639 for message in &execution_result.messages {
640 log.info(message);
641 }
642
643 log.debug(format!(
644 "Telnet got {} protocol results",
645 execution_result.protocol_results.len()
646 ));
647
648 // Classify *before* `protocol_results` is consumed.
649 //
650 // Three of these outcomes write nothing to the socket
651 // and leave the session open, so from the peer's seat
652 // they are identical: a deliberate `wait_for_more`, a
653 // model that returned no usable action at all, and an
654 // action the executor refused. Only the log can tell
655 // them apart, and only if it carries a token.
656 let waited = execution_result
657 .protocol_results
658 .iter()
659 .any(|r| matches!(r, ActionResult::WaitForMore));
660 let failure_summary = execution_result
661 .failures
662 .iter()
663 .map(|f| format!("{}: {}", f.action, f.error))
664 .collect::<Vec<_>>()
665 .join("; ");
666 let mut wrote = 0usize;
667 let mut closed = false;
668
669 for protocol_result in execution_result.protocol_results {
670 match protocol_result {
671 ActionResult::Output(data) => {
672 wrote += 1;
673 let mut write = write_half_arc.lock().await;
674
675 // Write the action's bytes verbatim. Going
676 // via String::from_utf8_lossy would replace
677 // any non-UTF-8 byte with U+FFFD before it
678 // reached the wire.
679 use tokio::io::AsyncWriteExt;
680 let _ = write.write_all(&data).await;
681 let _ = write.flush().await;
682 drop(write);
683 state_clone
684 .update_connection_stats(
685 server_id,
686 connection_id,
687 None,
688 Some(data.len() as u64),
689 None,
690 Some(1),
691 )
692 .await;
693
694 let response = String::from_utf8_lossy(&data);
695
696 // Summary + full payload FileOnly: the
697 // send_telnet_* action template already
698 // reports the send to the TUI.
699 let response_preview = preview(&response, 100);
700 log.debug(format!(
701 "Telnet sent {} bytes on connection {}: {}",
702 data.len(),
703 connection_id,
704 response_preview.trim()
705 ));
706 log.trace(format!(
707 "Telnet sent (text): {:?}",
708 response.trim()
709 ));
710 }
711 // Only flags the intent: `break` here would
712 // leave the read loop running and the socket
713 // open, so close_connection did nothing.
714 ActionResult::CloseConnection => {
715 close_requested = true;
716 closed = true;
717 }
718 _ => {}
719 }
720 }
721
722 // One decision line per line of input, whatever happened.
723 let subject = format!(
724 "Telnet line from {} (connection {})",
725 remote_addr, connection_id
726 );
727 if wrote > 0 {
728 log.info(format!(
729 "{} decision=model_answer ({} write(s){})",
730 subject,
731 wrote,
732 if closed { ", then close" } else { "" }
733 ));
734 } else if closed {
735 log.info(format!(
736 "{} decision=model_reject (close_connection with \
737 no reply: the model hung up rather than answer)",
738 subject
739 ));
740 } else if !failure_summary.is_empty() {
741 log.error(format!(
742 "{} decision=fail_closed_bad_action: nothing was \
743 written because the model's action(s) could not \
744 be executed ({})",
745 subject, failure_summary
746 ));
747 } else if waited {
748 log.info(format!(
749 "{} decision=model_wait_for_more (deliberately \
750 nothing yet; the session stays open)",
751 subject
752 ));
753 } else {
754 // Not a fail-closed: the peer gets no notice and the
755 // session stays open, so a human sits at a terminal
756 // that looks hung. It is the model's silence rather
757 // than the backend's, which is exactly why the two
758 // need different tokens.
759 log.warn(format!(
760 "{} decision=model_silent: the model was asked and \
761 returned no usable action, so nothing was written \
762 and the peer sees an idle session",
763 subject
764 ));
765 }
766 }
767 Err(e) => {
768 // Same reasoning as the greeting: a Telnet client that
769 // sent a line and gets nothing back cannot tell a
770 // broken server from a slow one. A notice is not a
771 // prompt and not a shell result, so nothing downstream
772 // can read it as the command having run.
773 //
774 // ERROR with a `decision=fail_closed_*` token: the notice
775 // on the wire is a category, not an answer, so this turn
776 // produced nothing the peer asked for. The token is what
777 // separates a dead backend from the model's own silence
778 // and from its `close_connection` — all three leave the
779 // request unanswered.
780 let decision = match crate::utils::WireFailure::classify(&e)
781 {
782 crate::utils::WireFailure::Overloaded => {
783 "fail_closed_llm_overloaded"
784 }
785 crate::utils::WireFailure::Unavailable => {
786 "fail_closed_llm_error"
787 }
788 };
789 log.error(format!(
790 "Telnet line from {} (connection {}) decision={}: {}",
791 remote_addr, connection_id, decision, e
792 ));
793 let notice = telnet_failure_notice(&e);
794 use tokio::io::AsyncWriteExt;
795 let mut write = write_half_arc.lock().await;
796 let _ = write.write_all(notice.as_bytes()).await;
797 let _ = write.flush().await;
798 }
799 }
800
801 if close_requested {
802 log.info(format!(
803 "Telnet connection {} closed by handler",
804 connection_id
805 ));
806 use tokio::io::AsyncWriteExt;
807 let mut write = write_half_arc.lock().await;
808 let _ = write.shutdown().await;
809 break;
810 }
811 }
812
813 // Connection closed - mark as closed
814 state_clone
815 .remove_peer_handle(server_id, connection_id.as_u32())
816 .await;
817 state_clone
818 .close_connection_on_server(server_id, connection_id)
819 .await;
820 let _ = status_clone.send("__UPDATE_UI__".to_string());
821 }).await;
822 }
823 Err(e) => {
824 Log::new(Some(&status_tx))
825 .error(format!("Failed to accept Telnet connection: {}", e));
826 break;
827 }
828 }
829 }
830 });
831
832 // Register the accept loop so stop_server can abort it and release the port.
833 task_registrar
834 .register_server_task(server_id, accept_handle)
835 .await;
836
837 Ok(local_addr)
838 }
839}
840
841/// The line to print when the LLM backend fails.
842///
843/// Telnet is an unstructured byte stream: there is no status code to send and no framing a
844/// client could key off, so the only useful thing to do is tell whoever is on the other end,
845/// in words, that the server could not answer - and to say so on its own line so it cannot be
846/// mistaken for the output of whatever they typed.
847///
848/// The text is a category, never the error: see `crate::utils::wire_failure`. This is the
849/// path that put netget's own retry message on a stranger's terminal.
850///
851/// CRLF because a raw Telnet client is usually in a mode where a bare LF does not return the
852/// carriage, which would leave the message stair-stepping across the terminal.
853fn telnet_failure_notice(err: &anyhow::Error) -> String {
854 format!(
855 "\r\n[netget] {}\r\n",
856 crate::utils::WireFailure::classify(err).text()
857 )
858}