tear-core 0.1.9

Runtime logic for the tear multiplexer — InProcess MultiplexerControl impl backed by portable-pty. The single source of truth for pane semantics across pleme-io GUI consumers (mado at tier 3).
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
//! Per-pane recording — captures every PTY byte with a relative
//! timestamp, exposes a ring-buffered snapshot, and exports as
//! asciinema v2 .cast (JSON-lines) so any external player handles
//! playback. Daemon-native: no `asciinema rec` wrapper needed.
//!
//! Storage shape
//! -------------
//! Each pane gets a `PaneRecording` instance keyed off the pane
//! id. Recording starts as Disabled; `enable()` flips it on and
//! marks the start instant. From then, every chunk fed via
//! [`Self::push`] is appended as `(millis_since_start, Vec<u8>)`.
//!
//! Size cap
//! --------
//! A long-lived pane could pile up arbitrarily many bytes. Each
//! recording has a configurable max event count (default 50_000 —
//! enough for ~1h of typical interactive output). When the cap is
//! exceeded, the oldest events are dropped (ring-buffer semantics)
//! so the recording always reflects the most recent N events.
//!
//! Export format
//! -------------
//! `to_cast_json()` emits the asciinema v2 header line + one
//! data line per event. Every byte chunk lands as `[t_s, "o",
//! "<utf-8 string>"]` — the v2 wire shape. Non-UTF-8 bytes are
//! losslessly preserved by the JSON string encoding (asciinema
//! players handle the raw bytes the same as terminal display
//! does).

use std::sync::Mutex;
use std::time::Instant;

use serde::{Deserialize, Serialize};

/// Wall-clock now, unix epoch ms. Saturates to 0 before the epoch rather
/// than panicking — a clock that far wrong is not this module's problem to
/// escalate.
fn now_unix_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
        .unwrap_or(0)
}

/// One captured PTY chunk, relative to the recording's start.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PaneEvent {
    /// Milliseconds since `enable()` was called.
    pub ts_ms: u64,
    /// Raw bytes pushed to the PTY's output side. The vte parser
    /// has already fed these into the grid; the recording stores
    /// a verbatim copy so replays match what a live viewer saw.
    pub bytes: Vec<u8>,
}

/// State of a per-pane recording. Cheap when disabled — the
/// `on_bytes` hook hits a single boolean check before deciding
/// whether to deep-copy the chunk.
pub struct PaneRecording {
    enabled: Mutex<RecordingState>,
}

struct RecordingState {
    /// `None` when disabled.
    started_at: Option<Instant>,
    /// Wall-clock start, unix epoch ms. `None` when disabled.
    ///
    /// `started_at` is an [`Instant`] — monotonic, with NO epoch — so it
    /// can measure elapsed time and can never answer "when did this
    /// begin?". This field is that answer, and it exists for two reasons:
    ///
    /// 1. The asciinema v2 header's `timestamp` is the RECORDING START.
    ///    Without an anchor `to_cast_json` had to reach for
    ///    `SystemTime::now()`, which stamps the EXPORT time — so a cast
    ///    exported a day later claimed to have been recorded a day later.
    /// 2. It is the join key to anything stamped in epoch time (a
    ///    `Block.started_at_unix_ms`, say). `read_around` takes ms since
    ///    `enable()`, so converting requires this anchor; passing an
    ///    epoch timestamp straight in is off by ~1.7e12 and silently
    ///    returns the buffer tail rather than erroring.
    started_at_unix_ms: Option<u64>,
    /// Captured events. Ring-buffered against `max_events`.
    events: std::collections::VecDeque<PaneEvent>,
    /// Max retained events. Default 50_000.
    max_events: usize,
    /// Recorded cols × rows at start — written into the asciinema
    /// cast header on export.
    cols: u16,
    rows: u16,
}

impl Default for PaneRecording {
    fn default() -> Self {
        Self::new(50_000)
    }
}

impl PaneRecording {
    #[must_use]
    pub fn new(max_events: usize) -> Self {
        Self {
            enabled: Mutex::new(RecordingState {
                started_at: None,
                started_at_unix_ms: None,
                events: std::collections::VecDeque::new(),
                max_events,
                cols: 80,
                rows: 24,
            }),
        }
    }

    /// Begin (or restart) recording. Clears prior events; stamps
    /// the start instant; remembers the pane dimensions for the
    /// cast header.
    pub fn enable(&self, cols: u16, rows: u16) {
        let mut g = self.enabled.lock().expect("recording state poisoned");
        g.started_at = Some(Instant::now());
        g.started_at_unix_ms = Some(now_unix_ms());
        g.events.clear();
        g.cols = cols;
        g.rows = rows;
    }

    /// Stop recording. Retains captured events so an operator can
    /// `export` them after stopping; a subsequent `enable()`
    /// clears + restarts.
    pub fn disable(&self) {
        let mut g = self.enabled.lock().expect("recording state poisoned");
        g.started_at = None;
        g.started_at_unix_ms = None;
    }

    /// Wall-clock start of the current recording, unix epoch ms.
    ///
    /// The join key for anything stamped in epoch time. [`Self::read_around`]
    /// takes ms since `enable()`, so a caller holding an epoch timestamp
    /// must convert through this anchor:
    ///
    /// ```text
    /// rel_ms = epoch_ms.saturating_sub(anchor)
    /// ```
    ///
    /// Passing an epoch timestamp straight to `read_around` is off by
    /// ~1.7e12 ms and silently returns the buffer tail — it does not error,
    /// which is why this accessor exists rather than leaving callers to
    /// guess.
    #[must_use]
    pub fn epoch_anchor(&self) -> Option<u64> {
        self.enabled
            .lock()
            .expect("recording state poisoned")
            .started_at_unix_ms
    }

    /// Backdate the epoch anchor. **Tests only.**
    ///
    /// Exists because the header-timestamp invariant cannot be tested
    /// without it: `enable()` and export happen in the same wall-clock
    /// second, so a correct implementation and one that stamps
    /// `SystemTime::now()` produce identical output. Backdating is what
    /// makes the two distinguishable without a `sleep`.
    #[cfg(test)]
    pub(crate) fn set_epoch_anchor_for_test(&self, unix_ms: u64) {
        self.enabled
            .lock()
            .expect("recording state poisoned")
            .started_at_unix_ms = Some(unix_ms);
    }

    /// Whether recording is currently capturing new events.
    #[must_use]
    pub fn is_enabled(&self) -> bool {
        self.enabled
            .lock()
            .expect("recording state poisoned")
            .started_at
            .is_some()
    }

    /// Number of currently-buffered events.
    #[must_use]
    pub fn event_count(&self) -> usize {
        self.enabled
            .lock()
            .expect("recording state poisoned")
            .events
            .len()
    }

    /// Append a PTY chunk. Cheap no-op when recording is disabled.
    /// Ring-evicts the oldest event if the cap is hit.
    pub fn push(&self, bytes: &[u8]) {
        let mut g = self.enabled.lock().expect("recording state poisoned");
        let Some(start) = g.started_at else {
            return;
        };
        let ts_ms = start.elapsed().as_millis() as u64;
        let cap = g.max_events;
        if g.events.len() == cap {
            g.events.pop_front();
        }
        g.events.push_back(PaneEvent {
            ts_ms,
            bytes: bytes.to_vec(),
        });
    }

    /// Export as asciinema v2 .cast (JSON-lines). The first line
    /// is the header object; every subsequent line is
    /// `[t_seconds, "o", "<utf-8 chunk>"]`. Returns the full
    /// string ready to write to disk or pipe to `asciinema play`.
    pub fn to_cast_json(&self) -> String {
        use std::time::{SystemTime, UNIX_EPOCH};
        let g = self.enabled.lock().expect("recording state poisoned");
        let header = serde_json::json!({
            "version": 2,
            "width": g.cols,
            "height": g.rows,
            // The RECORDING START, not the export time. See
            // `RecordingState::started_at_unix_ms`. Falls back to now()
            // only when nothing was ever recorded, where there is no
            // start to report and the header value is meaningless anyway.
            // The RECORDING START, not the export time. See
            // `RecordingState::started_at_unix_ms`. Falls back to now()
            // only when nothing was ever recorded, where there is no
            // start to report and the header value is meaningless anyway.
            "timestamp": g.started_at_unix_ms.map_or_else(
                || SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or(0),
                |ms| ms / 1000,
            ),
            "env": {
                "TERM": "xterm-256color",
                "SHELL": std::env::var("SHELL").unwrap_or_default(),
            },
        });
        let mut out = header.to_string();
        out.push('\n');
        for ev in &g.events {
            let secs = (ev.ts_ms as f64) / 1000.0;
            let chunk = String::from_utf8_lossy(&ev.bytes).into_owned();
            // asciinema v2 row: [<float-seconds>, "o", "<chunk>"]
            let row = serde_json::Value::Array(vec![
                serde_json::json!(secs),
                serde_json::json!("o"),
                serde_json::json!(chunk),
            ]);
            out.push_str(&row.to_string());
            out.push('\n');
        }
        out
    }

    /// Read the events at (or just before) a given `ts_ms` —
    /// returns up to `limit` events nearest to the cursor for the
    /// time-travel scrubber. Lets a replay renderer seek without
    /// re-reading the full recording.
    pub fn read_around(&self, ts_ms: u64, limit: usize) -> Vec<PaneEvent> {
        let g = self.enabled.lock().expect("recording state poisoned");
        // Binary-search-ish — events are sorted by ts_ms since
        // they're appended in order. Simple linear scan is fine
        // for the typical recording size (<10k events).
        let cursor = g
            .events
            .iter()
            .position(|e| e.ts_ms >= ts_ms)
            .unwrap_or(g.events.len());
        let start = cursor.saturating_sub(limit / 2);
        let end = (start + limit).min(g.events.len());
        g.events.range(start..end).cloned().collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// ★ RED AGAINST THE CODE AS IT SHIPPED. The asciinema v2 header's
    /// `timestamp` is the RECORDING START; `to_cast_json` stamped
    /// `SystemTime::now()`, so a cast exported an hour later claimed to
    /// have been recorded an hour later. `started_at` is an `Instant` —
    /// monotonic, no epoch — so it structurally could not answer the
    /// question, which is why the fix is a new field rather than a
    /// different expression.
    #[test]
    fn the_cast_header_timestamp_is_the_recording_start_not_the_export_time() {
        let r = PaneRecording::default();
        r.enable(80, 24);
        r.push(b"x");
        let anchor = r.epoch_anchor().expect("enabled recording has an anchor");

        // ★ The discrimination this test exists for. A first draft simply
        // exported and compared against the anchor — and PASSED with the
        // bug restored, because enable() and export land in the same
        // wall-clock second, so `now()` and the anchor are equal. A test
        // that cannot tell the two implementations apart proves nothing.
        //
        // So: backdate the anchor by an hour, standing in for "this cast
        // was exported an hour after it was recorded" without sleeping.
        // Only an implementation that READS the anchor can report it.
        let backdated = anchor - 3_600_000;
        r.set_epoch_anchor_for_test(backdated);

        let json = r.to_cast_json();
        let header: serde_json::Value =
            serde_json::from_str(json.lines().next().unwrap()).unwrap();
        let ts = header["timestamp"].as_u64().unwrap();

        assert_eq!(
            ts,
            backdated / 1000,
            "header timestamp must be the RECORDING START ({}), not the \
             export time ({}) — a cast that misreports when it was \
             recorded is worse than one carrying no timestamp at all",
            backdated / 1000,
            anchor / 1000
        );
    }

    /// The anchor is the join key to epoch-stamped data, and it must be
    /// absent when there is nothing to anchor.
    #[test]
    fn the_epoch_anchor_appears_on_enable_and_clears_on_disable() {
        let r = PaneRecording::default();
        assert!(r.epoch_anchor().is_none(), "nothing recorded, nothing to anchor");
        r.enable(80, 24);
        let a = r.epoch_anchor().expect("enabled");
        assert!(a > 1_700_000_000_000, "anchor must be epoch MS, got {a}");
        r.disable();
        assert!(
            r.epoch_anchor().is_none(),
            "a stopped recording must not keep advertising a live anchor"
        );
    }

    #[test]
    fn disabled_recording_drops_pushes() {
        let r = PaneRecording::default();
        r.push(b"hello");
        assert_eq!(r.event_count(), 0);
    }

    #[test]
    fn enable_then_push_captures_events() {
        let r = PaneRecording::default();
        r.enable(80, 24);
        r.push(b"hello");
        r.push(b" world");
        assert_eq!(r.event_count(), 2);
    }

    #[test]
    fn ring_buffer_caps_at_max() {
        let r = PaneRecording::new(3);
        r.enable(80, 24);
        r.push(b"a");
        r.push(b"b");
        r.push(b"c");
        r.push(b"d");
        r.push(b"e");
        assert_eq!(r.event_count(), 3);
    }

    #[test]
    fn cast_export_has_header_plus_one_line_per_event() {
        let r = PaneRecording::default();
        r.enable(120, 40);
        r.push(b"$ ls\n");
        r.push(b"file1 file2\n");
        let cast = r.to_cast_json();
        let lines: Vec<&str> = cast.lines().collect();
        assert_eq!(lines.len(), 3, "expected header + 2 events, got {cast}");
        let header: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
        assert_eq!(header["version"], 2);
        assert_eq!(header["width"], 120);
        assert_eq!(header["height"], 40);
        let ev: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
        let arr = ev.as_array().unwrap();
        assert_eq!(arr[1], "o");
        assert_eq!(arr[2].as_str().unwrap(), "$ ls\n");
    }

    #[test]
    fn disable_then_enable_clears_old_events() {
        let r = PaneRecording::default();
        r.enable(80, 24);
        r.push(b"x");
        assert_eq!(r.event_count(), 1);
        r.disable();
        r.enable(80, 24);
        assert_eq!(r.event_count(), 0);
    }

    #[test]
    fn read_around_returns_events_near_cursor() {
        let r = PaneRecording::default();
        r.enable(80, 24);
        // Manually craft 5 events at fixed timestamps for a
        // deterministic test (push uses Instant::now under the
        // hood which we'd race on).
        {
            let mut g = r.enabled.lock().unwrap();
            for i in 0..5u64 {
                g.events.push_back(PaneEvent {
                    ts_ms: i * 1000,
                    bytes: vec![b'a' + i as u8],
                });
            }
        }
        let around = r.read_around(2500, 4);
        // Cursor lands at index 3 (ts=3000 >= 2500). With limit=4
        // start = 3 - 4/2 = 1; end = 1 + 4 = 5. So we get events at
        // ts=1000, 2000, 3000, 4000.
        assert_eq!(around.len(), 4);
        assert_eq!(around[0].ts_ms, 1000);
        assert_eq!(around[3].ts_ms, 4000);
    }
}