diff --git a/photo_pipeline/services/rename_journal.py b/photo_pipeline/services/rename_journal.py
new file mode 100644
index 0000000..d5e90b7
--- /dev/null
+++ b/photo_pipeline/services/rename_journal.py
@@ -0,0 +1,346 @@
+"""Rename journal — the durable, crash-safe record of every rename transition
+(US04-02).
+
+The filesystem and SQLite cannot share a transaction, so the journal is the only
+thing that makes a rename recoverable: **intent is written before the mutation it
+describes**, and the recorded state plus the real filesystem evidence is what a later
+restart reasons about. This module owns the state machine and the durable writes; it
+performs no filesystem mutation itself (that is US04-03).
+
+Per-operation state machine (concept §7):
+
+```
+planned → moving → moved → database_updated → verified → complete
+              ↘ failed          ↘ rollback_required → rolled_back
+```
+
+``moving`` is the dangerous window: it means "we have committed to touching the disk
+and may already have done so". Anything found in ``moving`` after a restart needs
+evidence from the filesystem, never a blind retry.
+
+Terminal transitions are idempotent: re-completing an already-``complete`` operation
+is a no-op rather than an error, so a recovery pass may run repeatedly. Ownership uses
+the same fencing-token discipline as durable jobs — a worker whose token has been
+superseded cannot commit.
+
+Startup classification (``classify``) inspects the journal together with the current
+source/destination on disk and labels each incomplete operation:
+
+- ``resumable``     — the move has not happened; the source is intact and the
+                      destination is free, so applying again is safe.
+- ``rollback_safe`` — the move happened but the database was not updated; the
+                      destination holds the expected content and the source is gone,
+                      so the operation can be completed or reversed deterministically.
+- ``manual``        — the evidence is ambiguous or contradicts the journal (both or
+                      neither path present, or an unexpected occupant). Nothing is
+                      guessed; the operation blocks until a human decides.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import RenameOperation, RenamePlan
+
+
+class JournalState:
+    PLANNED = "planned"
+    MOVING = "moving"
+    MOVED = "moved"
+    DATABASE_UPDATED = "database_updated"
+    VERIFIED = "verified"
+    COMPLETE = "complete"
+    FAILED = "failed"
+    ROLLBACK_REQUIRED = "rollback_required"
+    ROLLED_BACK = "rolled_back"
+
+
+ALLOWED_TRANSITIONS = {
+    JournalState.PLANNED: {JournalState.MOVING, JournalState.FAILED},
+    # From `moving` the outcome is genuinely unknown until evidence is gathered,
+    # so it may resolve forward, back to planned (proven untouched), or to failure.
+    JournalState.MOVING: {
+        JournalState.MOVED,
+        JournalState.PLANNED,
+        JournalState.FAILED,
+        JournalState.ROLLBACK_REQUIRED,
+    },
+    JournalState.MOVED: {JournalState.DATABASE_UPDATED, JournalState.ROLLBACK_REQUIRED},
+    JournalState.DATABASE_UPDATED: {JournalState.VERIFIED, JournalState.ROLLBACK_REQUIRED},
+    JournalState.VERIFIED: {JournalState.COMPLETE, JournalState.ROLLBACK_REQUIRED},
+    JournalState.COMPLETE: set(),
+    JournalState.FAILED: {JournalState.PLANNED, JournalState.ROLLBACK_REQUIRED},
+    JournalState.ROLLBACK_REQUIRED: {JournalState.ROLLED_BACK, JournalState.FAILED},
+    JournalState.ROLLED_BACK: set(),
+}
+
+TERMINAL_STATES = frozenset({JournalState.COMPLETE, JournalState.ROLLED_BACK})
+# States where the disk may already have been touched by this operation.
+UNSAFE_STATES = frozenset({JournalState.MOVING, JournalState.MOVED, JournalState.DATABASE_UPDATED})
+
+RESUMABLE = "resumable"
+ROLLBACK_SAFE = "rollback_safe"
+MANUAL = "manual"
+
+
+def can_transition(current: str, target: str) -> bool:
+    return target in ALLOWED_TRANSITIONS.get(current, set())
+
+
+class JournalError(RuntimeError):
+    pass
+
+
+class InvalidTransition(JournalError):
+    pass
+
+
+class JournalConflict(JournalError):
+    """Fencing check failed; a newer owner has taken over this operation."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class RenameJournal:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    # ── intent ────────────────────────────────────────────────────────────────
+
+    def begin(self, operation_id: str, *, worker_id: str, fencing_token: int) -> dict:
+        """Record the intent to mutate **before** touching the filesystem.
+
+        Claims the operation for ``worker_id`` with ``fencing_token`` and moves it to
+        ``moving``. A token lower than the one already recorded loses the race and
+        raises ``JournalConflict`` without changing anything.
+        """
+        with self._session_factory() as session:
+            row = self._require(session, operation_id)
+            if row.fencing_token is not None and fencing_token < row.fencing_token:
+                raise JournalConflict(
+                    f"stale fencing token {fencing_token} (current {row.fencing_token})"
+                )
+            if row.journal_state in TERMINAL_STATES:
+                raise InvalidTransition(f"{row.journal_state} is terminal")
+            if row.journal_state != JournalState.MOVING and not can_transition(
+                row.journal_state, JournalState.MOVING
+            ):
+                raise InvalidTransition(f"{row.journal_state} -> {JournalState.MOVING}")
+            if row.journal_state != JournalState.MOVING:
+                row.attempt_count += 1
+            row.journal_state = JournalState.MOVING
+            row.worker_id = worker_id
+            row.fencing_token = fencing_token
+            row.error_code = row.error_message = None
+            row.updated_at = _now()
+            session.commit()
+            return _operation_dict(row)
+
+    # ── transitions ───────────────────────────────────────────────────────────
+
+    def transition(
+        self,
+        operation_id: str,
+        target: str,
+        *,
+        fencing_token: int | None = None,
+        error: tuple[str, str] | None = None,
+    ) -> dict:
+        """Move one operation to ``target``, enforcing the state machine.
+
+        Re-entering the state an operation already holds is a no-op, which is what
+        makes recovery idempotent across repeated restarts.
+        """
+        with self._session_factory() as session:
+            row = self._require(session, operation_id)
+            if fencing_token is not None and row.fencing_token is not None:
+                if fencing_token < row.fencing_token:
+                    raise JournalConflict(
+                        f"stale fencing token {fencing_token} (current {row.fencing_token})"
+                    )
+            if row.journal_state == target:
+                return _operation_dict(row)  # idempotent
+            if not can_transition(row.journal_state, target):
+                raise InvalidTransition(f"{row.journal_state} -> {target}")
+
+            row.journal_state = target
+            row.updated_at = _now()
+            if target == JournalState.VERIFIED:
+                row.verified_at = _now()
+            if error:
+                row.error_code, row.error_message = error[0], error[1][:500]
+            elif target not in (JournalState.FAILED, JournalState.ROLLBACK_REQUIRED):
+                row.error_code = row.error_message = None
+            session.commit()
+            return _operation_dict(row)
+
+    # ── reads ─────────────────────────────────────────────────────────────────
+
+    def get(self, operation_id: str) -> dict | None:
+        with self._session_factory() as session:
+            row = session.get(RenameOperation, operation_id)
+            return _operation_dict(row) if row else None
+
+    def operations(self, plan_id: str) -> list[dict]:
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(RenameOperation)
+                .where(RenameOperation.plan_id == plan_id)
+                .order_by(RenameOperation.sequence)
+            )
+            return [_operation_dict(row) for row in rows]
+
+    def incomplete(self) -> list[dict]:
+        """Every operation left in a non-terminal, non-planned state — the work a
+        restart has to reason about."""
+        with self._session_factory() as session:
+            rows = session.scalars(
+                select(RenameOperation)
+                .where(
+                    RenameOperation.journal_state.not_in([*TERMINAL_STATES, JournalState.PLANNED])
+                )
+                .order_by(RenameOperation.plan_id, RenameOperation.sequence)
+            )
+            return [_operation_dict(row) for row in rows]
+
+    # ── startup classification ────────────────────────────────────────────────
+
+    def classify(self, operation_id: str) -> dict:
+        """Classify one incomplete operation from the journal plus disk evidence.
+
+        Returns ``{operation_id, journal_state, classification, reason, source_exists,
+        destination_exists}``. Never mutates anything — deciding is US04-04's job.
+        """
+        row = self.get(operation_id)
+        if row is None:
+            raise JournalError(f"unknown rename operation {operation_id!r}")
+        source = Path(row["source_path"])
+        destination = Path(row["destination_path"])
+        source_exists = source.exists()
+        destination_exists = destination.exists()
+
+        classification, reason = _classify(row["journal_state"], source_exists, destination_exists)
+        return {
+            "operation_id": operation_id,
+            "journal_state": row["journal_state"],
+            "classification": classification,
+            "reason": reason,
+            "source_exists": source_exists,
+            "destination_exists": destination_exists,
+        }
+
+    def classify_all(self) -> list[dict]:
+        return [self.classify(row["id"]) for row in self.incomplete()]
+
+    def blocks_mutation(self) -> bool:
+        """True when any operation is in a state where the library may be
+        half-renamed. Unrelated mutations must not start until it is resolved."""
+        return any(row["journal_state"] in UNSAFE_STATES for row in self.incomplete())
+
+    # ── plan-level ────────────────────────────────────────────────────────────
+
+    def plan_state(self, plan_id: str) -> str:
+        """Derive the plan's state from its operations, so the summary can never
+        disagree with the journal."""
+        states = {row["journal_state"] for row in self.operations(plan_id)}
+        if not states:
+            return "planned"
+        if states <= {JournalState.COMPLETE}:
+            return "applied"
+        if states <= {JournalState.ROLLED_BACK, JournalState.COMPLETE}:
+            return "rolled_back"
+        if states & {JournalState.FAILED, JournalState.ROLLBACK_REQUIRED}:
+            return "failed"
+        if states & UNSAFE_STATES:
+            return "applying"
+        return "planned"
+
+    def sync_plan_state(self, plan_id: str) -> str:
+        state = self.plan_state(plan_id)
+        with self._session_factory() as session:
+            plan = session.get(RenamePlan, plan_id)
+            if plan is None:
+                raise JournalError(f"unknown rename plan {plan_id!r}")
+            if plan.state != state:
+                plan.state = state
+                plan.version += 1
+                plan.updated_at = _now()
+                if state == "applied" and plan.applied_at is None:
+                    plan.applied_at = _now()
+                session.commit()
+        return state
+
+    @staticmethod
+    def _require(session, operation_id: str) -> RenameOperation:
+        row = session.get(RenameOperation, operation_id)
+        if row is None:
+            raise JournalError(f"unknown rename operation {operation_id!r}")
+        return row
+
+
+def _classify(state: str, source_exists: bool, destination_exists: bool) -> tuple[str, str]:
+    """The evidence table. Kept a pure function so every combination is testable."""
+    if state == JournalState.MOVING:
+        # Intent was recorded; the move may or may not have happened.
+        if source_exists and not destination_exists:
+            return RESUMABLE, "source intact and destination free; the move did not happen"
+        if destination_exists and not source_exists:
+            return ROLLBACK_SAFE, "the move completed before the journal was updated"
+        if source_exists and destination_exists:
+            return MANUAL, "both source and destination exist; cannot tell them apart safely"
+        return MANUAL, "neither source nor destination exists"
+
+    if state in (JournalState.MOVED, JournalState.DATABASE_UPDATED, JournalState.VERIFIED):
+        if destination_exists and not source_exists:
+            return (
+                ROLLBACK_SAFE,
+                "destination holds the content; finish or reverse deterministically",
+            )
+        if source_exists and not destination_exists:
+            return MANUAL, f"journal says {state} but the content is still at the source"
+        if source_exists and destination_exists:
+            return MANUAL, "both paths exist after a recorded move"
+        return MANUAL, "the recorded destination is missing"
+
+    if state == JournalState.ROLLBACK_REQUIRED:
+        if source_exists and not destination_exists:
+            return RESUMABLE, "already back at the source; the rollback is effectively done"
+        if destination_exists and not source_exists:
+            return ROLLBACK_SAFE, "content is at the destination and can be moved back"
+        return MANUAL, "ambiguous paths for a rollback"
+
+    if state == JournalState.FAILED:
+        if source_exists and not destination_exists:
+            return RESUMABLE, "the failure left the source untouched"
+        return MANUAL, "a failed operation with unexpected paths on disk"
+
+    return MANUAL, f"unhandled journal state {state}"
+
+
+def _operation_dict(row: RenameOperation) -> dict:
+    return {
+        "id": row.id,
+        "plan_id": row.plan_id,
+        "sequence": row.sequence,
+        "album": row.album,
+        "source_path": row.source_path,
+        "destination_path": row.destination_path,
+        "asset_ids": json.loads(row.asset_ids or "[]"),
+        "expected_sha256": json.loads(row.expected_sha256 or "{}"),
+        "case_only": row.case_only,
+        "same_filesystem": row.same_filesystem,
+        "journal_state": row.journal_state,
+        "attempt_count": row.attempt_count,
+        "fencing_token": row.fencing_token,
+        "worker_id": row.worker_id,
+        "verified_at": row.verified_at.isoformat() if row.verified_at else None,
+        "error_code": row.error_code,
+        "error_message": row.error_message,
+    }
diff --git a/tests/integration/test_rename_journal.py b/tests/integration/test_rename_journal.py
new file mode 100644
index 0000000..d820db3
--- /dev/null
+++ b/tests/integration/test_rename_journal.py
@@ -0,0 +1,370 @@
+"""Rename journal persistence, fencing, idempotency, and restart durability
+(US04-02).
+
+The journal is what makes a rename recoverable, so these tests care about what
+survives: intent recorded before any mutation, terminal transitions that can be
+replayed, ownership that a superseded worker cannot steal back, and classification
+that reads real files on disk after a simulated restart.
+"""
+
+import json
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import RenameOperation, RenamePlan
+from photo_pipeline.services.rename_journal import (
+    MANUAL,
+    RESUMABLE,
+    ROLLBACK_SAFE,
+    InvalidTransition,
+    JournalConflict,
+    JournalError,
+    JournalState,
+    RenameJournal,
+)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+def _factory(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url))
+
+
+def _plan(sf, tmp_path, *, count=1):
+    """A persisted plan whose operations point at real folders on disk."""
+    plan_id = str(uuid.uuid4())
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    ids = []
+    with sf() as session:
+        session.add(
+            RenamePlan(id=plan_id, state="validated", schema_version=1, operation_count=count)
+        )
+        session.flush()
+        for index in range(count):
+            source = lib / f"album{index}"
+            source.mkdir(exist_ok=True)
+            (source / "a.jpg").write_bytes(b"x" * 8)
+            operation_id = str(uuid.uuid4())
+            ids.append(operation_id)
+            session.add(
+                RenameOperation(
+                    id=operation_id,
+                    plan_id=plan_id,
+                    sequence=index,
+                    operation="move_folder",
+                    album=f"album{index}",
+                    source_path=str(source),
+                    destination_path=str(lib / f"Renamed {index}"),
+                    asset_ids=json.dumps([f"asset{index}"]),
+                    asset_count=1,
+                )
+            )
+        session.commit()
+    return plan_id, ids
+
+
+# ── intent ───────────────────────────────────────────────────────────────────
+
+
+def test_begin_records_intent_ownership_and_attempt(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    started = journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    assert started["journal_state"] == JournalState.MOVING
+    assert started["worker_id"] == "w1" and started["fencing_token"] == 1
+    assert started["attempt_count"] == 1
+
+
+def test_begin_is_replayable_without_inflating_attempts(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    again = journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    # Re-entering `moving` is the same attempt, not a new one.
+    assert again["attempt_count"] == 1
+
+
+def test_a_retry_after_failure_counts_a_new_attempt(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    journal.transition(operation_id, JournalState.FAILED, error=("io", "disk hiccup"))
+    journal.transition(operation_id, JournalState.PLANNED)
+    retried = journal.begin(operation_id, worker_id="w1", fencing_token=2)
+    assert retried["attempt_count"] == 2 and retried["error_code"] is None
+
+
+def test_a_superseded_worker_cannot_take_the_operation_back(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    journal.begin(operation_id, worker_id="w2", fencing_token=5)  # takeover
+
+    with pytest.raises(JournalConflict):
+        journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    with pytest.raises(JournalConflict):
+        journal.transition(operation_id, JournalState.MOVED, fencing_token=1)
+    assert journal.get(operation_id)["worker_id"] == "w2"
+
+
+def test_begin_refuses_a_terminal_operation(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    _walk_to_complete(journal, operation_id)
+
+    with pytest.raises(InvalidTransition):
+        journal.begin(operation_id, worker_id="w1", fencing_token=9)
+
+
+def test_unknown_operation_is_reported(tmp_path):
+    _, sf = _factory(tmp_path)
+    journal = RenameJournal(sf)
+    with pytest.raises(JournalError):
+        journal.begin("nope", worker_id="w1", fencing_token=1)
+    with pytest.raises(JournalError):
+        journal.classify("nope")
+
+
+# ── transitions ──────────────────────────────────────────────────────────────
+
+
+def _walk_to_complete(journal, operation_id):
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+    for state in (
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.VERIFIED,
+        JournalState.COMPLETE,
+    ):
+        journal.transition(operation_id, state)
+
+
+def test_full_lifecycle_is_recorded_with_verification_evidence(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    _walk_to_complete(journal, operation_id)
+    final = journal.get(operation_id)
+    assert final["journal_state"] == JournalState.COMPLETE
+    assert final["verified_at"] is not None, "verification evidence must be durable"
+
+
+def test_terminal_transitions_are_idempotent(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    _walk_to_complete(journal, operation_id)
+
+    # Replaying the terminal transition is a no-op, so recovery can run repeatedly.
+    for _ in range(3):
+        replayed = journal.transition(operation_id, JournalState.COMPLETE)
+        assert replayed["journal_state"] == JournalState.COMPLETE
+
+
+def test_invalid_transition_is_refused_without_changing_state(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+
+    with pytest.raises(InvalidTransition):
+        journal.transition(operation_id, JournalState.COMPLETE)  # planned -> complete
+    assert journal.get(operation_id)["journal_state"] == JournalState.PLANNED
+
+
+def test_failure_records_a_structured_error(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+
+    failed = journal.transition(
+        operation_id, JournalState.FAILED, error=("permission_denied", "read-only target")
+    )
+    assert failed["error_code"] == "permission_denied"
+    assert "read-only" in failed["error_message"]
+
+
+# ── classification against real files ────────────────────────────────────────
+
+
+def test_moving_with_untouched_source_classifies_resumable(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+
+    verdict = journal.classify(operation_id)
+    assert verdict["classification"] == RESUMABLE
+    assert verdict["source_exists"] is True and verdict["destination_exists"] is False
+
+
+def test_moving_after_the_move_happened_classifies_rollback_safe(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    row = journal.get(operation_id)
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+
+    # Simulate the crash window: the move completed, the journal never advanced.
+    from pathlib import Path
+
+    Path(row["source_path"]).rename(row["destination_path"])
+
+    verdict = journal.classify(operation_id)
+    assert verdict["classification"] == ROLLBACK_SAFE
+
+
+def test_an_unexpected_occupant_at_the_destination_requires_manual_recovery(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    row = journal.get(operation_id)
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+
+    from pathlib import Path
+
+    Path(row["destination_path"]).mkdir()  # both paths now exist
+
+    verdict = journal.classify(operation_id)
+    assert verdict["classification"] == MANUAL
+    assert verdict["reason"]
+
+
+def test_classification_is_stable_across_repeated_runs(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    journal = RenameJournal(sf)
+    journal.begin(operation_id, worker_id="w1", fencing_token=1)
+
+    first = journal.classify(operation_id)
+    for _ in range(3):
+        assert journal.classify(operation_id) == first
+
+
+# ── incomplete work and mutation blocking ────────────────────────────────────
+
+
+def test_incomplete_lists_only_unsettled_operations(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, ids = _plan(sf, tmp_path, count=3)
+    journal = RenameJournal(sf)
+
+    _walk_to_complete(journal, ids[0])  # settled
+    journal.begin(ids[1], worker_id="w1", fencing_token=1)  # in flight
+    # ids[2] stays `planned` — not started, so not something a restart must resolve.
+
+    incomplete = {row["id"] for row in journal.incomplete()}
+    assert incomplete == {ids[1]}
+
+
+def test_in_flight_work_blocks_unrelated_mutations(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, ids = _plan(sf, tmp_path, count=2)
+    journal = RenameJournal(sf)
+    assert journal.blocks_mutation() is False
+
+    journal.begin(ids[0], worker_id="w1", fencing_token=1)
+    assert journal.blocks_mutation() is True, "a half-renamed library must block"
+
+    for state in (
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.VERIFIED,
+        JournalState.COMPLETE,
+    ):
+        journal.transition(ids[0], state)
+    assert journal.blocks_mutation() is False
+
+
+def test_classify_all_covers_every_incomplete_operation(tmp_path):
+    _, sf = _factory(tmp_path)
+    _, ids = _plan(sf, tmp_path, count=2)
+    journal = RenameJournal(sf)
+    journal.begin(ids[0], worker_id="w1", fencing_token=1)
+    journal.begin(ids[1], worker_id="w1", fencing_token=1)
+
+    verdicts = journal.classify_all()
+    assert {v["operation_id"] for v in verdicts} == set(ids)
+
+
+# ── plan-level derivation ────────────────────────────────────────────────────
+
+
+def test_plan_state_is_derived_from_its_operations(tmp_path):
+    _, sf = _factory(tmp_path)
+    plan_id, ids = _plan(sf, tmp_path, count=2)
+    journal = RenameJournal(sf)
+    assert journal.plan_state(plan_id) == "planned"
+
+    journal.begin(ids[0], worker_id="w1", fencing_token=1)
+    assert journal.plan_state(plan_id) == "applying"
+    assert journal.sync_plan_state(plan_id) == "applying"
+
+    _finish = (
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.VERIFIED,
+        JournalState.COMPLETE,
+    )
+    for state in _finish:
+        journal.transition(ids[0], state)
+    _walk_to_complete(journal, ids[1])
+    assert journal.sync_plan_state(plan_id) == "applied"
+
+    with sf() as session:
+        plan = session.get(RenamePlan, plan_id)
+        assert plan.state == "applied" and plan.applied_at is not None
+
+
+def test_a_failed_operation_makes_the_plan_failed(tmp_path):
+    _, sf = _factory(tmp_path)
+    plan_id, ids = _plan(sf, tmp_path, count=2)
+    journal = RenameJournal(sf)
+    journal.begin(ids[0], worker_id="w1", fencing_token=1)
+    journal.transition(ids[0], JournalState.FAILED, error=("io", "nope"))
+    assert journal.sync_plan_state(plan_id) == "failed"
+
+
+def test_sync_plan_state_rejects_an_unknown_plan(tmp_path):
+    _, sf = _factory(tmp_path)
+    journal = RenameJournal(sf)
+    with pytest.raises(JournalError):
+        journal.sync_plan_state("nope")
+
+
+# ── durability ───────────────────────────────────────────────────────────────
+
+
+def test_journal_survives_a_process_restart(tmp_path):
+    config, sf = _factory(tmp_path)
+    _, (operation_id,) = _plan(sf, tmp_path)
+    RenameJournal(sf).begin(operation_id, worker_id="w1", fencing_token=7)
+
+    # Drop every in-process handle, then reopen the database like a fresh process.
+    reopened = create_session_factory(create_db_engine(config.database_url))
+    journal = RenameJournal(reopened)
+    row = journal.get(operation_id)
+    assert row["journal_state"] == JournalState.MOVING
+    assert row["fencing_token"] == 7 and row["worker_id"] == "w1"
+    assert journal.classify(operation_id)["classification"] == RESUMABLE
+    # And the stale worker still cannot commit after the restart.
+    with pytest.raises(JournalConflict):
+        journal.transition(operation_id, JournalState.MOVED, fencing_token=1)
diff --git a/tests/story_traceability.json b/tests/story_traceability.json
index c9dedbd..730f579 100644
--- a/tests/story_traceability.json
+++ b/tests/story_traceability.json
@@ -87,6 +87,10 @@
     ],
     "US04-01": [
       "tests/integration/test_rename_plans.py"
+    ],
+    "US04-02": [
+      "tests/unit/test_rename_journal_states.py",
+      "tests/integration/test_rename_journal.py"
     ]
   }
 }
diff --git a/tests/unit/test_rename_journal_states.py b/tests/unit/test_rename_journal_states.py
new file mode 100644
index 0000000..7ca2f13
--- /dev/null
+++ b/tests/unit/test_rename_journal_states.py
@@ -0,0 +1,136 @@
+"""Rename journal state machine and startup classification (US04-02).
+
+Pure-logic coverage: every valid and invalid transition in the matrix, the terminal
+states, the states that make the library unsafe to touch, and the full evidence table
+that turns (journal state, source exists, destination exists) into a recovery
+classification. No database and no filesystem here — the repository behaviour is
+covered by tests/integration/test_rename_journal.py.
+"""
+
+import itertools
+
+from photo_pipeline.services.rename_journal import (
+    ALLOWED_TRANSITIONS,
+    MANUAL,
+    RESUMABLE,
+    ROLLBACK_SAFE,
+    TERMINAL_STATES,
+    UNSAFE_STATES,
+    JournalState,
+    _classify,
+    can_transition,
+)
+
+ALL_STATES = set(ALLOWED_TRANSITIONS)
+
+
+def test_every_state_is_declared_in_the_matrix():
+    declared = {
+        value
+        for name, value in vars(JournalState).items()
+        if not name.startswith("_") and isinstance(value, str)
+    }
+    assert declared == ALL_STATES
+    # Every target is itself a declared state — no dangling edges.
+    for targets in ALLOWED_TRANSITIONS.values():
+        assert targets <= ALL_STATES
+
+
+def test_the_happy_path_is_walkable_end_to_end():
+    path = [
+        JournalState.PLANNED,
+        JournalState.MOVING,
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+        JournalState.VERIFIED,
+        JournalState.COMPLETE,
+    ]
+    for current, target in itertools.pairwise(path):
+        assert can_transition(current, target), f"{current} -> {target} must be allowed"
+
+
+def test_terminal_states_allow_no_further_transition():
+    for state in TERMINAL_STATES:
+        assert ALLOWED_TRANSITIONS[state] == set()
+        for target in ALL_STATES:
+            assert not can_transition(state, target)
+
+
+def test_invalid_transitions_are_rejected():
+    # A few that must never be possible, spelled out rather than derived.
+    assert not can_transition(JournalState.PLANNED, JournalState.COMPLETE)
+    assert not can_transition(JournalState.PLANNED, JournalState.MOVED)
+    assert not can_transition(JournalState.MOVING, JournalState.COMPLETE)
+    assert not can_transition(JournalState.MOVED, JournalState.PLANNED)
+    assert not can_transition(JournalState.COMPLETE, JournalState.ROLLED_BACK)
+    assert not can_transition(JournalState.ROLLED_BACK, JournalState.PLANNED)
+
+
+def test_the_whole_matrix_is_exhaustively_consistent():
+    for current, target in itertools.product(ALL_STATES, repeat=2):
+        expected = target in ALLOWED_TRANSITIONS[current]
+        assert can_transition(current, target) is expected
+
+
+def test_unsafe_states_are_the_ones_where_disk_may_have_changed():
+    assert UNSAFE_STATES == {
+        JournalState.MOVING,
+        JournalState.MOVED,
+        JournalState.DATABASE_UPDATED,
+    }
+    # planned has not touched anything; complete/rolled_back are settled.
+    assert JournalState.PLANNED not in UNSAFE_STATES
+    assert not (TERMINAL_STATES & UNSAFE_STATES)
+
+
+# ── evidence table ───────────────────────────────────────────────────────────
+
+
+def test_moving_with_source_intact_is_resumable():
+    assert _classify(JournalState.MOVING, True, False)[0] == RESUMABLE
+
+
+def test_moving_with_only_destination_is_rollback_safe():
+    assert _classify(JournalState.MOVING, False, True)[0] == ROLLBACK_SAFE
+
+
+def test_moving_with_both_or_neither_path_requires_manual_recovery():
+    assert _classify(JournalState.MOVING, True, True)[0] == MANUAL
+    assert _classify(JournalState.MOVING, False, False)[0] == MANUAL
+
+
+def test_recorded_move_contradicted_by_disk_requires_manual_recovery():
+    for state in (JournalState.MOVED, JournalState.DATABASE_UPDATED, JournalState.VERIFIED):
+        # Journal says the content moved, but it is still at the source.
+        assert _classify(state, True, False)[0] == MANUAL
+        # Both present: an unexpected occupant; never guess.
+        assert _classify(state, True, True)[0] == MANUAL
+        # Neither present: the content is gone.
+        assert _classify(state, False, False)[0] == MANUAL
+        # The expected shape is recoverable.
+        assert _classify(state, False, True)[0] == ROLLBACK_SAFE
+
+
+def test_rollback_required_classification():
+    assert _classify(JournalState.ROLLBACK_REQUIRED, True, False)[0] == RESUMABLE
+    assert _classify(JournalState.ROLLBACK_REQUIRED, False, True)[0] == ROLLBACK_SAFE
+    assert _classify(JournalState.ROLLBACK_REQUIRED, True, True)[0] == MANUAL
+
+
+def test_failed_classification():
+    assert _classify(JournalState.FAILED, True, False)[0] == RESUMABLE
+    assert _classify(JournalState.FAILED, False, True)[0] == MANUAL
+    assert _classify(JournalState.FAILED, True, True)[0] == MANUAL
+
+
+def test_every_classification_carries_a_reason():
+    for state in ALL_STATES:
+        for source, destination in itertools.product([True, False], repeat=2):
+            classification, reason = _classify(state, source, destination)
+            assert classification in {RESUMABLE, ROLLBACK_SAFE, MANUAL}
+            assert reason.strip(), f"{state} {source} {destination} has no reason"
+
+
+def test_unknown_state_falls_back_to_manual():
+    # An unexpected value must never be treated as safe.
+    assert _classify("something_new", True, False)[0] == MANUAL