agentmux_remote/
ring.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! A session's output history: the last `capacity` bytes of everything its
5//! terminal wrote, addressed by offset in the whole stream (spec §7.2). A
6//! reattach asks for "everything after offset N" and gets it, or, when N has
7//! fallen out, everything still held plus how much was lost.
8//!
9//! Held in memory (8 MB by default per session): the spec's on-disk ring
10//! would outlive the daemon, but the sessions themselves cannot (their
11//! terminals are the daemon's), so memory loses nothing that disk would keep.
12
13/// The default history per session.
14#[cfg_attr(not(unix), allow(dead_code))]
15pub const DEFAULT_CAPACITY: usize = 8 * 1024 * 1024;
16
17#[derive(Debug)]
18pub struct Ring {
19    /// A deque, so dropping the oldest bytes is cheap even when full.
20    buf: std::collections::VecDeque<u8>,
21    capacity: usize,
22    /// Offset of the next byte to be written: the stream's length so far.
23    end: u64,
24}
25
26/// What a read from an offset gives back.
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct Replay {
29    /// Bytes `lost.0..lost.1` were asked for but are no longer held.
30    pub lost: Option<(u64, u64)>,
31    /// The offset of `data[0]`.
32    pub offset: u64,
33    pub data: Vec<u8>,
34}
35
36impl Ring {
37    pub fn new(capacity: usize) -> Self {
38        assert!(capacity > 0);
39        Ring {
40            buf: std::collections::VecDeque::new(),
41            capacity,
42            end: 0,
43        }
44    }
45
46    /// The offset the next byte will have.
47    pub fn end(&self) -> u64 {
48        self.end
49    }
50
51    /// The oldest offset still held.
52    pub fn start(&self) -> u64 {
53        self.end - self.buf.len() as u64
54    }
55
56    pub fn append(&mut self, bytes: &[u8]) {
57        self.end += bytes.len() as u64;
58        if bytes.len() >= self.capacity {
59            self.buf.clear();
60            self.buf.extend(&bytes[bytes.len() - self.capacity..]);
61            return;
62        }
63        let overflow = (self.buf.len() + bytes.len()).saturating_sub(self.capacity);
64        self.buf.drain(..overflow);
65        self.buf.extend(bytes);
66    }
67
68    /// Everything from `offset` on. An offset past the end is the end (nothing
69    /// to replay); one before the start replays from the start and says what
70    /// was lost.
71    pub fn read_from(&self, offset: u64) -> Replay {
72        let offset = offset.min(self.end);
73        let start = self.start();
74        if offset < start {
75            return Replay {
76                lost: Some((offset, start)),
77                offset: start,
78                data: self.buf.iter().copied().collect(),
79            };
80        }
81        let skip = (offset - start) as usize;
82        Replay {
83            lost: None,
84            offset,
85            data: self.buf.range(skip..).copied().collect(),
86        }
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93
94    #[test]
95    fn a_reattach_gets_exactly_what_it_missed() {
96        let mut r = Ring::new(100);
97        r.append(b"hello ");
98        r.append(b"world");
99        assert_eq!((r.start(), r.end()), (0, 11));
100        assert_eq!(
101            r.read_from(6),
102            Replay {
103                lost: None,
104                offset: 6,
105                data: b"world".to_vec()
106            }
107        );
108        assert_eq!(r.read_from(0).data, b"hello world");
109        assert_eq!(r.read_from(11).data, b"", "caught up: nothing");
110        assert_eq!(
111            r.read_from(500),
112            Replay {
113                lost: None,
114                offset: 11,
115                data: Vec::new()
116            },
117            "past the end is the end"
118        );
119    }
120
121    #[test]
122    fn what_fell_out_is_reported_not_silently_skipped() {
123        let mut r = Ring::new(8);
124        r.append(b"0123456789"); // longer than the ring
125        assert_eq!((r.start(), r.end()), (2, 10));
126        assert_eq!(
127            r.read_from(0),
128            Replay {
129                lost: Some((0, 2)),
130                offset: 2,
131                data: b"23456789".to_vec()
132            }
133        );
134        r.append(b"abc");
135        assert_eq!((r.start(), r.end()), (5, 13));
136        assert_eq!(
137            r.read_from(3),
138            Replay {
139                lost: Some((3, 5)),
140                offset: 5,
141                data: b"56789abc".to_vec()
142            }
143        );
144        assert_eq!(r.read_from(9).data, b"9abc");
145    }
146
147    #[test]
148    fn many_small_writes_keep_the_last_capacity_bytes() {
149        let mut r = Ring::new(16);
150        let mut all = Vec::new();
151        for i in 0..200u32 {
152            let s = format!("{i},");
153            r.append(s.as_bytes());
154            all.extend_from_slice(s.as_bytes());
155        }
156        let replay = r.read_from(r.start());
157        assert_eq!(replay.data, all[all.len() - 16..]);
158        assert_eq!(r.end(), all.len() as u64);
159    }
160}
agentmux_remote/ring.rs

agentmux_remote/
ring.rs
