@@ -17,6 +17,10 @@ We drop support for Python 3.10, which reached its end of life in October 2026.
 
 `StatisticsState.crawler_runtime` is computed from the crawler's start and finish timestamps. In v1, assigning to it only emitted a `DeprecationWarning` and the value was ignored. In v2, the property has no setter, so an assignment raises `AttributeError`. Remove any such assignments.
 
+## `StatisticsState.crawler_runtime_for_serialization` is removed
+
+`StatisticsState.crawler_runtime_for_serialization` is removed, and the persisted `crawlerRuntimeMillis` value is a number of milliseconds instead of an ISO 8601 duration string. Read the runtime from the `crawler_runtime` property instead.
+
 ## `RequestQueueClient.is_finished` is abstract
 
 `RequestQueueClient.is_finished` is now an abstract method. In v1, it fell back to `is_empty`, which ignores requests still being processed and could report the queue as finished too early. A subclass that doesn't implement it raises `TypeError` on instantiation. If you have a custom request queue client, implement `is_finished` so that it returns `True` only when the queue is empty and no requests are in progress. The built-in clients already implement it.
 
@@ -5,7 +5,15 @@
 from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Annotated, Any
 
-from pydantic import BaseModel, ConfigDict, Field, PlainSerializer, PlainValidator, computed_field
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    PlainSerializer,
+    PlainValidator,
+    computed_field,
+    field_serializer,
+)
 from pydantic.alias_generators import to_camel
 from typing_extensions import override
 
@@ -100,25 +108,31 @@ class StatisticsState(BaseModel):
         ),
     ] = {}
 
-    # Used to track the crawler runtime, that had already been persisted. This is the runtime from previous runs.
-    _runtime_offset: Annotated[timedelta, Field(exclude=True)] = timedelta()
+    accumulated_runtime: Annotated[timedelta_ms, Field(alias='crawlerRuntimeMillis')] = timedelta()
+    """The runtime accumulated by the previous runs of the crawler, excluding the run in progress.
 
-    def model_post_init(self, /, __context: Any) -> None:
-        self._runtime_offset = self.crawler_runtime or self._runtime_offset
+    When serialized, the runtime of the run in progress is included, so that a persisted state always carries
+    the total runtime accumulated so far. Do not assign to it: it is restored from the persisted state, and
+    serialization derives the live total from the timestamps.
+    """
 
     @property
     def crawler_runtime(self) -> timedelta:
-        if self.crawler_last_started_at:
-            finished_at = self.crawler_finished_at or datetime.now(UTC)
-            return self._runtime_offset + finished_at - self.crawler_last_started_at
-        return self._runtime_offset
-
-    @computed_field(alias='crawlerRuntimeMillis')
-    def crawler_runtime_for_serialization(self) -> timedelta:
-        if self.crawler_last_started_at:
-            finished_at = self.crawler_finished_at or datetime.now(UTC)
-            return self._runtime_offset + finished_at - self.crawler_last_started_at
-        return self._runtime_offset
+        """The total runtime of the crawler, accumulated over all its runs, excluding the downtime between them.
+
+        The run in progress is measured by the `crawler_last_started_at` and `crawler_finished_at` timestamps.
+        In a loaded state, the timestamps describe the previous run, whose runtime `accumulated_runtime` already
+        contains, so they must be reset before a new run is measured. `Statistics` resets them on enter, before
+        the runtime is first read.
+        """
+        if self.crawler_last_started_at is None:
+            return self.accumulated_runtime
+        finished_at = self.crawler_finished_at or datetime.now(UTC)
+        return self.accumulated_runtime + max(timedelta(), finished_at - self.crawler_last_started_at)
+
+    @field_serializer('accumulated_runtime')
+    def _serialize_accumulated_runtime(self, _value: timedelta) -> float:
+        return round(self.crawler_runtime.total_seconds() * 1000)
 
     @computed_field(alias='requestTotalDurationMillis', return_type=timedelta_ms)
     @property
 
@@ -169,10 +169,15 @@ async def __aenter__(self) -> Self:
             raise RuntimeError(f'The {self.__class__.__name__} is already active.')
 
         await self._state.initialize()
-        # Reset `crawler_finished_at` to indicate a new run in progress.
+
+        # The restored timestamps belong to the previous run and must not measure this one: left in place,
+        # they would count the downtime between the runs (e.g. after a migration or resurrection) towards
+        # the runtime. Clear them, so that the initial periodic log reports exactly the runtime accumulated
+        # by the previous runs (zero for a fresh start).
+        self.state.crawler_last_started_at = None
         self.state.crawler_finished_at = None
 
-        # Start periodic logging and let it print initial state before activation.
+        # Start periodic logging and let it print the restored state before the new run starts being measured.
         self._periodic_logger.start()
         await asyncio.sleep(0.01)
         self._active = True
@@ -195,10 +200,7 @@ async def __aexit__(
         if not self._active:
             raise RuntimeError(f'The {self.__class__.__name__} is not active.')
 
-        if not self.state.crawler_last_started_at:
-            raise RuntimeError('Statistics.state.crawler_last_started_at not set.')
-
-        # Stop logging and deactivate the statistics to prevent further changes to crawler_runtime
+        # Stop logging; the finish timestamp freezes the measured run before the final persist.
         await self._periodic_logger.stop()
         self.state.crawler_finished_at = datetime.now(UTC)
         self._active = False
 
@@ -1,6 +1,15 @@
 from __future__ import annotations
 
+import asyncio
+import logging
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
 from crawlee.statistics import Statistics
+from crawlee.storages import KeyValueStore
+
+if TYPE_CHECKING:
+    import pytest
 
 
 async def test_basic_persistence() -> None:
@@ -13,3 +22,155 @@ async def test_basic_persistence() -> None:
         pass
 
     assert statistics.state.requests_failed == 42
+
+
+async def test_first_periodic_log_of_fresh_run_reports_zero_runtime(caplog: pytest.LogCaptureFixture) -> None:
+    """The first periodic log of a fresh run reports a runtime of exactly zero."""
+    caplog.set_level(logging.INFO)
+    log_message = 'Fresh statistics'
+
+    async with Statistics.with_default_state(log_message=log_message, statistics_log_format='inline'):
+        pass
+
+    periodic_records = [record for record in caplog.records if record.message == log_message]
+    assert periodic_records
+    assert periodic_records[0].crawler_runtime == 0  # ty: ignore[unresolved-attribute]
+
+
+async def test_periodic_log_after_resume_excludes_downtime(caplog: pytest.LogCaptureFixture) -> None:
+    """The first periodic log of a resumed run reports exactly the previous runtime, without the downtime."""
+    caplog.set_level(logging.INFO)
+    key = 'statistics_downtime_clean'
+    log_message = 'Statistics after resume'
+    downtime = timedelta(hours=2)
+
+    async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key):
+        pass
+
+    # Simulate a resurrection after two hours of downtime by shifting the persisted timestamps into the past.
+    kvs = await KeyValueStore.open()
+    stored_state = await kvs.get_value(key)
+    for field in ('crawlerStartedAt', 'crawlerLastStartTimestamp', 'crawlerFinishedAt', 'statsPersistedAt'):
+        stored_state[field] = (datetime.fromisoformat(stored_state[field]) - downtime).isoformat()
+    await kvs.set_value(key, stored_state)
+
+    caplog.clear()
+    async with Statistics.with_default_state(
+        persistence_enabled=True,
+        persist_state_key=key,
+        log_message=log_message,
+        statistics_log_format='inline',
+    ):
+        pass
+
+    periodic_records = [record for record in caplog.records if record.message == log_message]
+    assert periodic_records
+    first_logged_runtime = timedelta(seconds=periodic_records[0].crawler_runtime)  # ty: ignore[unresolved-attribute]
+    previous_runtime = timedelta(milliseconds=stored_state['crawlerRuntimeMillis'])
+    assert abs(first_logged_runtime - previous_runtime) < timedelta(milliseconds=1)
+
+
+async def test_runtime_accumulates_over_multiple_resurrections() -> None:
+    """The persisted total runtime is restored on resume, so runs before the last one still count."""
+    key = 'statistics_accumulated_runtime'
+    now = datetime.now(UTC)
+    last_run_start = now - timedelta(hours=1)
+    last_run_duration = timedelta(seconds=5)
+    total_runtime = timedelta(seconds=30)
+
+    # State persisted after a clean run whose own segment took 5s, with 30s of runtime accumulated in total.
+    kvs = await KeyValueStore.open()
+    await kvs.set_value(
+        key,
+        {
+            'crawlerStartedAt': (now - timedelta(hours=2)).isoformat(),
+            'crawlerLastStartTimestamp': last_run_start.isoformat(),
+            'crawlerFinishedAt': (last_run_start + last_run_duration).isoformat(),
+            'statsPersistedAt': (last_run_start + last_run_duration).isoformat(),
+            'crawlerRuntimeMillis': total_runtime.total_seconds() * 1000,
+        },
+    )
+
+    async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics:
+        runtime = statistics.state.crawler_runtime
+
+    assert total_runtime <= runtime < total_runtime + timedelta(minutes=1)
+
+
+async def test_runtime_after_unclean_shutdown_excludes_downtime() -> None:
+    """A state checkpointed mid-run restores the runtime up to the last persist, without the downtime that follows."""
+    key = 'statistics_downtime_unclean'
+    now = datetime.now(UTC)
+    downtime = timedelta(hours=2)
+    previous_runtime = timedelta(seconds=10)
+
+    # State persisted by a periodic persist during a run that never finished cleanly (migration or abort):
+    # there is no finish timestamp, only the runtime checkpointed at the last persist.
+    kvs = await KeyValueStore.open()
+    await kvs.set_value(
+        key,
+        {
+            'crawlerStartedAt': (now - downtime - previous_runtime).isoformat(),
+            'crawlerLastStartTimestamp': (now - downtime - previous_runtime).isoformat(),
+            'crawlerFinishedAt': None,
+            'statsPersistedAt': (now - downtime).isoformat(),
+            'crawlerRuntimeMillis': previous_runtime.total_seconds() * 1000,
+        },
+    )
+
+    async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics:
+        runtime = statistics.state.crawler_runtime
+
+    assert previous_runtime <= runtime < previous_runtime + timedelta(minutes=1)
+
+
+async def test_serialized_state_includes_runtime_of_run_in_progress() -> None:
+    """A state serialized during a run carries the runtime up to that moment, so every persist is a checkpoint."""
+    run_in_progress_duration = timedelta(hours=1)
+
+    async with Statistics.with_default_state() as statistics:
+        # Simulate a run that has been in progress for an hour by shifting its start timestamp into the past.
+        assert statistics.state.crawler_last_started_at
+        statistics.state.crawler_last_started_at -= run_in_progress_duration
+        serialized = statistics.state.model_dump(by_alias=True)
+
+    assert serialized['crawlerRuntimeMillis'] >= run_in_progress_duration.total_seconds() * 1000
+
+
+async def test_backwards_clock_jump_does_not_regress_runtime() -> None:
+    """A backwards wall clock jump during a run does not push the runtime below the accumulated total."""
+    async with Statistics.with_default_state() as statistics:
+        # Simulate a backwards clock jump by shifting the start timestamp of the run into the future.
+        assert statistics.state.crawler_last_started_at
+        statistics.state.crawler_last_started_at += timedelta(hours=1)
+        runtime = statistics.state.crawler_runtime
+
+    assert runtime == timedelta()
+
+
+async def test_runtime_stops_growing_after_run_finishes() -> None:
+    """The runtime measured by a finished run stays frozen instead of following the clock."""
+    async with Statistics.with_default_state() as statistics:
+        pass
+
+    runtime_after_exit = statistics.state.crawler_runtime
+    await asyncio.sleep(0.05)
+
+    assert statistics.state.crawler_runtime == runtime_after_exit
+
+
+async def test_runtime_carries_over_between_runs() -> None:
+    """A second run continues accumulating the runtime from the total persisted by the first run."""
+    key = 'statistics_restored_runtime'
+
+    async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key):
+        pass
+
+    kvs = await KeyValueStore.open()
+    stored_state = await kvs.get_value(key)
+    previous_runtime = timedelta(milliseconds=stored_state['crawlerRuntimeMillis'])
+
+    async with Statistics.with_default_state(persistence_enabled=True, persist_state_key=key) as statistics:
+        pass
+
+    assert previous_runtime <= statistics.state.crawler_runtime < previous_runtime + timedelta(minutes=1)