feat: track currently watching videos

src/keys.rs

@@ -142,6 +142,7 @@ actions! {
     SavedVideoSave, "video.saved.save", "Save the selected video (also in previews)", VIEW, ["v"];
     SavedVideoBrowse, "view.saved.browse", "Choose saved videos across all categories or one category", VIEW, ["V"];
     SavedVideoBrowseCategory, "view.saved.category", "Browse the current category's saved videos", VIEW, ["alt+v"];
+    WatchingBrowse, "view.watching.browse", "Browse started, unfinished videos", VIEW, ["C"];
     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"];
@@ -538,16 +539,20 @@ impl Bindings {
                     ),
                 ],
             ),
+            (
+                "CURRENTLY WATCHING",
+                &[(&[WatchingBrowse], "Browse started, unfinished videos")],
+            ),
             (
                 "COLLECTION LISTS",
                 &[
                     (
                         &[CollectionRemoveSelected],
-                        "Remove selected from current watch-later / saved list only",
+                        "Remove selected from the current collection only",
                     ),
                     (
                         &[CollectionRemoveBulk],
-                        "Confirm removal of current category / filtered subset",
+                        "Confirm removal of current list / category / filtered subset",
                     ),
                     (
                         &[],

src/video_state.rs

@@ -59,28 +59,35 @@ impl StoredFile {
 pub enum CollectionKind {
     WatchLater,
     Saved,
+    Watching,
 }
 
 struct Collection {
     file: StoredFile,
     videos: Vec<Video>,
     ids: BTreeSet<String>,
+    require_channel: bool,
 }
 
 impl Collection {
-    fn open(path: PathBuf) -> io::Result<Self> {
+    fn open(path: PathBuf, require_channel: bool) -> io::Result<Self> {
         let (file, document) = StoredFile::open(path)?;
-        let (videos, ids) =
-            load_videos(&document).map_err(|e| invalid(format!("{}: {e}", file.path.display())))?;
-        Ok(Self { file, videos, ids })
+        let (videos, ids) = load_videos(&document, require_channel)
+            .map_err(|e| invalid(format!("{}: {e}", file.path.display())))?;
+        Ok(Self {
+            file,
+            videos,
+            ids,
+            require_channel,
+        })
     }
 
     fn save(&mut self, video: &Video) -> io::Result<()> {
         if self.ids.contains(&video.id) {
-            return Ok(());
+            return check_original(&self.file.path, self.file.original.as_deref());
         }
         let mut video = video.clone();
-        validate_saved(&mut video)?;
+        validate_snapshot(&mut video, self.require_channel)?;
         self.file.save(video_document(
             self.videos.iter().chain(std::iter::once(&video)),
         ))?;
@@ -91,7 +98,7 @@ impl Collection {
 
     fn remove(&mut self, ids: &[String]) -> io::Result<()> {
         if !ids.iter().any(|id| self.ids.contains(id)) {
-            return Ok(());
+            return check_original(&self.file.path, self.file.original.as_deref());
         }
         self.file.save(video_document(
             self.videos.iter().filter(|v| !ids.contains(&v.id)),
@@ -105,6 +112,7 @@ impl Collection {
 pub struct State {
     watch_later: Collection,
     saved_videos: Collection,
+    watching: Collection,
     watch_state: StoredFile,
     watched: BTreeSet<String>,
     pending: BTreeSet<String>,
@@ -121,14 +129,20 @@ impl State {
             .parent()
             .filter(|p| !p.as_os_str().is_empty())
             .unwrap_or(Path::new("."));
-        let watch_later = Collection::open(parent.join("watch-later.toml"))?;
-        let saved_videos = Collection::open(parent.join("saved-videos.toml"))?;
+        let watch_later = Collection::open(parent.join("watch-later.toml"), true)?;
+        let saved_videos = Collection::open(parent.join("saved-videos.toml"), true)?;
+        let mut watching = Collection::open(parent.join("currently-watching.toml"), false)?;
         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())))?;
+        let completed: Vec<_> = watching.ids.intersection(&watched).cloned().collect();
+        if !completed.is_empty() {
+            watching.remove(&completed)?;
+        }
         Ok(Self {
             watch_later,
             saved_videos,
+            watching,
             watch_state,
             watched,
             pending: BTreeSet::new(),
@@ -144,6 +158,7 @@ impl State {
         match kind {
             CollectionKind::WatchLater => &self.watch_later,
             CollectionKind::Saved => &self.saved_videos,
+            CollectionKind::Watching => &self.watching,
         }
     }
 
@@ -151,6 +166,7 @@ impl State {
         match kind {
             CollectionKind::WatchLater => &mut self.watch_later,
             CollectionKind::Saved => &mut self.saved_videos,
+            CollectionKind::Watching => &mut self.watching,
         }
     }
 
@@ -165,19 +181,43 @@ impl State {
     }
 
     pub fn save(&mut self, kind: CollectionKind, video: &Video) -> io::Result<()> {
+        if matches!(kind, CollectionKind::Watching) && self.is_watched(&video.id) {
+            return Ok(());
+        }
         self.collection_mut(kind).save(video)
     }
 
     pub fn remove(&mut self, kind: CollectionKind, ids: &[String]) -> io::Result<()> {
+        if matches!(kind, CollectionKind::Watching) {
+            self.discovery = None;
+            loop {
+                self.discover()?;
+                if self.discovery.is_none() {
+                    break;
+                }
+            }
+            for stream in &mut self.streams {
+                match stream.prepare() {
+                    Ok(()) => {}
+                    Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
+                    Err(e) => return Err(e),
+                }
+                if stream.id.as_ref().is_some_and(|id| ids.contains(id)) {
+                    stream.mark_seen()?;
+                }
+            }
+        }
         self.collection_mut(kind).remove(ids)
     }
 
-    pub fn progress_file(&mut self) -> io::Result<PathBuf> {
+    pub fn progress_file(&mut self, video: &Video) -> io::Result<PathBuf> {
         if self.streams.len() >= MAX_STREAMS {
             return Err(invalid(
                 "Playback session limit reached; remove inactive playback-sessions directories and restart",
             ));
         }
+        let mut video = video.clone();
+        validate_snapshot(&mut video, false)?;
         fs::create_dir_all(self.sessions.parent().unwrap_or(Path::new(".")))?;
         private_directory(&self.sessions, true)?;
         let directory = loop {
@@ -192,6 +232,8 @@ impl State {
                 Err(e) => return Err(e),
             }
         };
+        let (mut snapshot, _) = StoredFile::open(directory.join("video.toml"))?;
+        snapshot.save(video_document(std::iter::once(&video)))?;
         let path = directory.join("progress.jsonl");
         self.streams.push(ProgressStream::new(path.clone()));
         Ok(path)
@@ -220,6 +262,38 @@ impl State {
             self.watched.append(&mut self.pending);
             self.notify_changed = true;
         }
+        let completed: Vec<_> = self
+            .watching
+            .ids
+            .intersection(&self.watched)
+            .cloned()
+            .collect();
+        if !completed.is_empty() {
+            self.watching.remove(&completed)?;
+            self.notify_changed = true;
+        }
+        for stream in &mut self.streams {
+            if !stream.started || stream.seen {
+                continue;
+            }
+            let id = stream.id.as_ref().unwrap();
+            if !self.watched.contains(id) {
+                let video = stream.video.as_ref().or_else(|| {
+                    self.watch_later
+                        .videos
+                        .iter()
+                        .chain(self.saved_videos.videos.iter())
+                        .find(|video| &video.id == id)
+                });
+                if let Some(video) = video {
+                    let added = !self.watching.ids.contains(id);
+                    self.watching.save(video)?;
+                    self.notify_changed |= added;
+                }
+                stream.mark_seen()?;
+            }
+            stream.started = false;
+        }
         // Preserve notification after reporting a bad record; siblings were still processed.
         if let Some(error) = error {
             return Err(error);
@@ -258,7 +332,9 @@ impl State {
                 continue;
             }
             if self.streams.len() >= MAX_STREAMS {
-                break;
+                return Err(invalid(
+                    "Playback session limit reached; remove inactive playback-sessions directories and restart",
+                ));
             }
             self.streams.push(ProgressStream::new(path));
         }
@@ -268,7 +344,7 @@ impl State {
     }
 }
 
-fn validate_saved(video: &mut Video) -> io::Result<()> {
+fn validate_snapshot(video: &mut Video, require_channel: bool) -> io::Result<()> {
     let date = video.publication_date.clone();
     if let Some(date) = &date {
         let raw = date.strip_prefix('~').unwrap_or(date);
@@ -303,7 +379,7 @@ fn validate_saved(video: &mut Video) -> io::Result<()> {
     {
         return Err(invalid("Saved uploader_id must be a valid YouTube @handle"));
     }
-    if video.channel_id.is_none() && video.uploader_id.is_none() {
+    if require_channel && video.channel_id.is_none() && video.uploader_id.is_none() {
         return Err(invalid(
             "Video has no channel identity; refresh its metadata before saving",
         ));
@@ -311,7 +387,10 @@ fn validate_saved(video: &mut Video) -> io::Result<()> {
     Ok(())
 }
 
-fn load_videos(document: &DocumentMut) -> io::Result<(Vec<Video>, BTreeSet<String>)> {
+fn load_videos(
+    document: &DocumentMut,
+    require_channel: bool,
+) -> io::Result<(Vec<Video>, BTreeSet<String>)> {
     if document
         .iter()
         .any(|(key, _)| !matches!(key, "version" | "videos"))
@@ -350,7 +429,7 @@ fn load_videos(document: &DocumentMut) -> io::Result<(Vec<Video>, BTreeSet<Strin
         }
         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)?;
+        validate_snapshot(&mut video, require_channel)?;
         if ids.insert(video.id.clone()) {
             videos.push(video);
         }
@@ -472,6 +551,16 @@ fn read_bytes(path: &Path, limit: u64) -> io::Result<Option<Vec<u8>>> {
     Ok(Some(bytes))
 }
 
+fn check_original(path: &Path, original: Option<&[u8]>) -> io::Result<()> {
+    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()
+        )));
+    }
+    Ok(())
+}
+
 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"));
@@ -500,12 +589,7 @@ fn atomic_write(path: &Path, original: Option<&[u8]>, contents: &[u8]) -> io::Re
     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()
-            )));
-        }
+        check_original(path, original)?;
         if original.is_none() {
             fs::hard_link(&temporary, path)?;
         } else {
@@ -602,6 +686,10 @@ struct ProgressRecord {
 struct ProgressStream {
     path: PathBuf,
     id: Option<String>,
+    video: Option<Video>,
+    seen_file: Option<StoredFile>,
+    seen: bool,
+    started: bool,
     file: Option<File>,
     partial: Vec<u8>,
     discarding: bool,
@@ -614,6 +702,10 @@ impl ProgressStream {
         Self {
             path,
             id: None,
+            video: None,
+            seen_file: None,
+            seen: false,
+            started: false,
             file: None,
             partial: Vec::new(),
             discarding: false,
@@ -622,6 +714,74 @@ impl ProgressStream {
         }
     }
 
+    fn prepare(&mut self) -> io::Result<()> {
+        if self.id.is_some() {
+            return Ok(());
+        }
+        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"))?;
+        let snapshot = self.path.with_file_name("video.toml");
+        let video = match regular_file(&snapshot, true) {
+            Ok(_) => {
+                let mut collection = Collection::open(snapshot, false)?;
+                if collection.videos.len() != 1 || collection.videos[0].id != id {
+                    return Err(invalid(
+                        "Session video metadata must match its video-id marker",
+                    ));
+                }
+                collection.videos.pop()
+            }
+            Err(e) if e.kind() == io::ErrorKind::NotFound => None,
+            Err(e) => return Err(e),
+        };
+        let seen_path = self.path.with_file_name("watching-seen.toml");
+        match regular_file(&seen_path, true) {
+            Ok(_) => {}
+            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
+            Err(e) => return Err(e),
+        }
+        let (seen_file, document) = StoredFile::open(seen_path)?;
+        let seen = seen_file.original.is_some();
+        if seen
+            && (document
+                .iter()
+                .any(|(key, _)| !matches!(key, "version" | "id"))
+                || document.get("id").and_then(|item| item.as_str()) != Some(id))
+        {
+            return Err(invalid(
+                "Session watching admission must match its video-id marker",
+            ));
+        }
+        self.id = Some(id.into());
+        self.video = video;
+        self.seen_file = Some(seen_file);
+        self.seen = seen;
+        Ok(())
+    }
+
+    fn mark_seen(&mut self) -> io::Result<()> {
+        if self.seen {
+            let file = self.seen_file.as_ref().unwrap();
+            return check_original(&file.path, file.original.as_deref());
+        }
+        private_directory(self.path.parent().unwrap(), false)?;
+        let mut document = DocumentMut::new();
+        document["version"] = value(1);
+        document["id"] = value(self.id.as_deref().unwrap());
+        self.seen_file.as_mut().unwrap().save(document)?;
+        self.seen = true;
+        self.started = false;
+        Ok(())
+    }
+
     fn poll(
         &mut self,
         pending: &mut BTreeSet<String>,
@@ -649,17 +809,7 @@ impl ProgressStream {
         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.prepare()?;
             self.file = Some(regular_file(&self.path, true)?);
         }
         let mut buffer = [0; STREAM_BYTES];
@@ -699,7 +849,11 @@ impl ProgressStream {
         ))
     }
 
-    fn record(&self, pending: &mut BTreeSet<String>, watched: &BTreeSet<String>) -> io::Result<()> {
+    fn record(
+        &mut 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")
@@ -723,8 +877,13 @@ impl ProgressStream {
                 "position must be finite/nonnegative and duration positive/finite or null",
             ));
         }
-        if self.id.as_deref() == Some(id)
-            && record.completion_eligible
+        if self.id.as_deref() != Some(id) {
+            return Ok(());
+        }
+        if record.position_seconds > 0.0 && !self.seen && !watched.contains(id) {
+            self.started = true;
+        }
+        if record.completion_eligible
             && record
                 .duration_seconds
                 .is_some_and(|d| record.position_seconds / d >= 0.95)
@@ -807,7 +966,7 @@ mod tests {
             assert_eq!(state.videos(kind).len(), 1);
             assert!(state.contains(kind, &video.id));
         }
-        let report = state.progress_file().unwrap();
+        let report = state.progress_file(&video).unwrap();
         register_identity(&report, &video.id).unwrap();
         append(
             &report,
@@ -889,7 +1048,7 @@ mod tests {
     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();
+        let report = state.progress_file(&video()).unwrap();
         register_identity(&report, &video().id).unwrap();
         append(
             &report,
@@ -899,6 +1058,7 @@ mod tests {
         fs::write(&path, "malformed state").unwrap();
         assert!(state.poll_progress().is_err());
         assert!(!state.is_watched(&video().id));
+        assert!(state.videos(CollectionKind::Watching).is_empty());
         assert_eq!(fs::read_to_string(&path).unwrap(), "malformed state");
         fs::remove_file(&path).unwrap();
         assert!(state.poll_progress().unwrap());
@@ -910,7 +1070,7 @@ mod tests {
     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();
+        let report = state.progress_file(&video()).unwrap();
         register_identity(&report, &video().id).unwrap();
         let line = record(&video().id, "ended", true, 95.0);
         append(&report, &line.as_bytes()[..line.len() - 1]);
@@ -918,8 +1078,10 @@ mod tests {
         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();
+        let mut other = video();
+        other.id = "rQ_J9WH6CGk".into();
+        let bad = state.progress_file(&other).unwrap();
+        register_identity(&bad, &other.id).unwrap();
         append(&bad, b"not json\n");
         append(&report, b"\n");
         assert!(state.poll_progress().is_err());
@@ -932,7 +1094,7 @@ mod tests {
     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();
+        let report = state.progress_file(&video()).unwrap();
         register_identity(&report, &video().id).unwrap();
         for line in [
             record("rQ_J9WH6CGk", "progress", true, 100.0),
@@ -941,7 +1103,8 @@ mod tests {
         ] {
             append(&report, line.as_bytes());
         }
-        assert!(!state.poll_progress().unwrap());
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video().id));
         assert!(!state.is_watched(&video().id));
         assert!(!state.is_watched("rQ_J9WH6CGk"));
         for line in [
@@ -958,7 +1121,7 @@ mod tests {
         }
         #[cfg(unix)]
         {
-            let symlink = state.progress_file().unwrap();
+            let symlink = state.progress_file(&video()).unwrap();
             register_identity(&symlink, &video().id).unwrap();
             let external = directory.0.join("external.jsonl");
             append(
@@ -971,4 +1134,435 @@ mod tests {
             assert!(!state.poll_progress().unwrap());
         }
     }
+
+    #[test]
+    fn watching_requires_a_valid_positive_bound_acknowledgement() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        let report = state.progress_file(&video).unwrap();
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        assert!(!state.poll_progress().unwrap());
+        register_identity(&report, &video.id).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        for line in [
+            record("rQ_J9WH6CGk", "progress", true, 100.0),
+            record(&video.id, "progress", true, 0.0),
+        ] {
+            append(&report, line.as_bytes());
+            assert!(!state.poll_progress().unwrap());
+            assert!(state.videos(CollectionKind::Watching).is_empty());
+        }
+        for line in [
+            record(&video.id, "unknown", true, 10.0),
+            record(&video.id, "progress", true, -1.0),
+            record(&video.id, "progress", true, 10.0)
+                .replace("\"duration_seconds\":100", "\"duration_seconds\":0"),
+            record(&video.id, "progress", true, 10.0).replace("\"version\":1", "\"version\":2"),
+        ] {
+            append(&report, line.as_bytes());
+            assert!(state.poll_progress().is_err());
+            assert!(state.videos(CollectionKind::Watching).is_empty());
+            assert!(!state.is_watched(&video.id));
+        }
+        append(
+            &report,
+            record(&video.id, "progress", false, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(!state.is_watched(&video.id));
+    }
+
+    #[test]
+    fn starts_arriving_while_closed_preserve_metadata_without_a_creator() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let mut video = video();
+        video.channel = None;
+        video.channel_id = None;
+        video.uploader_id = None;
+        assert!(state.save(CollectionKind::Saved, &video).is_err());
+        assert!(state.save(CollectionKind::WatchLater, &video).is_err());
+        let report = state.progress_file(&video).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        drop(state);
+        append(
+            &report,
+            record(&video.id, "closed", true, 10.0)
+                .replace("\"duration_seconds\":100", "\"duration_seconds\":null")
+                .as_bytes(),
+        );
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(state.poll_progress().unwrap());
+        let stored = &state.videos(CollectionKind::Watching)[0];
+        assert_eq!(stored.id, video.id);
+        assert_eq!(stored.title, video.title);
+        assert_eq!(stored.channel, None);
+        assert_eq!(stored.channel_id, None);
+        assert_eq!(stored.uploader_id, None);
+        assert_eq!(stored.publication_date, video.publication_date);
+        assert_eq!(stored.timestamp, video.timestamp);
+        append(&report, record(&video.id, "ended", false, 100.0).as_bytes());
+        assert!(!state.poll_progress().unwrap());
+        assert!(!state.is_watched(&video.id));
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(!state.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn completion_wins_over_parallel_sessions_late_reports_and_replays() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        for kind in [CollectionKind::Saved, CollectionKind::WatchLater] {
+            state.save(kind, &video).unwrap();
+        }
+        let first = state.progress_file(&video).unwrap();
+        register_identity(&first, &video.id).unwrap();
+        append(&first, record(&video.id, "progress", true, 94.9).as_bytes());
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(!state.is_watched(&video.id));
+        let second = state.progress_file(&video).unwrap();
+        register_identity(&second, &video.id).unwrap();
+        append(
+            &second,
+            record(&video.id, "progress", true, 20.0).as_bytes(),
+        );
+        append(&first, record(&video.id, "ended", true, 95.0).as_bytes());
+        assert!(state.poll_progress().unwrap());
+        assert!(state.is_watched(&video.id));
+        assert!(!state.contains(CollectionKind::Watching, &video.id));
+        for report in [&first, &second] {
+            append(report, record(&video.id, "progress", true, 30.0).as_bytes());
+        }
+        let replay = state.progress_file(&video).unwrap();
+        register_identity(&replay, &video.id).unwrap();
+        append(
+            &replay,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.is_watched(&video.id));
+        state
+            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
+            .unwrap();
+        assert!(state.is_watched(&video.id));
+        assert!(state.contains(CollectionKind::Saved, &video.id));
+        assert!(state.contains(CollectionKind::WatchLater, &video.id));
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+    }
+
+    #[test]
+    fn manual_removal_suppresses_existing_sessions_across_restart_but_not_new_sessions() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        for kind in [CollectionKind::Saved, CollectionKind::WatchLater] {
+            state.save(kind, &video).unwrap();
+        }
+        let first = state.progress_file(&video).unwrap();
+        register_identity(&first, &video.id).unwrap();
+        append(&first, record(&video.id, "progress", true, 10.0).as_bytes());
+        assert!(state.poll_progress().unwrap());
+        // Simulate a crash after collection persistence but before its admission marker.
+        fs::remove_file(first.with_file_name("watching-seen.toml")).unwrap();
+        let second = state.progress_file(&video).unwrap();
+        register_identity(&second, &video.id).unwrap();
+        append(
+            &second,
+            record(&video.id, "progress", true, 20.0).as_bytes(),
+        );
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        state
+            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
+            .unwrap();
+        for report in [&first, &second] {
+            append(report, record(&video.id, "progress", true, 50.0).as_bytes());
+        }
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        assert!(!state.is_watched(&video.id));
+        assert!(state.contains(CollectionKind::Saved, &video.id));
+        assert!(state.contains(CollectionKind::WatchLater, &video.id));
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        let replay = state.progress_file(&video).unwrap();
+        register_identity(&replay, &video.id).unwrap();
+        append(
+            &replay,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(!state.is_watched(&video.id));
+    }
+
+    #[test]
+    fn watching_write_conflicts_retry_consumed_starts() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        let report = state.progress_file(&video).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        let path = directory.0.join("currently-watching.toml");
+        fs::write(&path, "external edit").unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().is_err());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        assert_eq!(fs::read_to_string(&path).unwrap(), "external edit");
+        assert!(state.poll_progress().is_err());
+        fs::remove_file(&path).unwrap();
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        let original = fs::read(&path).unwrap();
+        append(&path, b"\n# external removal conflict\n");
+        let replay = state.progress_file(&video).unwrap();
+        register_identity(&replay, &video.id).unwrap();
+        append(
+            &replay,
+            record(&video.id, "progress", true, 20.0).as_bytes(),
+        );
+        assert!(state.poll_progress().is_err());
+        assert!(!replay.with_file_name("watching-seen.toml").exists());
+        fs::write(&path, original).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(!state.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn failed_session_admission_retries_without_losing_starts_or_allowing_readdition() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        let report = state.progress_file(&video).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        state.streams[0].prepare().unwrap();
+        let marker = report.with_file_name("watching-seen.toml");
+        append(&marker, b"external edit");
+        append(
+            &report,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().is_err());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert!(
+            state
+                .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
+                .is_err()
+        );
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert_eq!(fs::read_to_string(&marker).unwrap(), "external edit");
+        fs::remove_file(&marker).unwrap();
+        assert!(state.poll_progress().unwrap());
+        state
+            .remove(CollectionKind::Watching, std::slice::from_ref(&video.id))
+            .unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 50.0).as_bytes(),
+        );
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        drop(state);
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+    }
+
+    #[test]
+    fn failed_completion_cleanup_retries_and_restart_prunes_stale_watching() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        let report = state.progress_file(&video).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().unwrap());
+        let path = directory.0.join("currently-watching.toml");
+        let original = fs::read(&path).unwrap();
+        append(&path, b"\n# external edit\n");
+        let changed = fs::read(&path).unwrap();
+        append(&report, record(&video.id, "ended", true, 95.0).as_bytes());
+        assert!(state.poll_progress().is_err());
+        assert!(state.is_watched(&video.id));
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        assert_eq!(fs::read(&path).unwrap(), changed);
+        append(
+            &report,
+            record(&video.id, "progress", true, 20.0).as_bytes(),
+        );
+        assert!(state.poll_progress().is_err());
+        fs::write(&path, original).unwrap();
+        assert!(state.poll_progress().unwrap());
+        assert!(!state.contains(CollectionKind::Watching, &video.id));
+        drop(state);
+        // Crash recovery must also handle a watched write committed before collection cleanup.
+        fs::write(&path, video_document(std::iter::once(&video)).to_string()).unwrap();
+        let mut state = State::open(&directory.config()).unwrap();
+        assert!(state.is_watched(&video.id));
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        assert!(!state.poll_progress().unwrap());
+    }
+
+    #[test]
+    fn legacy_sessions_use_saved_metadata_when_available_and_still_mark_completion_without_it() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        state.save(CollectionKind::Saved, &video).unwrap();
+        let report = state.progress_file(&video).unwrap();
+        fs::remove_file(report.with_file_name("video.toml")).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(state.poll_progress().unwrap());
+        assert!(state.contains(CollectionKind::Watching, &video.id));
+        append(&report, record(&video.id, "ended", true, 95.0).as_bytes());
+        assert!(state.poll_progress().unwrap());
+        assert!(state.is_watched(&video.id));
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        let mut other = video.clone();
+        other.id = "rQ_J9WH6CGk".into();
+        let report = state.progress_file(&other).unwrap();
+        fs::remove_file(report.with_file_name("video.toml")).unwrap();
+        register_identity(&report, &other.id).unwrap();
+        append(
+            &report,
+            record(&other.id, "progress", true, 10.0).as_bytes(),
+        );
+        assert!(!state.poll_progress().unwrap());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        append(&report, record(&other.id, "closed", true, 95.0).as_bytes());
+        assert!(state.poll_progress().unwrap());
+        assert!(state.is_watched(&other.id));
+    }
+
+    #[test]
+    fn session_metadata_and_admission_markers_must_match_the_bound_identity() {
+        for mismatched_snapshot in [true, false] {
+            let directory = TestDirectory::new();
+            let mut state = State::open(&directory.config()).unwrap();
+            let video = video();
+            let report = state.progress_file(&video).unwrap();
+            register_identity(&report, &video.id).unwrap();
+            if mismatched_snapshot {
+                let mut other = video.clone();
+                other.id = "rQ_J9WH6CGk".into();
+                fs::write(
+                    report.with_file_name("video.toml"),
+                    video_document(std::iter::once(&other)).to_string(),
+                )
+                .unwrap();
+            } else {
+                append(
+                    &report.with_file_name("watching-seen.toml"),
+                    b"version = 1\nid = \"rQ_J9WH6CGk\"\n",
+                );
+            }
+            append(&report, record(&video.id, "ended", true, 100.0).as_bytes());
+            assert!(state.poll_progress().is_err());
+            assert!(state.videos(CollectionKind::Watching).is_empty());
+            assert!(!state.is_watched(&video.id));
+        }
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn session_snapshots_reject_symlinks_and_shared_permissions() {
+        use std::os::unix::fs::PermissionsExt;
+        for symlink in [true, false] {
+            let directory = TestDirectory::new();
+            let mut state = State::open(&directory.config()).unwrap();
+            let video = video();
+            let report = state.progress_file(&video).unwrap();
+            register_identity(&report, &video.id).unwrap();
+            let snapshot = report.with_file_name("video.toml");
+            if symlink {
+                let external = directory.0.join("external.toml");
+                fs::write(&external, fs::read(&snapshot).unwrap()).unwrap();
+                fs::remove_file(&snapshot).unwrap();
+                std::os::unix::fs::symlink(&external, &snapshot).unwrap();
+            } else {
+                fs::set_permissions(&snapshot, fs::Permissions::from_mode(0o644)).unwrap();
+            }
+            append(
+                &report,
+                record(&video.id, "progress", true, 100.0).as_bytes(),
+            );
+            assert!(state.poll_progress().is_err());
+            assert!(state.videos(CollectionKind::Watching).is_empty());
+            assert!(!state.is_watched(&video.id));
+        }
+    }
+
+    #[test]
+    fn watching_and_session_snapshots_keep_date_handle_channel_and_schema_validation() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        for invalid_field in ["date", "handle", "channel"] {
+            let mut video = video();
+            match invalid_field {
+                "date" => video.publication_date = Some("2005-02-30".into()),
+                "handle" => video.uploader_id = Some("not-a-handle".into()),
+                "channel" => video.channel_id = Some("not-a-channel-id".into()),
+                _ => unreachable!(),
+            }
+            assert!(state.progress_file(&video).is_err());
+            assert!(state.save(CollectionKind::Watching, &video).is_err());
+            assert!(state.videos(CollectionKind::Watching).is_empty());
+        }
+        let path = directory.0.join("currently-watching.toml");
+        let contents = "version = 1\nvideos = []\nunexpected = true\n";
+        fs::write(&path, contents).unwrap();
+        assert!(State::open(&directory.config()).is_err());
+        assert_eq!(fs::read_to_string(&path).unwrap(), contents);
+    }
+
+    #[test]
+    fn watching_and_session_snapshots_reject_oversized_files() {
+        let directory = TestDirectory::new();
+        let mut state = State::open(&directory.config()).unwrap();
+        let video = video();
+        let report = state.progress_file(&video).unwrap();
+        register_identity(&report, &video.id).unwrap();
+        OpenOptions::new()
+            .write(true)
+            .open(report.with_file_name("video.toml"))
+            .unwrap()
+            .set_len(MAX_STATE_BYTES + 1)
+            .unwrap();
+        append(
+            &report,
+            record(&video.id, "progress", true, 100.0).as_bytes(),
+        );
+        assert!(state.poll_progress().is_err());
+        assert!(state.videos(CollectionKind::Watching).is_empty());
+        assert!(!state.is_watched(&video.id));
+        File::create(directory.0.join("currently-watching.toml"))
+            .unwrap()
+            .set_len(MAX_STATE_BYTES + 1)
+            .unwrap();
+        assert!(State::open(&directory.config()).is_err());
+    }
 }