feat(hook): 启用 segment_processor 阶段，新增 remux:mp4 内建钩子 (#1626)

* feat(hook): built-in remux:mp4 hook + activate segment_processor stage

B站 transcoder rejects raw .ts uploads with PTS jumps ("您的视频未能成功
转码 ... 该视频内部存在时间戳跳变"), which is common on HLS reconnects
mid-segment. The fix is to feed it an mp4 with monotonic PTS — a plain
ffmpeg copy-remux is enough.

Add a built-in `HookStep::Remux { remux: String }` variant that runs
`ffmpeg -fflags +genpts+igndts -c copy -bsf:a aac_adtstoasc
-movflags +faststart -avoid_negative_ts make_zero` on .ts/.m2ts inputs.
Lossless and idempotent (reuses existing .mp4 sibling). Currently
target="mp4" only.

To make this useful, segment_processor (defined in config but never
previously invoked) is now executed in pipeline_upload_videos BEFORE
each upload_single_file. That's the right stage for ts→mp4 — must
run before pre_upload, not after.

Path-mutating processor pipeline:
- new execute_paths(&mut Vec<PathBuf>) API on HookStep
- new process_video_paths helper
- Remux replaces .ts entries with the new .mp4 entries so downstream
  stages (upload, postprocessor) see the right paths.
The legacy execute(&[&Path]) path stays for postprocessor-style steps
that don't mutate the working set.

Config form:

  segment_processor:
    - {remux: "mp4"}

Tests cover YAML/JSON deserialization of all 4 HookStep variants
and Remux skipping non-.ts inputs.

* fix(upload): 单段 segment_processor / upload 失败不再拖死整批

之前 pipeline_upload_videos 里两个 ? 把单段失败直接抛到
process_with_upload，让后续 submit + execute_postprocessor（含 rm）
全部跳过。在磁盘紧的场景下，ffmpeg remux 写 mp4 header 时遇到
"No space left on device" → segment_processor Err → 整个直播会话已
成功上传的段也不会触发本地 rm → 文件堆积 → 磁盘更紧 → 下一段又失败，
正反馈把磁盘吃完。

把两处 ? 改成 log + continue：
- segment_processor 失败：跳过当前段，下个事件继续
- upload_single_file 失败：跳过，已上传的段照常进入 uploaded.{videos,paths}

直播会话结束（channel close）时，submit_to_bilibili 仍然能用收集到
的 videos 提交，execute_postprocessor 仍能 rm 已成功的本地文件。

修复对应残留 1.6 TB 历史文件的根因（35 次/5 天 "Process segment
event failed: ffmpeg remux failed ... No space left on device"）。

* test(hook): remux 在 ffmpeg 失败时返回 Err（供 pipeline 容错前置条件）

复现 pipeline_upload_videos 容错路径的触发条件：segment_processor
(Remux step) 在 ffmpeg 退非 0 时返回 Err。生产事故的具体表现就是磁盘
满让 ffmpeg 写 mp4 header 失败，调用方必须能从该 Err 优雅恢复
（log + continue），否则整批已上传的段都拿不到 postprocessor 的 rm。

完整 pipeline_upload_videos 的端到端测试要 mock UploadContext
(BiliBili client + 登录态)，这是后续重构 seam 的方向。

crates/biliup-cli/src/server/common/upload.rs

@@ -5,7 +5,9 @@ use crate::server::core::downloader::SegmentInfo;
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus};
 use crate::server::infrastructure::models::InsertFileItem;
-use crate::server::infrastructure::models::hook_step::process_video;
+use crate::server::infrastructure::models::hook_step::{
+    HookStep, process_video, process_video_paths,
+};
 use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 use async_channel::Receiver;
 use biliup::bilibili::{BiliBili, ResponseData, Studio, Video};
@@ -52,8 +54,15 @@ where
     let upload_context =
         initialize_upload_context(&ctx.config(), &ctx.stateless_client(), upload_config).await?;
 
-    // 2. 流水线处理视频上传
-    let uploaded_videos = pipeline_upload_videos(rx, &upload_context).await?;
+    // 2. 流水线处理视频上传（segment_processor 在每段上传前执行；用于 Remux 等
+    // 在原地改写路径的预处理）
+    let segment_processors: Vec<HookStep> = ctx
+        .live_streamer()
+        .segment_processor
+        .clone()
+        .unwrap_or_default();
+    let uploaded_videos =
+        pipeline_upload_videos(rx, &upload_context, &segment_processors).await?;
 
     // 3. 提交到B站
     if !uploaded_videos.videos.is_empty() {
@@ -122,27 +131,46 @@ async fn get_upload_line(client: &reqwest::Client, line: &str) -> AppResult<Line
 async fn pipeline_upload_videos<F>(
     rx: Inspect<Receiver<SegmentInfo>, F>,
     context: &UploadContext,
+    segment_processors: &[HookStep],
 ) -> AppResult<UploadedVideos>
 where
     F: FnMut(&SegmentInfo),
 {
-    // let mut desc_v2 = Vec::new();
-    // for credit in context.upload_config.desc_v2_credit {
-    //     desc_v2.push(Credit {
-    //         type_id: credit.type_id,
-    //         raw_text: credit.raw_text,
-    //         biz_id: credit.biz_id,
-    //     });
-    // }
-
     let mut uploaded = UploadedVideos::default();
     pin!(rx);
     // 流式处理后续事件
     while let Some(event) = rx.next().await {
-        let video = upload_single_file(&event.prev_file_path, context).await?;
-        uploaded.videos.push(video);
-        uploaded.paths.push(event.prev_file_path);
-        // 失败的文件不加入路径列表，避免后处理出错
+        // segment_processor 在上传前对路径列表做就地转换（如 Remux .ts→.mp4）。
+        // 单段失败（典型场景：磁盘满让 ffmpeg remux 写头失败）不应拖死整批——
+        // 否则已成功上传的段也无法到达 submit + postprocessor，本地 `rm` 不触发，
+        // 文件越堆越多，磁盘进一步紧张，形成正反馈。
+        let mut paths: Vec<PathBuf> = vec![event.prev_file_path.clone()];
+        if !segment_processors.is_empty()
+            && let Err(e) = process_video_paths(&mut paths, segment_processors).await
+        {
+            error!(
+                file = ?event.prev_file_path,
+                "segment_processor failed, skipping segment: {:?}", e
+            );
+            continue;
+        }
+        let upload_path = paths
+            .first()
+            .cloned()
+            .unwrap_or_else(|| event.prev_file_path.clone());
+        match upload_single_file(&upload_path, context).await {
+            Ok(video) => {
+                uploaded.videos.push(video);
+                // postprocessor 看到的是 segment_processor 转换后的路径
+                uploaded.paths.push(upload_path);
+            }
+            Err(e) => {
+                error!(
+                    file = ?upload_path,
+                    "upload_single_file failed, skipping segment: {:?}", e
+                );
+            }
+        }
     }
 
     Ok(uploaded)

crates/biliup-cli/src/server/infrastructure/models/hook_step.rs

@@ -2,11 +2,12 @@ use crate::server::errors::{AppError, AppResult};
 use error_stack::{ResultExt, bail};
 use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
-use std::path::Path;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
 use tokio::fs;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::Command;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 /// 钩子步骤枚举：支持多种操作格式
 /// 既支持 key-value 形式（如 {run: "..."}），也支持纯字符串（如 "rm"）
@@ -17,6 +18,12 @@ pub enum HookStep {
     Run { run: String },
     /// 移动文件格式：{mv: "target_dir"}
     Move { mv: String },
+    /// ts→mp4 无重编码 remux 格式：{remux: "mp4"}
+    /// 解决 B 站对 .ts 直传时常见的 "时间戳跳变" 转码失败。
+    /// 用 ffmpeg `-c copy -fflags +genpts+igndts -bsf:a aac_adtstoasc -movflags +faststart`，
+    /// 重新生成 PTS。仅当输入扩展名为 .ts/.m2ts 时生效；其它格式跳过。
+    /// 路径列表会被原地替换为新生成的 .mp4，原文件保留（postprocessor 中再 "rm"）。
+    Remux { remux: String },
     /// 删除文件格式："rm"
     Remove(String),
 }
@@ -44,6 +51,13 @@ impl HookStep {
                 // 移动文件到指定目录
                 self.move_file(video_paths, mv).await?;
             }
+            HookStep::Remux { remux } => {
+                bail!(AppError::Custom(format!(
+                    "remux step needs the path-mutating execute_paths API; \
+                     called via the legacy execute path with target={}",
+                    remux
+                )));
+            }
             HookStep::Remove(cmd) if cmd == "rm" => {
                 // 删除文件
                 HookStep::remove_file(video_paths).await?;
@@ -56,6 +70,110 @@ impl HookStep {
         Ok(())
     }
 
+    /// Path-mutating variant: lets steps swap entries in the working set
+    /// (e.g. Remux replaces .ts with the newly-created .mp4).
+    pub async fn execute_paths(&self, video_paths: &mut Vec<PathBuf>) -> AppResult<()> {
+        match self {
+            HookStep::Remux { remux } => {
+                let target = remux.to_lowercase();
+                if target != "mp4" {
+                    bail!(AppError::Custom(format!(
+                        "remux: only target=\"mp4\" supported, got {}",
+                        remux
+                    )));
+                }
+                for p in video_paths.iter_mut() {
+                    let ext = p
+                        .extension()
+                        .and_then(|s| s.to_str())
+                        .map(|s| s.to_lowercase())
+                        .unwrap_or_default();
+                    if ext != "ts" && ext != "m2ts" {
+                        info!("remux: skipping non-ts file {}", p.display());
+                        continue;
+                    }
+                    let mp4 = p.with_extension("mp4");
+                    if mp4.exists() && tokio::fs::metadata(&mp4).await
+                        .map(|m| m.len() > 0)
+                        .unwrap_or(false)
+                    {
+                        info!("remux: reusing existing {}", mp4.display());
+                        *p = mp4;
+                        continue;
+                    }
+                    Self::ffmpeg_remux_to_mp4(p, &mp4).await?;
+                    *p = mp4;
+                }
+            }
+            other => {
+                // For non-path-mutating steps, fall back to the read-only API.
+                let refs: Vec<&Path> = video_paths.iter().map(|p| p.as_path()).collect();
+                other.execute(&refs).await?;
+            }
+        }
+        Ok(())
+    }
+
+    async fn ffmpeg_remux_to_mp4(src: &Path, dst: &Path) -> AppResult<()> {
+        info!("remux ts→mp4: {} → {}", src.display(), dst.display());
+        let started = std::time::Instant::now();
+        let status = Command::new("ffmpeg")
+            .args([
+                "-hide_banner",
+                "-loglevel",
+                "warning",
+                "-y",
+                "-fflags",
+                "+genpts+igndts",
+                "-i",
+            ])
+            .arg(src)
+            .args([
+                "-c",
+                "copy",
+                "-bsf:a",
+                "aac_adtstoasc",
+                "-movflags",
+                "+faststart",
+                "-avoid_negative_ts",
+                "make_zero",
+            ])
+            .arg(dst)
+            .kill_on_drop(true)
+            .status()
+            .await
+            .change_context(AppError::Custom("failed to spawn ffmpeg".into()))?;
+        if !status.success() {
+            // Clean up partial output so a retry restarts cleanly.
+            let _ = tokio::fs::remove_file(dst).await;
+            bail!(AppError::Custom(format!(
+                "ffmpeg remux failed (status {:?}) for {}",
+                status,
+                src.display()
+            )));
+        }
+        let meta = tokio::fs::metadata(dst).await.change_context_lazy(|| {
+            AppError::Custom(format!("remux output missing: {}", dst.display()))
+        })?;
+        if meta.len() == 0 {
+            let _ = tokio::fs::remove_file(dst).await;
+            bail!(AppError::Custom(format!(
+                "ffmpeg produced empty mp4: {}",
+                dst.display()
+            )));
+        }
+        info!(
+            "remux done {} ({:.1} MiB) in {:.1}s",
+            dst.display(),
+            meta.len() as f64 / 1048576.0,
+            started.elapsed().as_secs_f64()
+        );
+        // Suppress unused-import warning for Duration on platforms that don't
+        // need it (kept for future timeout work).
+        let _ = Duration::from_secs(0);
+        Ok(())
+    }
+
     /// 执行钩子步骤操作
     ///
     /// # 参数
@@ -264,6 +382,85 @@ pub async fn process_video(video_path: &[&Path], processors: &[HookStep]) -> App
     Ok(())
 }
 
+/// Path-mutating processor pipeline. After each step the path list reflects any
+/// in-place transforms (e.g. Remux replaces .ts with .mp4). Callers can read the
+/// final paths to find the actual files to upload / postprocess.
+pub async fn process_video_paths(
+    video_paths: &mut Vec<PathBuf>,
+    processors: &[HookStep],
+) -> AppResult<()> {
+    info!("Starting video processing (path-mutating)...");
+    for processor in processors {
+        processor.execute_paths(video_paths).await?;
+    }
+    info!("Video processing completed");
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn deserialize_yaml_variants() {
+        let yaml = r#"
+- run: "echo hi"
+- mv: "/dst"
+- remux: "mp4"
+- "rm"
+"#;
+        let steps: Vec<HookStep> = serde_yaml::from_str(yaml).unwrap();
+        assert_eq!(steps.len(), 4);
+        assert!(matches!(steps[0], HookStep::Run { .. }));
+        assert!(matches!(steps[1], HookStep::Move { .. }));
+        assert!(matches!(steps[2], HookStep::Remux { .. }));
+        assert!(matches!(steps[3], HookStep::Remove(_)));
+    }
+
+    #[test]
+    fn deserialize_json_variants() {
+        let json = r#"[
+            {"run": "echo hi"},
+            {"mv": "/dst"},
+            {"remux": "mp4"},
+            "rm"
+        ]"#;
+        let steps: Vec<HookStep> = serde_json::from_str(json).unwrap();
+        assert_eq!(steps.len(), 4);
+        assert!(matches!(steps[2], HookStep::Remux { .. }));
+    }
+
+    #[tokio::test]
+    async fn remux_skips_non_ts() {
+        let dir = tempfile::tempdir().unwrap();
+        let p = dir.path().join("foo.mp4");
+        std::fs::write(&p, b"not a real mp4 but we shouldn't touch it").unwrap();
+        let mut paths = vec![p.clone()];
+        let step = HookStep::Remux { remux: "mp4".into() };
+        step.execute_paths(&mut paths).await.unwrap();
+        // Already .mp4 → unchanged.
+        assert_eq!(paths[0], p);
+    }
+
+    /// 复现 pipeline_upload_videos 的 fault-tolerance 触发条件：
+    /// segment_processor 在 ffmpeg 退非 0 时返回 Err。生产事故就是这一步因磁盘满
+    /// 而失败，调用方必须能从 Err 恢复（log + continue 而不是 ? 早退）。
+    #[tokio::test]
+    async fn remux_fails_when_ffmpeg_cannot_read_input() {
+        let dir = tempfile::tempdir().unwrap();
+        let bogus = dir.path().join("does_not_exist.ts");
+        let mut paths = vec![bogus.clone()];
+        let step = HookStep::Remux { remux: "mp4".into() };
+        let result = step.execute_paths(&mut paths).await;
+        assert!(
+            result.is_err(),
+            "ffmpeg 读不到输入应返回 Err，调用方才能在 pipeline 里做 log+continue"
+        );
+        // 失败时不要把破损的 mp4 留在磁盘上（idempotent retry 前提）
+        assert!(!bogus.with_extension("mp4").exists());
+    }
+}
+
 pub async fn process(input: &[u8], processors: &Option<Vec<HookStep>>) {
     if let Some(hooks) = processors {
         // 依次执行每个处理器步骤