fix(app): keep render state consistent

moss_transcribe_diarize/app/jobs.py

@@ -28,6 +28,10 @@
 TERMINAL_STATES = {"waiting_review", "done", "failed", "cancelled"}
 
 
+class JobConflictError(RuntimeError):
+    """Raised when an operation conflicts with the job's current state."""
+
+
 @dataclass(slots=True)
 class JobRecord:
     id: str
@@ -153,6 +157,7 @@ def __init__(
         self.temperature = temperature
         self._jobs: dict[str, JobRecord] = {}
         self._queue: queue.Queue[str] = queue.Queue()
+        self._state_lock = threading.RLock()
         self._render_lock = threading.Lock()
         self._progress_save_times: dict[str, float] = {}
         self._load_existing_jobs()
@@ -299,15 +304,18 @@ def update_segments(
         style_payload: dict[str, Any] | None = None,
     ) -> list[dict[str, Any]]:
         job = self.get_job(job_id)
-        segments = coerce_subtitle_segments(payload)
-        style = SubtitleStyle.from_dict(style_payload) if style_payload is not None else None
-        if style is not None:
-            job.subtitle_style = style.to_dict()
-        self._write_subtitle_files(job, segments, style=style)
-        if job.status == "done":
-            self._set_status(job, "waiting_review", 0.95, error=None)
-        else:
-            self._touch(job, error=None)
+        with self._state_lock:
+            if job.status == "rendering":
+                raise JobConflictError("Cannot update subtitles while the job is rendering.")
+            segments = coerce_subtitle_segments(payload)
+            style = SubtitleStyle.from_dict(style_payload) if style_payload is not None else None
+            if style is not None:
+                job.subtitle_style = style.to_dict()
+            self._write_subtitle_files(job, segments, style=style)
+            if job.status == "done":
+                self._set_status(job, "waiting_review", 0.95, error=None)
+            else:
+                self._touch(job, error=None)
         return [segment.to_dict() for segment in segments]
 
     def render(self, job_id: str, style_payload: dict[str, Any] | None = None) -> JobRecord:
@@ -316,12 +324,25 @@ def render(self, job_id: str, style_payload: dict[str, Any] | None = None) -> Jo
             raise RuntimeError("ffmpeg and ffprobe are not available on PATH.")
         if not job.segments_path.exists():
             raise RuntimeError("No subtitle segments are available for this job.")
-        threading.Thread(
+        style = SubtitleStyle.from_dict(style_payload)
+        thread = threading.Thread(
             target=self._render_job,
-            args=(job.id, SubtitleStyle.from_dict(style_payload)),
+            args=(job.id, style),
             name=f"mtd-render-{job.id}",
             daemon=True,
-        ).start()
+        )
+        with self._state_lock:
+            if job.status == "rendering":
+                raise JobConflictError("This job is already rendering.")
+            previous_status = job.status
+            previous_progress = job.progress
+            previous_error = job.error
+            self._set_status(job, "rendering", 0.97, error=None)
+            try:
+                thread.start()
+            except Exception:
+                self._set_status(job, previous_status, previous_progress, error=previous_error)
+                raise
         return job
 
     def download_path(self, job_id: str, kind: str) -> Path:
@@ -356,7 +377,13 @@ def _load_existing_jobs(self) -> None:
                 job = JobRecord.from_dict(data)
                 if not job.job_dir:
                     job.job_dir = str(path.parent)
-                if job.status in {"queued", "loading_model", "transcribing", "postprocessing", "rendering"}:
+                if job.status == "rendering" and self._has_complete_segments(job):
+                    job.status = "waiting_review"
+                    job.progress = 0.95
+                    job.error = "Rendering was interrupted by the previous server shutdown. You can retry rendering."
+                    job.updated_at = time.time()
+                    self._save_job(job)
+                elif job.status in {"queued", "loading_model", "transcribing", "postprocessing", "rendering"}:
                     job.status = "failed"
                     job.progress = 1.0
                     job.error = "Interrupted by previous server shutdown."
@@ -366,6 +393,17 @@ def _load_existing_jobs(self) -> None:
             except Exception:
                 continue
 
+    @staticmethod
+    def _has_complete_segments(job: JobRecord) -> bool:
+        try:
+            payload = json.loads(job.segments_path.read_text(encoding="utf-8"))
+            if not isinstance(payload, list):
+                return False
+            coerce_subtitle_segments(payload)
+        except (OSError, TypeError, ValueError, KeyError, AttributeError):
+            return False
+        return True
+
     def _process_job(self, job: JobRecord) -> None:
         try:
             def update(status: str, progress: float | None, generated_tokens: int | None = None) -> None:
@@ -402,7 +440,6 @@ def _render_job(self, job_id: str, style: SubtitleStyle) -> None:
         job = self.get_job(job_id)
         with self._render_lock:
             try:
-                self._set_status(job, "rendering", 0.97, error=None)
                 segments = [SubtitleSegment.from_dict(item) for item in self.list_segments(job.id)]
                 width, height = probe_video_size(job.input_path)
                 write_text(job.ass_path, export_ass(segments, style=style, video_width=width, video_height=height))
@@ -442,9 +479,10 @@ def _set_status(
         error: str | None = None,
         save: bool = True,
     ) -> None:
-        job.status = status
-        job.progress = max(0.0, min(1.0, progress))
-        self._touch(job, error=error, save=save)
+        with self._state_lock:
+            job.status = status
+            job.progress = max(0.0, min(1.0, progress))
+            self._touch(job, error=error, save=save)
 
     def _resolve_inference_options(
         self,

moss_transcribe_diarize/app/server.py

@@ -5,7 +5,7 @@
 from moss_transcribe_diarize.inference_utils import DEFAULT_PROMPT
 
 from .ffmpeg import detect_ffmpeg
-from .jobs import JobManager
+from .jobs import JobConflictError, JobManager
 from .model_runner import ModelRunner
 from .vllm_runner import VllmRunner
 
@@ -187,6 +187,8 @@ async def update_segments(job_id: str, request: Request):
             return {"segments": manager.update_segments(job_id, segments, style)}
         except KeyError as exc:
             raise HTTPException(status_code=404, detail=str(exc)) from exc
+        except JobConflictError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
         except Exception as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
 
@@ -201,6 +203,8 @@ async def render(job_id: str, request: Request):
             return job.to_dict()
         except KeyError as exc:
             raise HTTPException(status_code=404, detail=str(exc)) from exc
+        except JobConflictError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
         except RuntimeError as exc:
             return JSONResponse({"detail": str(exc)}, status_code=503)
         except Exception as exc:
@@ -1044,7 +1048,7 @@ def _runner_runtime_info(runner) -> dict[str, Any]:
   </main>
 <script>
 const RUNNING_STATES = new Set(['queued', 'loading_model', 'transcribing', 'postprocessing', 'rendering']);
-const EDIT_STATES = new Set(['waiting_review', 'rendering', 'done']);
+const EDIT_STATES = new Set(['waiting_review', 'done']);
 const TERMINAL_STATES = new Set(['waiting_review', 'done', 'failed', 'cancelled']);
 const fileInput = document.querySelector('#file');
 const importTitleEl = document.querySelector('#importTitle');

tests/test_app_api.py

@@ -213,6 +213,151 @@ def test_running_job_exposes_live_token_progress(self):
             self.assertEqual(finished["status"], "waiting_review")
             self.assertEqual(finished["usage"]["generated_tokens"], 5)
 
+    def test_rendering_job_rejects_subtitle_updates(self):
+        from fastapi.testclient import TestClient
+        from moss_transcribe_diarize.app.server import create_app
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8)
+            app.state.manager.model_runner = FakeRunner()
+            client = TestClient(app)
+
+            created = client.post(
+                "/api/jobs",
+                files={"file": ("sample.wav", b"audio", "audio/wav")},
+            )
+            job_id = created.json()["id"]
+            for _ in range(40):
+                job = client.get(f"/api/jobs/{job_id}").json()
+                if job["status"] == "waiting_review":
+                    break
+                time.sleep(0.05)
+            self.assertEqual(job["status"], "waiting_review")
+
+            original = client.get(f"/api/jobs/{job_id}/segments").json()["segments"]
+            edited = [dict(item) for item in original]
+            edited[0]["text"] = "must not be saved"
+            app.state.manager._set_status(
+                app.state.manager.get_job(job_id), "rendering", 0.97, error=None
+            )
+
+            updated = client.put(
+                f"/api/jobs/{job_id}/segments",
+                json={"segments": edited},
+            )
+
+            self.assertEqual(updated.status_code, 409)
+            self.assertIn("while the job is rendering", updated.json()["detail"])
+            self.assertEqual(
+                client.get(f"/api/jobs/{job_id}/segments").json()["segments"],
+                original,
+            )
+            self.assertIn(
+                "const EDIT_STATES = new Set(['waiting_review', 'done']);",
+                client.get("/").text,
+            )
+
+    def test_duplicate_render_request_is_rejected_before_starting_worker(self):
+        from fastapi.testclient import TestClient
+        from moss_transcribe_diarize.app.server import create_app
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8)
+            app.state.manager.model_runner = FakeRunner()
+            client = TestClient(app)
+
+            created = client.post(
+                "/api/jobs",
+                files={"file": ("sample.wav", b"audio", "audio/wav")},
+            )
+            job_id = created.json()["id"]
+            for _ in range(40):
+                job = client.get(f"/api/jobs/{job_id}").json()
+                if job["status"] == "waiting_review":
+                    break
+                time.sleep(0.05)
+            self.assertEqual(job["status"], "waiting_review")
+
+            class Available:
+                available = True
+
+            render_started = threading.Event()
+            release_render = threading.Event()
+
+            def block_render(*_args):
+                render_started.set()
+                release_render.wait(timeout=2)
+
+            with (
+                patch("moss_transcribe_diarize.app.jobs.detect_ffmpeg", return_value=Available()),
+                patch.object(app.state.manager, "_render_job", side_effect=block_render) as render_job,
+            ):
+                first = client.post(f"/api/jobs/{job_id}/render", json={"style": {}})
+                self.assertEqual(first.status_code, 200)
+                self.assertEqual(first.json()["status"], "rendering")
+                self.assertTrue(render_started.wait(timeout=2))
+
+                duplicate = client.post(f"/api/jobs/{job_id}/render", json={"style": {}})
+
+                self.assertEqual(duplicate.status_code, 409)
+                self.assertIn("already rendering", duplicate.json()["detail"])
+                self.assertEqual(render_job.call_count, 1)
+                release_render.set()
+
+    def test_restart_recovers_interrupted_render_for_review(self):
+        from fastapi.testclient import TestClient
+        from moss_transcribe_diarize.app.server import create_app
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8)
+            app.state.manager.model_runner = FakeRunner()
+            client = TestClient(app)
+
+            job_ids = []
+            for name in ("rendering.wav", "transcribing.wav"):
+                created = client.post(
+                    "/api/jobs",
+                    files={"file": (name, b"audio", "audio/wav")},
+                )
+                job_ids.append(created.json()["id"])
+            for job_id in job_ids:
+                for _ in range(40):
+                    job = client.get(f"/api/jobs/{job_id}").json()
+                    if job["status"] == "waiting_review":
+                        break
+                    time.sleep(0.05)
+                self.assertEqual(job["status"], "waiting_review")
+
+            rendering_id, transcribing_id = job_ids
+            app.state.manager._set_status(
+                app.state.manager.get_job(rendering_id), "rendering", 0.97, error=None
+            )
+            app.state.manager._set_status(
+                app.state.manager.get_job(transcribing_id), "transcribing", 0.5, error=None
+            )
+
+            restarted = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8)
+            restarted_client = TestClient(restarted)
+            recovered = restarted_client.get(f"/api/jobs/{rendering_id}").json()
+
+            self.assertEqual(recovered["status"], "waiting_review")
+            self.assertEqual(recovered["progress"], 0.95)
+            self.assertIn("interrupted", recovered["error"])
+            self.assertIn("retry rendering", recovered["error"])
+            segments = restarted_client.get(f"/api/jobs/{rendering_id}/segments").json()["segments"]
+            edited = [dict(item) for item in segments]
+            edited[0]["text"] = "editable after restart"
+            updated = restarted_client.put(
+                f"/api/jobs/{rendering_id}/segments",
+                json={"segments": edited},
+            )
+            self.assertEqual(updated.status_code, 200)
+
+            failed = restarted_client.get(f"/api/jobs/{transcribing_id}").json()
+            self.assertEqual(failed["status"], "failed")
+            self.assertEqual(failed["progress"], 1.0)
+            self.assertEqual(failed["error"], "Interrupted by previous server shutdown.")
+
 
 if __name__ == "__main__":
     unittest.main()