FIX: stabilize shared HLS playback, recovery and account handovers (#925)

* fix (hls): preserve playback reserve across cache eviction

keep a stable session media clock and completed segment timings so cache
eviction and origin rollovers cannot reset playback reserve.

update segment fixtures and cover reserve stability after collection.

* fix (hls): prevent playback stalls during origin manifest retries

mark the origin as degraded on the first retryable manifest failure, cap
retry-after delays, and reevaluate reserve independently of client polling.

use a bounded state commit after an expired splice deadline instead of
leaving the lease live when terminal media cannot be published safely.

* fix (hls): serve custom responses after the live splice deadline

publish a finite lease-bound channel_unavailable playlist on the existing
manifest route when a safe live splice is no longer possible.

renew lease and session activity on successful custom-segment requests
and retain concrete preparation failure diagnostics.

* fix(hls): persist origin qualification for short playlists

Share qualification and retry receipts across leases, preserve them while media publication is pending, and allow controlled recovery after expired acceptance episodes. Prefer eligible regular candidates before short fallbacks.

Add startup, revalidation, replay, cookie and endpoint regression coverage, and document short-window behavior.

Validation during implementation: Linux make test (7009 passed, 23 ignored), focused HLS and endpoint suites, format checks, Clippy and Markdown lint. Offline production library and backend checks passed after the import fix.

* fix(hls): reconcile cache write budgets and revision reclamation

Transfer reserved bytes atomically through staging and publication, preserve cleanup ownership across cancellation, and retire unpinned revisions during local and global reclamation.

Version temp registrations during inventory scans, remove production panic calls, and isolate debug diagnostics from admission results.

Validation: 1,162 HLS tests and 7,038 workspace tests passed; fmt-check, lint, test, markdown-lint and isolated production Clippy passed with dev-slim.

* feat (ai): add local codex files

* fix (hls): replace invalid origin accounts in shared HLS sessions

* fix (hls): synchronize origin account handovers across WebUI viewers

* fix (hls): preserve same-account resume work and backpressure

* fix (hls): resume reduced retries without crossing generations
Allow eligible exhausted acceptance episodes to select reduced retry candidates
while preserving generation, candidate, workload and timing guards. Keep
recovery chains bound to their generation, including changes between selection
and commit.

Add regressions for repaired and persistent resource conflicts, generation
isolation and media delivery through the app's local HTTP origin. Keep shared
manifest fixtures under test/fixtures/hls/reduced-retry.

backend/processing/src/processor/playlist/ingest/pipeline_transparency_tests/transport.rs

@@ -12,7 +12,7 @@ fn pipeline_transparency_telemetry_reports_non_cluster_cache_without_synthetic_c
 
     assert_eq!(telemetry.refresh_policy, InputRefreshPolicy::NORMAL);
     assert_eq!(telemetry.source, Some(PlaylistUpdateDataSource::Cache));
-    assert!(telemetry.clusters.is_empty());
+    assert_eq!(telemetry.clusters.len(), 0);
 }
 
 #[test]

backend/processing/src/processor/playlist/tests/curation_publication.rs

@@ -342,7 +342,7 @@ impl Publication {
                 .all(|item| text.contains(item.header.title.as_ref()) && text.contains(item.header.url.as_ref()))
         }));
         let strm = self.strm_contents();
-        assert!(!strm.is_empty());
+        assert_ne!(strm.len(), 0);
         let files = file_snapshot(self.directory.path());
         let live_title = expected[0].channels[0].header.title.as_bytes();
         assert!(
@@ -352,7 +352,7 @@ impl Publication {
             }),
             "new EPG programme must be published"
         );
-        assert!(!self.cache_signature().await.is_empty());
+        assert_ne!(self.cache_signature().await.len(), 0);
     }
 
     async fn identity_signature(&self) -> BTreeMap<String, (u32, u32)> {
@@ -455,7 +455,7 @@ async fn tmdb_https_selection_only_publishes_m3u_strm_without_an_xtream_output()
     let text = std::fs::read_to_string(run.directory.path().join("published.m3u")).unwrap();
     assert!(text.contains("First") && text.contains("Show S01E01") && text.contains("Live"));
     assert!(!text.contains("Unselected") && !text.contains("TMDB"));
-    assert!(!run.strm_contents().is_empty());
+    assert_ne!(run.strm_contents().len(), 0);
     let cache = run.context.playlist_state.as_ref().unwrap().data.read().await;
     assert!(cache[&run.target.name].xtream.is_none());
     assert_eq!(server.requests.lock().unwrap().len(), 3);
@@ -469,7 +469,7 @@ async fn tmdb_https_empty_and_no_match_clear_published_vod_series_but_keep_live(
         let server = DiscoveryServer::start().await;
         let run = Publication::new(&server, &target("curated", false, true));
         run.publish().await.unwrap();
-        assert!(!run.strm_contents().is_empty());
+        assert_ne!(run.strm_contents().len(), 0);
         server.reply(MOVIES, 200, body);
         server.reply(TV, 200, body);
         run.publish().await.unwrap();

backend/processing/src/processor/playlist/tests/curation_publication/item_limit.rs

@@ -248,7 +248,7 @@ async fn curation_production_wiring_does_not_fall_back_to_the_generic_client() {
         server.requests.lock().unwrap().is_empty(),
         "generic client has the fixture CA and could succeed, but is not used"
     );
-    assert!(!run.cache_signature().await.is_empty());
+    assert_ne!(run.cache_signature().await.len(), 0);
     assert!(!file_snapshot(run.directory.path()).is_empty());
 }
 
@@ -272,7 +272,7 @@ async fn curation_failure_respects_empty_input_authorization() {
         assert_eq!(result.is_err(), authorized, "ordinary M3U persistence still rejects an empty playlist");
         assert_eq!(server.requests.lock().unwrap().len(), if authorized { 3 } else { 0 });
         assert!(file_snapshot(run.directory.path()).is_empty());
-        assert!(run.cache_signature().await.is_empty());
+        assert_eq!(run.cache_signature().await.len(), 0);
     }
 }
 

backend/processing/src/processor/playlist/tests/mapping_stage.rs

@@ -151,7 +151,7 @@ async fn m3u_alias_playlist_is_downloaded_and_indexed_separately() {
     assert!(errors.is_empty(), "unexpected download errors: {errors:?}");
     assert!(storage_error.is_none(), "unexpected primary storage error: {storage_error:?}");
     assert!(!partial);
-    assert!(quality_rejections.is_empty());
+    assert_eq!(quality_rejections.len(), 0);
     assert!(!primary_playlist.is_empty());
     let alias_url = tuliprox_repository::load_input_m3u_stream_url(
         &ctx.config,
@@ -212,7 +212,7 @@ async fn failed_m3u_alias_is_retried_after_primary_input_is_processed() {
     assert!(!first_errors.is_empty(), "missing alias should report an error");
     assert!(first_storage_error.is_none(), "primary storage should succeed: {first_storage_error:?}");
     assert!(!first_partial);
-    assert!(first_quality_rejections.is_empty());
+    assert_eq!(first_quality_rejections.len(), 0);
     assert!(!first_playlist.is_empty());
     assert!(ctx.is_input_downloaded("main-account").await);
     assert!(!ctx.is_input_downloaded("retry-account").await);
@@ -236,7 +236,7 @@ async fn failed_m3u_alias_is_retried_after_primary_input_is_processed() {
     assert!(second_errors.is_empty(), "unexpected retry errors: {second_errors:?}");
     assert!(second_storage_error.is_none(), "primary storage should remain readable: {second_storage_error:?}");
     assert!(!second_partial);
-    assert!(second_quality_rejections.is_empty());
+    assert_eq!(second_quality_rejections.len(), 0);
     assert!(!second_playlist.is_empty());
     assert!(ctx.is_input_downloaded("retry-account").await);
     let alias_url = tuliprox_repository::load_input_m3u_stream_url(

backend/processing/src/processor/playlist/tests/quality_rejection_fallback.rs

@@ -313,8 +313,8 @@ async fn playlist_update_run_manual_and_automatic_execution_order_is_generated_a
     };
     let manual_progress = progress_for(&manual_run_id);
     let automatic_progress = progress_for(&automatic_run_id);
-    assert!(!manual_progress.is_empty());
-    assert!(!automatic_progress.is_empty());
+    assert_ne!(manual_progress.len(), 0);
+    assert_ne!(automatic_progress.len(), 0);
     assert!(manual_progress.iter().all(|progress| progress.execution_order == Some(manual_order)));
     assert!(automatic_progress.iter().all(|progress| progress.execution_order == Some(automatic_order)));
 }
@@ -687,11 +687,11 @@ async fn coderabbit_stalker_global_failures_mark_only_requested_non_skipped_clus
             assert_eq!(fetch.errors.len(), 1, "{failure:?}");
             assert!(!fetch.is_ok());
             assert!(fetch.quality_acceptances.is_empty() && fetch.quality_rejections.is_empty());
-            assert!(fetch.force_updates.is_empty());
+            assert_eq!(fetch.force_updates.len(), 0);
             let requests = server.finish();
             match failure {
-                FailurePoint::Handshake => assert!(!requests.is_empty()),
-                FailurePoint::Portal | FailurePoint::Client | FailurePoint::Storage => assert!(requests.is_empty()),
+                FailurePoint::Handshake => assert_ne!(requests.len(), 0),
+                FailurePoint::Portal | FailurePoint::Client | FailurePoint::Storage => assert_eq!(requests.len(), 0),
             }
         }
     }
@@ -725,7 +725,7 @@ async fn coderabbit_xtream_http_failure_preserves_successful_disk_clusters() {
         assert_eq!(fetch.failed_clusters, vec![XtreamCluster::Video]);
         assert!(!fetch.is_ok(), "the failed cluster must remain a technical failure");
         assert!(fetch.persisted);
-        assert!(fetch.quality_rejections.is_empty());
+        assert_eq!(fetch.quality_rejections.len(), 0);
         let accepted: Vec<_> = match quality {
             UpdateQualityPolicy::Enforce => fetch.quality_acceptances.iter().map(|value| value.cluster).collect(),
             UpdateQualityPolicy::Bypass => fetch.force_updates.iter().map(|value| value.cluster).collect(),
@@ -805,7 +805,7 @@ async fn pipeline_transparency_normal_policy_reuses_a_valid_cluster_cache() {
 
     assert!(result.errors.is_empty());
     assert!(result.storage_error.is_none());
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     assert_eq!(result.job_state(), InputJobState::Ready);
     assert_baseline(&result.source.take_groups());
     assert!(requests.is_empty(), "valid cache should avoid provider requests: {requests:?}");
@@ -1215,7 +1215,7 @@ async fn force_policy_retains_a_cluster_after_a_technical_parse_failure() {
 
     assert!(!result.errors.is_empty(), "malformed VOD response must remain a technical error");
     assert!(result.storage_error.is_none());
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     let groups = result.source.take_groups();
     let vod = groups.iter().find(|group| group.xtream_cluster == XtreamCluster::Video).expect("retained VOD");
     assert_eq!(vod.title.as_ref(), "old-vod");
@@ -1278,7 +1278,7 @@ async fn in_memory_series_quality_compares_catalog_rows_with_an_enriched_baselin
     let requests = server.finish();
 
     assert!(result.errors.is_empty(), "Series update failed: {:?}", result.errors);
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     let groups = result.source.take_groups();
     assert_eq!(groups.len(), 1);
     assert_eq!(groups[0].title.as_ref(), "candidate-series");
@@ -1388,7 +1388,7 @@ async fn in_memory_quality_accepts_unique_population_and_persists_the_last_dupli
 
     assert!(result.errors.is_empty(), "deduplicated acceptance failed: {:?}", result.errors);
     assert!(result.storage_error.is_none());
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     let groups = result.source.take_groups();
     assert_eq!(groups.iter().map(|group| group.channels.len()).sum::<usize>(), 100);
     let winner_group = groups.iter().find(|group| group.id == 2).expect("winning category");
@@ -1443,7 +1443,7 @@ async fn m3u_update_quality_enforces_the_existing_cluster_threshold() {
     let mut initial = download_input(&initial_ctx, &input, false).await;
 
     assert!(initial.errors.is_empty(), "initial M3U download failed: {:?}", initial.errors);
-    assert!(initial.quality_rejections.is_empty());
+    assert_eq!(initial.quality_rejections.len(), 0);
     assert_eq!(initial.source.get_channel_count(), 2);
 
     tokio::fs::write(&playlist_path, "#EXTM3U\n#EXTINF:-1,New\nhttp://new.example/1\n")

backend/processing/src/processor/playlist/tests/quality_rejection_fallback/m3u_update_quality.rs

@@ -105,7 +105,7 @@ async fn m3u_update_quality_all_clusters_accepted() {
     let temp = tempfile::tempdir().unwrap();
     let (ctx, input) = fixture(temp.path(), Some([10, 10, 10]), [90, 90, 90]).await;
     let mut result = run(&ctx, &input, [9, 11, 10]).await;
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     assert_eq!(result.update_state(), PlaylistUpdateState::Success);
     verify_effective(&ctx, &input, &mut result, [9, 11, 10]).await;
 }
@@ -257,7 +257,7 @@ async fn m3u_update_quality_disabled() {
     let temp = tempfile::tempdir().unwrap();
     let (ctx, input) = fixture(temp.path(), Some([10, 10, 10]), [0, 0, 0]).await;
     let mut result = run(&ctx, &input, [1, 1, 0]).await;
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     assert!(result
         .input_telemetry
         .as_ref()
@@ -273,7 +273,7 @@ async fn m3u_update_quality_first_import() {
     let temp = tempfile::tempdir().unwrap();
     let (ctx, input) = fixture(temp.path(), None, [90, 90, 90]).await;
     let mut result = run(&ctx, &input, [1, 1, 1]).await;
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     assert!(result.input_telemetry.as_ref().unwrap().clusters.iter().all(|c| c.quality.is_none()
         && c.baseline_count.is_none()
         && c.decision == Some(PlaylistUpdateClusterDecision::Accepted)));
@@ -292,7 +292,7 @@ async fn m3u_update_quality_download_or_parse_failure() {
         }
         let mut result = download_input(&ctx, &input, false).await;
         assert!(!result.errors.is_empty());
-        assert!(result.quality_rejections.is_empty());
+        assert_eq!(result.quality_rejections.len(), 0);
         assert_eq!(result.update_state(), PlaylistUpdateState::Failure);
         let mut saved = load_input_playlist(&ctx.config, &input, None).await.unwrap();
         assert_eq!(counts(&saved.take_groups()), [10, 10, 10]);
@@ -324,7 +324,7 @@ async fn m3u_update_quality_skip_preserves_previous_and_reuse_emits_no_new_facts
     Arc::make_mut(&mut input).options.as_mut().unwrap().flags.set(ConfigInputFlags::SkipSeries);
     let mut result = run(&ctx, &input, [10, 10, 0]).await;
     assert_eq!(result.update_state(), PlaylistUpdateState::Success);
-    assert!(result.quality_rejections.is_empty());
+    assert_eq!(result.quality_rejections.len(), 0);
     let mut saved = load_input_playlist(&ctx.config, &input, None).await.unwrap();
     assert_eq!(counts(&result.source.take_groups()), [10, 10, 0]);
     assert_eq!(counts(&saved.take_groups()), [10, 10, 10]);
@@ -388,7 +388,7 @@ async fn m3u_update_quality_force_keeps_parse_and_empty_publication_errors() {
         tokio::fs::write(&input.url, body).await.unwrap();
         let mut result = download_input(&ctx, &input, false).await;
         assert_eq!(result.update_state(), PlaylistUpdateState::Failure);
-        assert!(result.quality_rejections.is_empty());
+        assert_eq!(result.quality_rejections.len(), 0);
         let mut saved = load_input_playlist(&ctx.config, &input, None).await.unwrap();
         assert_eq!(counts(&saved.take_groups()), [10; 3]);
     }

backend/processing/src/processor/playlist/tests/quality_rejection_fallback/persisted_input_status.rs

@@ -93,7 +93,7 @@ async fn m3u_update_quality_playlist_update_status_staged_completion_is_independ
         let telemetry = completions[0].input_telemetry.as_ref().unwrap();
         assert_eq!(telemetry.source, None);
         assert!(telemetry.clusters.iter().all(|cluster| cluster.source == Some(PlaylistUpdateDataSource::Provider)));
-        assert!(staged_completion::input_completions(&events, parent.id).is_empty());
+        assert_eq!(staged_completion::input_completions(&events, parent.id).len(), 0);
     }
 }
 
@@ -243,7 +243,7 @@ async fn playlist_update_status_parent_download_persists_staged_own_per_cluster_
         let skipped = download_input(&ctx, &staged, false).await;
         assert!(skipped.input_telemetry.is_none());
         assert!(input_cache::load_input_status(&staged_path).last_input_update.is_none());
-        assert!(staged_completion::input_completions(&events, staged.id).is_empty());
+        assert_eq!(staged_completion::input_completions(&events, staged.id).len(), 0);
 
         let result = download_input(&ctx, &parent, false).await;
         let status = input_cache::load_input_status(&staged_path);

backend/processing/src/processor/stalker.rs

@@ -667,7 +667,7 @@ mod tests {
     fn playback_candidates_are_empty_without_any_command() {
         let item = stalker_item("", None);
 
-        assert!(playback_candidates(&item).is_empty());
+        assert_eq!(playback_candidates(&item).len(), 0);
     }
 
     #[test]

backend/processing/src/processor/stalker_refresh.rs

@@ -1314,7 +1314,7 @@ mod tests {
             finish_completed_refresh(&app_config, temp.path(), identity, &checkpoint, update_quality(0, 0, 100))
                 .await?;
 
-        assert!(rejections.is_empty());
+        assert_eq!(rejections.len(), 0);
         let published = load_active_manifest(temp.path(), identity).await?;
         assert_eq!(published.series.as_ref().map(|files| files.generation), Some(generation));
         Ok(())
@@ -1377,7 +1377,7 @@ mod tests {
         )
         .await?;
 
-        assert!(reports.quality_rejections.is_empty());
+        assert_eq!(reports.quality_rejections.len(), 0);
         assert_eq!(
             reports.force_updates,
             vec![ClusterForceUpdate { cluster: XtreamCluster::Video, candidate_count: 0, configured_threshold: 100 }]
@@ -1421,7 +1421,7 @@ mod tests {
         )
         .await?;
 
-        assert!(reports.quality_rejections.is_empty());
+        assert_eq!(reports.quality_rejections.len(), 0);
         assert_eq!(reports.force_updates.len(), 1);
         assert_eq!(reports.force_updates[0].cluster, XtreamCluster::Live);
         let published = load_active_manifest(temp.path(), identity).await?;