diff --git a/src/engine.rs b/src/engine.rs
index 1f1d55fd..87564997 100644
--- a/src/engine.rs
+++ b/src/engine.rs
@@ -2265,6 +2265,77 @@ pub(crate) mod tests {
         assert!(start_2 < start_3);
     }
 
+    #[test]
+    fn test_remove_recycled_files_left_by_unclean_shutdown() {
+        let dir = tempfile::Builder::new()
+            .prefix("test_unclean_recycled")
+            .tempdir()
+            .unwrap();
+        let entry_data = vec![b'x'; 16];
+        let cfg = Config {
+            dir: dir.path().to_str().unwrap().to_owned(),
+            target_file_size: ReadableSize(1),
+            purge_threshold: ReadableSize(50),
+            format_version: Version::V2,
+            enable_log_recycle: true,
+            prefill_for_recycle: true,
+            ..Default::default()
+        };
+        let fs = Arc::new(DefaultFileSystem);
+        let engine = RaftLogEngine::open_with_file_system(cfg, fs.clone()).unwrap();
+        for rid in 1..=10 {
+            engine.append(rid, 1, 11, Some(&entry_data));
+        }
+        for rid in 1..=10 {
+            engine.clean(rid);
+        }
+        engine.purge_manager.must_rewrite_append_queue(None, None);
+        // A clean shutdown renames recycled files to `*.raftlog.reserved`. Simulate
+        // an unclean shutdown (SIGKILL / OOM, `SinglePipe::drop` never runs) by
+        // renaming them back to plain append files, reproducing the leaked-files
+        // on-disk state reported in issue #407.
+        let engine = engine.reopen();
+        let dir_path = dir.path();
+        let mut reserved_names = Vec::new();
+        for entry in std::fs::read_dir(dir_path).unwrap() {
+            let name = entry.unwrap().file_name().into_string().unwrap();
+            if name.ends_with(".raftlog.reserved") {
+                reserved_names.push(name);
+            }
+        }
+        assert!(!reserved_names.is_empty());
+        let ghost_names: Vec<String> = reserved_names
+            .iter()
+            .map(|n| n.trim_end_matches(".reserved").to_owned())
+            .collect();
+        for (name, ghost) in reserved_names.iter().zip(ghost_names.iter()) {
+            std::fs::rename(dir_path.join(name), dir_path.join(ghost)).unwrap();
+        }
+        // Reopen: the leaked (ghost) files form a hole before the active range and
+        // must be physically removed instead of leaking on disk forever.
+        let engine = engine.reopen();
+        let disk_append_count = std::fs::read_dir(dir_path)
+            .unwrap()
+            .filter(|e| {
+                e.as_ref()
+                    .unwrap()
+                    .file_name()
+                    .into_string()
+                    .unwrap()
+                    .ends_with(".raftlog")
+            })
+            .count();
+        // Every plain `.raftlog` file on disk must belong to the active span; any
+        // leaked recycled file would show up as a stray file and break the equality.
+        assert_eq!(
+            engine.file_count(Some(LogQueue::Append)),
+            disk_append_count,
+            "leaked recycled files should be removed on restart"
+        );
+        let (start, end) = engine.file_span(LogQueue::Append);
+        assert!(start <= end);
+    }
+
     #[test]
     fn test_simple_write_perf_context() {
         let dir = tempfile::Builder::new()
diff --git a/src/file_pipe_log/pipe_builder.rs b/src/file_pipe_log/pipe_builder.rs
index 96af7dcb..b154afc1 100644
--- a/src/file_pipe_log/pipe_builder.rs
+++ b/src/file_pipe_log/pipe_builder.rs
@@ -177,7 +177,31 @@ impl<F: FileSystem> DualPipesBuilder<F> {
                     invalid_idx = i + 1;
                 }
             }
-            files.drain(..invalid_idx);
+            if invalid_idx > 0 {
+                // Files before the first hole are unreachable and are dropped from the
+                // in-memory list. Physically remove them as well, otherwise they leak
+                // on disk forever. This can happen when a process exits without running
+                // `SinglePipe::drop` (e.g. SIGKILL / OOM), leaving recycled append files
+                // that still use their original `.raftlog` name: on restart they are
+                // re-scanned as regular append files and form a hole before the active
+                // range, but only the metadata (not the log file) was cleaned before.
+                let invalid_files = files.drain(..invalid_idx).collect::<Vec<_>>();
+                if !is_recycled_file {
+                    // Collect the paths first so the file handles (and thus the files)
+                    // can be closed before deletion, which matters on Windows.
+                    let mut paths = Vec::with_capacity(invalid_files.len());
+                    for f in &invalid_files {
+                        let file_id = FileId { queue, seq: f.seq };
+                        paths.push(file_id.build_file_path(&self.dirs[f.path_id]));
+                    }
+                    drop(invalid_files);
+                    for path in paths {
+                        if let Err(e) = self.file_system.delete(&path) {
+                            error!("failed to delete leaked log file {}: {e}.", path.display());
+                        }
+                    }
+                }
+            }
             // Try to cleanup stale metadata left by the previous version.
             if files.is_empty() || is_recycled_file {
                 continue;