feat: add watch-later views and Terebi watched-state tracking

README.md

@@ -7,7 +7,7 @@ A modal terminal YouTube browser built with Rust and [Ratatui](https://ratatui.r
 - A recent stable Rust toolchain, including Cargo, to build and install.
 - [`yt-dlp`](https://github.com/yt-dlp/yt-dlp) on `PATH` for searches, feeds, and full descriptions.
 - `curl` on `PATH` for thumbnails.
-- The **Terebi CLI** installed on `PATH`, together with its app bundle and streaming dependencies, for playback.
+- The **Terebi CLI** installed on `PATH`, together with its app bundle and streaming dependencies, for playback. It must support `--new-instance --url … --progress-file PATH`.
 
 The default playback setup requires **macOS 15+**, since Terebi is a native macOS player. Install Terebi separately and follow its dependency/setup instructions; installing Tubular does not install the player.
 
@@ -43,10 +43,11 @@ Tubular starts in **normal mode**, with or without a command-line query. Press `
 Selecting a video launches:
 
 ```sh
-terebi --new-instance --url 'https://www.youtube.com/watch?v=VIDEO_ID'
+terebi --new-instance --url 'https://www.youtube.com/watch?v=VIDEO_ID' \
+  --progress-file '/private/session/progress.jsonl'
 ```
 
-Each selection opens an independent player instance without replacing existing playback. `--player` overrides the Terebi CLI executable; it is not a generic mpv/IINA adapter. Tubular reports launcher failures, while later streaming/playback errors appear in Terebi. Quitting Tubular leaves launched players running.
+Each selection opens an independent player instance without replacing existing playback. Tubular creates a private reporting session and consumes Terebi's progress records independently of the launcher. `--player` overrides the Terebi CLI executable; it is not a generic mpv/IINA adapter. Tubular reports launcher failures, while later streaming/playback errors appear in Terebi. Quitting Tubular leaves launched players running.
 
 ## Controls
 
@@ -69,6 +70,10 @@ These are the default bindings. Overrides in `[keys]` replace an action's bindin
 | `U` (Shift-U) | Open the selected video's channel |
 | `s` | Edit the selected channel's tags |
 | `S` | Edit tags for all saved channels in the current category |
+| `w` | Save the selected video to watch later (also works in previews) |
+| `W` | Browse watch later for the current category, or the combined collection elsewhere |
+| `f` | In watch later, cycle All → Unwatched → Watched |
+| `D` | In watch later, confirm removal of all videos matching the current list/filter |
 | Tab / Shift-Tab | Switch between visible views |
 | `r` | Reload configuration and refresh the current view |
 | `?` | Open / close the scrollable keybinding reference |
@@ -157,12 +162,40 @@ Category entries accept `@handles`, HTTPS YouTube channel URLs, or canonical `UC
 Video tables color channel names **cyan** when the creator belongs to at least one category. Untagged registrations keep the default text color; selected rows retain their black-on-cyan highlight. Membership is matched by channel ID or handle, not the displayed name, and updates after saving tags or reloading config.
 
 - **Search** and **Feed** are permanent tabs. Feed combines all `[channels]` registrations, including untagged channels, plus channels entered directly in categories, without duplicates.
-- Browsing an unpinned category opens one temporary tab; choosing another replaces it. Pinned categories keep their own tabs, marked with a dimmed `p`.
+- Browsing an unpinned category, channel, or watch-later list uses one shared temporary tab; choosing another replaces it. Pinned categories keep their own tabs, marked with a dimmed `p`.
 - Removing a channel from a category does **not** unsubscribe it from Feed. To unsubscribe, remove its `[channels]` mappings and category references, then press `r`.
 - Temporary tabs do not survive restarting; saved categories and pins do. Hidden views retain loaded rows, selection, and pagination during the session.
 
 Bulk editing refuses to clear all tags when a target has no `[channels]` registration, preventing accidental unsubscription of category-only channels. If source memberships change while a bulk dialog is open, close and reopen it before saving.
 
+## Watch later
+
+Watch later is **one collection**, deduplicated by YouTube video ID. Category lists are derived from the saved videos' channel identities and the **current** subscription categories: adding or removing a channel from a category changes its watch-later list after saving tags or reloading config with `r`. No category memberships are stored on individual videos.
+
+Browse a category with `c`, then use `W` for its watch-later list. From Search, Feed, or a channel view, `W` opens the combined collection. Watch-later views use saved title, creator, and publication metadata without fetching video lists, so they also work offline. Their filters are independent and retained during the session.
+
+Saved video titles are **magenta** in all video tables; categorized channel names remain cyan. A `✓` marks watched videos, including outside watch-later views. The selected row retains its usual highlight.
+
+### Watched status
+
+Terebi reports acknowledged playback positions. A video becomes watched when an eligible position reaches **95% of a known positive duration**. Seeking near the end counts; live/non-seekable, unknown-duration, and audio-only inputs do not. Launcher exit or window closure alone never marks a video watched. Watched status is global across categories and remains after removing a video from watch later.
+
+### Removal
+
+`D` snapshots all videos matching the current list and filter, then shows their count, category, and filter. **`y` confirms; `n` or Esc cancels.** Enter does not confirm. Removal deletes those videos from the shared collection, so they disappear from every matching category; watched status is untouched. An empty filtered list has nothing to remove.
+
+### Storage
+
+Files are siblings of `config.toml`:
+
+- `watch-later.toml`: versioned saved-video metadata.
+- `watch-state.toml`: versioned global watched video IDs.
+- `playback-sessions/`: private per-launch progress streams and their expected video identities.
+
+State writes are atomic and owner-only. Invalid state files or detected external edits are reported rather than overwritten; repair the file or restart before retrying. Each state file is limited to 8 MiB.
+
+Progress sessions survive Tubular exit so later player reports can be recovered on the next launch. They are retained because the GUI may still be writing; up to 4,096 sessions are monitored. If that limit is reached, close the corresponding players, remove their inactive session directories, and restart Tubular. Never remove an active player's reporting directory.
+
 ## Configurable keybindings
 
 Print a complete default configuration, including every action and its default bindings:

src/keys.rs

@@ -105,6 +105,10 @@ actions! {
     VideoChannel, "video.channel", "Open the selected video's channel", VIEW, ["U"];
     ChannelTags, "channel.tags", "Edit the selected channel's category tags", VIEW, ["s"];
     CategoryTags, "category.tags", "Edit category tags for all channels in this category", VIEW, ["S"];
+    WatchLaterSave, "video.watch_later.save", "Save the selected video to watch later", VIEW, ["w"];
+    WatchLaterBrowse, "view.watch_later.browse", "Browse this category's watch later, or the combined collection elsewhere", &[Context::Normal], ["W"];
+    WatchLaterFilter, "view.watch_later.filter", "Cycle watch-later filter: all, unwatched, watched", &[Context::Normal], ["f"];
+    WatchLaterRemove, "view.watch_later.remove", "Confirm removal of the current filtered watch-later subset", &[Context::Normal], ["D"];
     SelectorNext, "selector.select.next", "Select the next category or channel", &[Context::SelectorNormal], ["j", "down"];
     SelectorPrevious, "selector.select.previous", "Select the previous category or channel", &[Context::SelectorNormal], ["k", "up"];
     SelectorInputNext, "selector.input.next", "Select the next match while editing the filter", &[Context::SelectorInput], ["down", "ctrl+j"];
@@ -458,6 +462,21 @@ impl Bindings {
                     (&[Quit, Interrupt], "Quit"),
                 ],
             ),
+            (
+                "WATCH LATER",
+                &[
+                    (&[WatchLaterSave], "Save the selected video"),
+                    (
+                        &[WatchLaterBrowse],
+                        "Browse category watch later; combined elsewhere",
+                    ),
+                    (&[WatchLaterFilter], "Cycle all / unwatched / watched"),
+                    (
+                        &[WatchLaterRemove],
+                        "Confirm removal of filtered subset (combined: all matches)",
+                    ),
+                ],
+            ),
             (
                 "FULL DESCRIPTION",
                 &[
@@ -948,6 +967,86 @@ mod tests {
         );
     }
 
+    #[test]
+    fn watch_later_overrides_disable_and_respect_contexts() {
+        let defaults = Bindings::default();
+        for (key, action) in [
+            ('w', Action::WatchLaterBrowse),
+            ('d', Action::WatchLaterRemove),
+        ] {
+            assert_eq!(
+                press(
+                    &defaults,
+                    Context::Normal,
+                    KeyCode::Char(key),
+                    KeyModifiers::SHIFT
+                ),
+                Some(action),
+            );
+        }
+        assert_eq!(
+            press(
+                &defaults,
+                Context::Info,
+                KeyCode::Char('w'),
+                KeyModifiers::NONE
+            ),
+            Some(Action::WatchLaterSave),
+        );
+        assert_eq!(
+            press(
+                &defaults,
+                Context::Info,
+                KeyCode::Char('d'),
+                KeyModifiers::NONE
+            ),
+            Some(Action::InfoDescription),
+        );
+        let bindings = configured(
+            "[keys.video.watch_later]\nsave = 'alt+w'\n[keys.view.watch_later]\nbrowse = 'shift+x'\nfilter = 'ctrl+f'\nremove = []\n",
+        )
+        .unwrap();
+        for context in ALL {
+            assert_eq!(
+                press(&bindings, *context, KeyCode::Char('w'), KeyModifiers::ALT),
+                VIEW.contains(context).then_some(Action::WatchLaterSave),
+            );
+            for (code, modifiers, action) in [
+                (
+                    KeyCode::Char('x'),
+                    KeyModifiers::SHIFT,
+                    Action::WatchLaterBrowse,
+                ),
+                (
+                    KeyCode::Char('f'),
+                    KeyModifiers::CONTROL,
+                    Action::WatchLaterFilter,
+                ),
+            ] {
+                assert_eq!(
+                    press(&bindings, *context, code, modifiers),
+                    (*context == Context::Normal).then_some(action),
+                );
+            }
+            for key in ['w', 'W', 'f', 'D'] {
+                assert_eq!(
+                    press(&bindings, *context, KeyCode::Char(key), KeyModifiers::NONE),
+                    None,
+                );
+            }
+        }
+        let help = bindings.help_text().to_string();
+        assert!(help.contains("Alt+w") && help.contains("X") && help.contains("Ctrl+f"));
+        assert!(help.contains("unbound") && help.contains("WATCH LATER"));
+        let collision = configured("[keys]\nview.watch_later.browse = 'w'")
+            .unwrap_err()
+            .to_string();
+        assert!(
+            collision.contains("view.watch_later.browse")
+                && collision.contains("video.watch_later.save")
+        );
+    }
+
     #[test]
     fn invalid_entries_and_collisions_have_actionable_errors() {
         for (source, path) in [

src/watch_later.rs

@@ -0,0 +1,898 @@
+use std::{
+    collections::BTreeSet,
+    fs::{self, File, OpenOptions, ReadDir},
+    io::{self, Read, Write},
+    path::{Path, PathBuf},
+    sync::atomic::{AtomicU64, Ordering},
+};
+
+use serde::Deserialize;
+use toml_edit::{Array, ArrayOfTables, DocumentMut, Table, value};
+
+use crate::youtube::{Video, valid_video_id, validate_video};
+
+const MAX_STATE_BYTES: u64 = 8 * 1024 * 1024;
+const MAX_LINE: usize = 16 * 1024;
+const STREAM_BYTES: usize = 16 * 1024;
+const MAX_STREAMS: usize = 4096;
+static NEXT_NAME: AtomicU64 = AtomicU64::new(0);
+
+fn invalid(message: impl Into<String>) -> io::Error {
+    io::Error::new(io::ErrorKind::InvalidData, message.into())
+}
+
+struct StoredFile {
+    path: PathBuf,
+    original: Option<Vec<u8>>,
+}
+
+impl StoredFile {
+    fn open(path: PathBuf) -> io::Result<(Self, DocumentMut)> {
+        let original = read_bytes(&path, MAX_STATE_BYTES)?;
+        let document = match &original {
+            Some(bytes) => std::str::from_utf8(bytes)
+                .map_err(|e| invalid(format!("{} is not UTF-8: {e}", path.display())))?
+                .parse::<DocumentMut>()
+                .map_err(|e| {
+                    invalid(format!(
+                        "Invalid {}: {e}; repair this file before retrying",
+                        path.display()
+                    ))
+                })?,
+            None => DocumentMut::new(),
+        };
+        if original.is_some() && document.get("version").and_then(|v| v.as_integer()) != Some(1) {
+            return Err(invalid(format!("{} requires version = 1", path.display())));
+        }
+        Ok((Self { path, original }, document))
+    }
+
+    fn save(&mut self, document: DocumentMut) -> io::Result<()> {
+        let contents = document.to_string().into_bytes();
+        atomic_write(&self.path, self.original.as_deref(), &contents)?;
+        self.original = Some(contents);
+        Ok(())
+    }
+}
+
+pub struct State {
+    collection: StoredFile,
+    watch_state: StoredFile,
+    videos: Vec<Video>,
+    saved: BTreeSet<String>,
+    watched: BTreeSet<String>,
+    pending: BTreeSet<String>,
+    sessions: PathBuf,
+    streams: Vec<ProgressStream>,
+    discovery: Option<ReadDir>,
+    next_stream: usize,
+    notify_changed: bool,
+}
+
+impl State {
+    pub fn open(config_path: &Path) -> io::Result<Self> {
+        let parent = config_path
+            .parent()
+            .filter(|p| !p.as_os_str().is_empty())
+            .unwrap_or(Path::new("."));
+        let (collection, document) = StoredFile::open(parent.join("watch-later.toml"))?;
+        let (videos, saved) = load_videos(&document)
+            .map_err(|e| invalid(format!("{}: {e}", collection.path.display())))?;
+        let (watch_state, document) = StoredFile::open(parent.join("watch-state.toml"))?;
+        let watched = load_watched(&document)
+            .map_err(|e| invalid(format!("{}: {e}", watch_state.path.display())))?;
+        Ok(Self {
+            collection,
+            watch_state,
+            videos,
+            saved,
+            watched,
+            pending: BTreeSet::new(),
+            sessions: parent.join("playback-sessions"),
+            streams: Vec::new(),
+            discovery: None,
+            next_stream: 0,
+            notify_changed: false,
+        })
+    }
+
+    pub fn videos(&self) -> &[Video] {
+        &self.videos
+    }
+    pub fn contains(&self, id: &str) -> bool {
+        self.saved.contains(id)
+    }
+    pub fn is_watched(&self, id: &str) -> bool {
+        self.watched.contains(id)
+    }
+
+    pub fn save(&mut self, video: &Video) -> io::Result<()> {
+        if self.contains(&video.id) {
+            return Ok(());
+        }
+        let mut video = video.clone();
+        validate_saved(&mut video)?;
+        self.collection.save(video_document(
+            self.videos.iter().chain(std::iter::once(&video)),
+        ))?;
+        self.saved.insert(video.id.clone());
+        self.videos.push(video);
+        Ok(())
+    }
+
+    pub fn remove(&mut self, ids: &[String]) -> io::Result<()> {
+        if !ids.iter().any(|id| self.saved.contains(id)) {
+            return Ok(());
+        }
+        self.collection.save(video_document(
+            self.videos.iter().filter(|v| !ids.contains(&v.id)),
+        ))?;
+        self.videos.retain(|v| !ids.contains(&v.id));
+        self.saved.retain(|id| !ids.contains(id));
+        Ok(())
+    }
+
+    pub fn progress_file(&mut self) -> io::Result<PathBuf> {
+        if self.streams.len() >= MAX_STREAMS {
+            return Err(invalid(
+                "Playback session limit reached; remove inactive playback-sessions directories and restart",
+            ));
+        }
+        fs::create_dir_all(self.sessions.parent().unwrap_or(Path::new(".")))?;
+        private_directory(&self.sessions, true)?;
+        let directory = loop {
+            let path = self.sessions.join(format!(
+                "session-{}-{}",
+                std::process::id(),
+                NEXT_NAME.fetch_add(1, Ordering::Relaxed)
+            ));
+            match create_private_directory(&path) {
+                Ok(()) => break path,
+                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
+                Err(e) => return Err(e),
+            }
+        };
+        let path = directory.join("progress.jsonl");
+        self.streams.push(ProgressStream::new(path.clone()));
+        Ok(path)
+    }
+
+    pub fn poll_progress(&mut self) -> io::Result<bool> {
+        let mut error = self.discover().err();
+        let count = self.streams.len().min(16);
+        for _ in 0..count {
+            if self.next_stream >= self.streams.len() {
+                self.next_stream = 0;
+            }
+            let stream = &mut self.streams[self.next_stream];
+            self.next_stream += 1;
+            if let Err(e) = stream.poll(&mut self.pending, &self.watched)
+                && error.is_none()
+            {
+                error = Some(e);
+            }
+        }
+        self.pending.retain(|id| !self.watched.contains(id));
+        if !self.pending.is_empty() {
+            self.watch_state.save(watched_document(
+                self.watched.iter().chain(self.pending.iter()),
+            ))?;
+            self.watched.append(&mut self.pending);
+            self.notify_changed = true;
+        }
+        // Preserve notification after reporting a bad record; siblings were still processed.
+        if let Some(error) = error {
+            return Err(error);
+        }
+        Ok(std::mem::take(&mut self.notify_changed))
+    }
+
+    fn discover(&mut self) -> io::Result<()> {
+        match private_directory(&self.sessions, false) {
+            Ok(()) => {}
+            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
+            Err(e) => return Err(e),
+        }
+        if self.discovery.is_none() {
+            self.discovery = Some(fs::read_dir(&self.sessions)?);
+        }
+        for _ in 0..32 {
+            let Some(entry) = self.discovery.as_mut().unwrap().next() else {
+                self.discovery = None;
+                break;
+            };
+            let entry = entry?;
+            if !entry.file_name().to_string_lossy().starts_with("session-") {
+                continue;
+            }
+            let path = entry.path().join("progress.jsonl");
+            if self.streams.iter().any(|stream| stream.path == path) {
+                continue;
+            }
+            if private_directory(&entry.path(), false).is_err() {
+                continue;
+            }
+            // Failed launchers leave empty private directories; only bound identities are discoverable.
+            if !matches!(fs::symlink_metadata(entry.path().join("video-id")), Ok(m) if m.is_file())
+            {
+                continue;
+            }
+            if self.streams.len() >= MAX_STREAMS {
+                break;
+            }
+            self.streams.push(ProgressStream::new(path));
+        }
+        // ponytail: retain reports for live GUI writers, capped at 4096 monitored sessions;
+        // add explicit inactive-session archival if long-lived installations hit this ceiling.
+        Ok(())
+    }
+}
+
+fn validate_saved(video: &mut Video) -> io::Result<()> {
+    let date = video.publication_date.clone();
+    if let Some(date) = &date {
+        let raw = date.strip_prefix('~').unwrap_or(date);
+        let compact = raw.replace('-', "");
+        if !(raw.len() == 10 && raw.as_bytes()[4] == b'-' && raw.as_bytes()[7] == b'-')
+            || compact.len() != 8
+            || !compact.bytes().all(|b| b.is_ascii_digit())
+            || time::Date::from_calendar_date(
+                compact[..4]
+                    .parse()
+                    .map_err(|_| invalid("Invalid saved publication date"))?,
+                compact[4..6]
+                    .parse::<u8>()
+                    .ok()
+                    .and_then(|m| m.try_into().ok())
+                    .ok_or_else(|| invalid("Invalid saved publication month"))?,
+                compact[6..]
+                    .parse()
+                    .map_err(|_| invalid("Invalid saved publication day"))?,
+            )
+            .is_err()
+        {
+            return Err(invalid(
+                "Saved publication date must be a valid YYYY-MM-DD or ~YYYY-MM-DD",
+            ));
+        }
+    }
+    validate_video(video).map_err(invalid)?;
+    video.publication_date = date;
+    if let Some(handle) = &video.uploader_id
+        && (!handle.starts_with('@') || crate::config::normalize_channel(handle).is_err())
+    {
+        return Err(invalid("Saved uploader_id must be a valid YouTube @handle"));
+    }
+    if video.channel_id.is_none() && video.uploader_id.is_none() {
+        return Err(invalid(
+            "Video has no channel identity; refresh its metadata before saving",
+        ));
+    }
+    Ok(())
+}
+
+fn load_videos(document: &DocumentMut) -> io::Result<(Vec<Video>, BTreeSet<String>)> {
+    if document
+        .iter()
+        .any(|(key, _)| !matches!(key, "version" | "videos"))
+    {
+        return Err(invalid("Unknown watch-later field"));
+    }
+    let Some(item) = document.get("videos") else {
+        return Ok((Vec::new(), BTreeSet::new()));
+    };
+    if item.as_array().is_some_and(|a| a.is_empty()) {
+        return Ok((Vec::new(), BTreeSet::new()));
+    }
+    let tables = item
+        .as_array_of_tables()
+        .ok_or_else(|| invalid("videos must be an array of tables"))?;
+    let mut videos = Vec::new();
+    let mut ids = BTreeSet::new();
+    for table in tables {
+        let mut fields = serde_json::Map::new();
+        for (key, item) in table.iter() {
+            let field = match key {
+                "id" | "title" | "channel" | "channel_id" | "uploader_id" | "upload_date" => {
+                    serde_json::Value::String(
+                        item.as_str()
+                            .ok_or_else(|| invalid(format!("Video {key} must be a string")))?
+                            .into(),
+                    )
+                }
+                "timestamp" => serde_json::Value::from(
+                    item.as_integer()
+                        .ok_or_else(|| invalid("Video timestamp must be an integer"))?,
+                ),
+                _ => return Err(invalid(format!("Unknown saved video field {key:?}"))),
+            };
+            fields.insert(key.into(), field);
+        }
+        let mut video: Video = serde_json::from_value(serde_json::Value::Object(fields))
+            .map_err(|e| invalid(format!("Invalid saved video: {e}")))?;
+        validate_saved(&mut video)?;
+        if ids.insert(video.id.clone()) {
+            videos.push(video);
+        }
+    }
+    Ok((videos, ids))
+}
+
+fn load_watched(document: &DocumentMut) -> io::Result<BTreeSet<String>> {
+    if document
+        .iter()
+        .any(|(key, _)| !matches!(key, "version" | "watched"))
+    {
+        return Err(invalid("Unknown watch-state field"));
+    }
+    let Some(item) = document.get("watched") else {
+        return Ok(BTreeSet::new());
+    };
+    item.as_array()
+        .ok_or_else(|| invalid("watched must be an array of YouTube IDs"))?
+        .iter()
+        .map(|v| {
+            let id = v
+                .as_str()
+                .filter(|id| valid_video_id(id))
+                .ok_or_else(|| invalid("Invalid watched YouTube ID"))?;
+            Ok(id.to_owned())
+        })
+        .collect()
+}
+
+fn video_document<'a>(videos: impl Iterator<Item = &'a Video>) -> DocumentMut {
+    let mut document = DocumentMut::new();
+    document["version"] = value(1);
+    let mut tables = ArrayOfTables::new();
+    for video in videos {
+        let mut table = Table::new();
+        table["id"] = value(video.id.as_str());
+        table["title"] = value(video.title.as_str());
+        for (key, field) in [
+            ("channel", &video.channel),
+            ("channel_id", &video.channel_id),
+            ("uploader_id", &video.uploader_id),
+            ("upload_date", &video.publication_date),
+        ] {
+            if let Some(field) = field {
+                table[key] = value(field.as_str());
+            }
+        }
+        if let Some(timestamp) = video.timestamp {
+            table["timestamp"] = value(timestamp);
+        }
+        tables.push(table);
+    }
+    document["videos"] = if tables.is_empty() {
+        value(Array::new())
+    } else {
+        toml_edit::Item::ArrayOfTables(tables)
+    };
+    document
+}
+
+fn watched_document<'a>(watched: impl Iterator<Item = &'a String>) -> DocumentMut {
+    let mut document = DocumentMut::new();
+    document["version"] = value(1);
+    let mut ids = Array::new();
+    for id in watched {
+        ids.push(id.as_str());
+    }
+    document["watched"] = value(ids);
+    document
+}
+
+fn regular_file(path: &Path, private: bool) -> io::Result<File> {
+    let mut options = OpenOptions::new();
+    options.read(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK);
+    }
+    let file = options.open(path)?;
+    let metadata = file.metadata()?;
+    if !metadata.is_file() {
+        return Err(invalid(format!(
+            "{} must be a regular file",
+            path.display()
+        )));
+    }
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::MetadataExt;
+        // SAFETY: geteuid has no preconditions.
+        if metadata.uid() != unsafe { libc::geteuid() } || (private && metadata.mode() & 0o077 != 0)
+        {
+            return Err(invalid(format!(
+                "{} must be owned by you with owner-only permissions",
+                path.display()
+            )));
+        }
+    }
+    Ok(file)
+}
+
+fn read_bytes(path: &Path, limit: u64) -> io::Result<Option<Vec<u8>>> {
+    let file = match regular_file(path, false) {
+        Ok(file) => file,
+        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
+        Err(e) => return Err(e),
+    };
+    let mut bytes = Vec::new();
+    file.take(limit + 1).read_to_end(&mut bytes)?;
+    if bytes.len() as u64 > limit {
+        return Err(invalid(format!(
+            "{} exceeds the {} byte limit",
+            path.display(),
+            limit
+        )));
+    }
+    Ok(Some(bytes))
+}
+
+fn atomic_write(path: &Path, original: Option<&[u8]>, contents: &[u8]) -> io::Result<()> {
+    if contents.len() as u64 > MAX_STATE_BYTES {
+        return Err(invalid("Watch state exceeds the 8 MiB limit"));
+    }
+    let parent = path.parent().unwrap_or(Path::new("."));
+    fs::create_dir_all(parent)?;
+    let (temporary, mut file) = loop {
+        let path = parent.join(format!(
+            ".watch-{}-{}.tmp",
+            std::process::id(),
+            NEXT_NAME.fetch_add(1, Ordering::Relaxed)
+        ));
+        let mut options = OpenOptions::new();
+        options.write(true).create_new(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            options.mode(0o600);
+        }
+        match options.open(&path) {
+            Ok(file) => break (path, file),
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
+            Err(e) => return Err(e),
+        }
+    };
+    let result = (|| {
+        file.write_all(contents)?;
+        file.sync_all()?;
+        if read_bytes(path, MAX_STATE_BYTES)?.as_deref() != original {
+            return Err(io::Error::other(format!(
+                "{} changed on disk; restart Tubular before retrying (no changes written)",
+                path.display()
+            )));
+        }
+        if original.is_none() {
+            fs::hard_link(&temporary, path)?;
+        } else {
+            fs::rename(&temporary, path)?;
+        }
+        Ok(())
+    })();
+    drop(file);
+    let _ = fs::remove_file(temporary);
+    result
+}
+
+fn create_private_directory(path: &Path) -> io::Result<()> {
+    let mut builder = fs::DirBuilder::new();
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::DirBuilderExt;
+        builder.mode(0o700);
+    }
+    builder.create(path)
+}
+
+fn private_directory(path: &Path, create: bool) -> io::Result<()> {
+    if create {
+        match create_private_directory(path) {
+            Ok(()) => {}
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
+            Err(e) => return Err(e),
+        }
+    }
+    let metadata = fs::symlink_metadata(path)?;
+    if !metadata.is_dir() {
+        return Err(invalid(format!(
+            "{} must be a plain private directory, not a symlink",
+            path.display()
+        )));
+    }
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::MetadataExt;
+        // SAFETY: geteuid has no preconditions.
+        if metadata.uid() != unsafe { libc::geteuid() } || metadata.mode() & 0o077 != 0 {
+            return Err(invalid(format!(
+                "{} must be owned by you with permissions 0700",
+                path.display()
+            )));
+        }
+    }
+    Ok(())
+}
+
+pub(crate) fn register_identity(progress_file: &Path, id: &str) -> io::Result<()> {
+    if !valid_video_id(id)
+        || progress_file
+            .file_name()
+            .is_none_or(|name| name != "progress.jsonl")
+    {
+        return Err(invalid(
+            "Invalid playback reporting identity or destination",
+        ));
+    }
+    let directory = progress_file
+        .parent()
+        .ok_or_else(|| invalid("Missing private session directory"))?;
+    private_directory(directory, false)?;
+    private_directory(
+        directory
+            .parent()
+            .ok_or_else(|| invalid("Missing playback-sessions directory"))?,
+        false,
+    )?;
+    match fs::symlink_metadata(progress_file) {
+        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
+        Ok(_) => {
+            return Err(invalid(
+                "Playback report destination must not already exist",
+            ));
+        }
+        Err(e) => return Err(e),
+    }
+    atomic_write(&directory.join("video-id"), None, id.as_bytes())
+}
+
+#[derive(Deserialize)]
+struct ProgressRecord {
+    version: u32,
+    event: String,
+    media_url: String,
+    position_seconds: f64,
+    duration_seconds: Option<f64>,
+    completion_eligible: bool,
+}
+
+struct ProgressStream {
+    path: PathBuf,
+    id: Option<String>,
+    file: Option<File>,
+    partial: Vec<u8>,
+    discarding: bool,
+    failed: bool,
+    line: u64,
+}
+
+impl ProgressStream {
+    fn new(path: PathBuf) -> Self {
+        Self {
+            path,
+            id: None,
+            file: None,
+            partial: Vec::new(),
+            discarding: false,
+            failed: false,
+            line: 0,
+        }
+    }
+
+    fn poll(
+        &mut self,
+        pending: &mut BTreeSet<String>,
+        watched: &BTreeSet<String>,
+    ) -> io::Result<()> {
+        if self.failed {
+            return Ok(());
+        }
+        match self.read(pending, watched) {
+            Ok(error) => match error {
+                Some(e) => Err(e),
+                None => Ok(()),
+            },
+            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
+            Err(e) => {
+                self.failed = true;
+                Err(invalid(format!("Cannot read {}: {e}", self.path.display())))
+            }
+        }
+    }
+
+    fn read(
+        &mut self,
+        pending: &mut BTreeSet<String>,
+        watched: &BTreeSet<String>,
+    ) -> io::Result<Option<io::Error>> {
+        if self.file.is_none() {
+            private_directory(self.path.parent().unwrap(), false)?;
+            let marker = self.path.with_file_name("video-id");
+            let mut bytes = Vec::new();
+            regular_file(&marker, true)?
+                .take(65)
+                .read_to_end(&mut bytes)?;
+            let id = std::str::from_utf8(&bytes)
+                .ok()
+                .filter(|id| valid_video_id(id))
+                .ok_or_else(|| invalid("Invalid video-id session marker"))?;
+            self.id = Some(id.into());
+            self.file = Some(regular_file(&self.path, true)?);
+        }
+        let mut buffer = [0; STREAM_BYTES];
+        let size = self.file.as_mut().unwrap().read(&mut buffer)?;
+        let mut error = None;
+        for &byte in &buffer[..size] {
+            if byte == b'\n' {
+                self.line += 1;
+                if !self.discarding
+                    && let Err(e) = self.record(pending, watched)
+                    && error.is_none()
+                {
+                    error = Some(e);
+                }
+                self.partial.clear();
+                self.discarding = false;
+            } else if !self.discarding {
+                if self.partial.len() == MAX_LINE {
+                    self.partial.clear();
+                    self.discarding = true;
+                    if error.is_none() {
+                        error = Some(self.record_error("line exceeds 16 KiB"));
+                    }
+                } else {
+                    self.partial.push(byte);
+                }
+            }
+        }
+        Ok(error)
+    }
+
+    fn record_error(&self, detail: impl std::fmt::Display) -> io::Error {
+        invalid(format!(
+            "Invalid playback report {} line {}: {detail}; inspect this record (other sessions continue)",
+            self.path.display(),
+            self.line.max(1)
+        ))
+    }
+
+    fn record(&self, pending: &mut BTreeSet<String>, watched: &BTreeSet<String>) -> io::Result<()> {
+        let record: ProgressRecord =
+            serde_json::from_slice(&self.partial).map_err(|e| self.record_error(e))?;
+        if record.version != 1 || !matches!(record.event.as_str(), "progress" | "ended" | "closed")
+        {
+            return Err(
+                self.record_error("expected version 1 and progress, ended, or closed event")
+            );
+        }
+        let id = record
+            .media_url
+            .strip_prefix("https://www.youtube.com/watch?v=")
+            .filter(|id| valid_video_id(id))
+            .ok_or_else(|| self.record_error("expected canonical YouTube video URL"))?;
+        if !record.position_seconds.is_finite()
+            || record.position_seconds < 0.0
+            || record
+                .duration_seconds
+                .is_some_and(|d| !d.is_finite() || d <= 0.0)
+        {
+            return Err(self.record_error(
+                "position must be finite/nonnegative and duration positive/finite or null",
+            ));
+        }
+        if self.id.as_deref() == Some(id)
+            && record.completion_eligible
+            && record
+                .duration_seconds
+                .is_some_and(|d| record.position_seconds / d >= 0.95)
+            && !watched.contains(id)
+            && !pending.contains(id)
+        {
+            pending.insert(id.into());
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct TestDirectory(PathBuf);
+    impl TestDirectory {
+        fn new() -> Self {
+            loop {
+                let path = std::env::temp_dir().join(format!(
+                    "tubular-watch-test-{}-{}",
+                    std::process::id(),
+                    NEXT_NAME.fetch_add(1, Ordering::Relaxed)
+                ));
+                match create_private_directory(&path) {
+                    Ok(()) => return Self(path),
+                    Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,
+                    Err(e) => panic!("{e}"),
+                }
+            }
+        }
+        fn config(&self) -> PathBuf {
+            self.0.join("config.toml")
+        }
+    }
+    impl Drop for TestDirectory {
+        fn drop(&mut self) {
+            let _ = fs::remove_dir_all(&self.0);
+        }
+    }
+
+    fn video() -> Video {
+        Video {
+            id: "jNQXAC9IVRw".into(),
+            title: "Me at the zoo".into(),
+            channel: Some("jawed".into()),
+            channel_id: Some("UC4QobU6STFB0P71PMvOGN5A".into()),
+            uploader_id: Some("@jawed".into()),
+            timestamp: Some(1_114_300_000),
+            publication_date: Some("2005-04-23".into()),
+        }
+    }
+
+    fn append(path: &Path, bytes: &[u8]) {
+        let mut options = OpenOptions::new();
+        options.write(true).append(true).create(true);
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::OpenOptionsExt;
+            options.mode(0o600);
+        }
+        options.open(path).unwrap().write_all(bytes).unwrap();
+    }
+
+    fn record(id: &str, event: &str, eligible: bool, position: f64) -> String {
+        format!(
+            "{{\"version\":1,\"event\":\"{event}\",\"media_url\":\"https://www.youtube.com/watch?v={id}\",\"position_seconds\":{position},\"duration_seconds\":100,\"completion_eligible\":{eligible}}}\n"
+        )
+    }
+
+    #[test]
+    fn collection_is_idempotent_and_removal_preserves_global_watched_on_restart() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        state.save(&video).unwrap();
+        state.save(&video).unwrap();
+        assert_eq!(state.videos().len(), 1);
+        let report = state.progress_file().unwrap();
+        register_identity(&report, &video.id).unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 95.0).as_bytes(),
+        );
+        assert!(state.poll_progress().unwrap());
+        assert!(!state.poll_progress().unwrap());
+        let mut restarted = State::open(&directory.config()).unwrap();
+        assert_eq!(
+            restarted.videos()[0].publication_date,
+            video.publication_date
+        );
+        assert!(restarted.is_watched(&video.id));
+        restarted.remove(std::slice::from_ref(&video.id)).unwrap();
+        restarted.remove(std::slice::from_ref(&video.id)).unwrap();
+        let mut restarted = State::open(&directory.config()).unwrap();
+        assert!(restarted.videos().is_empty());
+        assert!(restarted.is_watched(&video.id));
+        assert!(!restarted.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn malformed_and_externally_changed_state_is_never_overwritten() {
+        let directory = TestDirectory::new();
+        let path = directory.0.join("watch-later.toml");
+        let mut state = State::open(&directory.config()).unwrap();
+        fs::write(&path, "not = [valid").unwrap();
+        assert!(state.save(&video()).is_err());
+        assert!(state.videos().is_empty());
+        assert_eq!(fs::read_to_string(&path).unwrap(), "not = [valid");
+        assert!(State::open(&directory.config()).is_err());
+        fs::remove_file(&path).unwrap();
+        fs::write(
+            directory.0.join("watch-state.toml"),
+            "version = 1\nwatched = [\"--unsafe\"]\n",
+        )
+        .unwrap();
+        assert!(State::open(&directory.config()).is_err());
+        assert_eq!(
+            fs::read_to_string(directory.0.join("watch-state.toml")).unwrap(),
+            "version = 1\nwatched = [\"--unsafe\"]\n"
+        );
+    }
+
+    #[test]
+    fn watched_write_failure_keeps_memory_unchanged_and_retries_consumed_reports() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let report = state.progress_file().unwrap();
+        register_identity(&report, &video().id).unwrap();
+        append(
+            &report,
+            record(&video().id, "progress", true, 95.0).as_bytes(),
+        );
+        let path = directory.0.join("watch-state.toml");
+        fs::write(&path, "malformed state").unwrap();
+        assert!(state.poll_progress().is_err());
+        assert!(!state.is_watched(&video().id));
+        assert_eq!(fs::read_to_string(&path).unwrap(), "malformed state");
+        fs::remove_file(&path).unwrap();
+        assert!(state.poll_progress().unwrap());
+        assert!(state.is_watched(&video().id));
+        assert!(!state.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn partial_reports_survive_restart_and_bad_siblings_do_not_block_completion() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let report = state.progress_file().unwrap();
+        register_identity(&report, &video().id).unwrap();
+        let line = record(&video().id, "ended", true, 95.0);
+        append(&report, &line.as_bytes()[..line.len() - 1]);
+        assert!(!state.poll_progress().unwrap());
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        let bad = state.progress_file().unwrap();
+        register_identity(&bad, "rQ_J9WH6CGk").unwrap();
+        append(&bad, b"not json\n");
+        append(&report, b"\n");
+        assert!(state.poll_progress().is_err());
+        assert!(state.is_watched(&video().id));
+        assert!(state.poll_progress().unwrap());
+        assert!(!state.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn wrong_identity_ineligible_positions_and_untrusted_reports_do_not_mark_watched() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let report = state.progress_file().unwrap();
+        register_identity(&report, &video().id).unwrap();
+        for line in [
+            record("rQ_J9WH6CGk", "progress", true, 100.0),
+            record(&video().id, "progress", false, 100.0),
+            record(&video().id, "closed", true, 94.9),
+        ] {
+            append(&report, line.as_bytes());
+        }
+        assert!(!state.poll_progress().unwrap());
+        assert!(!state.is_watched(&video().id));
+        assert!(!state.is_watched("rQ_J9WH6CGk"));
+        for line in [
+            record(&video().id, "unknown", true, 100.0),
+            record(&video().id, "progress", true, -1.0),
+            record(&video().id, "progress", true, 100.0).replace("\"version\":1", "\"version\":2"),
+            record(&video().id, "progress", true, 100.0)
+                .replace("\"duration_seconds\":100", "\"duration_seconds\":0"),
+        ] {
+            append(&report, line.as_bytes());
+            assert!(state.poll_progress().is_err());
+            assert!(!state.is_watched(&video().id));
+            assert!(!state.poll_progress().unwrap());
+        }
+        #[cfg(unix)]
+        {
+            let symlink = state.progress_file().unwrap();
+            register_identity(&symlink, &video().id).unwrap();
+            let external = directory.0.join("external.jsonl");
+            append(
+                &external,
+                record(&video().id, "progress", true, 100.0).as_bytes(),
+            );
+            std::os::unix::fs::symlink(&external, &symlink).unwrap();
+            assert!(state.poll_progress().is_err());
+            assert!(!state.is_watched(&video().id));
+            assert!(!state.poll_progress().unwrap());
+        }
+    }
+}

src/youtube.rs

@@ -1,12 +1,13 @@
 use std::{
     ffi::OsStr,
     io::{self, Read},
+    path::Path,
     process::{Child, Command, Stdio},
     sync::mpsc::{self, Receiver, TryRecvError},
     thread,
 };
 
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 
 pub const PAGE_SIZE: usize = 20;
 
@@ -16,7 +17,7 @@ pub struct Page {
     pub cutoff: Option<i64>,
 }
 
-#[derive(Deserialize)]
+#[derive(Clone, Deserialize, Serialize)]
 pub struct Video {
     pub id: String,
     pub title: String,
@@ -67,14 +68,14 @@ fn valid_channel_id(id: &str) -> bool {
             .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
 }
 
-fn valid_video_id(id: &str) -> bool {
+pub(crate) fn valid_video_id(id: &str) -> bool {
     id.len() == 11
         && id
             .bytes()
             .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
 }
 
-fn validate_video(video: &mut Video) -> Result<(), String> {
+pub(crate) fn validate_video(video: &mut Video) -> Result<(), String> {
     if !valid_video_id(&video.id) {
         return Err("an invalid YouTube video ID".into());
     }
@@ -463,19 +464,22 @@ pub struct PlayerLaunch {
 }
 
 impl PlayerLaunch {
-    pub fn start(id: &str, player: &OsStr) -> io::Result<Self> {
+    pub fn start(id: &str, player: &OsStr, progress_file: &Path) -> io::Result<Self> {
         if !valid_video_id(id) {
             return Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
                 "Invalid YouTube video ID",
             ));
         }
+        crate::watch_later::register_identity(progress_file, id)?;
         let mut player = isolated_command(player)
             .args([
                 "--new-instance",
                 "--url",
                 &format!("https://www.youtube.com/watch?v={id}"),
             ])
+            .arg("--progress-file")
+            .arg(progress_file)
             .stdin(Stdio::null())
             .stdout(Stdio::null())
             .stderr(Stdio::piped())
@@ -519,7 +523,7 @@ mod tests {
     fn player_launch_rejects_unsafe_video_ids_before_spawning() {
         for id in ["", "--help", "jNQXAC9IVRw&list=other", "日本語"] {
             assert!(matches!(
-                PlayerLaunch::start(id, OsStr::new("terebi")),
+                PlayerLaunch::start(id, OsStr::new("terebi"), Path::new("progress.jsonl")),
                 Err(error) if error.kind() == io::ErrorKind::InvalidInput
             ));
         }