diff --git a/docs/source/guides/configuration.rst b/docs/source/guides/configuration.rst
index 17cdfecc90..5ed0deba77 100644
--- a/docs/source/guides/configuration.rst
+++ b/docs/source/guides/configuration.rst
@@ -354,8 +354,9 @@ A list of all config options
    'ffmpeg_loglevel', 'flush_cache', 'frame_height', 'frame_rate',
    'frame_size', 'frame_width', 'frame_x_radius', 'frame_y_radius',
    'from_animation_number', `fullscreen`, 'images_dir', 'input_file', 'left_side',
-   'log_dir', 'log_to_file', 'max_files_cached', 'media_dir', 'media_width',
-   'movie_file_extension', 'notify_outdated_version', 'output_file', 'partial_movie_dir',
+   'log_dir', 'log_to_file', 'max_files_cached', 'max_inflight_encoders',
+   'media_dir', 'media_width', 'movie_file_extension', 'notify_outdated_version',
+   'output_file', 'partial_movie_dir',
    'pixel_height', 'pixel_width', 'plugins', 'preview',
    'progress_bar', 'quality', 'right_side', 'save_as_gif', 'save_last_frame',
    'save_pngs', 'scene_names', 'show_in_file_browser', 'sound', 'tex_dir',
diff --git a/manim/_config/default.cfg b/manim/_config/default.cfg
index 9155d28dbd..4fe2cfac1e 100644
--- a/manim/_config/default.cfg
+++ b/manim/_config/default.cfg
@@ -129,6 +129,13 @@ use_projection_stroke_shaders = False
 
 movie_file_extension = .mp4
 
+# Maximum number of partial movie files being encoded concurrently while the
+# scene continues rendering. 1 encodes each animation's file before the next
+# animation starts; values > 1 overlap encoding with rendering (bounds pipeline
+# depth, not CPU use; each encoder already slice-threads internally).
+# 4 is a good value on typical hardware.
+max_inflight_encoders = 1
+
 # These now override the --quality option.
 frame_rate = 60
 pixel_height = 1080
diff --git a/manim/_config/utils.py b/manim/_config/utils.py
index 3e45846539..44d29c8ee4 100644
--- a/manim/_config/utils.py
+++ b/manim/_config/utils.py
@@ -283,6 +283,7 @@ class MyScene(Scene): ...
         "log_dir",
         "log_to_file",
         "max_files_cached",
+        "max_inflight_encoders",
         "media_dir",
         "movie_file_extension",
         "notify_outdated_version",
@@ -605,6 +606,7 @@ def digest_parser(self, parser: configparser.ConfigParser) -> Self:
             "from_animation_number",
             "upto_animation_number",
             "max_files_cached",
+            "max_inflight_encoders",
             # the next two must be set BEFORE digesting frame_width and frame_height
             "pixel_height",
             "pixel_width",
@@ -1227,6 +1229,22 @@ def max_files_cached(self) -> int:
     def max_files_cached(self, value: int) -> None:
         self._set_pos_number("max_files_cached", value, True)
 
+    @property
+    def max_inflight_encoders(self) -> int:
+        """Maximum number of partial movie files encoded concurrently while the
+        scene continues rendering. 1 encodes each animation's file before the
+        next animation starts; values > 1 overlap encoding with rendering
+        (4 is a good value on typical hardware). No flag.
+        """
+        return self._d["max_inflight_encoders"]
+
+    @max_inflight_encoders.setter
+    def max_inflight_encoders(self, value: int) -> None:
+        if isinstance(value, int) and value >= 1:
+            self._d.__setitem__("max_inflight_encoders", value)
+        else:
+            raise ValueError("max_inflight_encoders must be a positive integer")
+
     @property
     def window_monitor(self) -> int:
         """The monitor on which the scene will be rendered."""
diff --git a/manim/scene/scene.py b/manim/scene/scene.py
index 845fafd0b9..68b9b6a22a 100644
--- a/manim/scene/scene.py
+++ b/manim/scene/scene.py
@@ -264,6 +264,9 @@ def render(self, preview: bool = False) -> bool:
             # TODO: The CairoRenderer does not have the method clear_screen()
             self.renderer.clear_screen()  # type: ignore[union-attr]
             self.renderer.num_plays = 0
+            # The rerun replaces the file writer; drain its encode jobs so no
+            # worker is still writing a partial file the new writer may reuse.
+            self.renderer.file_writer.join_all_encode_jobs()
             return True
         self.tear_down()
         # We have to reset these settings in case of multiple renders.
diff --git a/manim/scene/scene_file_writer.py b/manim/scene/scene_file_writer.py
index 21425af759..6dc8e67bf3 100644
--- a/manim/scene/scene_file_writer.py
+++ b/manim/scene/scene_file_writer.py
@@ -90,6 +90,93 @@ def convert_audio(
             output_audio.mux(packet)
 
 
+class _PartialMovieEncodeJob:
+    """Encode and write the frames for one partial movie file."""
+
+    def __init__(
+        self,
+        path: StrPath,
+        animation_index: int,
+        container: OutputContainer,
+        stream: Stream,
+    ) -> None:
+        self.path = path
+        self.animation_index = animation_index
+        self.container = container
+        self.stream = stream
+        # About 66 MB of 1080p RGBA frames per job. The worker drains through
+        # the sentinel after an exception, so a bounded queue cannot deadlock.
+        self.queue: Queue[tuple[int, PixelArray | None]] = Queue(maxsize=8)
+        self._exception: BaseException | None = None
+        self.thread = Thread(
+            target=self._listen_and_write,
+            name=f"partial-movie-encoder-{animation_index}",
+        )
+        self.thread.start()
+
+    def _capture_exception(self, exception: BaseException) -> None:
+        if self._exception is None:
+            self._exception = exception
+
+    def _listen_and_write(self) -> None:
+        while True:
+            num_frames, frame_data = self.queue.get()
+            if frame_data is None:
+                break
+            if self._exception is not None:
+                continue
+
+            try:
+                self._encode_and_write_frame(frame_data, num_frames)
+            except BaseException as exception:
+                self._capture_exception(exception)
+
+        try:
+            for packet in self.stream.encode():
+                self.container.mux(packet)
+        except BaseException as exception:
+            self._capture_exception(exception)
+        finally:
+            try:
+                self.container.close()
+            except BaseException as exception:
+                self._capture_exception(exception)
+
+    def _encode_and_write_frame(self, frame: PixelArray, num_frames: int) -> None:
+        for _ in range(num_frames):
+            # Notes: precomputing reusing packets does not work!
+            # I.e., you cannot do `packets = encode(...)`
+            # and reuse it, as it seems that `mux(...)`
+            # consumes the packet.
+            # The same issue applies for `av_frame`,
+            # reusing it renders weird-looking frames.
+            av_frame = av.VideoFrame.from_ndarray(frame, format="rgba")
+            for packet in self.stream.encode(av_frame):
+                self.container.mux(packet)
+
+    def put(self, num_frames: int, frame: PixelArray) -> None:
+        """Add a frame to the encoding queue."""
+        self.queue.put((num_frames, frame))
+
+    def seal(self) -> None:
+        """Signal that no more frames will be added."""
+        self.queue.put((-1, None))
+
+    def join(self) -> None:
+        """Wait for encoding to finish and propagate worker failures."""
+        self.thread.join()
+        if self._exception is not None:
+            # A failed encode may leave a structurally valid but truncated
+            # file behind; remove it so a later run cannot cache-hit it.
+            Path(self.path).unlink(missing_ok=True)
+            raise self._exception
+
+        logger.info(
+            f"Animation {self.animation_index} : Partial movie file written in %(path)s",
+            {"path": f"'{self.path}'"},
+        )
+
+
 class SceneFileWriter:
     """SceneFileWriter is the object that actually writes the animations
     played, into video files, using FFMPEG.
@@ -127,6 +214,9 @@ def __init__(
         **kwargs: Any,
     ) -> None:
         self.renderer = renderer
+        self._inflight_encode_jobs: list[_PartialMovieEncodeJob] = []
+        self._inflight_by_path: dict[str, _PartialMovieEncodeJob] = {}
+        self._current_encode_job: _PartialMovieEncodeJob | None = None
         self.init_output_directories(scene_name)
         self.init_audio()
         self.frame_count = 0
@@ -426,30 +516,6 @@ def end_animation(self, allow_write: bool = False) -> None:
         if write_to_movie() and allow_write:
             self.close_partial_movie_stream()
 
-    def listen_and_write(self) -> None:
-        """For internal use only: blocks until new frame is available on the queue."""
-        while True:
-            num_frames, frame_data = self.queue.get()
-            if frame_data is None:
-                break
-
-            self.encode_and_write_frame(frame_data, num_frames)
-
-    def encode_and_write_frame(self, frame: PixelArray, num_frames: int) -> None:
-        """For internal use only: takes a given frame in ``np.ndarray`` format and
-        writes it to the stream
-        """
-        for _ in range(num_frames):
-            # Notes: precomputing reusing packets does not work!
-            # I.e., you cannot do `packets = encode(...)`
-            # and reuse it, as it seems that `mux(...)`
-            # consumes the packet.
-            # The same issue applies for `av_frame`,
-            # reusing it renders weird-looking frames.
-            av_frame = av.VideoFrame.from_ndarray(frame, format="rgba")
-            for packet in self.video_stream.encode(av_frame):
-                self.video_container.mux(packet)
-
     def write_frame(
         self, frame_or_renderer: PixelArray | OpenGLRenderer, num_frames: int = 1
     ) -> None:
@@ -472,8 +538,12 @@ def write_frame(
                     else frame_or_renderer
                 )
 
-            msg = (num_frames, frame)
-            self.queue.put(msg)
+            job = self._current_encode_job
+            if job is None:
+                # Interactive OpenGL rendering emits frames outside an open
+                # partial movie stream; drop them silently.
+                return
+            job.put(num_frames, frame)
 
         if is_png_format() and not config["dry_run"]:
             if isinstance(frame_or_renderer, np.ndarray):
@@ -524,9 +594,11 @@ def finish(self) -> None:
         If save_last_frame is True, saves the last frame in the default image directory.
         """
         if write_to_movie():
+            self.join_all_encode_jobs()
             self.combine_to_movie()
             if config.save_sections:
                 self.combine_to_section_videos()
+            # Cache cleanup runs after the in-flight encode jobs have been drained.
             if config["flush_cache"]:
                 self.flush_cache_directory()
             else:
@@ -545,6 +617,14 @@ def open_partial_movie_stream(self, file_path: StrPath | None = None) -> None:
         """
         if file_path is None:
             file_path = self.partial_movie_files[self.renderer.num_plays]
+            if file_path is None:
+                raise RuntimeError(
+                    "open_partial_movie_stream() called for a play that has no "
+                    "partial movie file path.",
+                )
+        path_key = str(file_path)
+        if path_key in self._inflight_by_path:
+            self._join_job(self._inflight_by_path[path_key])
         self.partial_movie_file_path = file_path
 
         fps = to_av_frame_rate(config.frame_rate)
@@ -576,12 +656,34 @@ def open_partial_movie_stream(self, file_path: StrPath | None = None) -> None:
         stream.width = config.pixel_width
         stream.height = config.pixel_height
 
-        self.video_container: OutputContainer = video_container
-        self.video_stream: Stream = stream
+        self._current_encode_job = _PartialMovieEncodeJob(
+            path=file_path,
+            animation_index=self.renderer.num_plays,
+            container=video_container,
+            stream=stream,
+        )
 
-        self.queue: Queue[tuple[int, PixelArray | None]] = Queue()
-        self.writer_thread = Thread(target=self.listen_and_write, args=())
-        self.writer_thread.start()
+    def _join_job(self, job: _PartialMovieEncodeJob) -> None:
+        """Remove and join an in-flight partial movie encode job."""
+        if job in self._inflight_encode_jobs:
+            self._inflight_encode_jobs.remove(job)
+        self._inflight_by_path.pop(str(job.path), None)
+        job.join()
+
+    def join_all_encode_jobs(self) -> None:
+        """Join every in-flight encode job, re-raising the first failure."""
+        first_exception: BaseException | None = None
+        for job in list(self._inflight_encode_jobs):
+            try:
+                self._join_job(job)
+            except BaseException as exception:
+                if first_exception is None:
+                    first_exception = exception
+
+        self._inflight_encode_jobs.clear()
+        self._inflight_by_path.clear()
+        if first_exception is not None:
+            raise first_exception
 
     def close_partial_movie_stream(self) -> None:
         """Close the currently opened video container.
@@ -590,18 +692,19 @@ def close_partial_movie_stream(self) -> None:
         in the video stream holding a partial file, and then close
         the corresponding container.
         """
-        self.queue.put((-1, None))
-        self.writer_thread.join()
-
-        for packet in self.video_stream.encode():
-            self.video_container.mux(packet)
-
-        self.video_container.close()
+        job = self._current_encode_job
+        if job is None:
+            raise RuntimeError(
+                "close_partial_movie_stream() called without an open partial "
+                "movie stream.",
+            )
+        job.seal()
+        self._inflight_encode_jobs.append(job)
+        self._inflight_by_path[str(job.path)] = job
+        self._current_encode_job = None
 
-        logger.info(
-            f"Animation {self.renderer.num_plays} : Partial movie file written in %(path)s",
-            {"path": f"'{self.partial_movie_file_path}'"},
-        )
+        while len(self._inflight_encode_jobs) >= config.max_inflight_encoders:
+            self._join_job(self._inflight_encode_jobs[0])
 
     def is_already_cached(self, hash_invocation: str) -> bool:
         """Will check if a file named with `hash_invocation` exists.
@@ -622,6 +725,9 @@ def is_already_cached(self, hash_invocation: str) -> bool:
             self.partial_movie_directory
             / f"{hash_invocation}{config['movie_file_extension']}"
         )
+        path_key = str(path)
+        if path_key in self._inflight_by_path:
+            self._join_job(self._inflight_by_path[path_key])
         return path.exists()
 
     def combine_files(
diff --git a/tests/test_scene_rendering/test_parallel_encoding.py b/tests/test_scene_rendering/test_parallel_encoding.py
new file mode 100644
index 0000000000..eb96f073c0
--- /dev/null
+++ b/tests/test_scene_rendering/test_parallel_encoding.py
@@ -0,0 +1,404 @@
+from __future__ import annotations
+
+import sys
+import textwrap
+import threading
+from unittest.mock import Mock
+
+import av
+import numpy as np
+import pytest
+
+from manim import FadeIn, Scene, Square, capture, tempconfig
+
+_ENCODER_THREAD_PREFIX = "partial-movie-encoder-"
+_UNIQUE_PLAYS = 30
+_TOTAL_PLAYS = _UNIQUE_PLAYS + 2
+
+_SCENE_NAME = "ParallelEncodingCacheScene"
+# Animation hashing memoizes objects by mixed hash()/id() signatures, so a
+# rare collision (heap-address reuse, per-process str-hash seeds) can flip one
+# play's cache key. Rendering in fresh interpreters (like the neighboring
+# subprocess-based tests) keeps that rate low but not zero, so the cache
+# assertions below distinguish a rare flip from a systematic regression.
+_SCENE_SOURCE = textwrap.dedent(
+    f"""\
+    from manim import FadeIn, Scene, Square
+
+
+    class {_SCENE_NAME}(Scene):
+        def construct(self):
+            for index in range({_UNIQUE_PLAYS}):
+                square = Square(side_length=0.2 + index / 100)
+                self.play(FadeIn(square), run_time=0.1)
+                self.clear()
+
+            self.play(FadeIn(Square(side_length=0.75)), run_time=0.1)
+            self.clear()
+            self.play(FadeIn(Square(side_length=0.75)), run_time=0.1)
+    """
+)
+
+
+def _alive_encoder_threads():
+    return [
+        thread
+        for thread in threading.enumerate()
+        if thread.name.startswith(_ENCODER_THREAD_PREFIX) and thread.is_alive()
+    ]
+
+
+def _render_scene(media_dir, scene_file):
+    command = [
+        sys.executable,
+        "-m",
+        "manim",
+        "-ql",
+        "--media_dir",
+        str(media_dir),
+        str(scene_file),
+        _SCENE_NAME,
+    ]
+    _, err, exit_code = capture(command)
+    assert exit_code == 0, err
+    quality_directory = media_dir / "videos" / scene_file.stem / "480p15"
+    return quality_directory, quality_directory / "partial_movie_files" / _SCENE_NAME
+
+
+@pytest.mark.slow
+def test_parallel_encoding_cache_behavior(tmp_path):
+    scene_file = tmp_path / "parallel_encoding_scenes.py"
+    scene_file.write_text(_SCENE_SOURCE)
+
+    # The duplicate tail play must cache-hit within the run (one file fewer
+    # than the number of plays); that hit exercises the writer's in-flight
+    # lookup while the first tail file may still be encoding. A hash collision
+    # (see note above) can rarely split the duplicate pair, in which case that
+    # path was not exercised, so retry once in a fresh media dir; two splits
+    # in a row indicate a real caching regression.
+    for attempt in range(2):
+        media_dir = tmp_path / f"attempt{attempt}"
+        quality_directory, partial_directory = _render_scene(media_dir, scene_file)
+        partial_movies = sorted(partial_directory.glob("*.mp4"))
+        if len(partial_movies) == _TOTAL_PLAYS - 1:
+            break
+    assert len(partial_movies) == _TOTAL_PLAYS - 1
+    assert (quality_directory / f"{_SCENE_NAME}.mp4").exists()
+    for partial_movie in partial_movies:
+        with av.open(partial_movie) as container:
+            next(container.decode(video=0))
+
+    partial_snapshot = {
+        partial_movie.name: partial_movie.stat().st_mtime_ns
+        for partial_movie in partial_movies
+    }
+
+    _render_scene(media_dir, scene_file)
+
+    second_snapshot = {
+        partial_movie.name: partial_movie.stat().st_mtime_ns
+        for partial_movie in sorted(partial_directory.glob("*.mp4"))
+    }
+    # Every cached partial must survive the second render untouched (not withstanding
+    # the rare hash collisions mentioned above). A play or two as extra files
+    # might be expected, but systematic growth means caching is broken.
+    for name, mtime_ns in partial_snapshot.items():
+        assert second_snapshot.get(name) == mtime_ns
+    assert len(second_snapshot) <= len(partial_snapshot) + 2
+
+
+@pytest.mark.slow
+def test_no_encoder_threads_survive_render(config, tmp_path):
+    class ThreadSweepScene(Scene):
+        def construct(self):
+            for index in range(3):
+                square = Square(side_length=0.3 + index / 10)
+                self.play(FadeIn(square), run_time=0.1)
+                self.clear()
+
+    with tempconfig({"media_dir": tmp_path, "quality": "low_quality"}):
+        scene = ThreadSweepScene()
+        scene.render()
+
+    assert not _alive_encoder_threads()
+
+
+def _frame():
+    return np.zeros((4, 4, 4), dtype=np.uint8)
+
+
+def _new_encode_job(tmp_path, monkeypatch, name, stream, container):
+    from manim.scene.scene_file_writer import _PartialMovieEncodeJob
+
+    job = _PartialMovieEncodeJob(
+        path=tmp_path / f"{name}.mp4",
+        animation_index=0,
+        stream=Mock(),
+        container=Mock(),
+    )
+    monkeypatch.setattr(job, "stream", stream)
+    monkeypatch.setattr(job, "container", container)
+    return job
+
+
+def _assert_failed_join(job, expected_exception):
+    job.thread.join(timeout=5)
+    assert not job.thread.is_alive(), "Partial movie encoder did not finish"
+
+    with pytest.raises(type(expected_exception)) as exc_info:
+        job.join()
+
+    assert exc_info.value is expected_exception
+    assert not _alive_encoder_threads()
+
+
+def test_encode_failure_propagates_and_drains_bounded_queue(
+    tmp_path,
+    monkeypatch,
+    manim_caplog,
+):
+    expected_exception = RuntimeError("encode failed")
+    encode_failed = threading.Event()
+    stream = Mock()
+    container = Mock()
+
+    def encode(*args):
+        if args:
+            encode_failed.set()
+            raise expected_exception
+        return []
+
+    stream.encode.side_effect = encode
+    job = _new_encode_job(tmp_path, monkeypatch, "encode_failure", stream, container)
+    job.put(1, _frame())
+    assert encode_failed.wait(timeout=2), "Encode failure was not triggered"
+
+    def fill_queue_and_seal():
+        for _ in range(job.queue.maxsize + 1):
+            job.put(1, _frame())
+        job.seal()
+
+    producer = threading.Thread(target=fill_queue_and_seal, daemon=True)
+    producer.start()
+    producer.join(timeout=5)
+    assert not producer.is_alive(), "Producer deadlocked on the bounded queue"
+
+    _assert_failed_join(job, expected_exception)
+    container.close.assert_called_once_with()
+    assert "Partial movie file written" not in manim_caplog.text
+
+
+def test_flush_failure_propagates_and_closes_container(
+    tmp_path,
+    monkeypatch,
+    manim_caplog,
+):
+    expected_exception = RuntimeError("flush failed")
+    stream = Mock()
+    container = Mock()
+
+    def encode(*args):
+        if args:
+            return []
+        raise expected_exception
+
+    stream.encode.side_effect = encode
+    job = _new_encode_job(tmp_path, monkeypatch, "flush_failure", stream, container)
+    job.put(1, _frame())
+    job.seal()
+
+    _assert_failed_join(job, expected_exception)
+    container.close.assert_called_once_with()
+    assert "Partial movie file written" not in manim_caplog.text
+
+
+def test_close_failure_propagates_after_close_attempt(
+    tmp_path,
+    monkeypatch,
+    manim_caplog,
+):
+    expected_exception = RuntimeError("close failed")
+    stream = Mock()
+    stream.encode.return_value = []
+    container = Mock()
+    container.close.side_effect = expected_exception
+    job = _new_encode_job(tmp_path, monkeypatch, "close_failure", stream, container)
+    job.put(1, _frame())
+    job.seal()
+
+    _assert_failed_join(job, expected_exception)
+    container.close.assert_called_once_with()
+    assert "Partial movie file written" not in manim_caplog.text
+
+
+def test_encode_failure_precedes_close_failure_and_removes_partial(
+    tmp_path,
+    monkeypatch,
+    manim_caplog,
+):
+    expected_exception = RuntimeError("encode failed")
+    close_exception = RuntimeError("close failed")
+    stream = Mock()
+    container = Mock()
+
+    def encode(*args):
+        if args:
+            raise expected_exception
+        return []
+
+    stream.encode.side_effect = encode
+    container.close.side_effect = close_exception
+    job = _new_encode_job(
+        tmp_path,
+        monkeypatch,
+        "encode_and_close_failure",
+        stream,
+        container,
+    )
+    job.path.write_bytes(b"stale")
+    job.put(1, _frame())
+    job.seal()
+
+    _assert_failed_join(job, expected_exception)
+    container.close.assert_called_once_with()
+    assert not job.path.exists()
+    assert "Partial movie file written" not in manim_caplog.text
+
+
+def test_successful_encode_job_logs_partial_movie_written(
+    tmp_path,
+    monkeypatch,
+    manim_caplog,
+):
+    stream = Mock()
+    stream.encode.return_value = []
+    container = Mock()
+    job = _new_encode_job(tmp_path, monkeypatch, "encode_success", stream, container)
+    job.put(1, _frame())
+    job.seal()
+
+    job.join()
+
+    container.close.assert_called_once_with()
+    assert "Partial movie file written" in manim_caplog.text
+    assert not _alive_encoder_threads()
+
+
+@pytest.mark.parametrize("max_inflight_encoders", [1, 2, 3])
+def test_close_partial_movie_stream_respects_cap_and_joins_fifo(
+    config,
+    tmp_path,
+    max_inflight_encoders,
+):
+    from manim.scene.scene_file_writer import SceneFileWriter
+
+    config.max_inflight_encoders = max_inflight_encoders
+    renderer = Mock()
+    renderer.num_plays = 0
+    writer = SceneFileWriter(renderer, "EncoderCapScene")
+    jobs = [Mock(path=tmp_path / f"partial_{index}.mp4") for index in range(3)]
+
+    for index, job in enumerate(jobs):
+        writer._current_encode_job = job
+        writer.close_partial_movie_stream()
+
+        closed_jobs = jobs[: index + 1]
+        expected_joined = max(
+            0,
+            len(closed_jobs) - max_inflight_encoders + 1,
+        )
+        expected_inflight = closed_jobs[expected_joined:]
+        assert writer._inflight_encode_jobs == expected_inflight
+        assert writer._inflight_by_path == {
+            str(inflight_job.path): inflight_job for inflight_job in expected_inflight
+        }
+        assert len(writer._inflight_encode_jobs) < max_inflight_encoders
+
+        for joined_job in closed_jobs[:expected_joined]:
+            joined_job.join.assert_called_once_with()
+            assert str(joined_job.path) not in writer._inflight_by_path
+        for inflight_job in expected_inflight:
+            inflight_job.join.assert_not_called()
+            assert writer._inflight_by_path[str(inflight_job.path)] is inflight_job
+
+    for job in jobs:
+        job.seal.assert_called_once_with()
+
+
+def test_is_already_cached_joins_same_path_inflight_job(config, tmp_path):
+    from manim.scene.scene_file_writer import SceneFileWriter
+
+    renderer = Mock()
+    renderer.num_plays = 0
+    writer = SceneFileWriter(renderer, "CachedInflightScene")
+    hash_invocation = "same_path_hash"
+    path = (
+        writer.partial_movie_directory
+        / f"{hash_invocation}{config['movie_file_extension']}"
+    )
+    job = Mock(path=path)
+    writer._inflight_encode_jobs.append(job)
+    writer._inflight_by_path[str(path)] = job
+
+    writer.is_already_cached(hash_invocation)
+
+    job.join.assert_called_once_with()
+    assert writer._inflight_encode_jobs == []
+    assert writer._inflight_by_path == {}
+
+
+def test_open_partial_movie_stream_joins_same_path_inflight_job(config, tmp_path):
+    from manim.scene.scene_file_writer import SceneFileWriter
+
+    renderer = Mock()
+    renderer.num_plays = 0
+    writer = SceneFileWriter(renderer, "OpenInflightScene")
+    path = tmp_path / "same_path.mp4"
+    inflight_job = Mock(path=path)
+    writer._inflight_encode_jobs.append(inflight_job)
+    writer._inflight_by_path[str(path)] = inflight_job
+
+    writer.open_partial_movie_stream(file_path=path)
+    current_job = writer._current_encode_job
+    assert current_job is not None
+    try:
+        inflight_job.join.assert_called_once_with()
+        assert writer._inflight_encode_jobs == []
+        assert writer._inflight_by_path == {}
+    finally:
+        current_job.seal()
+        current_job.join()
+        writer._current_encode_job = None
+
+    assert not _alive_encoder_threads()
+
+
+def test_finish_propagates_join_failure_and_clears_inflight_state(
+    config,
+    tmp_path,
+    monkeypatch,
+):
+    from manim.scene.scene_file_writer import SceneFileWriter
+
+    expected_exception = RuntimeError("join failed")
+    renderer = Mock()
+    renderer.num_plays = 0
+    writer = SceneFileWriter(renderer, "JoinFailureScene")
+    failing_job = Mock(path=tmp_path / "failing.mp4")
+    failing_job.join.side_effect = expected_exception
+    succeeding_job = Mock(path=tmp_path / "succeeding.mp4")
+    writer._inflight_encode_jobs.extend([failing_job, succeeding_job])
+    writer._inflight_by_path[str(failing_job.path)] = failing_job
+    writer._inflight_by_path[str(succeeding_job.path)] = succeeding_job
+    combine_to_movie = Mock()
+    monkeypatch.setattr(writer, "combine_to_movie", combine_to_movie)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        writer.finish()
+
+    assert exc_info.value is expected_exception
+    failing_job.join.assert_called_once_with()
+    succeeding_job.join.assert_called_once_with()
+    assert writer._inflight_encode_jobs == []
+    assert writer._inflight_by_path == {}
+    combine_to_movie.assert_not_called()