netget/client/spice/
mod.rs

1//! SPICE client. Links the main channel with the password's RSA-OAEP ticket, learns the session
2//! id from `SPICE_MSG_MAIN_INIT`, asks for the channel list, then links the display and inputs
3//! channels into the same session. The display channel is decoded into a picture of the
4//! primary surface (`display.rs`); the model's actions become scancodes and mouse messages on
5//! the inputs channel; `spice_read_screen` describes the picture once the server stops drawing.
6//!
7//! Each channel has a reader task: it answers PING and SET_ACK itself (a client that does not
8//! ACK is throttled by the server after a window of messages), feeds the display model, and
9//! raises events. One session task owns the writers' use for input and the screen reads.
10pub mod actions;
11pub mod display;
12
13use crate::client::llm_budget::call_llm_for_client;
14use crate::llm::actions::client_trait::{Client, ClientActionResult};
15use crate::logging::emit::Log;
16use crate::protocol::{ConnectContext, Event, EventType};
17use crate::server::spice::wire::{self, channel, common_cap, link_err, msg, msgc, Cursor};
18use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
19use crate::state::{AccessLogOwner, ClientStatus};
20use actions::Input;
21pub use actions::SpiceClientProtocol;
22use anyhow::{bail, ensure, Context, Result};
23use serde_json::{json, Value};
24use std::net::SocketAddr;
25use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
26use std::sync::Arc;
27use std::time::Duration;
28use tokio::io::{AsyncReadExt, AsyncWriteExt};
29use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
30use tokio::net::TcpStream;
31use tokio::sync::{mpsc, Mutex};
32use tokio::time::Instant;
33
34/// Linking every channel, TCP connect to the last link result.
35pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
36/// How many model turns may follow from one another before the chain stops.
37pub const MAX_FOLLOWUP_DEPTH: u32 = 8;
38/// The largest message body accepted from the server: a full 4096x2160 32-bit bitmap and its
39/// headers. The size is a server-chosen `u32`, refused before the body is allocated.
40pub const MAX_SERVER_MESSAGE_BYTES: u32 = 36 << 20;
41/// Messages the connect path reads while waiting for MAIN_INIT and then CHANNELS_LIST.
42const MAX_SETUP_MESSAGES: usize = 256;
43const TURNS: usize = 64;
44
45/// One channel's write side.
46pub struct Channel {
47    writer: Mutex<OwnedWriteHalf>,
48    mini: bool,
49    serial: AtomicU64,
50}
51
52impl Channel {
53    async fn send(&self, msg_type: u16, body: &[u8]) -> std::io::Result<()> {
54        let mut w = self.writer.lock().await;
55        let serial = self.serial.fetch_add(1, Ordering::SeqCst) + 1;
56        let mut buf = wire::data_header(self.mini, serial, msg_type, body.len() as u32);
57        buf.extend_from_slice(body);
58        w.write_all(&buf).await?;
59        w.flush().await
60    }
61
62    async fn shutdown(&self) {
63        let _ = self.writer.lock().await.shutdown().await;
64    }
65}
66
67/// Link one channel: link message, reply, auth mechanism, ticket, result. A refused link is an
68/// error naming the server's `SPICE_LINK_ERR_*`.
69pub async fn link(
70    server: SocketAddr,
71    connection_id: u32,
72    channel_type: u8,
73    password: &str,
74    channel_caps: Vec<u32>,
75) -> Result<(OwnedReadHalf, Arc<Channel>, wire::LinkReply)> {
76    let stream = TcpStream::connect(server)
77        .await
78        .with_context(|| format!("cannot reach {server}"))?;
79    let _ = stream.set_nodelay(true);
80    let (mut r, mut w) = stream.into_split();
81    let mess = wire::LinkMess {
82        connection_id,
83        channel_type,
84        channel_id: 0,
85        common_caps: vec![wire::caps_word(&[
86            common_cap::AUTH_SELECTION,
87            common_cap::AUTH_SPICE,
88            common_cap::MINI_HEADER,
89        ])],
90        channel_caps,
91    };
92    w.write_all(&mess.encode()).await?;
93    let mut header = [0u8; wire::LINK_HEADER_BYTES];
94    r.read_exact(&mut header)
95        .await
96        .context("reading the link reply")?;
97    let size = wire::parse_link_header(&header, 4).map_err(|e| {
98        anyhow::anyhow!(
99            "the {} channel's link reply is not SPICE: {}",
100            channel::name(channel_type),
101            link_err::name(e.0)
102        )
103    })?;
104    let mut body = vec![0u8; size as usize];
105    r.read_exact(&mut body).await?;
106    let reply = wire::LinkReply::decode(&body)?;
107    if reply.error != link_err::OK {
108        bail!(
109            "the server refused the {} channel link: {}",
110            channel::name(channel_type),
111            link_err::name(reply.error)
112        );
113    }
114    if wire::has_cap(&reply.common_caps, common_cap::AUTH_SELECTION) {
115        w.write_all(&wire::AUTH_MECHANISM_SPICE.to_le_bytes())
116            .await?;
117    }
118    let ticket = wire::encrypt_ticket(&reply.pubkey, password)?;
119    w.write_all(&ticket).await?;
120    let result = r.read_u32_le().await.context("reading the link result")?;
121    if result != link_err::OK {
122        bail!(
123            "the server refused the {} channel: {}",
124            channel::name(channel_type),
125            link_err::name(result)
126        );
127    }
128    let mini = wire::has_cap(&reply.common_caps, common_cap::MINI_HEADER);
129    Ok((
130        r,
131        Arc::new(Channel {
132            writer: Mutex::new(w),
133            mini,
134            serial: AtomicU64::new(0),
135        }),
136        reply,
137    ))
138}
139
140/// What the connect path learned.
141struct Setup {
142    main: (OwnedReadHalf, Arc<Channel>),
143    display: Option<(OwnedReadHalf, Arc<Channel>)>,
144    inputs: Option<(OwnedReadHalf, Arc<Channel>)>,
145    session_id: u32,
146    mouse_mode: u32,
147    agent_connected: bool,
148    name: Option<String>,
149    channels: Vec<&'static str>,
150    /// Notifications that arrived while the channels were being linked.
151    notices: Vec<(&'static str, String)>,
152}
153
154/// A `SpiceMsgNotify` as (severity name, text with control characters removed).
155fn parse_notify(body: &[u8]) -> Result<(&'static str, String)> {
156    let mut c = Cursor::new(body);
157    let _stamp = c.u64("time stamp")?;
158    let severity = c.u32("severity")?;
159    let _visibility = c.u32("visibility")?;
160    let _what = c.u32("what")?;
161    let n = c.u32("message length")? as usize;
162    let text = c.bytes(n.min(c.remaining()), "message")?;
163    let text = crate::utils::sanitize::line_field(&String::from_utf8_lossy(text));
164    Ok((wire::severity_name(severity), text))
165}
166
167fn guest_name(body: &[u8]) -> Result<String> {
168    let mut c = Cursor::new(body);
169    let n = c.u32("name length")? as usize;
170    let bytes = c.bytes(n.min(c.remaining()), "name")?;
171    let end = bytes.iter().position(|b| *b == 0).unwrap_or(bytes.len());
172    Ok(crate::utils::sanitize::line_field(
173        &String::from_utf8_lossy(&bytes[..end]),
174    ))
175}
176
177async fn establish(server: SocketAddr, password: &str) -> Result<Setup> {
178    let (mut main_r, main, _) = link(
179        server,
180        0,
181        channel::MAIN,
182        password,
183        vec![wire::caps_word(&[wire::main_cap::NAME_AND_UUID])],
184    )
185    .await?;
186    let mut name = None;
187    let mut init = None;
188    let mut notices = Vec::new();
189    for _ in 0..MAX_SETUP_MESSAGES {
190        let (t, body) =
191            wire::read_message(&mut main_r, main.mini, MAX_SERVER_MESSAGE_BYTES).await?;
192        match t {
193            msg::PING => main.send(msgc::PONG, &wire::pong_for(&body)?).await?,
194            msg::MAIN_NAME => name = Some(guest_name(&body)?),
195            msg::NOTIFY => notices.push(parse_notify(&body)?),
196            msg::MAIN_INIT => {
197                let mut c = Cursor::new(&body);
198                let session_id = c.u32("session id")?;
199                let _hint = c.u32("display channels hint")?;
200                let supported = c.u32("supported mouse modes")?;
201                let current = c.u32("current mouse mode")?;
202                let agent = c.u32("agent connected")?;
203                init = Some((session_id, supported, current, agent != 0));
204                break;
205            }
206            _ => {}
207        }
208    }
209    let (session_id, supported, mut mouse_mode, agent_connected) =
210        init.context("the server sent no SPICE_MSG_MAIN_INIT")?;
211    main.send(msgc::MAIN_ATTACH_CHANNELS, &[]).await?;
212    if supported & wire::MOUSE_MODE_CLIENT != 0 && mouse_mode != wire::MOUSE_MODE_CLIENT {
213        main.send(
214            msgc::MAIN_MOUSE_MODE_REQUEST,
215            &(wire::MOUSE_MODE_CLIENT as u16).to_le_bytes(),
216        )
217        .await?;
218    }
219    let mut list = None;
220    for _ in 0..MAX_SETUP_MESSAGES {
221        let (t, body) =
222            wire::read_message(&mut main_r, main.mini, MAX_SERVER_MESSAGE_BYTES).await?;
223        match t {
224            msg::PING => main.send(msgc::PONG, &wire::pong_for(&body)?).await?,
225            msg::MAIN_NAME => name = Some(guest_name(&body)?),
226            msg::NOTIFY => notices.push(parse_notify(&body)?),
227            msg::MAIN_MOUSE_MODE => {
228                let mut c = Cursor::new(&body);
229                let _supported = c.u16("supported")?;
230                mouse_mode = u32::from(c.u16("current")?);
231            }
232            msg::MAIN_CHANNELS_LIST => {
233                let mut c = Cursor::new(&body);
234                let n = c.u32("channel count")?;
235                ensure!(
236                    (n as usize) * 2 <= c.remaining(),
237                    "{n} channels do not fit in the list"
238                );
239                let mut v = Vec::new();
240                for _ in 0..n {
241                    v.push((c.u8("channel type")?, c.u8("channel id")?));
242                }
243                list = Some(v);
244                break;
245            }
246            _ => {}
247        }
248    }
249    let list = list.context("the server sent no SPICE_MSG_MAIN_CHANNELS_LIST")?;
250    let channels: Vec<&'static str> = list.iter().map(|(t, _)| channel::name(*t)).collect();
251    let display = if list.contains(&(channel::DISPLAY, 0)) {
252        let (r, ch, reply) = link(server, session_id, channel::DISPLAY, password, vec![]).await?;
253        let cache_pixels = (display::MAX_CACHE_BYTES / 4) as i64;
254        ch.send(
255            msgc::DISPLAY_INIT,
256            &wire::display_init(1, cache_pixels, 1, 3 << 20),
257        )
258        .await?;
259        if wire::has_cap(&reply.channel_caps, wire::display_cap::PREF_COMPRESSION) {
260            // Ask for uncompressed images: they are the ones this client decodes exactly.
261            ch.send(
262                msgc::DISPLAY_PREFERRED_COMPRESSION,
263                &[wire::IMAGE_COMPRESSION_OFF],
264            )
265            .await?;
266        }
267        Some((r, ch))
268    } else {
269        None
270    };
271    let inputs = if list.contains(&(channel::INPUTS, 0)) {
272        let (r, ch, _) = link(server, session_id, channel::INPUTS, password, vec![]).await?;
273        Some((r, ch))
274    } else {
275        None
276    };
277    Ok(Setup {
278        main: (main_r, main),
279        display,
280        inputs,
281        session_id,
282        mouse_mode,
283        agent_connected,
284        name,
285        channels,
286        notices,
287    })
288}
289
290/// State the reader tasks and the session task share.
291struct Shared {
292    display: std::sync::Mutex<display::DisplayModel>,
293    mouse_mode: AtomicU32,
294}
295
296fn resolve_target(remote: &str) -> Result<String> {
297    let raw = remote
298        .trim()
299        .trim_start_matches("spice://")
300        .trim_end_matches('/');
301    let has_port = raw.parse::<SocketAddr>().is_ok()
302        || raw
303            .rsplit_once(':')
304            .is_some_and(|(h, p)| !h.ends_with(':') && p.parse::<u16>().is_ok());
305    ensure!(
306        has_port,
307        "SPICE has no default port; name one, as in 127.0.0.1:5930"
308    );
309    Ok(raw.to_string())
310}
311
312pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
313    let password = ctx
314        .startup_params
315        .as_ref()
316        .map(|p| p.get_optional_string("password"))
317        .transpose()?
318        .flatten()
319        .unwrap_or_default();
320    ensure!(
321        password.len() <= wire::MAX_PASSWORD_LENGTH,
322        "a SPICE password is at most {} bytes",
323        wire::MAX_PASSWORD_LENGTH
324    );
325    let target = resolve_target(&ctx.remote_addr)?;
326    let server = tokio::net::lookup_host(&target)
327        .await
328        .with_context(|| format!("cannot resolve {target}"))?
329        .next()
330        .with_context(|| format!("{target} resolved to no address"))?;
331    let setup = tokio::time::timeout(CONNECT_TIMEOUT, establish(server, &password))
332        .await
333        .context("linking the SPICE channels timed out")??;
334    ctx.state
335        .update_client_status(ctx.client_id, ClientStatus::Connected)
336        .await;
337    Log::new(Some(&ctx.status_tx)).info(format!(
338        "SPICE client {} connected to {server}: session {:#010x}, channels {:?}",
339        ctx.client_id, setup.session_id, setup.channels
340    ));
341    let external =
342        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
343    let (event_tx, event_rx) = mpsc::channel::<(Event, u32)>(TURNS);
344    let (internal_tx, internal) = mpsc::channel::<(Value, u32)>(64);
345    let (end_tx, end_rx) = mpsc::channel::<String>(8);
346    let shared = Arc::new(Shared {
347        display: std::sync::Mutex::new(display::DisplayModel::default()),
348        mouse_mode: AtomicU32::new(setup.mouse_mode),
349    });
350    let _ = event_tx.try_send((
351        Event::new(
352            &actions::CONNECTED_EVENT,
353            json!({
354                "server": server.to_string(),
355                "session_id": setup.session_id,
356                "channels": setup.channels,
357                "mouse_mode": mode_name(setup.mouse_mode),
358                "name": setup.name,
359                "agent_connected": setup.agent_connected,
360            }),
361        ),
362        0,
363    ));
364    for (severity, message) in &setup.notices {
365        let _ = event_tx.try_send((
366            Event::new(
367                &actions::NOTIFY_EVENT,
368                json!({"severity": severity, "message": message}),
369            ),
370            0,
371        ));
372    }
373    let dispatcher = tokio::spawn(run_turns(ctx.clone(), event_rx, internal_tx));
374    let dispatcher_abort = dispatcher.abort_handle();
375    ctx.state
376        .register_client_task(ctx.client_id, dispatcher)
377        .await;
378
379    let main = setup.main.1.clone();
380    let display_ch = setup.display.as_ref().map(|d| d.1.clone());
381    let inputs_ch = setup.inputs.as_ref().map(|i| i.1.clone());
382    let mut readers = Vec::new();
383    for (kind, half) in [
384        ("main", Some(setup.main)),
385        ("display", setup.display),
386        ("inputs", setup.inputs),
387    ] {
388        let Some((r, ch)) = half else { continue };
389        let task = tokio::spawn(read_channel(
390            kind,
391            r,
392            ch,
393            shared.clone(),
394            event_tx.clone(),
395            end_tx.clone(),
396            ctx.status_tx.clone(),
397        ));
398        readers.push(task.abort_handle());
399        ctx.state.register_client_task(ctx.client_id, task).await;
400    }
401    drop(end_tx);
402    let session_ctx = ctx.clone();
403    let task = tokio::spawn(async move {
404        let mut session = Session {
405            shared,
406            main,
407            display: display_ch,
408            inputs: inputs_ch,
409            pointer: (0, 0),
410            buttons: 0,
411        };
412        let reason = session
413            .run(&session_ctx, external, internal, end_rx, event_tx.clone())
414            .await;
415        for r in &readers {
416            r.abort();
417        }
418        session.close().await;
419        Log::new(Some(&session_ctx.status_tx)).info(format!(
420            "SPICE client {} disconnected: {reason}",
421            session_ctx.client_id
422        ));
423        let _ = event_tx
424            .send((
425                Event::new(&actions::DISCONNECTED_EVENT, json!({"reason": reason})),
426                0,
427            ))
428            .await;
429        drop(event_tx);
430        let _ = tokio::time::timeout(Duration::from_secs(30), async {
431            while !dispatcher_abort.is_finished() {
432                tokio::time::sleep(Duration::from_millis(50)).await;
433            }
434        })
435        .await;
436        dispatcher_abort.abort();
437        session_ctx
438            .state
439            .update_client_status(session_ctx.client_id, ClientStatus::Disconnected)
440            .await;
441        session_ctx
442            .state
443            .remove_client_handle(session_ctx.client_id)
444            .await;
445        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
446    });
447    ctx.state.register_client_task(ctx.client_id, task).await;
448    Ok(server)
449}
450
451fn mode_name(mode: u32) -> &'static str {
452    if mode == wire::MOUSE_MODE_CLIENT {
453        "client"
454    } else {
455        "server"
456    }
457}
458
459/// One channel's reader: flow control and PING for every channel, then what the channel is
460/// for. Ends the session (through `end`) when the channel ends.
461async fn read_channel(
462    kind: &'static str,
463    mut r: OwnedReadHalf,
464    ch: Arc<Channel>,
465    shared: Arc<Shared>,
466    events: mpsc::Sender<(Event, u32)>,
467    end: mpsc::Sender<String>,
468    status_tx: mpsc::UnboundedSender<String>,
469) {
470    let mut window = 0u32;
471    let mut since_ack = 0u32;
472    let emit = |event: &'static EventType, data: Value| {
473        if events.try_send((Event::new(event, data), 0)).is_err() {
474            tracing::warn!("SPICE client dropped an event: the model is {TURNS} behind decision=turn_queue_full");
475        }
476    };
477    let reason = loop {
478        let (t, body) = match wire::read_message(&mut r, ch.mini, MAX_SERVER_MESSAGE_BYTES).await {
479            Ok(m) => m,
480            Err(wire::ReadError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
481                break format!("the server closed the {kind} channel");
482            }
483            Err(e) => break format!("the {kind} channel ended: {e}"),
484        };
485        if window > 0 {
486            since_ack += 1;
487            if since_ack >= window {
488                since_ack = 0;
489                if ch.send(msgc::ACK, &[]).await.is_err() {
490                    break format!("the {kind} channel could not be written");
491                }
492            }
493        }
494        let handled: Result<()> = async {
495            match t {
496                msg::SET_ACK => {
497                    let mut c = Cursor::new(&body);
498                    let generation = c.u32("ack generation")?;
499                    window = c.u32("ack window")?;
500                    since_ack = 0;
501                    ch.send(msgc::ACK_SYNC, &wire::u32_body(generation)).await?;
502                }
503                msg::PING => ch.send(msgc::PONG, &wire::pong_for(&body)?).await?,
504                msg::NOTIFY => {
505                    let (severity, text) = parse_notify(&body)?;
506                    Log::new(Some(&status_tx))
507                        .info(format!("SPICE notification ({severity}): {text}"));
508                    emit(
509                        &actions::NOTIFY_EVENT,
510                        json!({"severity": severity, "message": text}),
511                    );
512                }
513                msg::MAIN_MOUSE_MODE if kind == "main" => {
514                    let mut c = Cursor::new(&body);
515                    let _supported = c.u16("supported")?;
516                    shared
517                        .mouse_mode
518                        .store(u32::from(c.u16("current")?), Ordering::SeqCst);
519                }
520                other if kind == "display" => {
521                    let change = match shared.display.lock() {
522                        Ok(mut d) => d.handle(other, &body)?,
523                        Err(_) => bail!("the display model is poisoned"),
524                    };
525                    if let Some(display::Change::SurfaceCreated {
526                        width,
527                        height,
528                        format,
529                    }) = change
530                    {
531                        emit(
532                            &actions::SURFACE_EVENT,
533                            json!({"width": width, "height": height, "format": format}),
534                        );
535                    }
536                }
537                _ => {}
538            }
539            Ok(())
540        }
541        .await;
542        if let Err(e) = handled {
543            break format!("the {kind} channel ended: {e:#}");
544        }
545    };
546    let _ = end.send(reason).await;
547}
548
549/// A screen read waiting for the server to stop drawing.
550struct PendingRead {
551    grid: u64,
552    points: Vec<(u32, u32)>,
553    settle: Duration,
554    give_up: Instant,
555    command: Option<ClientCommand>,
556    depth: u32,
557}
558
559struct Session {
560    shared: Arc<Shared>,
561    main: Arc<Channel>,
562    display: Option<Arc<Channel>>,
563    inputs: Option<Arc<Channel>>,
564    /// Where this client last put the pointer, for relative motion in server mouse mode.
565    pointer: (u32, u32),
566    buttons: u16,
567}
568
569impl Session {
570    fn last_update(&self) -> Instant {
571        self.shared
572            .display
573            .lock()
574            .map(|d| d.last_update)
575            .unwrap_or_else(|_| Instant::now())
576    }
577
578    fn reading(&self, p: &PendingRead, settled: bool) -> Value {
579        let Ok(d) = self.shared.display.lock() else {
580            return json!({"error": "the display model is poisoned"});
581        };
582        let mut data = d.describe(p.grid as usize);
583        let points: Vec<Value> = p
584            .points
585            .iter()
586            .map(|(x, y)| json!({"x": x, "y": y, "color": display::pixel_hex(&d, *x, *y)}))
587            .collect();
588        data["points"] = json!(points);
589        data["settled"] = json!(settled);
590        data["mouse_mode"] = json!(mode_name(self.shared.mouse_mode.load(Ordering::SeqCst)));
591        data
592    }
593
594    async fn close(&self) {
595        self.main.shutdown().await;
596        if let Some(d) = &self.display {
597            d.shutdown().await;
598        }
599        if let Some(i) = &self.inputs {
600            i.shutdown().await;
601        }
602    }
603
604    async fn run(
605        &mut self,
606        ctx: &ConnectContext,
607        mut external: mpsc::Receiver<ClientCommand>,
608        mut internal: mpsc::Receiver<(Value, u32)>,
609        mut end: mpsc::Receiver<String>,
610        events: mpsc::Sender<(Event, u32)>,
611    ) -> String {
612        let mut pending: Option<PendingRead> = None;
613        let emit = |event: &'static EventType, data: Value, depth: u32| {
614            if events.try_send((Event::new(event, data), depth)).is_err() {
615                tracing::warn!("SPICE client {} dropped an event: the model is {TURNS} behind decision=turn_queue_full", ctx.client_id);
616            }
617        };
618        loop {
619            let wake = pending
620                .as_ref()
621                .map(|p| (self.last_update() + p.settle).min(p.give_up))
622                .unwrap_or_else(|| Instant::now() + Duration::from_secs(3600));
623            let (action, depth, command) = tokio::select! {
624                reason = end.recv() => {
625                    return reason.unwrap_or_else(|| "every channel ended".into());
626                }
627                _ = tokio::time::sleep_until(wake), if pending.is_some() => {
628                    let now = Instant::now();
629                    let quiet = self.last_update() + pending.as_ref().map(|p| p.settle).unwrap_or_default() <= now;
630                    let expired = pending.as_ref().is_some_and(|p| p.give_up <= now);
631                    if !quiet && !expired {
632                        continue;
633                    }
634                    let Some(p) = pending.take() else { continue };
635                    let data = self.reading(&p, quiet);
636                    if let Some(c) = p.command {
637                        crate::client::command_support::reply(c, Ok(ClientSendOutcome::Executed { detail: data.to_string() }));
638                    }
639                    emit(&actions::SCREEN_EVENT, data, p.depth);
640                    continue;
641                }
642                c = external.recv() => match c {
643                    Some(c) => {
644                        ctx.state.record_access_log(AccessLogOwner::Client(ctx.client_id.as_u32()), "SPICE", None,
645                            "injected_action", c.action.clone(), vec![]).await;
646                        (c.action.clone(), 0, Some(c))
647                    }
648                    None => return "the client was stopped".into(),
649                },
650                a = internal.recv() => match a {
651                    Some((a, d)) => (a, d, None),
652                    None => return "the client was stopped".into(),
653                },
654            };
655            let reply = |command: Option<ClientCommand>, outcome: ClientSendOutcome| {
656                if let Some(c) = command {
657                    crate::client::command_support::reply(c, Ok(outcome));
658                }
659            };
660            let kind = action["type"].as_str().unwrap_or_default().to_string();
661            match SpiceClientProtocol.execute_action(action.clone()) {
662                Ok(ClientActionResult::Disconnect) => {
663                    reply(command, ClientSendOutcome::Disconnected);
664                    return "the client disconnected".into();
665                }
666                Ok(_) => {}
667                Err(e) => {
668                    reply(
669                        command,
670                        ClientSendOutcome::Rejected {
671                            error: e.to_string(),
672                        },
673                    );
674                    continue;
675                }
676            }
677            let Ok(input) = actions::check(&action) else {
678                continue;
679            };
680            if let Input::ReadScreen {
681                grid,
682                settle_ms,
683                points,
684            } = input
685            {
686                if let Some(old) = pending.take() {
687                    if let Some(c) = old.command {
688                        crate::client::command_support::reply(
689                            c,
690                            Ok(ClientSendOutcome::Rejected {
691                                error: "replaced by a newer spice_read_screen".into(),
692                            }),
693                        );
694                    }
695                }
696                let settle = Duration::from_millis(settle_ms);
697                pending = Some(PendingRead {
698                    grid,
699                    points,
700                    settle,
701                    give_up: Instant::now()
702                        + settle
703                        + Duration::from_millis(actions::MAX_SETTLE_WAIT_MS),
704                    command,
705                    depth,
706                });
707                continue;
708            }
709            match self.send_input(&input).await {
710                Ok(n) => {
711                    Log::new(Some(&ctx.status_tx)).info(format!(
712                        "SPICE client {}: {kind} ({n} input messages)",
713                        ctx.client_id
714                    ));
715                    reply(
716                        command,
717                        ClientSendOutcome::Executed {
718                            detail: json!({"action": kind, "input_messages": n}).to_string(),
719                        },
720                    );
721                }
722                Err(e) => {
723                    let message = format!("{e:#}");
724                    reply(
725                        command,
726                        ClientSendOutcome::Executed {
727                            detail: json!({"action": kind, "error": message}).to_string(),
728                        },
729                    );
730                    emit(
731                        &actions::ERROR_EVENT,
732                        json!({"action": kind, "message": message}),
733                        depth,
734                    );
735                }
736            }
737        }
738    }
739
740    /// Put the model's input on the inputs channel; returns how many messages it took.
741    async fn send_input(&mut self, input: &Input) -> Result<usize> {
742        let inputs = self
743            .inputs
744            .clone()
745            .context("the server offered no inputs channel")?;
746        let mut sent = 0;
747        match input {
748            Input::Keys(strokes) => {
749                for ((sc, ext), released) in strokes {
750                    let t = if *released {
751                        msgc::INPUTS_KEY_UP
752                    } else {
753                        msgc::INPUTS_KEY_DOWN
754                    };
755                    let code = crate::server::spice::keys::key_code(*sc, *ext, *released);
756                    inputs.send(t, &wire::u32_body(code)).await?;
757                    sent += 1;
758                }
759            }
760            Input::Mouse {
761                x,
762                y,
763                action,
764                button,
765            } => {
766                if let Some((w, h)) = self
767                    .shared
768                    .display
769                    .lock()
770                    .ok()
771                    .and_then(|d| d.primary_size())
772                {
773                    ensure!(
774                        *x < w && *y < h,
775                        "({x}, {y}) is outside the {w}x{h} display"
776                    );
777                }
778                let b = match button.as_str() {
779                    "right" => wire::button::RIGHT,
780                    "middle" => wire::button::MIDDLE,
781                    _ => wire::button::LEFT,
782                };
783                if self.shared.mouse_mode.load(Ordering::SeqCst) == wire::MOUSE_MODE_CLIENT {
784                    inputs
785                        .send(
786                            msgc::INPUTS_MOUSE_POSITION,
787                            &wire::mouse_position(*x, *y, self.buttons, 0),
788                        )
789                        .await?;
790                } else {
791                    let dx = *x as i32 - self.pointer.0 as i32;
792                    let dy = *y as i32 - self.pointer.1 as i32;
793                    inputs
794                        .send(
795                            msgc::INPUTS_MOUSE_MOTION,
796                            &wire::mouse_motion(dx, dy, self.buttons),
797                        )
798                        .await?;
799                }
800                self.pointer = (*x, *y);
801                sent += 1;
802                let mut presses: Vec<(u8, bool)> = Vec::new();
803                match action.as_str() {
804                    "click" => presses.extend([(b, true), (b, false)]),
805                    "double_click" => {
806                        presses.extend([(b, true), (b, false), (b, true), (b, false)])
807                    }
808                    "press" => presses.push((b, true)),
809                    "release" => presses.push((b, false)),
810                    "scroll_up" => {
811                        presses.extend([(wire::button::UP, true), (wire::button::UP, false)])
812                    }
813                    "scroll_down" => {
814                        presses.extend([(wire::button::DOWN, true), (wire::button::DOWN, false)])
815                    }
816                    _ => {}
817                }
818                for (button, down) in presses {
819                    if down {
820                        self.buttons |= wire::button::mask(button);
821                    } else {
822                        self.buttons &= !wire::button::mask(button);
823                    }
824                    let t = if down {
825                        msgc::INPUTS_MOUSE_PRESS
826                    } else {
827                        msgc::INPUTS_MOUSE_RELEASE
828                    };
829                    inputs
830                        .send(t, &wire::mouse_button(button, self.buttons))
831                        .await?;
832                    sent += 1;
833                }
834            }
835            Input::ReadScreen { .. } => {}
836        }
837        Ok(sent)
838    }
839}
840
841async fn run_turns(
842    ctx: ConnectContext,
843    mut events: mpsc::Receiver<(Event, u32)>,
844    internal: mpsc::Sender<(Value, u32)>,
845) {
846    let protocol = SpiceClientProtocol;
847    while let Some((event, depth)) = events.recv().await {
848        if depth >= MAX_FOLLOWUP_DEPTH {
849            tracing::warn!("SPICE client {} not asking the model about {}: {depth} turns deep decision=followup_depth", ctx.client_id, event.id());
850            continue;
851        }
852        let instruction = ctx
853            .state
854            .get_instruction_for_client(ctx.client_id)
855            .await
856            .unwrap_or_default();
857        let memory = ctx
858            .state
859            .get_memory_for_client(ctx.client_id)
860            .await
861            .unwrap_or_default();
862        match call_llm_for_client(
863            &ctx.llm_client,
864            &ctx.state,
865            ctx.client_id.to_string(),
866            &instruction,
867            &memory,
868            Some(&event),
869            &protocol,
870            &ctx.status_tx,
871        )
872        .await
873        {
874            Ok(result) => {
875                if let Some(memory) = result.memory_updates {
876                    ctx.state.set_memory_for_client(ctx.client_id, memory).await;
877                }
878                for action in result.actions {
879                    if internal.send((action, depth + 1)).await.is_err() {
880                        return;
881                    }
882                }
883            }
884            Err(e) => Log::new(Some(&ctx.status_tx)).warn(format!("SPICE client handler: {e}")),
885        }
886    }
887}
netget/client/spice/mod.rs

netget/client/spice/
mod.rs
