diff --git a/src/manga/PerfStats.py b/src/manga/PerfStats.py
new file mode 100644
index 0000000..714a46f
--- /dev/null
+++ b/src/manga/PerfStats.py
@@ -0,0 +1,242 @@
+"""
+perf_stats.py
+=============
+
+Lightweight performance profiler for the Suwayomi -> Kavita move pipeline.
+
+It records, per move run, how long each step of every chapter takes plus
+per-series and per-run totals, so a slowdown can be traced to the step
+responsible (cover download, image-dimension probing, CBZ packing, …).
+
+Data model (one entry per run, newest first)::
+
+    {
+      "runs": [
+        {
+          "startedAt":    1700000000,        # unix seconds
+          "finishedAt":   1700000123,
+          "totalSeconds": 123.4,             # wall clock of the whole run
+          "seriesCount":  2,
+          "chapterCount": 31,
+          "stepTotals":   {                  # summed over ALL chapters
+            "cover": 41.2, "image_dimensions": 55.8, "pack_cbz": 18.1, ...
+          },
+          "seriesStepTotals": {              # summed over ALL series
+            "fetch_metadata": 2.4, "person_sync": 9.7
+          },
+          "series": [
+            {
+              "title": "Call of the Night",
+              "totalSeconds": 60.2,
+              "chapterCount": 20,
+              "steps": {"fetch_metadata": 1.2, "person_sync": 3.4},
+              "chapters": [
+                {"chapter": "1", "ok": true, "totalSeconds": 11.5,
+                 "steps": {"cover": 1.8, "image_dimensions": 4.2, ...}}
+              ]
+            }
+          ]
+        }
+      ]
+    }
+
+Usage from the mover::
+
+    perf = PerfStats(path)                 # path=None -> disabled (no-op)
+    run = perf.begin_run()
+    series = run.begin_series("Title")
+    with series.measure("fetch_metadata"):
+        ...
+    chap = series.begin_chapter("1")
+    with chap.measure("pack_cbz"):
+        ...
+    chap.finish(ok=True)
+    series.finish()
+    run.finish()                           # persists the run to disk
+
+When ``path`` is None every recorder is a no-op and nothing is written,
+so the profiler can be left permanently wired in with negligible cost.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+from contextlib import contextmanager
+from pathlib import Path
+
+
+# Keep the JSON small: only the most recent runs are retained on disk.
+_MAX_RUNS = 30
+
+
+class _StepTimer:
+    """
+    Base recorder: accumulates ``{step_name: seconds}`` and tracks its own
+    wall-clock lifetime.  ``enabled=False`` turns every method into a no-op.
+    """
+
+    def __init__(self, enabled: bool = True):
+        self.steps: dict[str, float] = {}
+        self._enabled = enabled
+        self._t0 = time.monotonic()
+
+    @contextmanager
+    def measure(self, name: str):
+        """Context manager timing a named step (accumulates on repeat use)."""
+        if not self._enabled:
+            yield
+            return
+        start = time.monotonic()
+        try:
+            yield
+        finally:
+            self.steps[name] = round(
+                self.steps.get(name, 0.0) + (time.monotonic() - start), 4)
+
+    def elapsed(self) -> float:
+        return round(time.monotonic() - self._t0, 4)
+
+
+class ChapterRecorder(_StepTimer):
+    """Per-chapter step timer."""
+
+    def __init__(self, series: "SeriesRecorder", chapter: str,
+                 enabled: bool = True):
+        super().__init__(enabled)
+        self._series = series
+        self._chapter = chapter
+        self._ok = True
+
+    def finish(self, *, ok: bool = True) -> None:
+        self._ok = ok
+        if not self._enabled:
+            return
+        self._series._chapters.append({
+            "chapter":      self._chapter,
+            "ok":           ok,
+            "totalSeconds": self.elapsed(),
+            "steps":        self.steps,
+        })
+
+
+class SeriesRecorder(_StepTimer):
+    """Per-series step timer; also collects its chapters."""
+
+    def __init__(self, run: "RunRecorder", title: str, enabled: bool = True):
+        super().__init__(enabled)
+        self._run = run
+        self._title = title
+        self._chapters: list[dict] = []
+
+    def begin_chapter(self, chapter: str) -> ChapterRecorder:
+        return ChapterRecorder(self, chapter, enabled=self._enabled)
+
+    def finish(self) -> None:
+        if not self._enabled:
+            return
+        self._run._series.append({
+            "title":        self._title,
+            "totalSeconds": self.elapsed(),
+            "chapterCount": len(self._chapters),
+            "steps":        self.steps,
+            "chapters":     self._chapters,
+        })
+
+
+class RunRecorder:
+    """Top-level recorder for one full move run."""
+
+    def __init__(self, stats: "PerfStats", enabled: bool = True):
+        self._stats = stats
+        self._enabled = enabled
+        self._series: list[dict] = []
+        self._started = time.time()
+        self._t0 = time.monotonic()
+
+    def begin_series(self, title: str) -> SeriesRecorder:
+        return SeriesRecorder(self, title, enabled=self._enabled)
+
+    def finish(self) -> dict | None:
+        """Aggregates the run and persists it.  Returns the run dict."""
+        if not self._enabled:
+            return None
+
+        step_totals: dict[str, float] = {}
+        series_step_totals: dict[str, float] = {}
+        chapter_count = 0
+        for s in self._series:
+            for step, secs in s["steps"].items():
+                series_step_totals[step] = round(
+                    series_step_totals.get(step, 0.0) + secs, 4)
+            for ch in s["chapters"]:
+                chapter_count += 1
+                for step, secs in ch["steps"].items():
+                    step_totals[step] = round(
+                        step_totals.get(step, 0.0) + secs, 4)
+
+        run = {
+            "startedAt":         round(self._started),
+            "finishedAt":        round(time.time()),
+            "totalSeconds":      round(time.monotonic() - self._t0, 4),
+            "seriesCount":       len(self._series),
+            "chapterCount":      chapter_count,
+            "stepTotals":        step_totals,
+            "seriesStepTotals":  series_step_totals,
+            "series":            self._series,
+        }
+        self._stats._append_run(run)
+        return run
+
+
+class PerfStats:
+    """
+    Profiler facade + JSON persistence.
+
+    Parameters
+    ----------
+    path : Destination JSON file.  None disables the profiler entirely
+           (every recorder becomes a no-op and nothing is written).
+    """
+
+    def __init__(self, path=None):
+        self._path = Path(path) if path else None
+        self._lock = threading.Lock()
+
+    @property
+    def enabled(self) -> bool:
+        return self._path is not None
+
+    def begin_run(self) -> RunRecorder:
+        return RunRecorder(self, enabled=self.enabled)
+
+    # ------------------------------------------------------------------
+    # Read / write
+    # ------------------------------------------------------------------
+    def all(self) -> dict:
+        """Returns the persisted runs ({"runs": [...]}); newest first."""
+        if not self._path or not self._path.is_file():
+            return {"runs": []}
+        try:
+            with self._path.open("r", encoding="utf-8") as f:
+                data = json.load(f)
+        except (OSError, json.JSONDecodeError):
+            return {"runs": []}
+        if not isinstance(data, dict) or not isinstance(data.get("runs"), list):
+            return {"runs": []}
+        return data
+
+    def _append_run(self, run: dict) -> None:
+        if not self._path:
+            return
+        with self._lock:
+            data = self.all()
+            runs = data["runs"]
+            runs.insert(0, run)             # newest first
+            del runs[_MAX_RUNS:]            # cap history
+            self._path.parent.mkdir(parents=True, exist_ok=True)
+            tmp = self._path.with_suffix(self._path.suffix + ".tmp")
+            with tmp.open("w", encoding="utf-8") as f:
+                json.dump({"runs": runs}, f, ensure_ascii=False, indent=2)
+            tmp.replace(self._path)