netget/client/irc/
mod.rs

1//! IRC client implementation
2pub mod actions;
3
4pub use actions::IrcClientProtocol;
5
6use anyhow::{Context, Result};
7use std::net::SocketAddr;
8use std::sync::Arc;
9use tokio::io::{AsyncWriteExt, BufReader};
10use tokio::net::TcpStream;
11use tokio::sync::{mpsc, Mutex};
12use tracing::{debug, error, info, trace, warn};
13
14use crate::client::irc::actions::{IRC_CLIENT_CONNECTED_EVENT, IRC_CLIENT_MESSAGE_RECEIVED_EVENT};
15use crate::client::llm_budget::call_llm_for_client;
16use crate::server::irc::wire::{
17    cap_line, read_irc_line, reject_line_breaks, reject_not_a_word, IrcLine, MAX_IRC_READ_LINE,
18};
19
20use crate::llm::ollama_client::OllamaClient;
21use crate::llm::ClientLlmResult;
22use crate::protocol::Event;
23use crate::state::app_state::AppState;
24use crate::state::{ClientId, ClientStatus};
25
26/// Per-client data shared between the read loop and the injected-command loop.
27///
28/// There is deliberately no Idle/Processing/Accumulating state machine here. One used to
29/// exist, with a `queued_messages` vector, and it was worse than nothing twice over: the read
30/// loop awaits each LLM call inline before reading the next line, so `Processing` was
31/// unreachable and the queue was dead code - and the one path that touched it *cleared* the
32/// queue without ever handing it to the model, so had it been reachable it would have silently
33/// discarded every message that arrived during a call. Backpressure comes from TCP instead,
34/// which is the same choice `src/server/irc/` documents.
35struct ClientData {
36    memory: String,
37    nickname: String,
38}
39
40/// IRC client that connects to an IRC server
41pub struct IrcClient;
42
43impl IrcClient {
44    /// Connect to an IRC server with integrated LLM actions
45    pub async fn connect_with_llm_actions(
46        remote_addr: String,
47        llm_client: OllamaClient,
48        app_state: Arc<AppState>,
49        status_tx: mpsc::UnboundedSender<String>,
50        client_id: ClientId,
51        startup_params: Option<crate::protocol::StartupParams>,
52    ) -> Result<SocketAddr> {
53        // Parse startup params
54        let nickname = startup_params
55            .as_ref()
56            .map(|p| p.get_optional_string("nickname"))
57            .transpose()?
58            .flatten()
59            .unwrap_or_else(|| "netget_user".to_string());
60        let username = startup_params
61            .as_ref()
62            .map(|p| p.get_optional_string("username"))
63            .transpose()?
64            .flatten()
65            .unwrap_or_else(|| "netget".to_string());
66        let realname = startup_params
67            .as_ref()
68            .map(|p| p.get_optional_string("realname"))
69            .transpose()?
70            .flatten()
71            .unwrap_or_else(|| "NetGet IRC Client".to_string());
72
73        // These three are interpolated straight into NICK/USER lines below, so they are held
74        // to the same framing rules as anything the model produces: a `nickname` containing
75        // CRLF would forge a command before the session has even registered, and one
76        // containing a space would shift USER's positional parameters. `realname` is the
77        // trailing parameter, where spaces belong.
78        reject_not_a_word("nickname", &nickname)?;
79        reject_not_a_word("username", &username)?;
80        reject_line_breaks("realname", &realname)?;
81
82        // Resolve and connect
83        let stream = TcpStream::connect(&remote_addr)
84            .await
85            .context(format!("Failed to connect to {}", remote_addr))?;
86
87        let local_addr = stream.local_addr()?;
88        let remote_sock_addr = stream.peer_addr()?;
89
90        info!(
91            "IRC client {} connecting to {} (local: {})",
92            client_id, remote_sock_addr, local_addr
93        );
94
95        // Update client state
96        app_state
97            .update_client_status(client_id, ClientStatus::Connected)
98            .await;
99        let _ = status_tx.send(format!(
100            "[CLIENT] IRC client {} connected to {}",
101            client_id, remote_sock_addr
102        ));
103        let _ = status_tx.send("__UPDATE_UI__".to_string());
104
105        // Split stream
106        let (read_half, write_half) = tokio::io::split(stream);
107        let write_half_arc = Arc::new(Mutex::new(write_half));
108
109        // Send IRC registration
110        let mut writer = write_half_arc.lock().await;
111        writer
112            .write_all(format!("NICK {}\r\n", nickname).as_bytes())
113            .await?;
114        writer
115            .write_all(format!("USER {} 0 * :{}\r\n", username, realname).as_bytes())
116            .await?;
117        drop(writer);
118
119        debug!(
120            "IRC client {} sent registration (nick: {})",
121            client_id, nickname
122        );
123
124        // Initialize client data
125        let client_data = Arc::new(Mutex::new(ClientData {
126            memory: String::new(),
127            nickname: nickname.clone(),
128        }));
129
130        // Command channel for injected actions (the dashboard's [ send_privmsg ] /
131        // [ send_notice ] / [ send_raw ] rows). Registered BEFORE the read loop and therefore
132        // before the connected-event LLM call, which a manual `*` rule can park for minutes -
133        // the operator must be able to reach the client while it waits.
134        let command_rx =
135            crate::client::command_support::register_command_channel(&app_state, client_id).await;
136
137        // `read_line` is not cancellation-safe, so commands are drained by their own task
138        // rather than a `select!` arm in the read loop. Both tasks share the write half.
139        let cmd_state = app_state.clone();
140        let cmd_tx = status_tx.clone();
141        let cmd_write = write_half_arc.clone();
142        let cmd_data = client_data.clone();
143        let cmd_task = tokio::spawn(async move {
144            Self::command_loop(
145                command_rx, cmd_write, cmd_data, client_id, cmd_state, cmd_tx,
146            )
147            .await;
148        });
149        app_state.register_client_task(client_id, cmd_task).await;
150
151        // Clone for spawned task
152        let write_half_clone = write_half_arc.clone();
153        let client_data_clone = client_data.clone();
154        let nickname_clone = nickname.clone();
155
156        // Spawn read loop
157        // Registered with AppState so stop_client can abort this task —
158        // dropping a JoinHandle only detaches it in Tokio.
159        let task_registrar = app_state.clone();
160        let task_handle = tokio::spawn(async move {
161            let mut reader = BufReader::new(read_half);
162            let mut registered = false;
163
164            loop {
165                // Bounded. The peer here is a *server*, and an unbounded `read_line` lets one
166                // that never sends a newline grow this buffer until the netget process is out
167                // of memory - the same one-connection OOM the server side had.
168                let line = match read_irc_line(&mut reader, MAX_IRC_READ_LINE).await {
169                    Ok((IrcLine::Line(line), _)) => line,
170                    Ok((IrcLine::Eof, _)) => {
171                        info!("IRC client {} disconnected", client_id);
172                        app_state
173                            .update_client_status(client_id, ClientStatus::Disconnected)
174                            .await;
175                        let _ = status_tx
176                            .send(format!("[CLIENT] IRC client {} disconnected", client_id));
177                        let _ = status_tx.send("__UPDATE_UI__".to_string());
178                        break;
179                    }
180                    Ok((IrcLine::TooLong, n)) => {
181                        let msg = format!(
182                            "IRC server sent {n} bytes with no newline (limit \
183                             {MAX_IRC_READ_LINE})"
184                        );
185                        error!("IRC client {}: {}", client_id, msg);
186                        app_state
187                            .update_client_status(client_id, ClientStatus::Error(msg))
188                            .await;
189                        let _ = status_tx.send("__UPDATE_UI__".to_string());
190                        break;
191                    }
192                    Err(e) => {
193                        error!("IRC client {} read error: {}", client_id, e);
194                        app_state
195                            .update_client_status(client_id, ClientStatus::Error(e.to_string()))
196                            .await;
197                        let _ = status_tx.send("__UPDATE_UI__".to_string());
198                        break;
199                    }
200                };
201
202                {
203                    {
204                        let line = line.trim_end().to_string();
205                        trace!("IRC client {} received: {}", client_id, line);
206
207                        // Handle PING immediately.
208                        //
209                        // `line.replace("PING", "PONG")` was wrong in a way that only shows on
210                        // real servers: `replace` rewrites *every* occurrence, so a token
211                        // containing the letters PING - which ircds do send, they are opaque
212                        // cookies - came back corrupted and the server dropped the link for
213                        // failing its own ping. Only the command word is a command.
214                        if let Some(rest) = line.strip_prefix("PING ") {
215                            let pong = format!("PONG {rest}\r\n");
216                            if write_half_clone
217                                .lock()
218                                .await
219                                .write_all(pong.as_bytes())
220                                .await
221                                .is_ok()
222                            {
223                                trace!("IRC client {} sent PONG", client_id);
224                            }
225                            continue;
226                        }
227
228                        // Check for registration complete (001 welcome message)
229                        if !registered && line.contains(" 001 ") {
230                            registered = true;
231                            info!("IRC client {} registration complete", client_id);
232
233                            // Call LLM with connected event
234                            if let Some(instruction) =
235                                app_state.get_instruction_for_client(client_id).await
236                            {
237                                let protocol = Arc::new(IrcClientProtocol::new());
238                                let event = Event::new(
239                                    &IRC_CLIENT_CONNECTED_EVENT,
240                                    serde_json::json!({
241                                        "remote_addr": remote_sock_addr.to_string(),
242                                        "nickname": nickname_clone,
243                                    }),
244                                );
245
246                                if let Err(e) = Self::handle_llm_call(
247                                    &llm_client,
248                                    &app_state,
249                                    client_id,
250                                    &instruction,
251                                    &client_data_clone,
252                                    Some(&event),
253                                    protocol,
254                                    &write_half_clone,
255                                    &status_tx,
256                                )
257                                .await
258                                {
259                                    error!("IRC client {} LLM error on connect: {}", client_id, e);
260                                }
261                            }
262                            continue;
263                        }
264
265                        // Skip if not yet registered
266                        if !registered {
267                            continue;
268                        }
269
270                        // Parse IRC message
271                        let parsed = Self::parse_irc_message(&line);
272
273                        // Handle message with LLM. Awaited inline, so the next line is not read
274                        // until this one has been answered - the backpressure the removed
275                        // state machine pretended to provide.
276                        if let Some(instruction) =
277                            app_state.get_instruction_for_client(client_id).await
278                        {
279                            let protocol = Arc::new(IrcClientProtocol::new());
280                            let event = Event::new(
281                                &IRC_CLIENT_MESSAGE_RECEIVED_EVENT,
282                                serde_json::json!({
283                                    "source": parsed.source,
284                                    "command": parsed.command,
285                                    "target": parsed.target,
286                                    "message": parsed.message,
287                                    "raw_message": line,
288                                }),
289                            );
290
291                            if let Err(e) = Self::handle_llm_call(
292                                &llm_client,
293                                &app_state,
294                                client_id,
295                                &instruction,
296                                &client_data_clone,
297                                Some(&event),
298                                protocol,
299                                &write_half_clone,
300                                &status_tx,
301                            )
302                            .await
303                            {
304                                error!("IRC client {} LLM error: {}", client_id, e);
305                            }
306                        }
307                    }
308                }
309            }
310
311            // Every exit path (EOF - including the one an injected or LLM QUIT provokes - and
312            // read error) lands here: drop the command handle so the dashboard stops offering
313            // [ send ] on a dead connection (a late send then fails fast).
314            app_state.remove_client_handle(client_id).await;
315        });
316        task_registrar
317            .register_client_task(client_id, task_handle)
318            .await;
319
320        Ok(local_addr)
321    }
322
323    /// Drain injected commands until the channel closes (client removed) or an injected
324    /// `disconnect` ends the session.
325    ///
326    /// The generic `command_support::handle_stream_client_command` cannot run this client's
327    /// vocabulary because every wire verb (`send_privmsg`, `send_notice`, `send_raw`,
328    /// `join_channel`, ...) yields `ClientActionResult::Custom`, so the result goes through
329    /// [`Self::apply_action`] - the same function the LLM path uses - and the outcome is
330    /// recorded and replied exactly the way the generic arm does it.
331    async fn command_loop(
332        mut command_rx: mpsc::Receiver<crate::state::client_handles::ClientCommand>,
333        write_half: Arc<Mutex<tokio::io::WriteHalf<TcpStream>>>,
334        client_data: Arc<Mutex<ClientData>>,
335        client_id: ClientId,
336        app_state: Arc<AppState>,
337        status_tx: mpsc::UnboundedSender<String>,
338    ) {
339        use crate::llm::actions::client_trait::Client;
340        use crate::llm::actions::protocol_trait::Protocol;
341        use crate::state::client_handles::ClientSendOutcome;
342        use crate::state::AccessLogOwner;
343
344        let protocol = IrcClientProtocol::new();
345
346        while let Some(command) = command_rx.recv().await {
347            let action = command.action.clone();
348            let outcome = match protocol.execute_action(action.clone()) {
349                Err(e) => Ok(ClientSendOutcome::Rejected {
350                    error: e.to_string(),
351                }),
352                Ok(result) => Self::apply_action(result, &write_half, &client_data)
353                    .await
354                    .map(|applied| match applied {
355                        Applied::Disconnect => ClientSendOutcome::Disconnected,
356                        Applied::Sent(0) => ClientSendOutcome::Executed {
357                            detail: "executed (nothing to write)".to_string(),
358                        },
359                        Applied::Sent(bytes_sent) => ClientSendOutcome::Sent { bytes_sent },
360                    }),
361            };
362
363            let outcome_json = match &outcome {
364                Ok(outcome) => serde_json::to_value(outcome).unwrap_or(serde_json::Value::Null),
365                Err(e) => serde_json::json!({"error": e.to_string()}),
366            };
367            app_state
368                .record_access_log(
369                    AccessLogOwner::Client(client_id.as_u32()),
370                    protocol.protocol_name(),
371                    None,
372                    "injected_action",
373                    action,
374                    vec![outcome_json],
375                )
376                .await;
377
378            let disconnect = matches!(outcome, Ok(ClientSendOutcome::Disconnected));
379            if let Err(e) = &outcome {
380                error!("IRC client {} injected action failed: {}", client_id, e);
381                let _ = status_tx.send(format!(
382                    "[WARN] Client {} injected action failed: {}",
383                    client_id, e
384                ));
385            }
386            let _ = status_tx.send("__UPDATE_UI__".to_string());
387            crate::client::command_support::reply(command, outcome);
388
389            if disconnect {
390                // QUIT is already on the wire; half-close so the server reads EOF and the
391                // read loop runs its normal disconnect path.
392                let _ = write_half.lock().await.shutdown().await;
393                break;
394            }
395        }
396    }
397
398    /// Parse an IRC message into components
399    fn parse_irc_message(line: &str) -> ParsedMessage {
400        let mut parts = line.split_whitespace();
401
402        let source = if line.starts_with(':') {
403            parts.next().map(|s| s.trim_start_matches(':').to_string())
404        } else {
405            None
406        };
407
408        let command = parts.next().map(|s| s.to_uppercase()).unwrap_or_default();
409
410        let remaining: Vec<&str> = parts.collect();
411        let (target, message) = if command == "PRIVMSG" || command == "NOTICE" {
412            let target = remaining.first().map(|s| s.to_string());
413            let message = if let Some(idx) = remaining.iter().position(|s| s.starts_with(':')) {
414                let msg_parts: Vec<&str> = remaining[idx..].iter().map(|s| *s).collect();
415                Some(msg_parts.join(" ").trim_start_matches(':').to_string())
416            } else {
417                None
418            };
419            (target, message)
420        } else {
421            (None, None)
422        };
423
424        ParsedMessage {
425            source,
426            command,
427            target,
428            message,
429        }
430    }
431
432    /// Handle LLM call and execute actions
433    async fn handle_llm_call(
434        llm_client: &OllamaClient,
435        app_state: &Arc<AppState>,
436        client_id: ClientId,
437        instruction: &str,
438        client_data: &Arc<Mutex<ClientData>>,
439        event: Option<&Event>,
440        protocol: Arc<IrcClientProtocol>,
441        write_half: &Arc<Mutex<tokio::io::WriteHalf<TcpStream>>>,
442        status_tx: &mpsc::UnboundedSender<String>,
443    ) -> Result<()> {
444        let memory = client_data.lock().await.memory.clone();
445
446        match call_llm_for_client(
447            llm_client,
448            app_state,
449            client_id.to_string(),
450            instruction,
451            &memory,
452            event,
453            protocol.as_ref(),
454            status_tx,
455        )
456        .await
457        {
458            Ok(ClientLlmResult {
459                actions,
460                memory_updates,
461            }) => {
462                // Update memory
463                if let Some(mem) = memory_updates {
464                    client_data.lock().await.memory = mem;
465                }
466
467                // Execute actions
468                for action in actions {
469                    use crate::llm::actions::client_trait::Client;
470                    match protocol.as_ref().execute_action(action) {
471                        Ok(result) => {
472                            if let Applied::Disconnect =
473                                Self::apply_action(result, write_half, client_data).await?
474                            {
475                                info!("IRC client {} disconnecting", client_id);
476                                return Err(anyhow::anyhow!("Disconnect requested"));
477                            }
478                        }
479                        Err(e) => {
480                            warn!("IRC client {} action execution error: {}", client_id, e);
481                        }
482                    }
483                }
484            }
485            Err(e) => {
486                return Err(e);
487            }
488        }
489
490        Ok(())
491    }
492
493    /// Put one executed action on the wire. Shared by the LLM path and injected commands so
494    /// the encoding of every IRC verb exists exactly once.
495    async fn apply_action(
496        result: crate::llm::actions::client_trait::ClientActionResult,
497        write_half: &Arc<Mutex<tokio::io::WriteHalf<TcpStream>>>,
498        client_data: &Arc<Mutex<ClientData>>,
499    ) -> Result<Applied> {
500        use crate::llm::actions::client_trait::ClientActionResult;
501
502        match result {
503            ClientActionResult::Custom { name, data } => {
504                Self::execute_irc_action(&name, data, write_half, client_data).await
505            }
506            ClientActionResult::Disconnect => {
507                Self::execute_irc_action(
508                    "disconnect",
509                    serde_json::Value::Null,
510                    write_half,
511                    client_data,
512                )
513                .await
514            }
515            // WaitForMore, NoAction, SendData (unused by this vocabulary), nested Multiple.
516            _ => Ok(Applied::Sent(0)),
517        }
518    }
519
520    /// Execute IRC-specific actions.
521    ///
522    /// The fields arriving here have already been through [`crate::server::irc::wire`]'s
523    /// checks in `IrcClientProtocol::execute_action`, which is the gate both the LLM path and
524    /// the injected-command path pass through. This function's own contribution is the RFC
525    /// 1459 512-byte cap, which is about the finished line rather than any one field.
526    async fn execute_irc_action(
527        name: &str,
528        data: serde_json::Value,
529        write_half: &Arc<Mutex<tokio::io::WriteHalf<TcpStream>>>,
530        client_data: &Arc<Mutex<ClientData>>,
531    ) -> Result<Applied> {
532        // Build and validate the line *before* taking the write lock: a rejected action must
533        // not hold the socket, and must not have written half a line first.
534        let (wire, disconnect) = match name {
535            "join_channel" => {
536                let channel = data["channel"].as_str().context("Missing channel")?;
537                debug!("IRC: JOIN {}", channel);
538                (format!("JOIN {}\r\n", channel), false)
539            }
540            "part_channel" => {
541                let channel = data["channel"].as_str().context("Missing channel")?;
542                debug!("IRC: PART {}", channel);
543                match data["message"].as_str() {
544                    Some(msg) => (format!("PART {} :{}\r\n", channel, msg), false),
545                    None => (format!("PART {}\r\n", channel), false),
546                }
547            }
548            "change_nick" => {
549                let new_nick = data["new_nick"].as_str().context("Missing new_nick")?;
550                client_data.lock().await.nickname = new_nick.to_string();
551                debug!("IRC: NICK {}", new_nick);
552                (format!("NICK {}\r\n", new_nick), false)
553            }
554            "send_privmsg" => {
555                let target = data["target"].as_str().context("Missing target")?;
556                let message = data["message"].as_str().context("Missing message")?;
557                debug!("IRC: PRIVMSG {} :{}", target, message);
558                (format!("PRIVMSG {} :{}\r\n", target, message), false)
559            }
560            "send_notice" => {
561                let target = data["target"].as_str().context("Missing target")?;
562                let message = data["message"].as_str().context("Missing message")?;
563                debug!("IRC: NOTICE {} :{}", target, message);
564                (format!("NOTICE {} :{}\r\n", target, message), false)
565            }
566            "send_raw" => {
567                let command = data["command"].as_str().context("Missing command")?;
568                debug!("IRC: RAW {}", command);
569                (format!("{}\r\n", command), false)
570            }
571            "disconnect" => {
572                let quit_message = data["quit_message"].as_str().unwrap_or("Leaving");
573                debug!("IRC: QUIT");
574                (format!("QUIT :{}\r\n", quit_message), true)
575            }
576            _ => {
577                warn!("Unknown IRC action: {}", name);
578                return Ok(Applied::Sent(0));
579            }
580        };
581
582        // RFC 1459 caps a message at 512 bytes including the CRLF. A server that receives a
583        // longer one truncates or drops it, so sending the full thing loses the tail either
584        // way; cutting it here keeps the line well-formed and puts the loss in our own log.
585        let wire = cap_line(name, wire);
586
587        let mut writer = write_half.lock().await;
588        writer.write_all(&wire).await?;
589        writer.flush().await?;
590
591        Ok(if disconnect {
592            Applied::Disconnect
593        } else {
594            Applied::Sent(wire.len())
595        })
596    }
597}
598
599/// What [`IrcClient::apply_action`] did with one action.
600enum Applied {
601    /// Bytes written (0 when the action produced no wire output).
602    Sent(usize),
603    /// QUIT was written and the session should end.
604    Disconnect,
605}
606
607/// Parsed IRC message
608#[derive(Debug)]
609struct ParsedMessage {
610    source: Option<String>,
611    command: String,
612    target: Option<String>,
613    message: Option<String>,
614}
netget/client/irc/mod.rs

netget/client/irc/
mod.rs
