feat(errors): add exception hierarchy with actionable hints

Splits failures into FatalError (abort the run) and JobError (record and
continue the batch), which is the distinction the batch runner needs.
DiskFull and SchemaTooNew are fatal on purpose: continuing past ENOSPC
only produces more corrupt artifacts.

Only NetworkError is retryable; retrying a private video is noise.

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

src/ccn_transcribe/errors.py

@@ -0,0 +1,164 @@
+"""Exception hierarchy for the pipeline.
+
+Every error carries an actionable ``hint``; the CLI prints ``hint`` instead of a
+traceback. The split that matters is ``FatalError`` (abort the whole run) versus
+``JobError`` (record it and move to the next URL).
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+EXIT_OK = 0
+EXIT_JOB_FAILED = 1
+EXIT_FATAL = 2
+EXIT_INTERRUPTED = 130
+
+
+class CcnError(Exception):
+    code: ClassVar[str] = "ccn_error"
+    exit_code: ClassVar[int] = EXIT_JOB_FAILED
+    retryable: ClassVar[bool] = False
+
+    def __init__(self, message: str, *, hint: str | None = None) -> None:
+        super().__init__(message)
+        self.message = message
+        self.hint = hint
+
+    def __str__(self) -> str:
+        return self.message
+
+
+class FatalError(CcnError):
+    """Aborts the run. Continuing would only produce more bad artifacts."""
+
+    code: ClassVar[str] = "fatal"
+    exit_code: ClassVar[int] = EXIT_FATAL
+
+
+class ConfigError(FatalError):
+    code: ClassVar[str] = "config"
+
+
+class PreflightError(FatalError):
+    code: ClassVar[str] = "preflight"
+
+
+class DiskFullError(FatalError):
+    code: ClassVar[str] = "disk_full"
+
+
+class SchemaTooNewError(FatalError):
+    code: ClassVar[str] = "schema_too_new"
+
+
+class JobError(CcnError):
+    """Scoped to one job; the batch continues."""
+
+    code: ClassVar[str] = "job"
+
+    def __init__(self, message: str, *, job_id: str | None = None, hint: str | None = None) -> None:
+        super().__init__(message, hint=hint)
+        self.job_id = job_id
+
+
+class LockedError(JobError):
+    code: ClassVar[str] = "locked"
+
+
+class SourceError(JobError):
+    """The video itself cannot be obtained."""
+
+    code: ClassVar[str] = "source"
+
+
+class GeoBlockedError(SourceError):
+    code: ClassVar[str] = "geo_blocked"
+
+
+class PrivateVideoError(SourceError):
+    code: ClassVar[str] = "private_video"
+
+
+class AgeGatedError(SourceError):
+    code: ClassVar[str] = "age_gated"
+
+
+class MembersOnlyError(SourceError):
+    code: ClassVar[str] = "members_only"
+
+
+class NotFoundError(SourceError):
+    code: ClassVar[str] = "not_found"
+
+
+class BotCheckError(SourceError):
+    code: ClassVar[str] = "bot_check"
+
+
+class NetworkError(SourceError):
+    code: ClassVar[str] = "network"
+    retryable: ClassVar[bool] = True
+
+
+class DownloadError(JobError):
+    code: ClassVar[str] = "download"
+
+
+class Aria2Error(DownloadError):
+    code: ClassVar[str] = "aria2"
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        aria2_exit_code: int | None = None,
+        job_id: str | None = None,
+        hint: str | None = None,
+    ) -> None:
+        super().__init__(message, job_id=job_id, hint=hint)
+        self.aria2_exit_code = aria2_exit_code
+
+
+class IncompleteDownloadError(DownloadError):
+    code: ClassVar[str] = "incomplete_download"
+
+
+class MediaError(JobError):
+    code: ClassVar[str] = "media"
+
+
+class NoAudioStreamError(MediaError):
+    code: ClassVar[str] = "no_audio_stream"
+
+
+class DecodeError(MediaError):
+    code: ClassVar[str] = "decode"
+
+
+class ProbeError(MediaError):
+    code: ClassVar[str] = "probe"
+
+
+class TranscribeError(JobError):
+    code: ClassVar[str] = "transcribe"
+
+
+class BackendUnavailableError(TranscribeError):
+    code: ClassVar[str] = "backend_unavailable"
+
+
+class BackendRuntimeError(TranscribeError):
+    code: ClassVar[str] = "backend_runtime"
+
+
+class BackendNotImplementedError(TranscribeError):
+    code: ClassVar[str] = "backend_not_implemented"
+
+
+class ModelUnavailableError(TranscribeError):
+    code: ClassVar[str] = "model_unavailable"
+
+
+class OutOfMemoryError(TranscribeError):
+    code: ClassVar[str] = "out_of_memory"

tests/test_errors.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import pytest
+
+from ccn_transcribe import errors
+
+
+def test_base_carries_message_and_hint() -> None:
+    err = errors.CcnError("something broke", hint="try this")
+    assert str(err) == "something broke"
+    assert err.hint == "try this"
+    assert err.exit_code == errors.EXIT_JOB_FAILED
+
+
+def test_hint_defaults_to_none() -> None:
+    assert errors.CcnError("bare").hint is None
+
+
+@pytest.mark.parametrize(
+    "cls",
+    [errors.ConfigError, errors.PreflightError, errors.DiskFullError, errors.SchemaTooNewError],
+)
+def test_fatal_errors_abort_the_run(cls: type[errors.CcnError]) -> None:
+    err = cls("nope")
+    assert err.exit_code == errors.EXIT_FATAL
+    assert isinstance(err, errors.FatalError)
+    # A fatal error must never be mistaken for a per-job failure.
+    assert not isinstance(err, errors.JobError)
+
+
+def test_job_error_carries_job_id() -> None:
+    err = errors.JobError("failed", job_id="youtube-abc123")
+    assert err.job_id == "youtube-abc123"
+    assert err.exit_code == errors.EXIT_JOB_FAILED
+
+
+def test_job_error_job_id_optional() -> None:
+    assert errors.JobError("failed").job_id is None
+
+
+@pytest.mark.parametrize(
+    "cls",
+    [
+        errors.GeoBlockedError,
+        errors.PrivateVideoError,
+        errors.AgeGatedError,
+        errors.MembersOnlyError,
+        errors.NotFoundError,
+        errors.BotCheckError,
+        errors.NetworkError,
+    ],
+)
+def test_source_errors_are_job_errors(cls: type[errors.SourceError]) -> None:
+    assert issubclass(cls, errors.SourceError)
+    assert issubclass(cls, errors.JobError)
+
+
+def test_only_network_errors_are_retryable() -> None:
+    assert errors.NetworkError("flaky").retryable is True
+    assert errors.PrivateVideoError("nope").retryable is False
+    assert errors.NotFoundError("gone").retryable is False
+
+
+def test_every_error_class_declares_a_distinct_code() -> None:
+    classes = [
+        v for v in vars(errors).values() if isinstance(v, type) and issubclass(v, errors.CcnError)
+    ]
+    codes = [c.code for c in classes]
+    assert len(codes) == len(set(codes)), "duplicate .code values"
+
+
+def test_aria2_error_carries_process_exit_code() -> None:
+    err = errors.Aria2Error(
+        "aria2c failed", aria2_exit_code=5, job_id="youtube-x", hint="slow link"
+    )
+    assert err.aria2_exit_code == 5
+    assert err.job_id == "youtube-x"
+    assert err.hint == "slow link"
+    assert isinstance(err, errors.DownloadError)
+
+
+def test_aria2_error_exit_code_optional() -> None:
+    assert errors.Aria2Error("boom").aria2_exit_code is None