migrations/versions/0004_durable_jobs.py

@@ -0,0 +1,85 @@
+"""Durable jobs: jobs, job_items, job_events.
+
+Revision ID: 0004_durable_jobs
+Revises: 0003_thumbnails
+Create Date: 2026-07-15
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0004_durable_jobs"
+down_revision = "0003_thumbnails"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("job_type", sa.String(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("lock_key", sa.String(), nullable=True),
+        sa.Column("idempotency_key", sa.String(), nullable=True, unique=True),
+        sa.Column("lease_owner", sa.String(), nullable=True),
+        sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("fencing_token", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+    )
+    op.create_index("ix_jobs_state", "jobs", ["state"])
+    op.create_index("ix_jobs_lock_key", "jobs", ["lock_key"])
+
+    op.create_table(
+        "job_items",
+        sa.Column("job_id", sa.String(), sa.ForeignKey("jobs.id"), primary_key=True),
+        sa.Column("item_key", sa.String(), primary_key=True),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+
+    op.create_table(
+        "job_events",
+        sa.Column("id", sa.String(), primary_key=True),
+        sa.Column("job_id", sa.String(), sa.ForeignKey("jobs.id"), nullable=False),
+        sa.Column("event_type", sa.String(), nullable=False),
+        sa.Column("message", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_job_events_job_id", "job_events", ["job_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("job_events")
+    op.drop_table("job_items")
+    op.drop_table("jobs")

photo_pipeline/models/__init__.py

@@ -10,6 +10,7 @@ from photo_pipeline.models.duplicates import (
     DuplicateMember,
     DuplicateNegativeLink,
 )
+from photo_pipeline.models.jobs import Job, JobEvent, JobItem
 from photo_pipeline.models.thumbnails import Thumbnail
 
 __all__ = [
@@ -18,5 +19,8 @@ __all__ = [
     "DuplicateCluster",
     "DuplicateMember",
     "DuplicateNegativeLink",
+    "Job",
+    "JobItem",
+    "JobEvent",
     "Thumbnail",
 ]

photo_pipeline/models/jobs.py

@@ -0,0 +1,71 @@
+"""Durable jobs, their per-item progress, and their event history.
+
+A job is a unit of long-running work whose lifecycle survives browser, API, and
+worker interruption. Progress is derived from ``job_items`` (the durable source of
+truth), never from browser memory. ``job_events`` is an append-only activity log.
+Leases (owner + expiry + heartbeat) and a monotonic ``fencing_token`` let a
+recovered worker be fenced out after a newer worker takes ownership.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class Job(Base):
+    __tablename__ = "jobs"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    job_type: Mapped[str] = mapped_column(String, nullable=False)
+    state: Mapped[str] = mapped_column(String, nullable=False)
+    lock_key: Mapped[str | None] = mapped_column(String)
+    idempotency_key: Mapped[str | None] = mapped_column(String, unique=True)
+
+    lease_owner: Mapped[str | None] = mapped_column(String)
+    lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    fencing_token: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    error_code: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+
+
+class JobItem(Base):
+    __tablename__ = "job_items"
+
+    job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), primary_key=True)
+    item_key: Mapped[str] = mapped_column(String, primary_key=True)
+    state: Mapped[str] = mapped_column(String, nullable=False)
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    error_code: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+
+class JobEvent(Base):
+    __tablename__ = "job_events"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id"), nullable=False)
+    event_type: Mapped[str] = mapped_column(String, nullable=False)
+    message: Mapped[str | None] = mapped_column(String)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )

photo_pipeline/services/jobs.py

@@ -0,0 +1,359 @@
+"""JobService — durable job coordinator (no execution; the worker lands in US02-03).
+
+Responsibilities: enqueue with idempotency keys and one-mutating-job locks, hand a
+queued job to a worker atomically (compare-and-set claim with a fencing token),
+track heartbeats/leases, validate every state transition, derive progress from
+durable item state, request cancellation, and recover jobs whose worker vanished.
+
+State machine (concept §3):
+
+    queued ─▶ running ─▶ succeeded
+       │        │   ╲──▶ failed ─▶ retry_queued ─▶ running
+       │        ╲──▶ cancelling ─▶ cancelled
+       ╰──▶ cancelled
+
+Terminal states (succeeded, cancelled) have no outgoing transition, so a duplicate
+terminal transition is rejected.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections import Counter
+from datetime import datetime, timedelta, timezone
+from collections.abc import Iterable, Sequence
+
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import Job, JobEvent, JobItem
+
+
+class JobState:
+    QUEUED = "queued"
+    RUNNING = "running"
+    CANCELLING = "cancelling"
+    CANCELLED = "cancelled"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    RETRY_QUEUED = "retry_queued"
+
+
+class ItemState:
+    QUEUED = "queued"
+    RUNNING = "running"
+    SUCCEEDED = "succeeded"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+# Jobs that occupy a lock or are eligible to run are "active" for lock/one-job checks.
+ACTIVE_STATES = frozenset(
+    {JobState.QUEUED, JobState.RUNNING, JobState.CANCELLING, JobState.RETRY_QUEUED}
+)
+TERMINAL_STATES = frozenset({JobState.SUCCEEDED, JobState.CANCELLED})
+CLAIMABLE_STATES = frozenset({JobState.QUEUED, JobState.RETRY_QUEUED})
+
+ALLOWED_TRANSITIONS = {
+    JobState.QUEUED: {JobState.RUNNING, JobState.CANCELLED, JobState.CANCELLING},
+    JobState.RUNNING: {JobState.SUCCEEDED, JobState.FAILED, JobState.CANCELLING},
+    JobState.CANCELLING: {JobState.CANCELLED, JobState.SUCCEEDED, JobState.FAILED},
+    JobState.FAILED: {JobState.RETRY_QUEUED},
+    JobState.RETRY_QUEUED: {JobState.RUNNING, JobState.CANCELLED, JobState.CANCELLING},
+    JobState.SUCCEEDED: set(),
+    JobState.CANCELLED: set(),
+}
+
+ITEM_TRANSITIONS = {
+    ItemState.QUEUED: {ItemState.RUNNING, ItemState.CANCELLED},
+    ItemState.RUNNING: {ItemState.SUCCEEDED, ItemState.FAILED, ItemState.CANCELLED},
+    ItemState.FAILED: {ItemState.QUEUED, ItemState.RUNNING},
+    ItemState.SUCCEEDED: set(),
+    ItemState.CANCELLED: set(),
+}
+
+
+def can_transition(current: str, target: str) -> bool:
+    return target in ALLOWED_TRANSITIONS.get(current, set())
+
+
+class JobError(RuntimeError):
+    pass
+
+
+class JobBlocked(JobError):
+    """Enqueue refused by a prerequisite or the one-mutating-job lock policy."""
+
+    def __init__(self, code: str, detail: str = "") -> None:
+        super().__init__(f"{code}: {detail}" if detail else code)
+        self.code = code
+        self.detail = detail
+
+
+class InvalidTransition(JobError):
+    pass
+
+
+class JobConflict(JobError):
+    """Optimistic/fencing check failed; another owner or version won the race."""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class JobService:
+    def __init__(self, session_factory: sessionmaker) -> None:
+        self._session_factory = session_factory
+
+    # ── enqueue ────────────────────────────────────────────────────────────────
+    def enqueue(
+        self,
+        job_type: str,
+        *,
+        lock: str | None = None,
+        idempotency_key: str | None = None,
+        items: Iterable[str] = (),
+    ) -> dict:
+        with self._session_factory() as session:
+            if idempotency_key:
+                existing = session.scalar(
+                    select(Job).where(Job.idempotency_key == idempotency_key)
+                )
+                if existing:
+                    return self._snapshot(session, existing.id)
+            if lock:
+                blocker = session.scalar(
+                    select(Job).where(Job.lock_key == lock, Job.state.in_(ACTIVE_STATES))
+                )
+                if blocker:
+                    raise JobBlocked("lock_held", f"{lock} held by job {blocker.id}")
+
+            job = Job(
+                id=str(uuid.uuid4()),
+                job_type=job_type,
+                state=JobState.QUEUED,
+                lock_key=lock,
+                idempotency_key=idempotency_key,
+                fencing_token=0,
+                version=1,
+            )
+            session.add(job)
+            for item_key in items:
+                session.add(
+                    JobItem(job_id=job.id, item_key=item_key, state=ItemState.QUEUED)
+                )
+            self._event(session, job.id, "queued")
+            try:
+                session.commit()
+            except IntegrityError:
+                # Concurrent enqueue with the same idempotency key won the insert.
+                session.rollback()
+                existing = session.scalar(
+                    select(Job).where(Job.idempotency_key == idempotency_key)
+                )
+                if existing:
+                    return self._snapshot(session, existing.id)
+                raise
+            return self._snapshot(session, job.id)
+
+    # ── claim / lease ──────────────────────────────────────────────────────────
+    def claim(
+        self, job_types: Sequence[str], worker_id: str, *, lease_seconds: int = 60
+    ) -> dict | None:
+        now = _now()
+        expires = now + timedelta(seconds=lease_seconds)
+        with self._session_factory() as session:
+            candidates = (
+                session.execute(
+                    select(Job)
+                    .where(Job.state.in_(CLAIMABLE_STATES), Job.job_type.in_(list(job_types)))
+                    .order_by(Job.created_at)
+                )
+                .scalars()
+                .all()
+            )
+            for job in candidates:
+                # Compare-and-set: only the worker whose UPDATE matches the observed
+                # state wins; concurrent claimers see rowcount 0 and try the next job.
+                result = session.execute(
+                    update(Job)
+                    .where(Job.id == job.id, Job.state == job.state)
+                    .values(
+                        state=JobState.RUNNING,
+                        lease_owner=worker_id,
+                        lease_expires_at=expires,
+                        heartbeat_at=now,
+                        started_at=job.started_at or now,
+                        fencing_token=Job.fencing_token + 1,
+                        version=Job.version + 1,
+                        updated_at=now,
+                    )
+                )
+                if result.rowcount == 1:
+                    self._event(session, job.id, "claimed", worker_id)
+                    session.commit()
+                    return self._snapshot(session, job.id)
+            return None
+
+    def heartbeat(self, job_id: str, worker_id: str, *, lease_seconds: int = 60) -> bool:
+        now = _now()
+        with self._session_factory() as session:
+            result = session.execute(
+                update(Job)
+                .where(
+                    Job.id == job_id,
+                    Job.lease_owner == worker_id,
+                    Job.state == JobState.RUNNING,
+                )
+                .values(heartbeat_at=now, lease_expires_at=now + timedelta(seconds=lease_seconds))
+            )
+            session.commit()
+            return result.rowcount == 1
+
+    # ── transitions ────────────────────────────────────────────────────────────
+    def transition(
+        self,
+        job_id: str,
+        to_state: str,
+        *,
+        worker_id: str | None = None,
+        error: tuple[str, str] | None = None,
+    ) -> dict:
+        now = _now()
+        with self._session_factory() as session:
+            job = session.get(Job, job_id)
+            if job is None:
+                raise JobError(f"unknown job {job_id}")
+            if not can_transition(job.state, to_state):
+                raise InvalidTransition(f"{job.state} -> {to_state}")
+            if worker_id is not None and job.lease_owner not in (None, worker_id):
+                raise JobConflict(f"job {job_id} owned by {job.lease_owner}, not {worker_id}")
+            job.state = to_state
+            job.version += 1
+            job.updated_at = now
+            if error:
+                job.error_code, job.error_message = error
+            if to_state in TERMINAL_STATES:
+                job.finished_at = now
+                job.lease_owner = None
+                job.lease_expires_at = None
+            self._event(session, job_id, f"state:{to_state}", error[1] if error else None)
+            session.commit()
+            return self._snapshot(session, job_id)
+
+    def cancel(self, job_id: str) -> dict:
+        with self._session_factory() as session:
+            job = session.get(Job, job_id)
+            if job is None:
+                raise JobError(f"unknown job {job_id}")
+        # queued work can be cancelled outright; running work drains via cancelling.
+        target = (
+            JobState.CANCELLED
+            if job.state in (JobState.QUEUED, JobState.RETRY_QUEUED)
+            else JobState.CANCELLING
+        )
+        return self.transition(job_id, target)
+
+    # ── items / progress ───────────────────────────────────────────────────────
+    def set_item(
+        self, job_id: str, item_key: str, state: str, *, error: tuple[str, str] | None = None
+    ) -> None:
+        now = _now()
+        with self._session_factory() as session:
+            item = session.get(JobItem, {"job_id": job_id, "item_key": item_key})
+            if item is None:
+                raise JobError(f"unknown item {job_id}/{item_key}")
+            if state != item.state and state not in ITEM_TRANSITIONS.get(item.state, set()):
+                raise InvalidTransition(f"item {item.state} -> {state}")
+            if state == ItemState.RUNNING and item.state != ItemState.RUNNING:
+                item.attempt_count += 1
+            item.state = state
+            item.updated_at = now
+            if error:
+                item.error_code, item.error_message = error
+            session.commit()
+
+    def progress(self, job_id: str) -> dict:
+        with self._session_factory() as session:
+            states = session.execute(
+                select(JobItem.state).where(JobItem.job_id == job_id)
+            ).scalars().all()
+        counts = Counter(states)
+        total = len(states)
+        done = counts.get(ItemState.SUCCEEDED, 0) + counts.get(ItemState.FAILED, 0) + counts.get(
+            ItemState.CANCELLED, 0
+        )
+        return {"total": total, "done": done, "by_state": dict(counts)}
+
+    # ── recovery ───────────────────────────────────────────────────────────────
+    def recover_stale(self, *, now: datetime | None = None) -> list[str]:
+        """Return running jobs whose lease expired to ``retry_queued`` for safe resume."""
+        now = now or _now()
+        recovered: list[str] = []
+        with self._session_factory() as session:
+            stale = (
+                session.execute(
+                    select(Job).where(
+                        Job.state == JobState.RUNNING, Job.lease_expires_at < now
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            for job in stale:
+                job.state = JobState.RETRY_QUEUED
+                job.lease_owner = None
+                job.lease_expires_at = None
+                job.version += 1
+                job.updated_at = now
+                self._event(session, job.id, "interrupted", "lease expired")
+                recovered.append(job.id)
+            session.commit()
+        return recovered
+
+    def blockers(self, lock: str) -> list[dict]:
+        with self._session_factory() as session:
+            held = (
+                session.execute(
+                    select(Job).where(Job.lock_key == lock, Job.state.in_(ACTIVE_STATES))
+                )
+                .scalars()
+                .all()
+            )
+            return [{"id": j.id, "state": j.state, "job_type": j.job_type} for j in held]
+
+    def get(self, job_id: str) -> dict | None:
+        with self._session_factory() as session:
+            if session.get(Job, job_id) is None:
+                return None
+            return self._snapshot(session, job_id)
+
+    def events(self, job_id: str) -> list[dict]:
+        with self._session_factory() as session:
+            rows = session.execute(
+                select(JobEvent).where(JobEvent.job_id == job_id).order_by(JobEvent.created_at)
+            ).scalars()
+            return [{"type": e.event_type, "message": e.message} for e in rows]
+
+    # ── helpers ──────────────────────────────────────────────────────────────────
+    @staticmethod
+    def _event(session, job_id: str, event_type: str, message: str | None = None) -> None:
+        session.add(
+            JobEvent(id=str(uuid.uuid4()), job_id=job_id, event_type=event_type, message=message)
+        )
+
+    def _snapshot(self, session, job_id: str) -> dict:
+        job = session.get(Job, job_id)
+        return {
+            "id": job.id,
+            "job_type": job.job_type,
+            "state": job.state,
+            "lock_key": job.lock_key,
+            "idempotency_key": job.idempotency_key,
+            "lease_owner": job.lease_owner,
+            "fencing_token": job.fencing_token,
+            "version": job.version,
+            "error_code": job.error_code,
+        }

tests/integration/test_jobs.py

@@ -0,0 +1,147 @@
+"""Durable job coordinator: idempotent submission, atomic claim, locks, progress
+from durable item state, cancellation, stale recovery, and restart persistence."""
+
+from concurrent.futures import ThreadPoolExecutor
+from datetime import timedelta
+
+import pytest
+
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.services.jobs import (
+    ItemState,
+    JobBlocked,
+    JobService,
+    JobState,
+    _now,
+)
+
+
+@pytest.fixture
+def db_url(tmp_path):
+    url = f"sqlite:///{tmp_path / 'jobs.db'}"
+    run_migrations(url)
+    return url
+
+
+@pytest.fixture
+def make_factory():
+    engines = []
+
+    def _make(url):
+        engine = create_db_engine(url)
+        engines.append(engine)
+        return create_session_factory(engine)
+
+    yield _make
+    for engine in engines:
+        engine.dispose()
+
+
+@pytest.fixture
+def jobs(db_url, make_factory):
+    return JobService(make_factory(db_url))
+
+
+def test_enqueue_persists_items_and_event(jobs):
+    job = jobs.enqueue("scan", items=["a", "b", "c"])
+    assert job["state"] == JobState.QUEUED
+    assert jobs.progress(job["id"]) == {
+        "total": 3,
+        "done": 0,
+        "by_state": {ItemState.QUEUED: 3},
+    }
+    assert any(e["type"] == "queued" for e in jobs.events(job["id"]))
+
+
+def test_idempotency_key_returns_same_job(jobs):
+    a = jobs.enqueue("scan", idempotency_key="key-1")
+    b = jobs.enqueue("scan", idempotency_key="key-1")
+    assert a["id"] == b["id"]
+
+
+def test_concurrent_idempotent_submission_creates_one_job(db_url, make_factory):
+    service = JobService(make_factory(db_url))
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        ids = list(
+            pool.map(lambda _: service.enqueue("scan", idempotency_key="race")["id"], range(8))
+        )
+    assert len(set(ids)) == 1
+
+
+def test_one_mutating_job_lock_blocks(jobs):
+    jobs.enqueue("scan", lock="library_write")
+    with pytest.raises(JobBlocked) as excinfo:
+        jobs.enqueue("rename", lock="library_write")
+    assert excinfo.value.code == "lock_held"
+    assert jobs.blockers("library_write")
+
+
+def test_lock_frees_after_terminal(jobs):
+    first = jobs.enqueue("scan", lock="library_write")
+    jobs.claim(["scan"], "w1")
+    jobs.transition(first["id"], JobState.SUCCEEDED, worker_id="w1")
+    # Lock is free now; a new mutating job enqueues fine.
+    assert jobs.enqueue("rename", lock="library_write")["state"] == JobState.QUEUED
+
+
+def test_atomic_claim_gives_job_to_one_worker(db_url, make_factory):
+    service = JobService(make_factory(db_url))
+    service.enqueue("scan")
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        claims = list(pool.map(lambda i: service.claim(["scan"], f"w{i}"), range(4)))
+    winners = [c for c in claims if c is not None]
+    assert len(winners) == 1
+    assert winners[0]["state"] == JobState.RUNNING
+    assert winners[0]["fencing_token"] == 1
+
+
+def test_progress_derives_from_item_state(jobs):
+    job = jobs.enqueue("scan", items=["a", "b"])
+    jobs.set_item(job["id"], "a", ItemState.RUNNING)
+    jobs.set_item(job["id"], "a", ItemState.SUCCEEDED)
+    jobs.set_item(job["id"], "b", ItemState.RUNNING)
+    jobs.set_item(job["id"], "b", ItemState.FAILED, error=("boom", "kaboom"))
+    progress = jobs.progress(job["id"])
+    assert progress["total"] == 2 and progress["done"] == 2
+    assert progress["by_state"] == {ItemState.SUCCEEDED: 1, ItemState.FAILED: 1}
+
+
+def test_cancel_running_job_drains_via_cancelling(jobs):
+    job = jobs.enqueue("scan")
+    jobs.claim(["scan"], "w1")
+    after = jobs.cancel(job["id"])
+    assert after["state"] == JobState.CANCELLING
+
+
+def test_cancel_queued_job_is_immediate(jobs):
+    job = jobs.enqueue("scan")
+    assert jobs.cancel(job["id"])["state"] == JobState.CANCELLED
+
+
+def test_recover_stale_requeues_expired_leases(jobs):
+    job = jobs.enqueue("scan")
+    jobs.claim(["scan"], "w1", lease_seconds=30)
+    recovered = jobs.recover_stale(now=_now() + timedelta(hours=1))
+    assert job["id"] in recovered
+    assert jobs.get(job["id"])["state"] == JobState.RETRY_QUEUED
+    assert jobs.get(job["id"])["lease_owner"] is None
+
+
+def test_heartbeat_only_from_lease_owner(jobs):
+    job = jobs.enqueue("scan")
+    jobs.claim(["scan"], "w1")
+    assert jobs.heartbeat(job["id"], "w1") is True
+    assert jobs.heartbeat(job["id"], "intruder") is False
+
+
+def test_state_durable_across_restart(db_url, make_factory):
+    first = JobService(make_factory(db_url))
+    job = first.enqueue("scan", items=["a", "b"])
+    first.claim(["scan"], "w1")
+    first.set_item(job["id"], "a", ItemState.RUNNING)
+    first.set_item(job["id"], "a", ItemState.SUCCEEDED)
+
+    # Fresh service/engine against the same database.
+    reopened = JobService(make_factory(db_url))
+    assert reopened.get(job["id"])["state"] == JobState.RUNNING
+    assert reopened.progress(job["id"])["done"] == 1

tests/story_traceability.json

@@ -44,6 +44,10 @@
       "tests/unit/test_safety.py",
       "tests/unit/test_design_tokens.py",
       "tests/integration/test_safety_parity.py"
+    ],
+    "US02-02": [
+      "tests/unit/test_job_state_machine.py",
+      "tests/integration/test_jobs.py"
     ]
   }
 }

tests/unit/test_job_state_machine.py

@@ -0,0 +1,51 @@
+"""Job/item state-machine rules: only declared transitions are legal, and terminal
+states are truly terminal (no duplicate terminal transition)."""
+
+import pytest
+
+from photo_pipeline.services.jobs import (
+    ALLOWED_TRANSITIONS,
+    ITEM_TRANSITIONS,
+    TERMINAL_STATES,
+    ItemState,
+    JobState,
+    can_transition,
+)
+
+
+def test_happy_path_transitions_are_allowed():
+    assert can_transition(JobState.QUEUED, JobState.RUNNING)
+    assert can_transition(JobState.RUNNING, JobState.SUCCEEDED)
+    assert can_transition(JobState.RUNNING, JobState.FAILED)
+    assert can_transition(JobState.FAILED, JobState.RETRY_QUEUED)
+    assert can_transition(JobState.RETRY_QUEUED, JobState.RUNNING)
+    assert can_transition(JobState.RUNNING, JobState.CANCELLING)
+    assert can_transition(JobState.CANCELLING, JobState.CANCELLED)
+
+
+@pytest.mark.parametrize(
+    "current,target",
+    [
+        (JobState.QUEUED, JobState.SUCCEEDED),  # can't skip running
+        (JobState.SUCCEEDED, JobState.RUNNING),  # terminal, no revive
+        (JobState.CANCELLED, JobState.RUNNING),  # terminal
+        (JobState.RUNNING, JobState.QUEUED),  # no backward
+    ],
+)
+def test_invalid_transitions_are_rejected(current, target):
+    assert not can_transition(current, target)
+
+
+def test_terminal_states_have_no_outgoing_transition():
+    for state in TERMINAL_STATES:
+        assert ALLOWED_TRANSITIONS[state] == set()
+
+
+def test_duplicate_terminal_transition_is_rejected():
+    assert not can_transition(JobState.SUCCEEDED, JobState.SUCCEEDED)
+    assert not can_transition(JobState.CANCELLED, JobState.CANCELLED)
+
+
+def test_item_states_are_consistent():
+    assert ItemState.SUCCEEDED in ITEM_TRANSITIONS[ItemState.RUNNING]
+    assert ITEM_TRANSITIONS[ItemState.SUCCEEDED] == set()  # terminal item