From 7623b2dc1b57b2f8beb1d77957a4652e8ce59527 Mon Sep 17 00:00:00 2001
From: "pullfrog[bot]" <226033991+pullfrog[bot]@users.noreply.github.com>
Date: Sun, 11 Oct 2026 02:44:50 +0000
Subject: [PATCH 1/2] fix: serialize export session operations on a lifecycle
 lock

---
 CHANGELOG.md         |  18 ++-
 CLAUDE.md            |  23 ++-
 plugin.json          |   2 +-
 routes.py            | 224 +++++++++++++++++++-----------
 tests/test_routes.py | 324 +++++++++++++++++++++++++++++++++++++++++++
 5 files changed, 504 insertions(+), 87 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index abf1b23..f0b5bfc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,9 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Export sessions let the browser stream encoded H.264 chunks to the server as
   they are produced instead of keeping the whole elementary stream in memory.
   The session's total encoded bytes stay capped by the existing 2 GB limit
-  across all appends, including appends that overlap. Cancelling an export still
-  deletes the session immediately, and abandoned sessions are expired
-  server-side.
+  across all appends, including appends that overlap. Cancelling an export
+  deletes the session, and abandoned sessions are expired server-side; both wait
+  for any upload or mux still using the session to finish before its files are
+  removed.
 - Set `FEEDBACK_PUBLIC_ORIGIN` (for example `http://127.0.0.1:5173`) to let the
   server fetch a same-host song mix itself, so the browser no longer downloads
   and re-uploads it. The server tells the browser whether it can resolve the
@@ -74,6 +75,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Export session operations are serialized on one lifecycle lock.** The
+  previous `append_lock` only protected concurrent video appends: a mux could
+  run while an upload was still streaming, so FFmpeg read an incomplete
+  elementary stream, and the session could be consumed (and its work dir
+  removed) before the upload finished. Two overlapping mux requests could also
+  both pass the audio-resolution step that runs before the session is popped.
+  Every operation that touches a session's work dir — video and audio uploads,
+  mux, cancel, and the TTL/size sweep — now takes the session's lifecycle
+  lock: mux waits for in-flight uploads so FFmpeg only ever reads finalized
+  media, only a single mux can own a session, and cancel/eviction wait for or
+  skip a session in use instead of deleting files underneath it.
 - **An interrupted session upload no longer leaves partial bytes behind.** A
   client disconnect (`ClientDisconnect`) or a cancelled task
   (`CancelledError`) bypassed the `OSError`/`ValueError` handlers, so a
diff --git a/CLAUDE.md b/CLAUDE.md
index 7679fa7..ba6a66a 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -97,11 +97,12 @@ it is the whole reason runs are cut: a live canvas must be composited between
 the static pixels above and below it.
 
 **Layer bounds are snapshot-based, under the lock.** `video_bytes` is the
-file's size, so it may only be read *inside* `append_lock`. Snapshotting before
-the acquire gives every in-flight append the same stale base, each gets a full
-`MAX_VIDEO_BYTES` of headroom, and a rollback can truncate below the accepted
-prefix. A rejected append truncates back to that snapshot — never unlink, or a
-single over-limit chunk destroys the whole render.
+file's size, so it may only be read *inside* the session's lifecycle lock
+(`state["lock"]`). Snapshotting before the acquire gives every in-flight append
+the same stale base, each gets a full `MAX_VIDEO_BYTES` of headroom, and a
+rollback can truncate below the accepted prefix. A rejected append truncates
+back to that snapshot — never unlink, or a single over-limit chunk destroys the
+whole render.
 
 **Fixed: Staff View (alphaTab) notation no longer goes stale.** Staff View
 renders sheet music as `<svg>`, swapped out by alphaTab as playback scrolls to
@@ -141,6 +142,18 @@ misses `ClientDisconnect` (a client that drops mid-stream) and `CancelledError`
 re-raise is what keeps cancellation working, don't swallow it to return a
 response.
 
+**Every operation on a session's work dir takes that session's single lifecycle
+lock** (`state["lock"]`), so ownership of the temp dir is unambiguous. Video and
+audio uploads hold it for their whole stream, mux holds it across audio
+resolution and the FFmpeg run, and cancel plus the TTL/size sweep take it before
+deleting. That is what makes mux wait for in-flight uploads (FFmpeg only reads
+finalized media), admits a single mux owner, and stops cleanup from deleting
+files an operation still has open; the sweep skips a locked session rather than
+block. A waiter re-checks `sessions.get(session_id) is state` under the lock, so
+an operation that lost the race against a consume/evict returns 404 instead of
+acting on a stale reference. Don't narrow this back to a video-append-only lock
+— that is exactly how mux came to read a half-written stream.
+
 **Server-side audio resolution is opt-in and origin-pinned.** The browser only
 ever sends a site-relative `audio_url`, but that constrains the path, not the
 destination: `urlopen` follows cross-origin 30x, and `request.base_url` is built
diff --git a/plugin.json b/plugin.json
index 13b17ee..81d8c75 100644
--- a/plugin.json
+++ b/plugin.json
@@ -1,7 +1,7 @@
 {
   "id": "visual_export",
   "name": "Visual Export",
-  "version": "0.3.2",
+  "version": "0.3.3",
   "description": "Renders the current player layout to a video file for the entire song, without a live performance.",
   "type": "tools",
   "script": "screen.js",
diff --git a/routes.py b/routes.py
index 3cff6e5..08dbe9b 100644
--- a/routes.py
+++ b/routes.py
@@ -215,20 +215,48 @@ def capabilities() -> JSONResponse:
     # fallback for hosts or sources that cannot use a session.
     sessions: dict[str, dict] = {}
 
-    def _drop_session(session_id: str) -> None:
-        state = sessions.pop(session_id, None)
-        if state:
+    async def _drop_session(session_id: str) -> None:
+        """Retire a session and delete its work files under its lifecycle lock.
+
+        Every operation that touches a session's work dir holds that session's
+        lock, so waiting for it here is what stops cancel and eviction from
+        deleting a directory out from under an active upload, mux, or audio
+        fetch. The membership re-check matters because a caller may have read
+        the session before an earlier operation consumed it while we waited.
+        """
+        state = sessions.get(session_id)
+        if state is None:
+            return
+        async with state["lock"]:
+            if sessions.get(session_id) is not state:
+                return
+            sessions.pop(session_id, None)
             shutil.rmtree(state["work"], ignore_errors=True)
 
-    def _sweep_sessions() -> None:
+    async def _sweep_sessions() -> None:
         now = time.monotonic()
         for session_id, state in list(sessions.items()):
+            # A locked session has an operation in flight, so its files are in
+            # use: skip it rather than delete underneath it. It stays a
+            # candidate for the next sweep, once it has gone idle. A client that
+            # holds an upload open without ever finishing it (rather than
+            # disconnecting, which the upload's own cleanup handles) therefore
+            # pins that one session; deleting in-use files is the worse failure,
+            # so this wait is deliberate.
+            if state["lock"].locked():
+                continue
             if now - state["touched"] > SESSION_TTL_SECONDS:
                 log.info("%s: expiring abandoned export session %s", PLUGIN_ID, session_id)
-                _drop_session(session_id)
+                await _drop_session(session_id)
+        # Best-effort cap: it can only evict idle sessions, so a burst of
+        # simultaneously locked sessions can briefly exceed MAX_SESSIONS until
+        # they go idle and the next create sweeps them.
         while len(sessions) >= MAX_SESSIONS:
-            oldest = min(sessions, key=lambda key: sessions[key]["touched"])
-            _drop_session(oldest)
+            idle = [key for key, state in sessions.items() if not state["lock"].locked()]
+            if not idle:
+                break
+            oldest = min(idle, key=lambda key: sessions[key]["touched"])
+            await _drop_session(oldest)
 
     async def _json_body(request: Request) -> dict | None:
         raw = await request.body()
@@ -284,7 +312,7 @@ async def create_session(request: Request) -> JSONResponse:
         audio_url = data.get("audio_url")
         if audio_url is not None and _same_host_path(audio_url) is None:
             return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
-        _sweep_sessions()
+        await _sweep_sessions()
         session_id = uuid.uuid4().hex
         sessions[session_id] = {
             "work": Path(tempfile.mkdtemp(prefix="feedback-visual-export-")),
@@ -294,9 +322,13 @@ async def create_session(request: Request) -> JSONResponse:
             # Cumulative across every append, so the session as a whole is
             # bounded by MAX_VIDEO_BYTES the way the single-request upload was.
             "video_bytes": 0,
-            # Serializes concurrent chunk appends so two overlapping requests
-            # can never interleave writes into the same elementary stream.
-            "append_lock": asyncio.Lock(),
+            # One lifecycle lock per session, held for the whole of every
+            # operation that touches its work dir: video/audio appends, mux,
+            # cancel and eviction. It is what makes mux wait for an upload to
+            # finish (so FFmpeg only ever reads finalized media), admits a
+            # single mux owner, and keeps cancel/eviction from deleting files an
+            # operation still has open.
+            "lock": asyncio.Lock(),
         }
         return JSONResponse({"session": session_id,
                              # Only true when this session can actually resolve a
@@ -312,7 +344,15 @@ async def append_session_video(session_id: str, request: Request) -> JSONRespons
         video_path = state["work"] / "frames.h264"
         total = 0
         try:
-            async with state["append_lock"]:
+            # Hold the lifecycle lock for the whole append. It is the same lock
+            # mux, cancel and eviction take, which is the point: FFmpeg can only
+            # start once every in-flight append has finished, and a cancel or
+            # eviction cannot delete frames.h264 while it is being written.
+            async with state["lock"]:
+                # Re-check under the lock: the session may have been consumed or
+                # evicted while this request waited for it.
+                if sessions.get(session_id) is not state:
+                    return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
                 # Snapshot under the lock. Reading it before would let every
                 # in-flight append take the same pre-lock base, so each would
                 # get a full MAX_VIDEO_BYTES of headroom, the cap would bind at
@@ -363,28 +403,36 @@ async def upload_session_audio(session_id: str, request: Request) -> JSONRespons
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
         state["touched"] = time.monotonic()
         audio_path = state["work"] / "audio.bin"
-        if audio_path.exists():
-            return JSONResponse({"error": "audio was already uploaded for this session"}, status_code=409)
         total = 0
         try:
-            with audio_path.open("wb") as target:
-                async for chunk in request.stream():
-                    if not chunk:
-                        continue
-                    total += len(chunk)
-                    if total > MAX_AUDIO_BYTES:
-                        raise ValueError("song audio is too large")
-                    target.write(chunk)
+            async with state["lock"]:
+                if sessions.get(session_id) is not state:
+                    return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
+                # "Exists" means "fully uploaded": an in-progress upload holds
+                # the lock, so a concurrent mux or a second upload can never
+                # observe the file before it is complete.
+                if audio_path.exists():
+                    return JSONResponse({"error": "audio was already uploaded for this session"}, status_code=409)
+                try:
+                    with audio_path.open("wb") as target:
+                        async for chunk in request.stream():
+                            if not chunk:
+                                continue
+                            total += len(chunk)
+                            if total > MAX_AUDIO_BYTES:
+                                raise ValueError("song audio is too large")
+                            target.write(chunk)
+                except BaseException:
+                    # A client disconnect (ClientDisconnect) or task
+                    # cancellation (CancelledError) bypasses the handler below.
+                    # Remove the incomplete mix while still holding the lock, so
+                    # no mux can see it as complete, and on every unsuccessful
+                    # exit so a retry is not refused with 409 for a file that
+                    # was never finished. Re-raising preserves cancellation.
+                    audio_path.unlink(missing_ok=True)
+                    raise
         except (OSError, ValueError) as exc:
-            audio_path.unlink(missing_ok=True)
             return JSONResponse({"error": str(exc)}, status_code=400)
-        except BaseException:
-            # A client disconnect or task cancellation bypasses the handler
-            # above. Remove the incomplete mix on every unsuccessful exit so a
-            # retry is not refused with 409 for a file that was never finished,
-            # then re-raise to preserve cancellation.
-            audio_path.unlink(missing_ok=True)
-            raise
         return JSONResponse({"bytes": total})
 
     @app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/mux")
@@ -400,62 +448,82 @@ async def mux_session(background_tasks: BackgroundTasks, session_id: str, reques
             return JSONResponse({"error": "invalid session request"}, status_code=400)
         if not _valid_timing(data.get("fps"), data.get("duration")):
             return JSONResponse({"error": "invalid export timing"}, status_code=400)
-        work = state["work"]
-        video_path, audio_path, output_path = work / "frames.h264", work / "audio.bin", work / "export.mp4"
-        if not video_path.is_file() or video_path.stat().st_size == 0:
-            return JSONResponse({"error": "no encoded video was uploaded for this session"}, status_code=400)
-
-        # Resolve the audio BEFORE consuming the session. A mix the server
-        # cannot fetch (no configured origin, a 404, an auth wall, or any
-        # redirect now refused) leaves the session intact so the browser can
-        # upload the audio and ask again, instead of losing a full render at
-        # the last step.
-        # `dict.get(key, default)` only falls back when the key is ABSENT, and
-        # the browser always sends `audio_url` -- as null when it has already
-        # uploaded the mix or wants the session's stored path. So an explicit
-        # null means "no opinion", not "no audio".
-        audio_url = data.get("audio_url") or state["audio_url"]
-        origin = state["origin"]
-        if not audio_path.is_file():
-            relative = _same_host_path(audio_url) if audio_url is not None else None
-            if audio_url is not None and origin is not None and relative is None:
-                return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
-            if relative is None or origin is None:
-                return JSONResponse({"error": "no song audio was recorded for this session",
-                                     "audio_required": True}, status_code=409)
+
+        # Hold the session's lifecycle lock for the whole mux. It is the single
+        # mux owner: a second concurrent mux waits here, then finds the session
+        # consumed and gets a 404 instead of racing the first for the same work
+        # dir. Waiting also means any upload still in flight finishes first, so
+        # FFmpeg only ever reads finalized media, and cancel/eviction cannot
+        # delete the files while FFmpeg or the audio fetch is using them.
+        async with state["lock"]:
+            if sessions.get(session_id) is not state:
+                return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
+            work = state["work"]
+            video_path, audio_path, output_path = work / "frames.h264", work / "audio.bin", work / "export.mp4"
+            if not video_path.is_file() or video_path.stat().st_size == 0:
+                return JSONResponse({"error": "no encoded video was uploaded for this session"}, status_code=400)
+
+            # Resolve the audio BEFORE consuming the session. A mix the server
+            # cannot fetch (no configured origin, a 404, an auth wall, or any
+            # redirect now refused) leaves the session intact so the browser can
+            # upload the audio and ask again, instead of losing a full render at
+            # the last step.
+            # `dict.get(key, default)` only falls back when the key is ABSENT,
+            # and the browser always sends `audio_url` -- as null when it has
+            # already uploaded the mix or wants the session's stored path. So an
+            # explicit null means "no opinion", not "no audio".
+            audio_url = data.get("audio_url") or state["audio_url"]
+            origin = state["origin"]
+            if not audio_path.is_file():
+                relative = _same_host_path(audio_url) if audio_url is not None else None
+                if audio_url is not None and origin is not None and relative is None:
+                    return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
+                if relative is None or origin is None:
+                    return JSONResponse({"error": "no song audio was recorded for this session",
+                                         "audio_required": True}, status_code=409)
+                try:
+                    await _fetch_host_audio(origin, relative, audio_path)
+                except (OSError, ValueError, urllib.error.URLError) as exc:
+                    audio_path.unlink(missing_ok=True)
+                    log.warning("%s: could not fetch song audio, asking the browser to upload it: %s", PLUGIN_ID, exc)
+                    return JSONResponse({"error": str(exc), "audio_required": True}, status_code=409)
+
+            # Past this point the session is consumed: every exit path below
+            # removes its temporary directory, and a mux must not be repeatable.
+            sessions.pop(session_id, None)
             try:
-                await _fetch_host_audio(origin, relative, audio_path)
-            except (OSError, ValueError, urllib.error.URLError) as exc:
-                audio_path.unlink(missing_ok=True)
-                log.warning("%s: could not fetch song audio, asking the browser to upload it: %s", PLUGIN_ID, exc)
-                return JSONResponse({"error": str(exc), "audio_required": True}, status_code=409)
-
-        # Past this point the session is consumed: every exit path below removes
-        # its temporary directory, and a mux must not be repeatable.
-        sessions.pop(session_id, None)
-        try:
-            proc = await asyncio.to_thread(
-                _run_ffmpeg, ffmpeg, video_path, audio_path, output_path, data["fps"], float(data["duration"])
-            )
-            if proc.returncode != 0 or not output_path.is_file():
-                detail = (proc.stderr or "FFmpeg failed").strip()[-2000:]
-                log.error("%s: mux failed: %s", PLUGIN_ID, detail)
+                proc = await asyncio.to_thread(
+                    _run_ffmpeg, ffmpeg, video_path, audio_path, output_path, data["fps"], float(data["duration"])
+                )
+                if proc.returncode != 0 or not output_path.is_file():
+                    detail = (proc.stderr or "FFmpeg failed").strip()[-2000:]
+                    log.error("%s: mux failed: %s", PLUGIN_ID, detail)
+                    shutil.rmtree(work, ignore_errors=True)
+                    return JSONResponse({"error": detail}, status_code=500)
+            except (OSError, ValueError, urllib.error.URLError, subprocess.TimeoutExpired) as exc:
                 shutil.rmtree(work, ignore_errors=True)
-                return JSONResponse({"error": detail}, status_code=500)
-        except (OSError, ValueError, urllib.error.URLError, subprocess.TimeoutExpired) as exc:
-            shutil.rmtree(work, ignore_errors=True)
-            return JSONResponse({"error": str(exc)}, status_code=400)
+                return JSONResponse({"error": str(exc)}, status_code=400)
+            except BaseException:
+                # The session is already popped, so a disconnect or task
+                # cancellation here has nowhere else to reclaim the work dir
+                # from; the FileResponse background task is only registered on
+                # success. Clean up on any unsuccessful exit, then re-raise so
+                # cancellation still propagates.
+                shutil.rmtree(work, ignore_errors=True)
+                raise
 
-        safe_name = _safe_filename(data.get("filename")) or "feedback-export"
-        background_tasks.add_task(shutil.rmtree, work, ignore_errors=True)
-        return FileResponse(output_path, media_type="video/mp4", filename=safe_name + ".mp4",
-                            background=background_tasks)
+            safe_name = _safe_filename(data.get("filename")) or "feedback-export"
+            background_tasks.add_task(shutil.rmtree, work, ignore_errors=True)
+            return FileResponse(output_path, media_type="video/mp4", filename=safe_name + ".mp4",
+                                background=background_tasks)
 
     @app.delete(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}")
     async def cancel_session(session_id: str) -> JSONResponse:
         if session_id not in sessions:
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
-        _drop_session(session_id)
+        # Waits for any in-flight upload/mux to release the session's lifecycle
+        # lock before the work dir is removed.
+        await _drop_session(session_id)
         return JSONResponse({"status": "cancelled"})
 
     @app.post(f"/api/plugins/{PLUGIN_ID}/mux")
diff --git a/tests/test_routes.py b/tests/test_routes.py
index ae00e70..7957040 100644
--- a/tests/test_routes.py
+++ b/tests/test_routes.py
@@ -889,3 +889,327 @@ def fake_mkdtemp(**kwargs):
     assert not (work / "work0").exists()
     assert (work / "work1").exists()
     assert (work / "work2").exists()
+
+
+SESSIONS_END = "/api/plugins/visual_export/sessions"
+
+
+def _async_client(tmp_path):
+    import httpx
+
+    app = FastAPI()
+    routes.setup(app, {"config_dir": tmp_path})
+    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver")
+
+
+@pytest.mark.asyncio
+async def test_mux_waits_for_an_in_flight_upload_before_running_ffmpeg(tmp_path, monkeypatch):
+    """FFmpeg must not read a stream that is still being appended.
+
+    A video append holds the session lifecycle lock for its whole body, so a
+    concurrent mux blocks on it and only then sees the finalized media. Before
+    the lock was shared, mux ignored the append in progress, invoked FFmpeg on
+    the partial elementary stream, and consumed the session out from under the
+    still-running upload.
+    """
+    import asyncio
+
+    work = tmp_path / "work"
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes.tempfile, "mkdtemp", lambda **kwargs: str(work.mkdir() or work))
+    _stub_audio(monkeypatch)
+
+    seen = []
+
+    def fake_ffmpeg(cmd, **kwargs):
+        seen.append((work / "frames.h264").read_bytes())
+        Path(cmd[-1]).write_bytes(b"mp4")
+        return SimpleNamespace(returncode=0, stderr="")
+
+    monkeypatch.setattr(routes.subprocess, "run", fake_ffmpeg)
+
+    first_chunk = asyncio.Event()
+    release = asyncio.Event()
+
+    async def blocked_body():
+        yield b"firsthalf"
+        first_chunk.set()
+        await release.wait()
+        yield b"secondhalf"
+
+    async with _async_client(tmp_path) as client:
+        session = (
+            await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0, "audio_url": "/songs/pack/full_mix.wav"})
+        ).json()["session"]
+        upload = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/video", content=blocked_body())
+        )
+        await first_chunk.wait()
+        mux = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/mux", json={"fps": 30, "duration": 1.0})
+        )
+        # The mux is parked on the lifecycle lock: give it a real chance to run
+        # before asserting it has not reached FFmpeg.
+        for _ in range(20):
+            await asyncio.sleep(0.005)
+        assert seen == []
+
+        release.set()
+        upload_response = await upload
+        mux_response = await mux
+
+    assert upload_response.status_code == 200
+    assert mux_response.status_code == 200
+    # FFmpeg saw both halves, not just the chunk that was flushed pre-release.
+    assert seen == [b"firsthalfsecondhalf"]
+
+
+@pytest.mark.asyncio
+async def test_mux_waits_for_an_in_flight_audio_upload(tmp_path, monkeypatch):
+    """A half-written audio.bin must not be muxed as if it were complete.
+
+    The audio upload holds the lifecycle lock while it streams, so a mux waits
+    for it instead of treating the file's mere existence as a finished mix.
+    """
+    import asyncio
+
+    work = tmp_path / "work"
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes.tempfile, "mkdtemp", lambda **kwargs: str(work.mkdir() or work))
+
+    seen = []
+
+    def fake_ffmpeg(cmd, **kwargs):
+        seen.append((work / "audio.bin").read_bytes())
+        Path(cmd[-1]).write_bytes(b"mp4")
+        return SimpleNamespace(returncode=0, stderr="")
+
+    monkeypatch.setattr(routes.subprocess, "run", fake_ffmpeg)
+
+    first_chunk = asyncio.Event()
+    release = asyncio.Event()
+
+    async def blocked_body():
+        yield b"firsthalf"
+        first_chunk.set()
+        await release.wait()
+        yield b"secondhalf"
+
+    async with _async_client(tmp_path) as client:
+        session = (await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})).json()["session"]
+        await client.post(f"{SESSIONS_END}/{session}/video", content=b"video")
+        upload = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/audio", content=blocked_body())
+        )
+        await first_chunk.wait()
+        mux = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/mux", json={"fps": 30, "duration": 1.0})
+        )
+        for _ in range(20):
+            await asyncio.sleep(0.005)
+        assert seen == []
+
+        release.set()
+        upload_response = await upload
+        mux_response = await mux
+
+    assert upload_response.status_code == 200
+    assert mux_response.status_code == 200
+    # FFmpeg saw the whole mix, not the first half that existed pre-release.
+    assert seen == [b"firsthalfsecondhalf"]
+
+
+@pytest.mark.asyncio
+async def test_two_concurrent_muxes_consume_the_session_once(tmp_path, monkeypatch):
+    """A session admits a single mux owner even while resolving host audio.
+
+    Audio is resolved before the session is popped, so two overlapping muxes
+    could both pass that point and each run FFmpeg on the same work dir. The
+    lifecycle lock serializes them: the first consumes the session, the second
+    then finds it gone and gets a 404.
+    """
+    import asyncio
+    import threading
+
+    work = tmp_path / "work"
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes.tempfile, "mkdtemp", lambda **kwargs: str(work.mkdir() or work))
+    monkeypatch.setenv("FEEDBACK_PUBLIC_ORIGIN", "http://feedback.test")
+
+    release = threading.Event()
+    started = threading.Event()
+
+    class _BlockingOpener:
+        def open(self, request, timeout=None):
+            started.set()
+            release.wait(5)
+
+            class _Response(io.BytesIO):
+                def __init__(self):
+                    super().__init__(b"audio")
+                    self._url = str(request.full_url)
+
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, *exc):
+                    self.close()
+                    return False
+
+                def geturl(self):
+                    return self._url
+
+            return _Response()
+
+    monkeypatch.setattr(routes, "_AUDIO_OPENER", _BlockingOpener())
+
+    ffmpeg_calls = []
+
+    def fake_ffmpeg(cmd, **kwargs):
+        ffmpeg_calls.append(1)
+        Path(cmd[-1]).write_bytes(b"mp4")
+        return SimpleNamespace(returncode=0, stderr="")
+
+    monkeypatch.setattr(routes.subprocess, "run", fake_ffmpeg)
+
+    async with _async_client(tmp_path) as client:
+        session = (
+            await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0, "audio_url": "/songs/pack/full_mix.wav"})
+        ).json()["session"]
+        await client.post(f"{SESSIONS_END}/{session}/video", content=b"video")
+        mux_url = f"{SESSIONS_END}/{session}/mux"
+        first = asyncio.create_task(client.post(mux_url, json={"fps": 30, "duration": 1.0}))
+        await asyncio.to_thread(started.wait, 5)
+        second = asyncio.create_task(client.post(mux_url, json={"fps": 30, "duration": 1.0}))
+        # Let the second mux reach the lock while the first is mid-fetch.
+        for _ in range(20):
+            await asyncio.sleep(0.005)
+        assert ffmpeg_calls == []
+        release.set()
+        responses = [await first, await second]
+
+    assert len(ffmpeg_calls) == 1
+    assert sorted(r.status_code for r in responses) == [200, 404]
+
+
+@pytest.mark.asyncio
+async def test_cancel_waits_for_an_active_upload_before_removing_the_work_dir(tmp_path, monkeypatch):
+    """Cancelling mid-upload must not delete files the upload is still writing.
+
+    The append holds the lifecycle lock, so cancel blocks until it finishes and
+    only then removes the work dir, instead of pulling frames.h264 out from
+    under the writer.
+    """
+    import asyncio
+
+    work = tmp_path / "work"
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes.tempfile, "mkdtemp", lambda **kwargs: str(work.mkdir() or work))
+
+    first_chunk = asyncio.Event()
+    release = asyncio.Event()
+
+    async def blocked_body():
+        yield b"half"
+        first_chunk.set()
+        await release.wait()
+        yield b"rest"
+
+    async with _async_client(tmp_path) as client:
+        session = (await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})).json()["session"]
+        upload = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/video", content=blocked_body())
+        )
+        await first_chunk.wait()
+        cancel = asyncio.create_task(client.delete(f"{SESSIONS_END}/{session}"))
+        for _ in range(20):
+            await asyncio.sleep(0.005)
+        assert not cancel.done()
+        assert work.exists()
+
+        release.set()
+        upload_response = await upload
+        cancel_response = await cancel
+
+    assert upload_response.status_code == 200
+    assert cancel_response.status_code == 200
+    assert not work.exists()
+
+
+@pytest.mark.asyncio
+async def test_eviction_skips_a_session_mid_operation(tmp_path, monkeypatch):
+    """A sweep must not expire a session whose files an operation is using."""
+    import asyncio
+
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes, "SESSION_TTL_SECONDS", 0)
+    _isolated_work(tmp_path, monkeypatch)
+
+    first_chunk = asyncio.Event()
+    release = asyncio.Event()
+
+    async def blocked_body():
+        yield b"half"
+        first_chunk.set()
+        await release.wait()
+        yield b"rest"
+
+    async with _async_client(tmp_path) as client:
+        session = (await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})).json()["session"]
+        first_work = tmp_path / "work1"
+        upload = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/video", content=blocked_body())
+        )
+        await first_chunk.wait()
+        # Creating another session sweeps expired ones; the in-flight session is
+        # locked, so it must be skipped rather than deleted mid-append.
+        await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})
+        assert first_work.exists()
+
+        release.set()
+        await upload
+        # Idle now, the next sweep collects it.
+        await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})
+        assert not first_work.exists()
+
+
+@pytest.mark.asyncio
+async def test_cancelled_mux_removes_the_work_dir(tmp_path, monkeypatch):
+    """A mux cancelled mid-FFmpeg must not leak the consumed session's files.
+
+    The session is popped before FFmpeg runs, so a cancellation there has no
+    other cleanup path — the response's background removal is only registered
+    on success.
+    """
+    import asyncio
+    import threading
+
+    work = tmp_path / "work"
+    monkeypatch.setattr(routes, "_ffmpeg_cmd", lambda: "ffmpeg")
+    monkeypatch.setattr(routes.tempfile, "mkdtemp", lambda **kwargs: str(work.mkdir() or work))
+
+    entered = threading.Event()
+    release = threading.Event()
+
+    def stuck_ffmpeg(cmd, **kwargs):
+        entered.set()
+        release.wait(5)
+        Path(cmd[-1]).write_bytes(b"mp4")
+        return SimpleNamespace(returncode=0, stderr="")
+
+    monkeypatch.setattr(routes.subprocess, "run", stuck_ffmpeg)
+
+    async with _async_client(tmp_path) as client:
+        session = (await client.post(SESSIONS_END, json={"fps": 30, "duration": 1.0})).json()["session"]
+        await client.post(f"{SESSIONS_END}/{session}/video", content=b"video")
+        await client.post(f"{SESSIONS_END}/{session}/audio", content=b"audio")
+        task = asyncio.create_task(
+            client.post(f"{SESSIONS_END}/{session}/mux", json={"fps": 30, "duration": 1.0})
+        )
+        await asyncio.to_thread(entered.wait, 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    release.set()
+    assert not work.exists()

From 98dfd5498611a5afb876b0ac961d2d1e0fa0d975 Mon Sep 17 00:00:00 2001
From: Carolina <79524656+carochacs@users.noreply.github.com>
Date: Sun, 11 Oct 2026 03:52:25 -0600
Subject: [PATCH 2/2] refactor: reduce route setup and session mux complexity

---
 routes.py | 249 +++++++++++++++++++++++++++++-------------------------
 1 file changed, 133 insertions(+), 116 deletions(-)

diff --git a/routes.py b/routes.py
index 08dbe9b..56963cf 100644
--- a/routes.py
+++ b/routes.py
@@ -159,28 +159,46 @@ def _expected_origin() -> str | None:
 
 
 def setup(app: FastAPI, context: dict) -> None:
-    config_dir = Path(context["config_dir"])
-    log = context.get("log") or logging.getLogger(f"feedBack.plugin.{PLUGIN_ID}")
-    config_file = config_dir / f"{PLUGIN_ID}.json"
-
-    def _read() -> dict:
+    _ExportRoutes(app, context)
+
+
+class _ExportRoutes:
+    """Route handlers and export sessions isolated to one host application."""
+
+    def __init__(self, app: FastAPI, context: dict) -> None:
+        self.config_dir = Path(context["config_dir"])
+        self.log = context.get("log") or logging.getLogger(f"feedBack.plugin.{PLUGIN_ID}")
+        self.config_file = self.config_dir / f"{PLUGIN_ID}.json"
+        # H.264 chunks stream into per-app sessions; /mux remains the fallback
+        # for hosts or audio sources that cannot use streaming sessions.
+        self.sessions: dict[str, dict] = {}
+        app.get(f"/api/plugins/{PLUGIN_ID}/settings")(self.get_settings)
+        app.post(f"/api/plugins/{PLUGIN_ID}/settings")(self.set_settings)
+        app.get(f"/api/plugins/{PLUGIN_ID}/capabilities")(self.capabilities)
+        app.post(f"/api/plugins/{PLUGIN_ID}/sessions")(self.create_session)
+        app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/video")(self.append_session_video)
+        app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/audio")(self.upload_session_audio)
+        app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/mux")(self.mux_session)
+        app.delete(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}")(self.cancel_session)
+        app.post(f"/api/plugins/{PLUGIN_ID}/mux")(self.mux_export)
+        self.log.info("%s: routes registered", PLUGIN_ID)
+
+    def _read(self) -> dict:
         settings = dict(_DEFAULTS)
         try:
-            data = json.loads(config_file.read_text(encoding="utf-8"))
+            data = json.loads(self.config_file.read_text(encoding="utf-8"))
             if isinstance(data, dict):
                 settings.update({k: v for k, v in data.items() if _is_valid_setting(k, v)})
         except FileNotFoundError:
             pass
         except (OSError, ValueError) as exc:
-            log.warning("%s: unreadable config, using defaults: %s", PLUGIN_ID, exc)
+            self.log.warning("%s: unreadable config, using defaults: %s", PLUGIN_ID, exc)
         return settings
 
-    @app.get(f"/api/plugins/{PLUGIN_ID}/settings")
-    def get_settings() -> JSONResponse:
-        return JSONResponse(_read())
+    def get_settings(self) -> JSONResponse:
+        return JSONResponse(self._read())
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/settings")
-    async def set_settings(request: Request) -> JSONResponse:
+    async def set_settings(self, request: Request) -> JSONResponse:
         raw = await request.body()
         if len(raw) > MAX_SETTINGS_BODY_BYTES:
             return JSONResponse({"error": "request body too large"}, status_code=413)
@@ -192,30 +210,21 @@ async def set_settings(request: Request) -> JSONResponse:
             return JSONResponse({"error": "unknown settings"}, status_code=400)
         if not all(_is_valid_setting(k, v) for k, v in incoming.items()):
             return JSONResponse({"error": "invalid setting values"}, status_code=400)
-        merged = {**_read(), **incoming}
+        merged = {**self._read(), **incoming}
         try:
-            config_dir.mkdir(parents=True, exist_ok=True)
+            self.config_dir.mkdir(parents=True, exist_ok=True)
             await asyncio.to_thread(
-                config_file.write_text, json.dumps(merged, indent=2), encoding="utf-8"
+                self.config_file.write_text, json.dumps(merged, indent=2), encoding="utf-8"
             )
         except OSError as exc:
-            log.error("%s: failed to save settings: %s", PLUGIN_ID, exc)
+            self.log.error("%s: failed to save settings: %s", PLUGIN_ID, exc)
             return JSONResponse({"error": "failed to save settings"}, status_code=500)
         return JSONResponse(merged)
 
-    @app.get(f"/api/plugins/{PLUGIN_ID}/capabilities")
-    def capabilities() -> JSONResponse:
+    def capabilities(self) -> JSONResponse:
         return JSONResponse({"ffmpeg": bool(_ffmpeg_cmd()), "format": "mp4", "sessions": True})
 
-    # Issue #9, phase 3: streaming export sessions. The browser posts encoded
-    # H.264 chunks as the encoder produces them, so the complete elementary
-    # stream is never held in browser memory, and the server resolves the
-    # song mix from its own host instead of receiving an uploaded copy. The
-    # single-request /mux route below stays available and unchanged as the
-    # fallback for hosts or sources that cannot use a session.
-    sessions: dict[str, dict] = {}
-
-    async def _drop_session(session_id: str) -> None:
+    async def _drop_session(self, session_id: str) -> None:
         """Retire a session and delete its work files under its lifecycle lock.
 
         Every operation that touches a session's work dir holds that session's
@@ -224,18 +233,18 @@ async def _drop_session(session_id: str) -> None:
         fetch. The membership re-check matters because a caller may have read
         the session before an earlier operation consumed it while we waited.
         """
-        state = sessions.get(session_id)
+        state = self.sessions.get(session_id)
         if state is None:
             return
         async with state["lock"]:
-            if sessions.get(session_id) is not state:
+            if self.sessions.get(session_id) is not state:
                 return
-            sessions.pop(session_id, None)
+            self.sessions.pop(session_id, None)
             shutil.rmtree(state["work"], ignore_errors=True)
 
-    async def _sweep_sessions() -> None:
+    async def _sweep_sessions(self) -> None:
         now = time.monotonic()
-        for session_id, state in list(sessions.items()):
+        for session_id, state in list(self.sessions.items()):
             # A locked session has an operation in flight, so its files are in
             # use: skip it rather than delete underneath it. It stays a
             # candidate for the next sweep, once it has gone idle. A client that
@@ -246,19 +255,19 @@ async def _sweep_sessions() -> None:
             if state["lock"].locked():
                 continue
             if now - state["touched"] > SESSION_TTL_SECONDS:
-                log.info("%s: expiring abandoned export session %s", PLUGIN_ID, session_id)
-                await _drop_session(session_id)
+                self.log.info("%s: expiring abandoned export session %s", PLUGIN_ID, session_id)
+                await self._drop_session(session_id)
         # Best-effort cap: it can only evict idle sessions, so a burst of
         # simultaneously locked sessions can briefly exceed MAX_SESSIONS until
         # they go idle and the next create sweeps them.
-        while len(sessions) >= MAX_SESSIONS:
-            idle = [key for key, state in sessions.items() if not state["lock"].locked()]
+        while len(self.sessions) >= MAX_SESSIONS:
+            idle = [key for key, state in self.sessions.items() if not state["lock"].locked()]
             if not idle:
                 break
-            oldest = min(idle, key=lambda key: sessions[key]["touched"])
-            await _drop_session(oldest)
+            oldest = min(idle, key=lambda key: self.sessions[key]["touched"])
+            await self._drop_session(oldest)
 
-    async def _json_body(request: Request) -> dict | None:
+    async def _json_body(self, request: Request) -> dict | None:
         raw = await request.body()
         if len(raw) > MAX_SESSION_BODY_BYTES:
             return None
@@ -268,7 +277,7 @@ async def _json_body(request: Request) -> dict | None:
             return None
         return data if isinstance(data, dict) else None
 
-    async def _fetch_host_audio(origin: str, relative: str, target: Path) -> None:
+    async def _fetch_host_audio(self, origin: str, relative: str, target: Path) -> None:
         """Download `relative` from `origin` without ever leaving that origin.
 
         Redirects are refused rather than followed: validating the *path* only
@@ -299,12 +308,11 @@ def _download() -> None:
 
         await asyncio.to_thread(_download)
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/sessions")
-    async def create_session(request: Request) -> JSONResponse:
+    async def create_session(self, request: Request) -> JSONResponse:
         ffmpeg = _ffmpeg_cmd()
         if not ffmpeg:
             return JSONResponse({"error": "FFmpeg is not installed"}, status_code=503)
-        data = await _json_body(request)
+        data = await self._json_body(request)
         if data is None:
             return JSONResponse({"error": "invalid session request"}, status_code=400)
         if not _valid_timing(data.get("fps"), data.get("duration")):
@@ -312,9 +320,9 @@ async def create_session(request: Request) -> JSONResponse:
         audio_url = data.get("audio_url")
         if audio_url is not None and _same_host_path(audio_url) is None:
             return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
-        await _sweep_sessions()
+        await self._sweep_sessions()
         session_id = uuid.uuid4().hex
-        sessions[session_id] = {
+        self.sessions[session_id] = {
             "work": Path(tempfile.mkdtemp(prefix="feedback-visual-export-")),
             "touched": time.monotonic(),
             "origin": _expected_origin(),
@@ -333,11 +341,10 @@ async def create_session(request: Request) -> JSONResponse:
         return JSONResponse({"session": session_id,
                              # Only true when this session can actually resolve a
                              # mix: both a configured origin and a path to use.
-                             "audio_fetch": bool(sessions[session_id]["origin"] and audio_url)})
+                             "audio_fetch": bool(self.sessions[session_id]["origin"] and audio_url)})
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/video")
-    async def append_session_video(session_id: str, request: Request) -> JSONResponse:
-        state = sessions.get(session_id)
+    async def append_session_video(self, session_id: str, request: Request) -> JSONResponse:
+        state = self.sessions.get(session_id)
         if state is None:
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
         state["touched"] = time.monotonic()
@@ -351,7 +358,7 @@ async def append_session_video(session_id: str, request: Request) -> JSONRespons
             async with state["lock"]:
                 # Re-check under the lock: the session may have been consumed or
                 # evicted while this request waited for it.
-                if sessions.get(session_id) is not state:
+                if self.sessions.get(session_id) is not state:
                     return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
                 # Snapshot under the lock. Reading it before would let every
                 # in-flight append take the same pre-lock base, so each would
@@ -396,9 +403,8 @@ async def append_session_video(session_id: str, request: Request) -> JSONRespons
             return JSONResponse({"error": str(exc)}, status_code=400)
         return JSONResponse({"bytes": total})
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/audio")
-    async def upload_session_audio(session_id: str, request: Request) -> JSONResponse:
-        state = sessions.get(session_id)
+    async def upload_session_audio(self, session_id: str, request: Request) -> JSONResponse:
+        state = self.sessions.get(session_id)
         if state is None:
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
         state["touched"] = time.monotonic()
@@ -406,7 +412,7 @@ async def upload_session_audio(session_id: str, request: Request) -> JSONRespons
         total = 0
         try:
             async with state["lock"]:
-                if sessions.get(session_id) is not state:
+                if self.sessions.get(session_id) is not state:
                     return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
                 # "Exists" means "fully uploaded": an in-progress upload holds
                 # the lock, so a concurrent mux or a second upload can never
@@ -435,15 +441,44 @@ async def upload_session_audio(session_id: str, request: Request) -> JSONRespons
             return JSONResponse({"error": str(exc)}, status_code=400)
         return JSONResponse({"bytes": total})
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}/mux")
-    async def mux_session(background_tasks: BackgroundTasks, session_id: str, request: Request):
+    async def _resolve_session_audio(self, state: dict, data: dict, audio_path: Path) -> JSONResponse | None:
+        """Resolve audio under the lifecycle lock without consuming the session."""
+        # Resolve the audio BEFORE consuming the session. A mix the server
+        # cannot fetch (no configured origin, a 404, an auth wall, or any
+        # redirect now refused) leaves the session intact so the browser can
+        # upload the audio and ask again, instead of losing a full render at
+        # the last step.
+        # `dict.get(key, default)` only falls back when the key is ABSENT,
+        # and the browser always sends `audio_url` -- as null when it has
+        # already uploaded the mix or wants the session's stored path. So an
+        # explicit null means "no opinion", not "no audio".
+        audio_url = data.get("audio_url") or state["audio_url"]
+        origin = state["origin"]
+        if audio_path.is_file():
+            return None
+        relative = _same_host_path(audio_url) if audio_url is not None else None
+        if audio_url is not None and origin is not None and relative is None:
+            return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
+        if relative is None or origin is None:
+            return JSONResponse({"error": "no song audio was recorded for this session",
+                                 "audio_required": True}, status_code=409)
+        try:
+            await self._fetch_host_audio(origin, relative, audio_path)
+        except (OSError, ValueError, urllib.error.URLError) as exc:
+            audio_path.unlink(missing_ok=True)
+            self.log.warning("%s: could not fetch song audio, asking the browser to upload it: %s", PLUGIN_ID, exc)
+            return JSONResponse({"error": str(exc), "audio_required": True}, status_code=409)
+
+        return None
+
+    async def mux_session(self, background_tasks: BackgroundTasks, session_id: str, request: Request):
         ffmpeg = _ffmpeg_cmd()
         if not ffmpeg:
             return JSONResponse({"error": "FFmpeg is not installed"}, status_code=503)
-        state = sessions.get(session_id)
+        state = self.sessions.get(session_id)
         if state is None:
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
-        data = await _json_body(request)
+        data = await self._json_body(request)
         if data is None:
             return JSONResponse({"error": "invalid session request"}, status_code=400)
         if not _valid_timing(data.get("fps"), data.get("duration")):
@@ -456,78 +491,62 @@ async def mux_session(background_tasks: BackgroundTasks, session_id: str, reques
         # FFmpeg only ever reads finalized media, and cancel/eviction cannot
         # delete the files while FFmpeg or the audio fetch is using them.
         async with state["lock"]:
-            if sessions.get(session_id) is not state:
+            if self.sessions.get(session_id) is not state:
                 return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
             work = state["work"]
-            video_path, audio_path, output_path = work / "frames.h264", work / "audio.bin", work / "export.mp4"
+            video_path, audio_path = work / "frames.h264", work / "audio.bin"
             if not video_path.is_file() or video_path.stat().st_size == 0:
                 return JSONResponse({"error": "no encoded video was uploaded for this session"}, status_code=400)
 
-            # Resolve the audio BEFORE consuming the session. A mix the server
-            # cannot fetch (no configured origin, a 404, an auth wall, or any
-            # redirect now refused) leaves the session intact so the browser can
-            # upload the audio and ask again, instead of losing a full render at
-            # the last step.
-            # `dict.get(key, default)` only falls back when the key is ABSENT,
-            # and the browser always sends `audio_url` -- as null when it has
-            # already uploaded the mix or wants the session's stored path. So an
-            # explicit null means "no opinion", not "no audio".
-            audio_url = data.get("audio_url") or state["audio_url"]
-            origin = state["origin"]
-            if not audio_path.is_file():
-                relative = _same_host_path(audio_url) if audio_url is not None else None
-                if audio_url is not None and origin is not None and relative is None:
-                    return JSONResponse({"error": "audio_url must be a same-host path"}, status_code=400)
-                if relative is None or origin is None:
-                    return JSONResponse({"error": "no song audio was recorded for this session",
-                                         "audio_required": True}, status_code=409)
-                try:
-                    await _fetch_host_audio(origin, relative, audio_path)
-                except (OSError, ValueError, urllib.error.URLError) as exc:
-                    audio_path.unlink(missing_ok=True)
-                    log.warning("%s: could not fetch song audio, asking the browser to upload it: %s", PLUGIN_ID, exc)
-                    return JSONResponse({"error": str(exc), "audio_required": True}, status_code=409)
+            audio_error = await self._resolve_session_audio(state, data, audio_path)
+            if audio_error is not None:
+                return audio_error
 
             # Past this point the session is consumed: every exit path below
             # removes its temporary directory, and a mux must not be repeatable.
-            sessions.pop(session_id, None)
-            try:
-                proc = await asyncio.to_thread(
-                    _run_ffmpeg, ffmpeg, video_path, audio_path, output_path, data["fps"], float(data["duration"])
-                )
-                if proc.returncode != 0 or not output_path.is_file():
-                    detail = (proc.stderr or "FFmpeg failed").strip()[-2000:]
-                    log.error("%s: mux failed: %s", PLUGIN_ID, detail)
-                    shutil.rmtree(work, ignore_errors=True)
-                    return JSONResponse({"error": detail}, status_code=500)
-            except (OSError, ValueError, urllib.error.URLError, subprocess.TimeoutExpired) as exc:
-                shutil.rmtree(work, ignore_errors=True)
-                return JSONResponse({"error": str(exc)}, status_code=400)
-            except BaseException:
-                # The session is already popped, so a disconnect or task
-                # cancellation here has nowhere else to reclaim the work dir
-                # from; the FileResponse background task is only registered on
-                # success. Clean up on any unsuccessful exit, then re-raise so
-                # cancellation still propagates.
+            self.sessions.pop(session_id, None)
+            return await self._mux_session_files(background_tasks, ffmpeg, work, data)
+
+    async def _mux_session_files(self, background_tasks: BackgroundTasks, ffmpeg: str,
+                                 work: Path, data: dict):
+        """Mux a consumed session; remove its files on every unsuccessful exit."""
+        video_path, audio_path, output_path = work / "frames.h264", work / "audio.bin", work / "export.mp4"
+        try:
+            proc = await asyncio.to_thread(
+                _run_ffmpeg, ffmpeg, video_path, audio_path, output_path, data["fps"], float(data["duration"])
+            )
+            if proc.returncode != 0 or not output_path.is_file():
+                detail = (proc.stderr or "FFmpeg failed").strip()[-2000:]
+                self.log.error("%s: mux failed: %s", PLUGIN_ID, detail)
                 shutil.rmtree(work, ignore_errors=True)
-                raise
+                return JSONResponse({"error": detail}, status_code=500)
+        except (OSError, ValueError, urllib.error.URLError, subprocess.TimeoutExpired) as exc:
+            shutil.rmtree(work, ignore_errors=True)
+            return JSONResponse({"error": str(exc)}, status_code=400)
+        except BaseException:
+            # The session is already popped, so a disconnect or task
+            # cancellation here has nowhere else to reclaim the work dir
+            # from; the FileResponse background task is only registered on
+            # success. Clean up on any unsuccessful exit, then re-raise so
+            # cancellation still propagates.
+            shutil.rmtree(work, ignore_errors=True)
+            raise
 
-            safe_name = _safe_filename(data.get("filename")) or "feedback-export"
-            background_tasks.add_task(shutil.rmtree, work, ignore_errors=True)
-            return FileResponse(output_path, media_type="video/mp4", filename=safe_name + ".mp4",
-                                background=background_tasks)
+        safe_name = _safe_filename(data.get("filename")) or "feedback-export"
+        background_tasks.add_task(shutil.rmtree, work, ignore_errors=True)
+        return FileResponse(output_path, media_type="video/mp4", filename=safe_name + ".mp4",
+                            background=background_tasks)
 
-    @app.delete(f"/api/plugins/{PLUGIN_ID}/sessions/{{session_id}}")
-    async def cancel_session(session_id: str) -> JSONResponse:
-        if session_id not in sessions:
+    async def cancel_session(self, session_id: str) -> JSONResponse:
+        if session_id not in self.sessions:
             return JSONResponse({"error": "unknown or expired export session"}, status_code=404)
         # Waits for any in-flight upload/mux to release the session's lifecycle
         # lock before the work dir is removed.
-        await _drop_session(session_id)
+        await self._drop_session(session_id)
         return JSONResponse({"status": "cancelled"})
 
-    @app.post(f"/api/plugins/{PLUGIN_ID}/mux")
     async def mux_export(
+        self,
         background_tasks: BackgroundTasks,
         video: UploadFile = File(...),
         audio: UploadFile = File(...),
@@ -550,7 +569,7 @@ async def mux_export(
             )
             if proc.returncode != 0 or not output_path.is_file():
                 detail = (proc.stderr or "FFmpeg failed").strip()[-2000:]
-                log.error("%s: mux failed: %s", PLUGIN_ID, detail)
+                self.log.error("%s: mux failed: %s", PLUGIN_ID, detail)
                 shutil.rmtree(work, ignore_errors=True)
                 return JSONResponse({"error": detail}, status_code=500)
         except (OSError, ValueError, subprocess.TimeoutExpired) as exc:
@@ -565,5 +584,3 @@ async def mux_export(
         return FileResponse(output_path, media_type="video/mp4",
                             filename=safe_name + ".mp4",
                             background=background_tasks)
-
-    log.info("%s: routes registered", PLUGIN_ID)