agentmux_remote/
fsproto.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The file protocol between srv and `agentmux-remote serve --stdio`
5//! (SPEC_REMOTE_TERMINALS_AND_DURABLE_SESSIONS_2026_10_02.md §6.1, §6.3),
6//! carried over one `ssh -T host -- agentmux-remote serve --stdio` per
7//! connection. Shared by both ends, so they can never disagree on it.
8//!
9//! A message is `[len: u32 big-endian][body: len bytes]`. A request body is
10//! `[id: u32][op: u8][fields]`, a reply body `[id: u32][status: u8][fields]`;
11//! the reply carries its request's id. Strings and byte runs are
12//! `[len: u32][bytes]`; numbers are big-endian. Binary, not JSON lines: the
13//! helper stays dependency-free (a few hundred KB on the host), and a file's
14//! bytes need no escaping.
15
16/// The protocol version both ends must share (`Hello`). 2 added `Procs`,
17/// 3 `PaneProcs`.
18pub const PROTOCOL: u32 = 3;
19
20/// Largest message accepted; a longer length means a broken stream.
21pub const MAX_MESSAGE: usize = 40 << 20;
22
23/// Largest `Read` answered at once: a bigger file is read in ranges.
24pub const MAX_READ: u32 = 4 << 20;
25
26/// Largest file `Write` takes in one message.
27pub const MAX_WRITE: usize = 32 << 20;
28
29/// What a request asks for.
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub enum Request {
32    /// The handshake: answered with [`Reply::Hello`].
33    Hello,
34    Stat {
35        path: String,
36    },
37    /// Up to `limit` entries of a directory from `offset`, sorted by name.
38    List {
39        path: String,
40        offset: u32,
41        limit: u32,
42    },
43    /// Up to `len` bytes from `offset` (at most [`MAX_READ`]).
44    Read {
45        path: String,
46        offset: u64,
47        len: u32,
48    },
49    /// Replace the file with `data`, atomically: a temp file beside it,
50    /// synced, then renamed over it, keeping an existing file's mode.
51    Write {
52        path: String,
53        data: Vec<u8>,
54    },
55    /// Add `data` to the end of an existing file (at most [`MAX_WRITE`]): a
56    /// big upload, a piece at a time, into a temp file that is then renamed
57    /// into place. Never creates the file.
58    Append {
59        path: String,
60        data: Vec<u8>,
61    },
62    Mkdir {
63        path: String,
64        parents: bool,
65    },
66    Rename {
67        from: String,
68        to: String,
69    },
70    Delete {
71        path: String,
72        recursive: bool,
73    },
74    /// Rename the file `from` over the file `to` in one step: what was at
75    /// `to` is gone only once `from` is there (never a folder, either side).
76    Replace {
77        from: String,
78        to: String,
79    },
80    /// `path` with every link and `..` resolved: [`Reply::Path`].
81    Realpath {
82        path: String,
83    },
84    /// Give a file its permission bits (`mode`, 0 to leave them) and its
85    /// modification time (`mtime_ms`, 0 to leave it).
86    SetMeta {
87        path: String,
88        mode: u32,
89        mtime_ms: i64,
90    },
91    /// Flush a file's contents to disk (before a moved source goes).
92    Sync {
93        path: String,
94    },
95    /// The host's processes for Tower (`crate::procs`): the `top` busiest by
96    /// CPU and the `top` largest by memory among those whose name or PID
97    /// matches every word of `filter`. [`Reply::Procs`].
98    Procs {
99        top: u32,
100        filter: String,
101    },
102    /// What one AgentMux pane's processes on this host use, for its stats
103    /// badge (`crate::procs::Sampler::pane`): the shell of durable `session`
104    /// and what it started, else the processes whose environment holds
105    /// `AGENTMUX_BLOCKID=<block_id>`. Either may be empty. [`Reply::PaneProcs`].
106    PaneProcs {
107        block_id: String,
108        session: String,
109    },
110}
111
112/// How [`Request::PaneProcs`] found the pane's processes.
113#[derive(Debug, Clone, Copy, PartialEq, Eq)]
114pub enum PaneFound {
115    /// Neither way found any: the frame has the host's totals only.
116    None = 0,
117    /// From the durable session's shell, which the daemon spawned.
118    Session = 1,
119    /// From `AGENTMUX_BLOCKID` in the processes' environments.
120    Marker = 2,
121}
122
123impl PaneFound {
124    fn from_u8(v: u8) -> Self {
125        match v {
126            1 => Self::Session,
127            2 => Self::Marker,
128            _ => Self::None,
129        }
130    }
131}
132
133/// The answer to [`Request::PaneProcs`]. Absent values are ones the host's
134/// OS wouldn't give, never zeros.
135#[derive(Debug, Clone, PartialEq, Eq)]
136pub struct PaneFrame {
137    pub found: PaneFound,
138    pub processes: u32,
139    /// Each executable's name and how many of it run, the pane's shell first.
140    pub names: Vec<(String, u32)>,
141    /// When the pane's shell started, Unix ms.
142    pub started_at_ms: Option<u64>,
143    /// The pane's processes' CPU since the previous request, thousandths of
144    /// one core; absent on a connection's first.
145    pub cpu_milli: Option<u64>,
146    /// Their private memory, bytes (`memory_metric`).
147    pub mem: Option<u64>,
148    /// Their resident memory, bytes, shared pages included.
149    pub mem_resident: Option<u64>,
150    /// Of `processes`, those the host wouldn't measure (another user's, as
151    /// `sudo` starts, on macOS): `cpu_milli`, `mem` and `mem_resident` leave
152    /// them out.
153    pub unmeasured: u32,
154    pub memory_metric: String,
155    /// The whole host, as [`ProcFrame`]'s totals.
156    pub cpu_count: u32,
157    pub host_cpu_milli: Option<u64>,
158    pub host_mem: u64,
159    pub host_processes: u32,
160    /// Of `host_processes`, those listed without CPU or memory (other users'
161    /// on macOS): the host's CPU and memory leave them out.
162    pub host_unmeasured: u32,
163}
164
165/// One process in a [`ProcFrame`]. Absent values are ones the host's OS
166/// wouldn't give without elevation, never zeros.
167#[derive(Debug, Clone, PartialEq, Eq)]
168pub struct ProcRow {
169    pub pid: u32,
170    pub ppid: Option<u32>,
171    /// With `pid`, tells the process apart from a later one with its PID.
172    pub start_key: u64,
173    pub started_at_ms: Option<u64>,
174    /// The executable's name. Never its command line.
175    pub name: String,
176    /// CPU since the previous `Procs`, in thousandths of one core.
177    pub cpu_milli: Option<u32>,
178    /// Private memory, bytes (`ProcFrame::memory_metric`).
179    pub mem: Option<u64>,
180    pub mem_resident: Option<u64>,
181}
182
183/// The answer to [`Request::Procs`].
184#[derive(Debug, Clone, PartialEq, Eq)]
185pub struct ProcFrame {
186    /// Logical CPUs on the host.
187    pub cpu_count: u32,
188    pub memory_metric: String,
189    /// Every process the host listed.
190    pub total: u32,
191    /// Those matching the filter (all of them without one).
192    pub matched: u32,
193    /// Listed without CPU or memory (other users' on macOS).
194    pub unmeasured: u32,
195    /// Every process's CPU, thousandths of one core; absent on a connection's
196    /// first frame, before there is anything to measure against.
197    pub cpu_milli: Option<u64>,
198    /// Every process's private memory, bytes.
199    pub mem: u64,
200    pub rows: Vec<ProcRow>,
201}
202
203/// What kind of thing an entry is (a symlink is reported as what it points
204/// to, with [`Entry::symlink`] set).
205#[derive(Debug, Clone, Copy, PartialEq, Eq)]
206pub enum Kind {
207    File = 1,
208    Dir = 2,
209    Other = 3,
210}
211
212#[derive(Debug, Clone, PartialEq, Eq)]
213pub struct Entry {
214    /// The file name (the last part of the path); lossy if not UTF-8.
215    pub name: String,
216    pub kind: Kind,
217    pub size: u64,
218    /// Last modified, in milliseconds since the Unix epoch (0 if unknown).
219    pub mtime_ms: i64,
220    /// Unix permission bits, `st_mode & 0o7777` (0 where there are none).
221    pub mode: u32,
222    pub symlink: bool,
223    /// Where a symlink points, as stored in the link (empty otherwise).
224    pub link_target: String,
225}
226
227/// Why a request failed, for srv to report the way a local error would be.
228#[derive(Debug, Clone, Copy, PartialEq, Eq)]
229pub enum ErrKind {
230    NotFound = 1,
231    PermissionDenied = 2,
232    AlreadyExists = 3,
233    NotADirectory = 4,
234    IsADirectory = 5,
235    DirectoryNotEmpty = 6,
236    TooLarge = 7,
237    Invalid = 8,
238    Other = 255,
239}
240
241impl ErrKind {
242    pub fn from_io(e: &std::io::Error) -> Self {
243        use std::io::ErrorKind as K;
244        match e.kind() {
245            K::NotFound => ErrKind::NotFound,
246            K::PermissionDenied => ErrKind::PermissionDenied,
247            K::AlreadyExists => ErrKind::AlreadyExists,
248            K::InvalidInput | K::InvalidData => ErrKind::Invalid,
249            // std's own mapping of each platform's codes, never raw numbers:
250            // the same number means different things on different hosts.
251            K::NotADirectory => ErrKind::NotADirectory,
252            K::IsADirectory => ErrKind::IsADirectory,
253            K::DirectoryNotEmpty => ErrKind::DirectoryNotEmpty,
254            _ => ErrKind::Other,
255        }
256    }
257
258    fn from_u8(b: u8) -> Self {
259        match b {
260            1 => ErrKind::NotFound,
261            2 => ErrKind::PermissionDenied,
262            3 => ErrKind::AlreadyExists,
263            4 => ErrKind::NotADirectory,
264            5 => ErrKind::IsADirectory,
265            6 => ErrKind::DirectoryNotEmpty,
266            7 => ErrKind::TooLarge,
267            8 => ErrKind::Invalid,
268            _ => ErrKind::Other,
269        }
270    }
271}
272
273/// The answer to one request.
274#[derive(Debug, Clone, PartialEq, Eq)]
275pub enum Reply {
276    Hello {
277        protocol: u32,
278        /// The user's home directory on the host (`~` in paths).
279        home: String,
280        /// `std::env::consts::OS` there: `linux`, `macos`, `windows`.
281        os: String,
282    },
283    Stat(Entry),
284    List {
285        entries: Vec<Entry>,
286        /// How many entries the directory has in all.
287        total: u32,
288    },
289    Read {
290        data: Vec<u8>,
291        /// The file ends at or before this range's end.
292        eof: bool,
293    },
294    Done,
295    /// A path ([`Request::Realpath`]).
296    Path(String),
297    Procs(ProcFrame),
298    PaneProcs(PaneFrame),
299    Err {
300        kind: ErrKind,
301        message: String,
302    },
303}
304
305mod op {
306    pub const HELLO: u8 = 1;
307    pub const STAT: u8 = 2;
308    pub const LIST: u8 = 3;
309    pub const READ: u8 = 4;
310    pub const WRITE: u8 = 5;
311    pub const MKDIR: u8 = 6;
312    pub const RENAME: u8 = 7;
313    pub const DELETE: u8 = 8;
314    pub const APPEND: u8 = 9;
315    pub const REPLACE: u8 = 10;
316    pub const REALPATH: u8 = 11;
317    pub const SETMETA: u8 = 12;
318    pub const SYNC: u8 = 13;
319    pub const PROCS: u8 = 14;
320    pub const PANE_PROCS: u8 = 15;
321}
322
323mod status {
324    pub const HELLO: u8 = 1;
325    pub const STAT: u8 = 2;
326    pub const LIST: u8 = 3;
327    pub const READ: u8 = 4;
328    pub const DONE: u8 = 5;
329    pub const PATH: u8 = 6;
330    pub const PROCS: u8 = 7;
331    pub const PANE_PROCS: u8 = 8;
332    pub const ERR: u8 = 255;
333}
334
335/// Builds a body.
336struct W(Vec<u8>);
337
338impl W {
339    fn u8(&mut self, v: u8) -> &mut Self {
340        self.0.push(v);
341        self
342    }
343    fn u32(&mut self, v: u32) -> &mut Self {
344        self.0.extend_from_slice(&v.to_be_bytes());
345        self
346    }
347    fn u64(&mut self, v: u64) -> &mut Self {
348        self.0.extend_from_slice(&v.to_be_bytes());
349        self
350    }
351    fn i64(&mut self, v: i64) -> &mut Self {
352        self.0.extend_from_slice(&v.to_be_bytes());
353        self
354    }
355    fn bytes(&mut self, v: &[u8]) -> &mut Self {
356        self.u32(v.len() as u32);
357        self.0.extend_from_slice(v);
358        self
359    }
360    fn str(&mut self, v: &str) -> &mut Self {
361        self.bytes(v.as_bytes())
362    }
363    fn entry(&mut self, e: &Entry) -> &mut Self {
364        self.str(&e.name)
365            .u8(e.kind as u8)
366            .u64(e.size)
367            .i64(e.mtime_ms)
368            .u32(e.mode)
369            .u8(u8::from(e.symlink))
370            .str(&e.link_target)
371    }
372    /// An optional number: the type's maximum stands for "absent".
373    fn opt_u32(&mut self, v: Option<u32>) -> &mut Self {
374        self.u32(v.unwrap_or(u32::MAX))
375    }
376    fn opt_u64(&mut self, v: Option<u64>) -> &mut Self {
377        self.u64(v.unwrap_or(u64::MAX))
378    }
379    fn proc_row(&mut self, p: &ProcRow) -> &mut Self {
380        self.u32(p.pid)
381            .opt_u32(p.ppid)
382            .u64(p.start_key)
383            .opt_u64(p.started_at_ms)
384            .str(&p.name)
385            .opt_u32(p.cpu_milli)
386            .opt_u64(p.mem)
387            .opt_u64(p.mem_resident)
388    }
389    /// The message: length, then body.
390    fn message(self) -> Vec<u8> {
391        let mut out = Vec::with_capacity(4 + self.0.len());
392        out.extend_from_slice(&(self.0.len() as u32).to_be_bytes());
393        out.extend_from_slice(&self.0);
394        out
395    }
396}
397
398/// Reads a body.
399struct R<'a> {
400    p: &'a [u8],
401    i: usize,
402}
403
404impl<'a> R<'a> {
405    fn take(&mut self, n: usize) -> Result<&'a [u8], String> {
406        let end = self.i.checked_add(n).filter(|&e| e <= self.p.len());
407        let end = end.ok_or("message too short")?;
408        let s = &self.p[self.i..end];
409        self.i = end;
410        Ok(s)
411    }
412    fn u8(&mut self) -> Result<u8, String> {
413        Ok(self.take(1)?[0])
414    }
415    fn u32(&mut self) -> Result<u32, String> {
416        Ok(u32::from_be_bytes(self.take(4)?.try_into().unwrap()))
417    }
418    fn u64(&mut self) -> Result<u64, String> {
419        Ok(u64::from_be_bytes(self.take(8)?.try_into().unwrap()))
420    }
421    fn i64(&mut self) -> Result<i64, String> {
422        Ok(i64::from_be_bytes(self.take(8)?.try_into().unwrap()))
423    }
424    fn bytes(&mut self) -> Result<Vec<u8>, String> {
425        let n = self.u32()? as usize;
426        Ok(self.take(n)?.to_vec())
427    }
428    fn str(&mut self) -> Result<String, String> {
429        String::from_utf8(self.bytes()?).map_err(|_| "a string is not UTF-8".to_string())
430    }
431    fn entry(&mut self) -> Result<Entry, String> {
432        Ok(Entry {
433            name: self.str()?,
434            kind: match self.u8()? {
435                1 => Kind::File,
436                2 => Kind::Dir,
437                _ => Kind::Other,
438            },
439            size: self.u64()?,
440            mtime_ms: self.i64()?,
441            mode: self.u32()?,
442            symlink: self.u8()? != 0,
443            link_target: self.str()?,
444        })
445    }
446    fn opt_u32(&mut self) -> Result<Option<u32>, String> {
447        Ok(Some(self.u32()?).filter(|&v| v != u32::MAX))
448    }
449    fn opt_u64(&mut self) -> Result<Option<u64>, String> {
450        Ok(Some(self.u64()?).filter(|&v| v != u64::MAX))
451    }
452    fn proc_row(&mut self) -> Result<ProcRow, String> {
453        Ok(ProcRow {
454            pid: self.u32()?,
455            ppid: self.opt_u32()?,
456            start_key: self.u64()?,
457            started_at_ms: self.opt_u64()?,
458            name: self.str()?,
459            cpu_milli: self.opt_u32()?,
460            mem: self.opt_u64()?,
461            mem_resident: self.opt_u64()?,
462        })
463    }
464    fn done(&self) -> Result<(), String> {
465        if self.i == self.p.len() {
466            Ok(())
467        } else {
468            Err("trailing bytes in a message".into())
469        }
470    }
471}
472
473impl Request {
474    /// The message for this request, with `id`.
475    pub fn encode(&self, id: u32) -> Vec<u8> {
476        let mut w = W(Vec::new());
477        w.u32(id);
478        match self {
479            Request::Hello => {
480                w.u8(op::HELLO);
481            }
482            Request::Stat { path } => {
483                w.u8(op::STAT).str(path);
484            }
485            Request::List {
486                path,
487                offset,
488                limit,
489            } => {
490                w.u8(op::LIST).str(path).u32(*offset).u32(*limit);
491            }
492            Request::Read { path, offset, len } => {
493                w.u8(op::READ).str(path).u64(*offset).u32(*len);
494            }
495            Request::Write { path, data } => {
496                w.u8(op::WRITE).str(path).bytes(data);
497            }
498            Request::Append { path, data } => {
499                w.u8(op::APPEND).str(path).bytes(data);
500            }
501            Request::Mkdir { path, parents } => {
502                w.u8(op::MKDIR).str(path).u8(u8::from(*parents));
503            }
504            Request::Rename { from, to } => {
505                w.u8(op::RENAME).str(from).str(to);
506            }
507            Request::Delete { path, recursive } => {
508                w.u8(op::DELETE).str(path).u8(u8::from(*recursive));
509            }
510            Request::Replace { from, to } => {
511                w.u8(op::REPLACE).str(from).str(to);
512            }
513            Request::Realpath { path } => {
514                w.u8(op::REALPATH).str(path);
515            }
516            Request::SetMeta {
517                path,
518                mode,
519                mtime_ms,
520            } => {
521                w.u8(op::SETMETA).str(path).u32(*mode).i64(*mtime_ms);
522            }
523            Request::Sync { path } => {
524                w.u8(op::SYNC).str(path);
525            }
526            Request::Procs { top, filter } => {
527                w.u8(op::PROCS).u32(*top).str(filter);
528            }
529            Request::PaneProcs { block_id, session } => {
530                w.u8(op::PANE_PROCS).str(block_id).str(session);
531            }
532        }
533        w.message()
534    }
535
536    /// A request body: its id and the request.
537    pub fn decode(body: &[u8]) -> Result<(u32, Request), String> {
538        let mut r = R { p: body, i: 0 };
539        let id = r.u32()?;
540        let req = match r.u8()? {
541            op::HELLO => Request::Hello,
542            op::STAT => Request::Stat { path: r.str()? },
543            op::LIST => Request::List {
544                path: r.str()?,
545                offset: r.u32()?,
546                limit: r.u32()?,
547            },
548            op::READ => Request::Read {
549                path: r.str()?,
550                offset: r.u64()?,
551                len: r.u32()?,
552            },
553            op::WRITE => Request::Write {
554                path: r.str()?,
555                data: r.bytes()?,
556            },
557            op::APPEND => Request::Append {
558                path: r.str()?,
559                data: r.bytes()?,
560            },
561            op::MKDIR => Request::Mkdir {
562                path: r.str()?,
563                parents: r.u8()? != 0,
564            },
565            op::RENAME => Request::Rename {
566                from: r.str()?,
567                to: r.str()?,
568            },
569            op::DELETE => Request::Delete {
570                path: r.str()?,
571                recursive: r.u8()? != 0,
572            },
573            op::REPLACE => Request::Replace {
574                from: r.str()?,
575                to: r.str()?,
576            },
577            op::REALPATH => Request::Realpath { path: r.str()? },
578            op::SYNC => Request::Sync { path: r.str()? },
579            op::PROCS => Request::Procs {
580                top: r.u32()?,
581                filter: r.str()?,
582            },
583            op::PANE_PROCS => Request::PaneProcs {
584                block_id: r.str()?,
585                session: r.str()?,
586            },
587            op::SETMETA => Request::SetMeta {
588                path: r.str()?,
589                mode: r.u32()?,
590                mtime_ms: r.i64()?,
591            },
592            other => return Err(format!("unknown request {other}")),
593        };
594        r.done()?;
595        Ok((id, req))
596    }
597}
598
599impl Reply {
600    /// The message for the reply to request `id`.
601    pub fn encode(&self, id: u32) -> Vec<u8> {
602        let mut w = W(Vec::new());
603        w.u32(id);
604        match self {
605            Reply::Hello { protocol, home, os } => {
606                w.u8(status::HELLO).u32(*protocol).str(home).str(os);
607            }
608            Reply::Stat(e) => {
609                w.u8(status::STAT).entry(e);
610            }
611            Reply::List { entries, total } => {
612                w.u8(status::LIST).u32(*total).u32(entries.len() as u32);
613                for e in entries {
614                    w.entry(e);
615                }
616            }
617            Reply::Read { data, eof } => {
618                w.u8(status::READ).u8(u8::from(*eof)).bytes(data);
619            }
620            Reply::Done => {
621                w.u8(status::DONE);
622            }
623            Reply::Path(p) => {
624                w.u8(status::PATH).str(p);
625            }
626            Reply::Procs(f) => {
627                w.u8(status::PROCS)
628                    .u32(f.cpu_count)
629                    .str(&f.memory_metric)
630                    .u32(f.total)
631                    .u32(f.matched)
632                    .u32(f.unmeasured)
633                    .opt_u64(f.cpu_milli)
634                    .u64(f.mem)
635                    .u32(f.rows.len() as u32);
636                for p in &f.rows {
637                    w.proc_row(p);
638                }
639            }
640            Reply::PaneProcs(f) => {
641                w.u8(status::PANE_PROCS)
642                    .u8(f.found as u8)
643                    .u32(f.processes)
644                    .u32(f.names.len() as u32);
645                for (name, count) in &f.names {
646                    w.str(name).u32(*count);
647                }
648                w.opt_u64(f.started_at_ms)
649                    .opt_u64(f.cpu_milli)
650                    .opt_u64(f.mem)
651                    .opt_u64(f.mem_resident)
652                    .u32(f.unmeasured)
653                    .str(&f.memory_metric)
654                    .u32(f.cpu_count)
655                    .opt_u64(f.host_cpu_milli)
656                    .u64(f.host_mem)
657                    .u32(f.host_processes)
658                    .u32(f.host_unmeasured);
659            }
660            Reply::Err { kind, message } => {
661                w.u8(status::ERR).u8(*kind as u8).str(message);
662            }
663        }
664        w.message()
665    }
666
667    /// A reply body: its request's id and the reply.
668    pub fn decode(body: &[u8]) -> Result<(u32, Reply), String> {
669        let mut r = R { p: body, i: 0 };
670        let id = r.u32()?;
671        let reply = match r.u8()? {
672            status::HELLO => Reply::Hello {
673                protocol: r.u32()?,
674                home: r.str()?,
675                os: r.str()?,
676            },
677            status::STAT => Reply::Stat(r.entry()?),
678            status::LIST => {
679                let total = r.u32()?;
680                let n = r.u32()? as usize;
681                // Each entry is at least 30 bytes: a count past that is a lie.
682                if n > (body.len() / 30) {
683                    return Err("list: impossible entry count".into());
684                }
685                let mut entries = Vec::with_capacity(n);
686                for _ in 0..n {
687                    entries.push(r.entry()?);
688                }
689                Reply::List { entries, total }
690            }
691            status::READ => Reply::Read {
692                eof: r.u8()? != 0,
693                data: r.bytes()?,
694            },
695            status::DONE => Reply::Done,
696            status::PATH => Reply::Path(r.str()?),
697            status::PROCS => {
698                let (cpu_count, memory_metric) = (r.u32()?, r.str()?);
699                let (total, matched, unmeasured) = (r.u32()?, r.u32()?, r.u32()?);
700                let (cpu_milli, mem) = (r.opt_u64()?, r.u64()?);
701                let n = r.u32()? as usize;
702                // Each row is at least 48 bytes: a count past that is a lie.
703                if n > body.len() / 48 {
704                    return Err("procs: impossible row count".into());
705                }
706                let mut rows = Vec::with_capacity(n);
707                for _ in 0..n {
708                    rows.push(r.proc_row()?);
709                }
710                Reply::Procs(ProcFrame {
711                    cpu_count,
712                    memory_metric,
713                    total,
714                    matched,
715                    unmeasured,
716                    cpu_milli,
717                    mem,
718                    rows,
719                })
720            }
721            status::PANE_PROCS => {
722                let (found, processes) = (PaneFound::from_u8(r.u8()?), r.u32()?);
723                let n = r.u32()? as usize;
724                // Each name is at least 8 bytes: a count past that is a lie.
725                if n > body.len() / 8 {
726                    return Err("pane procs: impossible name count".into());
727                }
728                let mut names = Vec::with_capacity(n);
729                for _ in 0..n {
730                    names.push((r.str()?, r.u32()?));
731                }
732                Reply::PaneProcs(PaneFrame {
733                    found,
734                    processes,
735                    names,
736                    started_at_ms: r.opt_u64()?,
737                    cpu_milli: r.opt_u64()?,
738                    mem: r.opt_u64()?,
739                    mem_resident: r.opt_u64()?,
740                    unmeasured: r.u32()?,
741                    memory_metric: r.str()?,
742                    cpu_count: r.u32()?,
743                    host_cpu_milli: r.opt_u64()?,
744                    host_mem: r.u64()?,
745                    host_processes: r.u32()?,
746                    host_unmeasured: r.u32()?,
747                })
748            }
749            status::ERR => Reply::Err {
750                kind: ErrKind::from_u8(r.u8()?),
751                message: r.str()?,
752            },
753            other => return Err(format!("unknown reply {other}")),
754        };
755        r.done()?;
756        Ok((id, reply))
757    }
758}
759
760/// Splits a byte stream that arrives in arbitrary pieces into message bodies.
761#[derive(Debug, Default)]
762pub struct Splitter {
763    buf: Vec<u8>,
764}
765
766impl Splitter {
767    pub fn new() -> Self {
768        Self::default()
769    }
770
771    /// Add bytes; return every body now complete. `Err` for an impossible
772    /// length: the stream is broken and the connection should be dropped.
773    pub fn push(&mut self, bytes: &[u8]) -> Result<Vec<Vec<u8>>, String> {
774        self.buf.extend_from_slice(bytes);
775        let mut bodies = Vec::new();
776        let mut at = 0;
777        while self.buf.len() - at >= 4 {
778            let len = u32::from_be_bytes(self.buf[at..at + 4].try_into().unwrap()) as usize;
779            if len > MAX_MESSAGE {
780                return Err(format!("a message of {len} bytes is over the limit"));
781            }
782            if self.buf.len() - at - 4 < len {
783                break;
784            }
785            bodies.push(self.buf[at + 4..at + 4 + len].to_vec());
786            at += 4 + len;
787        }
788        self.buf.drain(..at);
789        Ok(bodies)
790    }
791}
792
793#[cfg(test)]
794mod tests {
795    use super::*;
796
797    fn entry(name: &str) -> Entry {
798        Entry {
799            name: name.into(),
800            kind: Kind::File,
801            size: 42,
802            mtime_ms: 1_700_000_000_000,
803            mode: 0o644,
804            symlink: true,
805            link_target: "../elsewhere".into(),
806        }
807    }
808
809    #[test]
810    fn every_request_and_reply_survives_the_wire_in_any_pieces() {
811        let requests = vec![
812            Request::Hello,
813            Request::Stat {
814                path: "~/a b".into(),
815            },
816            Request::List {
817                path: "/etc".into(),
818                offset: 100,
819                limit: 50,
820            },
821            Request::Read {
822                path: "/x".into(),
823                offset: 1 << 40,
824                len: 4096,
825            },
826            Request::Write {
827                path: "/x".into(),
828                data: vec![0, 1, 2, 255],
829            },
830            Request::Append {
831                path: "/x".into(),
832                data: vec![3, 4],
833            },
834            Request::Mkdir {
835                path: "/d".into(),
836                parents: true,
837            },
838            Request::Rename {
839                from: "/a".into(),
840                to: "/b".into(),
841            },
842            Request::Delete {
843                path: "/d".into(),
844                recursive: false,
845            },
846            Request::Replace {
847                from: "/t".into(),
848                to: "/x".into(),
849            },
850            Request::Realpath { path: "~/l".into() },
851            Request::Sync { path: "/x".into() },
852            Request::SetMeta {
853                path: "/x".into(),
854                mode: 0o755,
855                mtime_ms: 1_700_000_000_000,
856            },
857            Request::Procs {
858                top: 100,
859                filter: "node agent".into(),
860            },
861            Request::PaneProcs {
862                block_id: "0b6c-block".into(),
863                session: "".into(),
864            },
865        ];
866        let replies = vec![
867            Reply::Hello {
868                protocol: PROTOCOL,
869                home: "/home/u".into(),
870                os: "linux".into(),
871            },
872            Reply::Stat(entry("a")),
873            Reply::List {
874                entries: vec![entry("a"), entry("ü")],
875                total: 7,
876            },
877            Reply::Read {
878                data: b"hello".to_vec(),
879                eof: true,
880            },
881            Reply::Done,
882            Reply::Path("/home/u/a".into()),
883            Reply::Err {
884                kind: ErrKind::NotFound,
885                message: "no such file".into(),
886            },
887            Reply::Procs(ProcFrame {
888                cpu_count: 8,
889                memory_metric: "anonymous resident + swap".into(),
890                total: 300,
891                matched: 2,
892                unmeasured: 1,
893                cpu_milli: Some(1500),
894                mem: 1 << 30,
895                rows: vec![
896                    ProcRow {
897                        pid: 42,
898                        ppid: Some(1),
899                        start_key: 98_765,
900                        started_at_ms: Some(1_700_000_000_000),
901                        name: "node".into(),
902                        cpu_milli: Some(1000),
903                        mem: Some(200 << 20),
904                        mem_resident: Some(300 << 20),
905                    },
906                    // Another user's process on macOS: nothing measured.
907                    ProcRow {
908                        pid: 1,
909                        ppid: None,
910                        start_key: 0,
911                        started_at_ms: None,
912                        name: "launchd".into(),
913                        cpu_milli: None,
914                        mem: None,
915                        mem_resident: None,
916                    },
917                ],
918            }),
919            Reply::PaneProcs(PaneFrame {
920                found: PaneFound::Marker,
921                processes: 3,
922                names: vec![("bash".into(), 1), ("node".into(), 2)],
923                started_at_ms: Some(1_700_000_000_000),
924                cpu_milli: Some(250),
925                mem: Some(80 << 20),
926                mem_resident: Some(120 << 20),
927                unmeasured: 1,
928                memory_metric: "anonymous resident + swap".into(),
929                cpu_count: 16,
930                host_cpu_milli: None,
931                host_mem: 6 << 30,
932                host_processes: 412,
933                host_unmeasured: 0,
934            }),
935            // Nothing found: the host's totals only.
936            Reply::PaneProcs(PaneFrame {
937                found: PaneFound::None,
938                processes: 0,
939                names: vec![],
940                started_at_ms: None,
941                cpu_milli: None,
942                mem: None,
943                mem_resident: None,
944                unmeasured: 0,
945                memory_metric: "memory footprint".into(),
946                cpu_count: 8,
947                host_cpu_milli: Some(3_100),
948                host_mem: 9 << 30,
949                host_processes: 600,
950                host_unmeasured: 210,
951            }),
952        ];
953        let mut stream = Vec::new();
954        for (i, r) in requests.iter().enumerate() {
955            stream.extend(r.encode(i as u32));
956        }
957        let mut s = Splitter::new();
958        let mut bodies = Vec::new();
959        for piece in stream.chunks(3) {
960            bodies.extend(s.push(piece).unwrap());
961        }
962        let got: Vec<(u32, Request)> = bodies.iter().map(|b| Request::decode(b).unwrap()).collect();
963        assert_eq!(
964            got,
965            requests
966                .into_iter()
967                .enumerate()
968                .map(|(i, r)| (i as u32, r))
969                .collect::<Vec<_>>()
970        );
971
972        for (i, r) in replies.iter().enumerate() {
973            let msg = r.encode(i as u32 + 9);
974            let (id, back) = Reply::decode(&msg[4..]).unwrap();
975            assert_eq!((id, &back), (i as u32 + 9, r));
976        }
977    }
978
979    #[test]
980    fn a_broken_stream_is_refused_not_guessed_at() {
981        assert!(Splitter::new().push(&u32::MAX.to_be_bytes()).is_err());
982        assert!(Request::decode(&[0, 0, 0, 1, 99]).is_err());
983        assert!(Request::decode(&[0, 0, 0, 1]).is_err());
984        // A string longer than the message.
985        assert!(Request::decode(&[0, 0, 0, 1, 2, 0, 0, 0, 9, b'a']).is_err());
986        // Trailing bytes.
987        assert!(Request::decode(&[0, 0, 0, 1, 1, 7]).is_err());
988        // A list claiming more entries than could fit.
989        let mut body = vec![0, 0, 0, 1, 3];
990        body.extend_from_slice(&5u32.to_be_bytes());
991        body.extend_from_slice(&1_000_000u32.to_be_bytes());
992        assert!(Reply::decode(&body).is_err());
993    }
994}
agentmux_remote/fsproto.rs

agentmux_remote/
fsproto.rs
