feat(jobs): rebuild a job record from its artifacts when state.json is unreadable

store.load already logged "rebuilding from artifacts" on a damaged state.json,
but nothing rebuilt anything: run_job fell back to JobState.new, so every
artifact path was None and verification reported MISSING for files sitting on
disk. Measured on a finished job with video, audio and all four transcripts
present, a one-byte corruption of state.json planned
"download -> audio -> transcribe -> outputs". It now plans nothing.

recover.rebuild scans the job directory for the merged video, the FLAC and the
written outputs, then recovers the two things the filesystem cannot answer:
events.jsonl says *why* an artifact is absent, so a policy deletion is not
mistaken for data loss and re-fetched, and transcript.json says how the
transcript was produced, so a complete job is not re-transcribed. A deletion
event for a file that is present again is treated as history, not truth.

transcript.json gains task and num_beams. It already claimed to record how the
transcript was made while omitting two parameters that change the result, and
recovery needs them to decide whether a re-run is warranted. Transcripts written
before this default to transcribe/1, which can only cost a re-transcription
rather than mislabel one.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/ccn_transcribe/backends/openvino_backend.py

@@ -206,6 +206,8 @@ class _OpenvinoBackend:
             model_revision=self.revision,
             audio_s=audio_s,
             wall_s=wall,
+            task=request.task,
+            num_beams=request.num_beams,
         )
 
     def close(self) -> None:

src/ccn_transcribe/formats.py

@@ -61,6 +61,8 @@ def to_json(result: TranscriptResult, source: Mapping[str, object] | None = None
         "backend": result.backend,
         "device": result.device,
         "model": {"id": result.model_id, "revision": result.model_revision},
+        "task": result.task,
+        "num_beams": result.num_beams,
         "audio_s": result.audio_s,
         "wall_s": result.wall_s,
         "rtf": result.rtf,

src/ccn_transcribe/jobs/recover.py

@@ -0,0 +1,120 @@
+"""Rebuilding a job record when state.json is gone or unreadable.
+
+Artifacts are the source of truth, so most of the record can be read back off
+the filesystem. Two things the filesystem cannot answer are recovered from the
+files that do record them: *why* an artifact is absent comes from events.jsonl,
+and how the transcript was produced comes from transcript.json. Without this a
+damaged state.json costs a full re-download and re-transcription of work that is
+sitting on disk.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING, Any
+
+from ccn_transcribe import errors, formats
+from ccn_transcribe.jobs import store
+from ccn_transcribe.jobs.state import Artifact, ArtifactStatus, JobState, Outputs
+from ccn_transcribe.jobs.state import TranscriptionRecord as Record
+from ccn_transcribe.media import ffmpeg
+from ccn_transcribe.stages import download as download_stage
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from ccn_transcribe.paths import JobPaths
+
+log = logging.getLogger(__name__)
+
+
+def rebuild(job: JobPaths, *, job_id: str, source_url: str) -> JobState:
+    """Reconstruct as much of the record as the job directory can support."""
+    state = JobState.new(job_id=job_id, source_url=source_url)
+    _scan_media(job, state)
+    _scan_outputs(job, state)
+    _restore_transcription(job, state)
+    _replay_events(job, state)
+    return state
+
+
+def _safe_duration(path: Path) -> float | None:
+    try:
+        return ffmpeg.probe_duration(path)
+    except errors.ProbeError:
+        return None
+
+
+def _found(job: JobPaths, path: Path) -> Artifact:
+    return Artifact(
+        status=ArtifactStatus.PRESENT,
+        path=str(path.relative_to(job.root)),
+        size=path.stat().st_size,
+        duration_s=_safe_duration(path),
+    )
+
+
+def _scan_media(job: JobPaths, state: JobState) -> None:
+    video = download_stage.find_video(job)
+    if video is not None:
+        state.video = _found(job, video)
+        state.duration_s = state.video.duration_s
+    if job.audio_file.exists():
+        state.audio = _found(job, job.audio_file)
+
+
+def _scan_outputs(job: JobPaths, state: JobState) -> None:
+    written = {
+        path.suffix.lstrip("."): str(path.relative_to(job.root))
+        for path in sorted(job.out_dir.glob("transcript.*"))
+        if path.suffix.lstrip(".") in formats.FORMATS
+    }
+    if written:
+        state.outputs = Outputs(status=ArtifactStatus.PRESENT, paths=written)
+
+
+def _restore_transcription(job: JobPaths, state: JobState) -> None:
+    path = job.transcript_file("json")
+    if not path.exists():
+        return
+    try:
+        doc: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
+        model: dict[str, Any] = doc["model"]
+        state.transcription = Record(
+            model_id=model["id"],
+            model_revision=model.get("revision"),
+            backend=doc["backend"],
+            device=doc["device"],
+            language=doc.get("language"),
+            # Absent in transcripts written before these were recorded. Defaulting
+            # can only disagree with what the user asks for next, and a disagreement
+            # costs a re-transcription rather than a wrong answer.
+            task=doc.get("task", "transcribe"),
+            num_beams=doc.get("num_beams", 1),
+            segments=len(doc.get("segments", ())),
+            audio_s=doc.get("audio_s"),
+            wall_s=doc.get("wall_s"),
+            rtf=doc.get("rtf"),
+        )
+    except (OSError, ValueError, KeyError, TypeError):
+        log.warning("%s is unreadable; the transcription record cannot be recovered", path.name)
+        return
+
+    state.duration_s = state.duration_s or state.transcription.audio_s
+    source: dict[str, Any] = doc.get("source") or {}
+    state.title = source.get("title") or state.title
+
+
+def _replay_events(job: JobPaths, state: JobState) -> None:
+    for event in store.read_events(job):
+        if event.get("event") != "deleted":
+            continue
+        artifact = {"video": state.video, "audio": state.audio}.get(str(event.get("artifact")))
+        # A deletion explains an absence. If the file is back, it was re-fetched
+        # since, and the event is history rather than the current truth.
+        if artifact is None or artifact.status is ArtifactStatus.PRESENT:
+            continue
+        artifact.status = ArtifactStatus.DELETED_BY_POLICY
+        artifact.reason = event.get("reason")
+        artifact.deleted_at = event.get("at")

src/ccn_transcribe/jobs/runner.py

@@ -8,7 +8,7 @@ from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
 from ccn_transcribe import errors
-from ccn_transcribe.jobs import store, verify
+from ccn_transcribe.jobs import recover, store, verify
 from ccn_transcribe.jobs.plan import Stage, Verdicts, plan_stages
 from ccn_transcribe.jobs.state import Artifact, ArtifactStatus, JobState, Outputs
 from ccn_transcribe.jobs.state import TranscriptionRecord as Record
@@ -96,7 +96,7 @@ def run_job(
 ) -> JobOutcome:
     """Run whatever stages this job still needs."""
     job.ensure()
-    state = store.load(job) or JobState.new(job_id=job.root.name, source_url=url)
+    state = store.load(job) or recover.rebuild(job, job_id=job.root.name, source_url=url)
     if entry:
         state.title = entry.get("title") or state.title
         duration = entry.get("duration")
@@ -256,6 +256,8 @@ def _rerender(job: JobPaths, config: RunConfig, state: JobState) -> dict[str, st
         model_revision=record.model_revision,
         audio_s=record.audio_s or 0.0,
         wall_s=record.wall_s or 0.0,
+        task=record.task,
+        num_beams=record.num_beams,
     )
     return outputs_stage.write(job, result, config.formats, source=_source(state))
 

src/ccn_transcribe/transcript.py

@@ -36,6 +36,8 @@ class TranscriptResult:
     model_revision: str | None
     audio_s: float
     wall_s: float
+    task: str = "transcribe"
+    num_beams: int = 1
 
     @property
     def text(self) -> str:

tests/test_jobs_recover.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+import pytest
+
+from ccn_transcribe import paths
+from ccn_transcribe.jobs import recover, store
+from ccn_transcribe.jobs.state import ArtifactStatus
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+@pytest.fixture
+def job(tmp_path: Path) -> paths.JobPaths:
+    j = paths.Workspace(tmp_path).job("youtube-abc")
+    j.ensure()
+    return j
+
+
+def _transcript(job: paths.JobPaths, **over: object) -> None:
+    doc: dict[str, object] = {
+        "text": "hello",
+        "language": "en",
+        "backend": "openvino",
+        "device": "GPU",
+        "model": {"id": "OpenVINO/whisper", "revision": "abc123"},
+        "task": "translate",
+        "num_beams": 3,
+        "audio_s": 212.8,
+        "wall_s": 52.9,
+        "rtf": 4.02,
+        "segments": [{"start": 0.0, "end": 1.0, "text": "hello"}],
+        "source": {"id": "youtube-abc", "title": "A Title", "url": "https://a.test/v"},
+    }
+    doc.update(over)
+    job.transcript_file("json").write_text(json.dumps(doc), encoding="utf-8")
+
+
+class TestRebuildFromArtifacts:
+    def test_an_empty_job_recovers_an_empty_record(self, job: paths.JobPaths) -> None:
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="https://a.test/v")
+        assert state.video.status is ArtifactStatus.MISSING
+        assert state.audio.status is ArtifactStatus.MISSING
+        assert state.transcription is None
+
+    def test_a_downloaded_video_is_found(self, job: paths.JobPaths, sine_video: Path) -> None:
+        # Without this a corrupt state.json re-downloads a file already on disk.
+        (job.media_dir / "video.mkv").write_bytes(sine_video.read_bytes())
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.PRESENT
+        assert state.video.path == "media/video.mkv"
+        assert state.video.size == sine_video.stat().st_size
+        assert state.video.duration_s == pytest.approx(1.0, abs=0.2)
+
+    def test_an_unmerged_stream_is_not_mistaken_for_the_video(
+        self, job: paths.JobPaths, sine_video: Path
+    ) -> None:
+        (job.media_dir / "video.f251.webm").write_bytes(sine_video.read_bytes())
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.MISSING
+
+    def test_extracted_audio_is_found(self, job: paths.JobPaths, sine_wav: Path) -> None:
+        from ccn_transcribe.media import ffmpeg
+
+        ffmpeg.extract_flac(sine_wav, job.audio_file)
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.audio.status is ArtifactStatus.PRESENT
+        assert state.audio.path == "media/audio.flac"
+        assert state.audio.duration_s == pytest.approx(1.0, abs=0.2)
+
+    def test_unreadable_media_recovers_without_a_duration(self, job: paths.JobPaths) -> None:
+        (job.media_dir / "video.mkv").write_bytes(b"not a video")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.PRESENT
+        assert state.video.duration_s is None
+
+    def test_written_outputs_are_found(self, job: paths.JobPaths) -> None:
+        for name in ("txt", "srt", "json"):
+            job.transcript_file(name).write_text("x", encoding="utf-8")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.outputs.status is ArtifactStatus.PRESENT
+        assert state.outputs.paths == {
+            "json": "out/transcript.json",
+            "srt": "out/transcript.srt",
+            "txt": "out/transcript.txt",
+        }
+
+    def test_a_stray_temp_file_is_not_an_output(self, job: paths.JobPaths) -> None:
+        job.transcript_file("txt").with_suffix(".txt.tmp").write_text("x", encoding="utf-8")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.outputs.paths == {}
+
+
+class TestRestoreTranscription:
+    def test_the_record_comes_back_from_the_transcript(self, job: paths.JobPaths) -> None:
+        # Without it params_changed sees no record, and a completed job is
+        # re-transcribed from scratch despite every segment being on disk.
+        _transcript(job)
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.transcription is not None
+        assert state.transcription.model_id == "OpenVINO/whisper"
+        assert state.transcription.model_revision == "abc123"
+        assert state.transcription.language == "en"
+        assert state.transcription.task == "translate"
+        assert state.transcription.num_beams == 3
+        assert state.transcription.device == "GPU"
+        assert state.transcription.segments == 1
+        assert state.transcription.rtf == pytest.approx(4.02)
+
+    def test_the_title_comes_back_for_out_dir_naming(self, job: paths.JobPaths) -> None:
+        _transcript(job)
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.title == "A Title"
+        assert state.duration_s == pytest.approx(212.8)
+
+    def test_a_transcript_from_before_task_was_recorded(self, job: paths.JobPaths) -> None:
+        # Older transcripts omit task/num_beams; defaulting them is the safe
+        # direction, since a mismatch only costs a re-transcription.
+        _transcript(job)
+        doc = json.loads(job.transcript_file("json").read_text(encoding="utf-8"))
+        del doc["task"], doc["num_beams"]
+        job.transcript_file("json").write_text(json.dumps(doc), encoding="utf-8")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.transcription is not None
+        assert state.transcription.task == "transcribe"
+        assert state.transcription.num_beams == 1
+
+    def test_an_unreadable_transcript_leaves_no_record(self, job: paths.JobPaths) -> None:
+        job.transcript_file("json").write_text("{ truncated", encoding="utf-8")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.transcription is None
+
+    def test_a_transcript_missing_its_model_block(self, job: paths.JobPaths) -> None:
+        _transcript(job, model="not a mapping")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.transcription is None
+
+
+def _deleted(job: paths.JobPaths, which: str, reason: str) -> None:
+    store.append_event(job, {"event": "deleted", "artifact": which, "reason": reason})
+
+
+class TestReplayEvents:
+    def test_a_policy_deletion_is_recovered(self, job: paths.JobPaths) -> None:
+        # The whole point of events.jsonl: without it this reads as data loss and
+        # triggers a re-download of media the user asked us to delete.
+        _deleted(job, "video", "--no-retain-video")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.DELETED_BY_POLICY
+        assert state.video.reason == "--no-retain-video"
+        assert state.video.deleted_at is not None
+
+    def test_audio_deletions_too(self, job: paths.JobPaths) -> None:
+        _deleted(job, "audio", "--no-retain-audio")
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.audio.status is ArtifactStatus.DELETED_BY_POLICY
+
+    def test_a_superseded_deletion_does_not_hide_a_present_file(
+        self, job: paths.JobPaths, sine_video: Path
+    ) -> None:
+        # Deleted once, downloaded again later. The event is history, not truth:
+        # the file on disk wins.
+        _deleted(job, "video", "--no-retain-video")
+        (job.media_dir / "video.mkv").write_bytes(sine_video.read_bytes())
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.PRESENT
+
+    def test_unrelated_events_are_ignored(self, job: paths.JobPaths) -> None:
+        store.append_event(job, {"event": "started"})
+        store.append_event(job, {"event": "deleted", "artifact": "outputs"})
+        state = recover.rebuild(job, job_id="youtube-abc", source_url="u")
+        assert state.video.status is ArtifactStatus.MISSING
+        assert state.audio.status is ArtifactStatus.MISSING