diff --git a/src/ccn_transcribe/errors.py b/src/ccn_transcribe/errors.py
new file mode 100644
index 0000000..b8d3abb
--- /dev/null
+++ b/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"
diff --git a/tests/test_errors.py b/tests/test_errors.py
new file mode 100644
index 0000000..ec7f5e4
--- /dev/null
+++ b/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