fix: atomically persist subtitle job state

moss_transcribe_diarize/app/jobs.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 import queue
 import shutil
 import threading
@@ -490,7 +491,13 @@ def _touch(self, job: JobRecord, *, error: str | None = None, save: bool = True)
             self._save_job(job)
 
     def _save_job(self, job: JobRecord) -> None:
-        job.job_path.write_text(json.dumps(job.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8")
+        payload = json.dumps(job.to_dict(), ensure_ascii=False, indent=2)
+        temp_path = job.job_path.with_name(f".{job.job_path.name}.{uuid.uuid4().hex}.tmp")
+        try:
+            temp_path.write_text(payload, encoding="utf-8")
+            os.replace(temp_path, job.job_path)
+        finally:
+            temp_path.unlink(missing_ok=True)
 
     def _should_save_live_progress(self, job_id: str) -> bool:
         now = time.time()

tests/test_job_persistence.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from moss_transcribe_diarize.app.jobs import JobManager, JobRecord
+
+
+class FakeRunner:
+    model_path = "fake-model"
+
+
+class JobPersistenceTest(unittest.TestCase):
+    def _make_job(self, runs_dir: Path) -> JobRecord:
+        job_dir = runs_dir / "test-job"
+        job_dir.mkdir()
+        return JobRecord(
+            id="test-job",
+            status="waiting_review",
+            media_name="interview.wav",
+            input_path=str(job_dir / "input.wav"),
+            job_dir=str(job_dir),
+            inference_prompt="transcribe",
+            max_length=4096,
+            max_new_tokens=1024,
+            decoding="greedy",
+            temperature=None,
+        )
+
+    def test_interrupted_temp_write_preserves_last_valid_job(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runs_dir = Path(tmpdir)
+            manager = JobManager.__new__(JobManager)
+            job = self._make_job(runs_dir)
+            manager._save_job(job)
+            previous_payload = job.job_path.read_text(encoding="utf-8")
+
+            job.status = "done"
+            original_write_text = Path.write_text
+
+            def interrupted_write(path, data, *args, **kwargs):
+                original_write_text(path, data[:24], *args, **kwargs)
+                raise OSError("simulated interrupted write")
+
+            with patch.object(Path, "write_text", new=interrupted_write):
+                with self.assertRaisesRegex(OSError, "simulated interrupted write"):
+                    manager._save_job(job)
+
+            self.assertEqual(job.job_path.read_text(encoding="utf-8"), previous_payload)
+            self.assertEqual(json.loads(previous_payload)["status"], "waiting_review")
+            self.assertEqual(list(job.job_path.parent.glob(".job.json.*.tmp")), [])
+
+    def test_save_atomically_replaces_from_the_job_directory(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runs_dir = Path(tmpdir)
+            manager = JobManager.__new__(JobManager)
+            job = self._make_job(runs_dir)
+
+            with patch("moss_transcribe_diarize.app.jobs.os.replace", wraps=os.replace) as replace:
+                manager._save_job(job)
+
+            source, destination = replace.call_args.args
+            self.assertEqual(Path(source).parent, job.job_path.parent)
+            self.assertEqual(Path(destination), job.job_path)
+            self.assertEqual(json.loads(job.job_path.read_text(encoding="utf-8"))["id"], job.id)
+            self.assertEqual(list(job.job_path.parent.glob(".job.json.*.tmp")), [])
+
+    def test_restart_ignores_orphan_temp_file_and_loads_last_valid_job(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            runs_dir = Path(tmpdir)
+            writer = JobManager.__new__(JobManager)
+            job = self._make_job(runs_dir)
+            writer._save_job(job)
+            job.job_path.with_name(".job.json.orphan.tmp").write_text("{incomplete", encoding="utf-8")
+
+            manager = JobManager(
+                runs_dir,
+                FakeRunner(),
+                prompt="transcribe",
+                max_length=4096,
+                max_new_tokens=1024,
+            )
+
+            self.assertEqual(manager.get_job(job.id).status, "waiting_review")
+
+
+if __name__ == "__main__":
+    unittest.main()