pixelflux/recorder/
mod.rs

1/*
2 * This Source Code Form is subject to the terms of the Mozilla Public
3 * License, v. 2.0. If a copy of the MPL was not distributed with this
4 * file, You can obtain one at https://mozilla.org/MPL/2.0/.
5 */
6
7//! Built-in MP4 recorder: an independent pipeline that captures the desktop to a
8//! ready-to-play fragmented MP4 without needing any connected client.
9//!
10//! One implementation serves three control surfaces — the Python module functions
11//! (`start_recording` / `stop_recording` / `recording_status`), the `PIXELFLUX_RECORD*`
12//! environment variables (record from process start), and the `record_start` /
13//! `record_stop` / `record_status` endpoints on the Computer Use HTTP server — so their
14//! behavior cannot drift.
15//!
16//! The recorder is a consumer of the existing capture machinery, never a hook inside an
17//! encoder:
18//!
19//! * **X11**: it always runs its own [`crate::x11::run_capture`] instance against the root
20//!   window (a second capture of the same root is independent of any streaming session), with
21//!   the encoded frames delivered straight into the recorder's queue.
22//! * **Wayland**: the compositor lives in this process and allows one capture per output, so
23//!   the recorder attaches at the delivery layer. When the output is already being captured
24//!   for a streaming client it taps that stream (and paces recovery keyframes through the
25//!   standard `RequestIdr` command); when the output is idle it starts its own capture with no
26//!   Python callback and taps the identical delivery point.
27//!
28//! Frames cross into the recorder through one bounded queue, mirroring the Unix-socket sink's
29//! isolation contract: the delivery thread only clones an `Arc` and `try_send`s, an overflowing
30//! queue drops frames (never blocks the pipeline), and all muxing happens on the recorder's own
31//! writer thread. Timestamps are wall-clock, so the damage-driven, variable-rate frame flow
32//! lands at its true times in the MP4.
33
34pub mod mp4;
35
36use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
37use std::sync::{mpsc, Arc, Mutex};
38use std::thread;
39use std::time::{Duration, Instant};
40
41use crossbeam_channel::{bounded, Receiver, Sender, TrySendError};
42
43use crate::encoders::software::EncodedStripe;
44use crate::RustCaptureSettings;
45use crate::ThreadCommand;
46
47/// Recorder queue bound, matching the socket sink's stalled-consumer policy: a writer that
48/// falls this far behind loses frames instead of growing memory or blocking the encoder.
49const QUEUE_CAP: usize = 256;
50
51/// A queued frame: `Arc`-shared encoded payload, byte offset where the Annex-B stream
52/// starts (past the wire header when present), and the wall-clock capture time.
53type TapFrame = (Arc<Vec<u8>>, usize, u64);
54
55/// Which capture system feeds the recording.
56#[derive(Clone, Copy, PartialEq, Eq, Debug)]
57pub enum PreferredBackend {
58    X11,
59    Wayland,
60}
61
62/// Resolved recording parameters. Environment variables provide the defaults; explicit
63/// Python/REST arguments override them.
64#[derive(Clone)]
65pub struct RecordOptions {
66    pub path: String,
67    pub display_id: u32,
68    /// Capture fps cap for a recorder-owned capture (`<= 0` = default 30). An attached
69    /// recording follows the live session's rate.
70    pub fps: f64,
71    /// Bitrate override in kbps for a recorder-owned capture (`<= 0` = settings default).
72    pub bitrate_kbps: i32,
73    /// Recovery-keyframe cadence in seconds (`<= 0` = default 2.0). Recorder-owned captures
74    /// carry it in their settings (scheduled IDRs); attached recordings pace standard
75    /// request-IDR commands at this interval.
76    pub keyframe_interval_s: f64,
77    pub backend: Option<PreferredBackend>,
78    /// Full capture settings for a recorder-owned capture; `None` derives them (full root /
79    /// current output geometry, H.264 full-frame).
80    pub capture: Option<RustCaptureSettings>,
81}
82
83impl RecordOptions {
84    /// Options seeded entirely from `PIXELFLUX_RECORD_*` environment variables.
85    pub fn from_env(path: String) -> Self {
86        let f = |k: &str| std::env::var(k).ok().and_then(|v| v.parse::<f64>().ok());
87        let backend = match std::env::var("PIXELFLUX_RECORD_BACKEND").ok().as_deref() {
88            Some("x11") => Some(PreferredBackend::X11),
89            Some("wayland") => Some(PreferredBackend::Wayland),
90            _ => None,
91        };
92        Self {
93            path,
94            display_id: f("PIXELFLUX_RECORD_DISPLAY").map(|v| v as u32).unwrap_or(0),
95            fps: f("PIXELFLUX_RECORD_FPS").unwrap_or(0.0),
96            bitrate_kbps: f("PIXELFLUX_RECORD_BITRATE").map(|v| v as i32).unwrap_or(0),
97            keyframe_interval_s: f("PIXELFLUX_RECORD_KEYFRAME_S").unwrap_or(0.0),
98            backend,
99            capture: None,
100        }
101    }
102
103    fn effective_fps(&self) -> f64 {
104        if self.fps > 0.0 { self.fps } else { 30.0 }
105    }
106
107    fn effective_keyframe_s(&self) -> f64 {
108        if self.keyframe_interval_s > 0.0 { self.keyframe_interval_s } else { 2.0 }
109    }
110}
111
112/// Live counters shared between the feeding side (delivery threads), the writer thread and
113/// the status surfaces.
114struct RecShared {
115    start: Instant,
116    enqueued: AtomicU64,
117    dropped: AtomicU64,
118    skipped_non_h264: AtomicU64,
119    muxed: AtomicU64,
120    sync_frames: AtomicU64,
121    bytes: AtomicU64,
122    width: AtomicU32,
123    height: AtomicU32,
124    last_idr_req_us: AtomicU64,
125    error: Mutex<Option<String>>,
126}
127
128impl RecShared {
129    fn new() -> Arc<Self> {
130        Arc::new(Self {
131            start: Instant::now(),
132            enqueued: AtomicU64::new(0),
133            dropped: AtomicU64::new(0),
134            skipped_non_h264: AtomicU64::new(0),
135            muxed: AtomicU64::new(0),
136            sync_frames: AtomicU64::new(0),
137            bytes: AtomicU64::new(0),
138            width: AtomicU32::new(0),
139            height: AtomicU32::new(0),
140            last_idr_req_us: AtomicU64::new(0),
141            error: Mutex::new(None),
142        })
143    }
144
145    fn set_error(&self, msg: String) {
146        let mut g = self.error.lock().unwrap();
147        if g.is_none() {
148            eprintln!("[recorder] {msg}");
149            *g = Some(msg);
150        }
151    }
152}
153
154/// Point-in-time view of the recorder, identical across the Python, env and REST surfaces.
155#[derive(Clone, Debug)]
156pub struct RecordingStatus {
157    pub active: bool,
158    pub path: String,
159    pub backend: &'static str,
160    pub mode: &'static str,
161    pub frames: u64,
162    pub sync_frames: u64,
163    pub dropped: u64,
164    pub skipped_non_h264: u64,
165    pub bytes: u64,
166    pub duration_s: f64,
167    pub width: u32,
168    pub height: u32,
169    pub error: Option<String>,
170}
171
172/// The wayland-side tap handle consulted by [`wayland_tap`]; cloneable so the delivery
173/// thread can drop the registry lock before doing any work.
174#[derive(Clone)]
175struct WlTap {
176    display_id: u32,
177    tx: Sender<TapFrame>,
178    shared: Arc<RecShared>,
179    /// Standard request-IDR path for attached recordings (`None` when the recorder owns the
180    /// capture and scheduled keyframes ride in its settings).
181    idr: Option<IdrRequester>,
182    keyframe_interval_us: u64,
183}
184
185/// Sends `ThreadCommand::RequestIdr` for one display over the compositor's command channel.
186#[derive(Clone)]
187struct IdrRequester {
188    tx: smithay::reexports::calloop::channel::Sender<ThreadCommand>,
189    display_id: u32,
190}
191
192impl IdrRequester {
193    fn request(&self) {
194        let _ = self.tx.send(ThreadCommand::RequestIdr { display_id: self.display_id });
195    }
196}
197
198/// How the active recording is fed and what must be torn down on stop.
199enum RecordingMode {
200    /// Recorder-owned X11 capture: its controls and capture thread.
201    X11Own {
202        controls: Arc<crate::x11::Controls>,
203        join: thread::JoinHandle<()>,
204    },
205    /// Recorder-owned Wayland capture on `display_id` (stopped unless a streaming client has
206    /// since taken ownership of the display).
207    WaylandOwn { display_id: u32 },
208    /// Tap on a streaming client's live Wayland capture; nothing to stop.
209    WaylandAttached,
210}
211
212struct ActiveRecording {
213    path: String,
214    backend: &'static str,
215    mode_name: &'static str,
216    mode: RecordingMode,
217    tx: Sender<TapFrame>,
218    shared: Arc<RecShared>,
219    writer: thread::JoinHandle<()>,
220}
221
222static ACTIVE: Mutex<Option<ActiveRecording>> = Mutex::new(None);
223static LAST_FINISHED: Mutex<Option<RecordingStatus>> = Mutex::new(None);
224/// Fast idle guard for [`wayland_tap`]: the delivery threads pay one relaxed atomic load
225/// per frame while no recording is armed.
226static WL_TAP_ARMED: AtomicBool = AtomicBool::new(false);
227static WL_TAP: Mutex<Option<WlTap>> = Mutex::new(None);
228
229/// Delivery-layer feed from the Wayland pipelines (readback encode loop and zero-copy
230/// tick). Near-zero cost when idle; while recording it clones an `Arc` into the bounded
231/// queue and never blocks.
232pub(crate) fn wayland_tap(display_id: u32, stripes: &[EncodedStripe]) {
233    if !WL_TAP_ARMED.load(Ordering::Relaxed) {
234        return;
235    }
236    let Some(tap) = WL_TAP.lock().unwrap().clone() else { return };
237    if tap.display_id != display_id {
238        return;
239    }
240    offer_frame(&tap.shared, &tap.tx, stripes);
241    if let Some(ref idr) = tap.idr {
242        pace_idr_requests(&tap.shared, idr, tap.keyframe_interval_us);
243    }
244}
245
246/// Issue a standard request-IDR when the cadence interval has elapsed. Compare-and-swap on
247/// the last-request stamp so concurrent delivery threads emit one request per interval.
248fn pace_idr_requests(shared: &RecShared, idr: &IdrRequester, interval_us: u64) {
249    if interval_us == 0 {
250        return;
251    }
252    let now = shared.start.elapsed().as_micros() as u64;
253    let last = shared.last_idr_req_us.load(Ordering::Relaxed);
254    if now.saturating_sub(last) >= interval_us
255        && shared
256            .last_idr_req_us
257            .compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed)
258            .is_ok()
259    {
260        idr.request();
261    }
262}
263
264/// Enqueue one delivered frame for muxing. Only a single full-frame H.264 stripe is
265/// recordable; JPEG or striped output is counted and skipped so the surfaces can report a
266/// clear "nothing recordable" error instead of writing a corrupt file.
267fn offer_frame(shared: &RecShared, tx: &Sender<TapFrame>, stripes: &[EncodedStripe]) {
268    if stripes.is_empty() {
269        return;
270    }
271    if stripes.len() != 1 || stripes[0].data_type != 2 || stripes[0].stripe_y_start != 0 {
272        shared.skipped_non_h264.fetch_add(1, Ordering::Relaxed);
273        return;
274    }
275    let s = &stripes[0];
276    if s.data.is_empty() {
277        return;
278    }
279    let offset = if s.data.len() >= 10 && s.data[0] == 0x04 { 10 } else { 0 };
280    if s.data.len() == offset {
281        return;
282    }
283    let pts_us = shared.start.elapsed().as_micros() as u64;
284    match tx.try_send((s.data.clone(), offset, pts_us)) {
285        Ok(()) => {
286            shared.enqueued.fetch_add(1, Ordering::Relaxed);
287        }
288        Err(TrySendError::Full(_)) => {
289            shared.dropped.fetch_add(1, Ordering::Relaxed);
290        }
291        Err(TrySendError::Disconnected(_)) => {}
292    }
293}
294
295/// Writer-thread body: drain the queue, convert Annex-B to AVCC, and mux. Output starts at
296/// the first IDR with parameter sets; earlier frames are discarded. Runs until every sender
297/// is dropped (stop) or a write error occurs, then finalizes the file and publishes the
298/// final status.
299fn writer_thread(
300    rx: Receiver<TapFrame>,
301    file: std::fs::File,
302    path: String,
303    backend: &'static str,
304    mode_name: &'static str,
305    shared: Arc<RecShared>,
306) {
307    let mut writer = mp4::FragmentWriter::new(std::io::BufWriter::new(file));
308    let mut builder = mp4::H264SampleBuilder::new();
309
310    'recv: for (buf, offset, pts_us) in rx.iter() {
311        let Some(sample) = builder.build_sample(&buf[offset..]) else { continue };
312        if !writer.init_written() {
313            if !sample.sync || !builder.have_parameter_sets() {
314                continue;
315            }
316            let Some(cfg) = builder.track_config() else {
317                shared.set_error("failed to parse H.264 SPS for MP4 init".to_string());
318                break 'recv;
319            };
320            shared.width.store(cfg.width, Ordering::Relaxed);
321            shared.height.store(cfg.height, Ordering::Relaxed);
322            if let Err(e) = writer.write_init(&cfg) {
323                shared.set_error(format!("MP4 init write failed: {e}"));
324                break 'recv;
325            }
326        }
327        if let Err(e) = writer.push_sample(sample.data, sample.sync, pts_us) {
328            shared.set_error(format!("MP4 write failed: {e}"));
329            break 'recv;
330        }
331        let st = writer.stats();
332        // The writer always holds one sample back to learn its duration, so it counts as muxed.
333        shared.muxed.store(st.samples + 1, Ordering::Relaxed);
334        shared.sync_frames.store(st.sync_samples, Ordering::Relaxed);
335        shared.bytes.store(st.bytes, Ordering::Relaxed);
336    }
337
338    let final_stats = match writer.finish() {
339        Ok(st) => st,
340        Err(e) => {
341            shared.set_error(format!("MP4 finalize failed: {e}"));
342            mp4::Mp4Stats::default()
343        }
344    };
345    if final_stats.samples == 0 {
346        shared.set_error(
347            "no recordable H.264 frames received (the session must produce a single \
348             full-frame H.264 stream; JPEG and striped modes cannot be recorded)"
349                .to_string(),
350        );
351        let _ = std::fs::remove_file(&path);
352    }
353    let status = RecordingStatus {
354        active: false,
355        path,
356        backend,
357        mode: mode_name,
358        frames: final_stats.samples,
359        sync_frames: final_stats.sync_samples,
360        dropped: shared.dropped.load(Ordering::Relaxed),
361        skipped_non_h264: shared.skipped_non_h264.load(Ordering::Relaxed),
362        bytes: final_stats.bytes,
363        duration_s: final_stats.duration_us as f64 / 1e6,
364        width: shared.width.load(Ordering::Relaxed),
365        height: shared.height.load(Ordering::Relaxed),
366        error: shared.error.lock().unwrap().clone(),
367    };
368    println!(
369        "[recorder] finished {}: {} frames ({} sync), {:.2}s, {} bytes",
370        status.path, status.frames, status.sync_frames, status.duration_s, status.bytes
371    );
372    *LAST_FINISHED.lock().unwrap() = Some(status);
373}
374
375/// Capture settings for a recorder-owned capture: explicit settings when given (validated
376/// H.264), otherwise derived defaults, always forced to the one recordable shape (full-frame
377/// H.264, no socket rebind, scheduled recovery keyframes).
378fn own_capture_settings(opts: &RecordOptions) -> Result<RustCaptureSettings, String> {
379    let mut s = match &opts.capture {
380        Some(explicit) => {
381            if explicit.output_mode != 1 {
382                return Err(
383                    "recording requires H.264 capture settings (output_mode=1); JPEG cannot be recorded"
384                        .to_string(),
385                );
386            }
387            explicit.clone()
388        }
389        None => RustCaptureSettings {
390            width: 0,
391            height: 0,
392            output_mode: 1,
393            capture_cursor: true,
394            target_fps: opts.effective_fps(),
395            ..Default::default()
396        },
397    };
398    s.video_fullframe = true;
399    s.recording_socket = String::new();
400    if opts.fps > 0.0 {
401        s.target_fps = opts.fps;
402    }
403    if opts.bitrate_kbps > 0 {
404        s.video_bitrate_kbps = opts.bitrate_kbps;
405    }
406    if s.keyframe_interval_s <= 0.0 {
407        s.keyframe_interval_s = opts.effective_keyframe_s();
408    }
409    Ok(s)
410}
411
412/// Start a recording. Exactly one may be active per process; returns the initial status.
413pub fn start(opts: RecordOptions) -> Result<RecordingStatus, String> {
414    let mut guard = ACTIVE.lock().unwrap();
415    if guard.is_some() {
416        return Err("a recording is already active".to_string());
417    }
418    if opts.path.is_empty() {
419        return Err("recording path is empty".to_string());
420    }
421    if let Some(cap) = &opts.capture
422        && cap.output_mode != 1 {
423            return Err(
424                "recording requires H.264 capture settings (output_mode=1); JPEG cannot be recorded"
425                    .to_string(),
426            );
427        }
428
429    let wl_tx = crate::computer_use::wayland_command_sender();
430    let backend = match opts.backend {
431        Some(PreferredBackend::X11) => PreferredBackend::X11,
432        Some(PreferredBackend::Wayland) => {
433            if wl_tx.is_none() {
434                return Err("no Wayland compositor is running in this process".to_string());
435            }
436            PreferredBackend::Wayland
437        }
438        None => {
439            if wl_tx.is_some() {
440                PreferredBackend::Wayland
441            } else if std::env::var("DISPLAY").map(|v| !v.is_empty()).unwrap_or(false) {
442                PreferredBackend::X11
443            } else {
444                return Err(
445                    "no capture backend available: no in-process Wayland compositor and DISPLAY is unset"
446                        .to_string(),
447                );
448            }
449        }
450    };
451
452    let file = std::fs::File::create(&opts.path)
453        .map_err(|e| format!("cannot create {}: {e}", opts.path))?;
454    let shared = RecShared::new();
455    let (tx, rx) = bounded::<TapFrame>(QUEUE_CAP);
456
457    let (mode, mode_name, backend_name) = match backend {
458        PreferredBackend::Wayland => {
459            let cmd_tx = wl_tx.unwrap();
460            let display_id = opts.display_id;
461            let attached = crate::wayland_alive().lock().unwrap().contains(&display_id);
462            if attached {
463                *WL_TAP.lock().unwrap() = Some(WlTap {
464                    display_id,
465                    tx: tx.clone(),
466                    shared: shared.clone(),
467                    idr: Some(IdrRequester { tx: cmd_tx.clone(), display_id }),
468                    keyframe_interval_us: (opts.effective_keyframe_s() * 1e6) as u64,
469                });
470                WL_TAP_ARMED.store(true, Ordering::Relaxed);
471                // The stream is mid-GOP: a standard request-IDR gives the file its first
472                // decodable frame promptly.
473                let _ = cmd_tx.send(ThreadCommand::RequestIdr { display_id });
474                (RecordingMode::WaylandAttached, "attached", "wayland")
475            } else {
476                let mut settings = own_capture_settings(&opts)?;
477                if settings.width <= 0 || settings.height <= 0 {
478                    let (reply_tx, reply_rx) = mpsc::channel();
479                    cmd_tx
480                        .send(ThreadCommand::ListOutputs { reply: reply_tx })
481                        .map_err(|_| "wayland compositor is not accepting commands".to_string())?;
482                    let outputs = reply_rx
483                        .recv_timeout(Duration::from_secs(5))
484                        .map_err(|_| "wayland compositor did not report its outputs".to_string())?;
485                    let out = outputs
486                        .iter()
487                        .find(|o| o.0 == display_id)
488                        .ok_or_else(|| format!("no wayland output with display id {display_id}"))?;
489                    settings.width = out.3;
490                    settings.height = out.4;
491                    settings.scale = out.5;
492                }
493                *WL_TAP.lock().unwrap() = Some(WlTap {
494                    display_id,
495                    tx: tx.clone(),
496                    shared: shared.clone(),
497                    idr: None,
498                    keyframe_interval_us: 0,
499                });
500                WL_TAP_ARMED.store(true, Ordering::Relaxed);
501                cmd_tx
502                    .send(ThreadCommand::StartCapture { display_id, callback: None, settings })
503                    .map_err(|_| {
504                        disarm_tap();
505                        "wayland compositor is not accepting commands".to_string()
506                    })?;
507                // StartCapture has no reply; liveness shows up in the alive set.
508                let deadline = Instant::now() + Duration::from_secs(3);
509                loop {
510                    if crate::wayland_alive().lock().unwrap().contains(&display_id) {
511                        break;
512                    }
513                    if Instant::now() >= deadline {
514                        disarm_tap();
515                        return Err(format!(
516                            "wayland capture on display {display_id} failed to start"
517                        ));
518                    }
519                    thread::sleep(Duration::from_millis(20));
520                }
521                (RecordingMode::WaylandOwn { display_id }, "own-capture", "wayland")
522            }
523        }
524        PreferredBackend::X11 => {
525            let settings = own_capture_settings(&opts)?;
526            let controls = Arc::new(crate::x11::Controls::new(&settings));
527            crate::live_x11().lock().unwrap().push(controls.clone());
528            let err_slot: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
529            let err_slot2 = err_slot.clone();
530            let (etid_tx, etid_rx) = mpsc::channel();
531            let cap_controls = controls.clone();
532            let feed_tx = tx.clone();
533            let feed_shared = shared.clone();
534            let join = thread::spawn(move || {
535                let on_frame = move |stripes: Vec<EncodedStripe>| {
536                    offer_frame(&feed_shared, &feed_tx, &stripes);
537                };
538                if let Err(e) = crate::x11::run_capture(settings, cap_controls, etid_tx, on_frame) {
539                    eprintln!("[recorder] x11 capture error: {e}");
540                    *err_slot2.lock().unwrap() = Some(e);
541                }
542            });
543            match etid_rx.recv_timeout(Duration::from_secs(3)) {
544                Ok(_) => {}
545                Err(mpsc::RecvTimeoutError::Disconnected) => {
546                    let _ = join.join();
547                    crate::live_x11().lock().unwrap().retain(|c| !Arc::ptr_eq(c, &controls));
548                    let msg = err_slot
549                        .lock()
550                        .unwrap()
551                        .clone()
552                        .unwrap_or_else(|| "X11 capture exited during start".to_string());
553                    return Err(msg);
554                }
555                // Slow X server setup: the capture is still coming up; proceed.
556                Err(mpsc::RecvTimeoutError::Timeout) => {}
557            }
558            (RecordingMode::X11Own { controls, join }, "own-capture", "x11")
559        }
560    };
561
562    let writer = {
563        let shared = shared.clone();
564        let path = opts.path.clone();
565        thread::Builder::new()
566            .name("pf-recorder".to_string())
567            .spawn(move || writer_thread(rx, file, path, backend_name, mode_name, shared))
568            .map_err(|e| format!("failed to spawn recorder writer thread: {e}"))?
569    };
570
571    println!(
572        "[recorder] recording to {} ({} {})",
573        opts.path, backend_name, mode_name
574    );
575    let status = RecordingStatus {
576        active: true,
577        path: opts.path.clone(),
578        backend: backend_name,
579        mode: mode_name,
580        frames: 0,
581        sync_frames: 0,
582        dropped: 0,
583        skipped_non_h264: 0,
584        bytes: 0,
585        duration_s: 0.0,
586        width: 0,
587        height: 0,
588        error: None,
589    };
590    *guard = Some(ActiveRecording {
591        path: opts.path,
592        backend: backend_name,
593        mode_name,
594        mode,
595        tx,
596        shared,
597        writer,
598    });
599    Ok(status)
600}
601
602fn disarm_tap() {
603    WL_TAP_ARMED.store(false, Ordering::Relaxed);
604    *WL_TAP.lock().unwrap() = None;
605}
606
607/// Stop the active recording, finalize the MP4, and return the final status. Errors when no
608/// recording is active or when nothing recordable was ever received.
609pub fn stop() -> Result<RecordingStatus, String> {
610    let active = ACTIVE
611        .lock()
612        .unwrap()
613        .take()
614        .ok_or_else(|| "no recording is active".to_string())?;
615    disarm_tap();
616    match active.mode {
617        RecordingMode::X11Own { controls, join } => {
618            controls.stop.store(true, Ordering::Relaxed);
619            let _ = join.join();
620            crate::live_x11().lock().unwrap().retain(|c| !Arc::ptr_eq(c, &controls));
621        }
622        RecordingMode::WaylandOwn { display_id } => {
623            // A streaming client that reconfigured this display now owns it; the capture
624            // must survive the recorder's exit.
625            let client_owns = crate::wayland_owners().lock().unwrap().contains_key(&display_id);
626            if !client_owns
627                && let Some(tx) = crate::computer_use::wayland_command_sender() {
628                    let _ = tx.send(ThreadCommand::StopCapture { display_id });
629                    // The next start classifies attached-vs-own against wayland_alive:
630                    // wait for the compositor to drain the stop, or an immediate restart
631                    // attaches to the dying capture and records nothing.
632                    let (ack_tx, ack_rx) = mpsc::channel();
633                    if tx.send(ThreadCommand::Barrier { reply: ack_tx }).is_ok() {
634                        let _ = ack_rx.recv_timeout(Duration::from_secs(2));
635                    }
636                }
637        }
638        RecordingMode::WaylandAttached => {}
639    }
640    drop(active.tx);
641    let _ = active.writer.join();
642    let finished = LAST_FINISHED.lock().unwrap().clone().ok_or_else(|| {
643        format!("recording of {} produced no final status", active.path)
644    })?;
645    match finished.error {
646        Some(ref e) => Err(e.clone()),
647        None => Ok(finished),
648    }
649}
650
651/// The current status: the live recording when one is active, otherwise the last finished
652/// one; `None` when the process has never recorded.
653pub fn status() -> Option<RecordingStatus> {
654    let guard = ACTIVE.lock().unwrap();
655    if let Some(a) = guard.as_ref() {
656        return Some(RecordingStatus {
657            active: true,
658            path: a.path.clone(),
659            backend: a.backend,
660            mode: a.mode_name,
661            frames: a.shared.muxed.load(Ordering::Relaxed),
662            sync_frames: a.shared.sync_frames.load(Ordering::Relaxed),
663            dropped: a.shared.dropped.load(Ordering::Relaxed),
664            skipped_non_h264: a.shared.skipped_non_h264.load(Ordering::Relaxed),
665            bytes: a.shared.bytes.load(Ordering::Relaxed),
666            duration_s: a.shared.start.elapsed().as_secs_f64(),
667            width: a.shared.width.load(Ordering::Relaxed),
668            height: a.shared.height.load(Ordering::Relaxed),
669            error: a.shared.error.lock().unwrap().clone(),
670        });
671    }
672    drop(guard);
673    LAST_FINISHED.lock().unwrap().clone()
674}
675
676/// Finalize any active recording at interpreter shutdown so the MP4's last buffered sample
677/// is flushed. Best-effort; the fragmented layout keeps even an unflushed file playable.
678pub fn finalize_on_exit() {
679    if ACTIVE.lock().unwrap().is_some() {
680        let _ = stop();
681    }
682}
683
684/// `PIXELFLUX_RECORD=<path>`: record from process start. X11 (via `DISPLAY`) starts
685/// immediately; otherwise a background thread waits for the in-process Wayland compositor to
686/// come up and then starts. Retries transient failures until the deadline.
687pub fn autostart_from_env() {
688    use std::sync::OnceLock;
689    static STARTED: OnceLock<()> = OnceLock::new();
690    let Ok(path) = std::env::var("PIXELFLUX_RECORD") else { return };
691    if path.is_empty() {
692        return;
693    }
694    let mut first = false;
695    STARTED.get_or_init(|| first = true);
696    if !first {
697        return;
698    }
699    thread::spawn(move || {
700        let opts = RecordOptions::from_env(path);
701        let deadline = Instant::now() + Duration::from_secs(300);
702        let mut last_err = String::new();
703        loop {
704            let wayland_up = crate::computer_use::wayland_command_sender().is_some();
705            let x11_up = std::env::var("DISPLAY").map(|v| !v.is_empty()).unwrap_or(false);
706            let ready = match opts.backend {
707                Some(PreferredBackend::X11) => x11_up,
708                Some(PreferredBackend::Wayland) => wayland_up,
709                None => wayland_up || x11_up,
710            };
711            if ready {
712                match start(opts.clone()) {
713                    Ok(_) => return,
714                    Err(e) => last_err = e,
715                }
716            }
717            if Instant::now() >= deadline {
718                eprintln!(
719                    "[recorder] PIXELFLUX_RECORD autostart gave up: {}",
720                    if last_err.is_empty() { "no capture backend appeared" } else { &last_err }
721                );
722                return;
723            }
724            thread::sleep(Duration::from_millis(250));
725        }
726    });
727}
728
729/// Serialize a status into the JSON shape shared by the REST endpoints.
730pub fn status_to_json(s: &RecordingStatus) -> serde_json::Value {
731    serde_json::json!({
732        "active": s.active,
733        "path": s.path,
734        "backend": s.backend,
735        "mode": s.mode,
736        "frames": s.frames,
737        "sync_frames": s.sync_frames,
738        "dropped": s.dropped,
739        "skipped_non_h264": s.skipped_non_h264,
740        "bytes": s.bytes,
741        "duration_s": s.duration_s,
742        "width": s.width,
743        "height": s.height,
744        "error": s.error,
745    })
746}
pixelflux/recorder/mod.rs

pixelflux/recorder/
mod.rs
