diff --git a/kopf/reactor/handling.py b/kopf/reactor/handling.py
index ef466050..2c5fb860 100644
--- a/kopf/reactor/handling.py
+++ b/kopf/reactor/handling.py
@@ -153,7 +153,7 @@ async def execute(
     cause_handlers = subregistry.get_handlers(cause=cause)
     storage = settings.persistence.progress_storage
     state = states.State.from_storage(body=cause.body, storage=storage, handlers=owned_handlers)
-    state = state.with_handlers(cause_handlers)
+    state = state.with_purpose(cause.reason).with_handlers(cause_handlers)
     outcomes = await execute_handlers_once(
         lifecycle=lifecycle,
         settings=settings,
diff --git a/kopf/reactor/processing.py b/kopf/reactor/processing.py
index b5741d24..9d9adebe 100644
--- a/kopf/reactor/processing.py
+++ b/kopf/reactor/processing.py
@@ -337,16 +337,38 @@ async def process_resource_changing_cause(
     # Regular causes invoke the handlers.
     if cause.reason in handlers_.HANDLER_REASONS:
         title = handlers_.TITLES.get(cause.reason, repr(cause.reason))
-        logger.debug(f"{title.capitalize()} event: %r", body)
-        if cause.diff and cause.old is not None and cause.new is not None:
-            logger.debug(f"{title.capitalize()} diff: %r", cause.diff)
 
         resource_registry = registry.resource_changing_handlers[cause.resource]
-        cause_handlers = resource_registry.get_handlers(cause=cause)
         owned_handlers = resource_registry.get_all_handlers()
+        cause_handlers = resource_registry.get_handlers(cause=cause)
         storage = settings.persistence.progress_storage
         state = states.State.from_storage(body=cause.body, storage=storage, handlers=owned_handlers)
-        state = state.with_handlers(cause_handlers)
+        state = state.with_purpose(cause.reason).with_handlers(cause_handlers)
+
+        # Report the causes that have been superseded (intercepted, overridden) by the current one.
+        # The mix-in causes (i.e. resuming) is re-purposed if its handlers are still selected.
+        # To the next cycle, all extras are purged or re-purposed, so the message does not repeat.
+        for extra_reason, counters in state.extras.items():  # usually 0..1 items, rarely 2+.
+            extra_title = handlers_.TITLES.get(extra_reason, repr(extra_reason))
+            logger.info(f"{extra_title.capitalize()} is superseded by {title.lower()}: "
+                        f"{counters.success} succeeded; "
+                        f"{counters.failure} failed; "
+                        f"{counters.running} left to the moment.")
+            state = state.with_purpose(purpose=cause.reason, handlers=cause_handlers)
+
+        # Purge the now-irrelevant handlers if they were not re-purposed (extras are recalculated!).
+        # The current cause continues afterwards, and overrides its own pre-purged handler states.
+        # TODO: purge only the handlers that fell out of current purpose; but it is not critical
+        if state.extras:
+            state.purge(body=cause.body, patch=cause.patch,
+                        storage=storage, handlers=owned_handlers)
+
+        # Inform on the current cause/event on every processing cycle. Even if there are
+        # no handlers -- to show what has happened and why the diff-base is patched.
+        logger.debug(f"{title.capitalize()} event: %r", body)
+        if cause.diff and cause.old is not None and cause.new is not None:
+            logger.debug(f"{title.capitalize()} diff: %r", cause.diff)
+
         if cause_handlers:
             outcomes = await handling.execute_handlers_once(
                 lifecycle=lifecycle,
@@ -360,10 +382,10 @@ async def process_resource_changing_cause(
             states.deliver_results(outcomes=outcomes, patch=cause.patch)
 
             if state.done:
-                success_count, failure_count = state.counts
+                counters = state.counts  # calculate only once
                 logger.info(f"{title.capitalize()} event is processed: "
-                            f"{success_count} succeeded; "
-                            f"{failure_count} failed.")
+                            f"{counters.success} succeeded; "
+                            f"{counters.failure} failed.")
                 state.purge(body=cause.body, patch=cause.patch,
                             storage=storage, handlers=owned_handlers)
 
diff --git a/kopf/storage/progress.py b/kopf/storage/progress.py
index 68e9f402..295d5d5d 100644
--- a/kopf/storage/progress.py
+++ b/kopf/storage/progress.py
@@ -54,6 +54,7 @@ class ProgressRecord(TypedDict, total=True):
     started: Optional[str]
     stopped: Optional[str]
     delayed: Optional[str]
+    purpose: Optional[str]
     retries: Optional[int]
     success: Optional[bool]
     failure: Optional[bool]
diff --git a/kopf/storage/states.py b/kopf/storage/states.py
index 2392aaef..446f925c 100644
--- a/kopf/storage/states.py
+++ b/kopf/storage/states.py
@@ -15,7 +15,8 @@
 import copy
 import dataclasses
 import datetime
-from typing import Any, Collection, Dict, Iterable, Iterator, Mapping, Optional, Tuple, overload
+from typing import Any, Collection, Dict, Iterable, Iterator, \
+                   Mapping, NamedTuple, Optional, overload
 
 from kopf.storage import progress
 from kopf.structs import bodies, callbacks, handlers as handlers_, patches
@@ -52,6 +53,7 @@ class HandlerState:
     started: Optional[datetime.datetime] = None  # None means this information was lost.
     stopped: Optional[datetime.datetime] = None  # None means it is still running (e.g. delayed).
     delayed: Optional[datetime.datetime] = None  # None means it is finished (succeeded/failed).
+    purpose: Optional[handlers_.Reason] = None  # None is a catch-all marker for upgrades/rollbacks.
     retries: int = 0
     success: bool = False
     failure: bool = False
@@ -60,9 +62,10 @@ class HandlerState:
     _origin: Optional[progress.ProgressRecord] = None  # to check later if it has actually changed.
 
     @classmethod
-    def from_scratch(cls) -> "HandlerState":
+    def from_scratch(cls, *, purpose: Optional[handlers_.Reason] = None) -> "HandlerState":
         return cls(
             started=datetime.datetime.utcnow(),
+            purpose=purpose,
         )
 
     @classmethod
@@ -71,6 +74,7 @@ def from_storage(cls, __d: progress.ProgressRecord) -> "HandlerState":
             started=_datetime_fromisoformat(__d.get('started')) or datetime.datetime.utcnow(),
             stopped=_datetime_fromisoformat(__d.get('stopped')),
             delayed=_datetime_fromisoformat(__d.get('delayed')),
+            purpose=handlers_.Reason(__d.get('purpose')) if __d.get('purpose') else None,
             retries=__d.get('retries') or 0,
             success=__d.get('success') or False,
             failure=__d.get('failure') or False,
@@ -84,6 +88,7 @@ def for_storage(self) -> progress.ProgressRecord:
             started=None if self.started is None else _datetime_toisoformat(self.started),
             stopped=None if self.stopped is None else _datetime_toisoformat(self.stopped),
             delayed=None if self.delayed is None else _datetime_toisoformat(self.delayed),
+            purpose=None if self.purpose is None else str(self.purpose.value),
             retries=None if self.retries is None else int(self.retries),
             success=None if self.success is None else bool(self.success),
             failure=None if self.failure is None else bool(self.failure),
@@ -95,6 +100,12 @@ def as_in_storage(self) -> Mapping[str, Any]:
         # Nones are not stored by Kubernetes, so we filter them out for comparison.
         return {key: val for key, val in self.for_storage().items() if val is not None}
 
+    def with_purpose(
+            self,
+            purpose: Optional[handlers_.Reason],
+    ) -> "HandlerState":
+        return dataclasses.replace(self, purpose=purpose)
+
     def with_outcome(
             self,
             outcome: HandlerOutcome,
@@ -102,6 +113,7 @@ def with_outcome(
         now = datetime.datetime.utcnow()
         cls = type(self)
         return cls(
+            purpose=self.purpose,
             started=self.started if self.started else now,
             stopped=self.stopped if self.stopped else now if outcome.final else None,
             delayed=now + datetime.timedelta(seconds=outcome.delay) if outcome.delay is not None else None,
@@ -133,6 +145,12 @@ def runtime(self) -> datetime.timedelta:
         return now - (self.started if self.started else now)
 
 
+class StateCounters(NamedTuple):
+    success: int
+    failure: int
+    running: int
+
+
 class State(Mapping[handlers_.HandlerId, HandlerState]):
     """
     A state of selected handlers, as persisted in the object's status.
@@ -149,9 +167,12 @@ class State(Mapping[handlers_.HandlerId, HandlerState]):
     def __init__(
             self,
             __src: Mapping[handlers_.HandlerId, HandlerState],
+            *,
+            purpose: Optional[handlers_.Reason] = None,
     ):
         super().__init__()
         self._states = dict(__src)
+        self.purpose = purpose
 
     @classmethod
     def from_scratch(cls) -> "State":
@@ -173,17 +194,27 @@ def from_storage(
                 handler_states[handler_id] = HandlerState.from_storage(content)
         return cls(handler_states)
 
+    def with_purpose(
+            self,
+            purpose: Optional[handlers_.Reason],
+            handlers: Iterable[handlers_.BaseHandler] = (),  # to be re-purposed
+    ) -> "State":
+        handler_states: Dict[handlers_.HandlerId, HandlerState] = dict(self)
+        for handler in handlers:
+            handler_states[handler.id] = handler_states[handler.id].with_purpose(purpose)
+        cls = type(self)
+        return cls(handler_states, purpose=purpose)
+
     def with_handlers(
             self,
             handlers: Iterable[handlers_.BaseHandler],
     ) -> "State":
-        handler_ids = {handler.id for handler in handlers}
         handler_states: Dict[handlers_.HandlerId, HandlerState] = dict(self)
-        for handler_id in handler_ids:
-            if handler_id not in handler_states:
-                handler_states[handler_id] = HandlerState.from_scratch()
+        for handler in handlers:
+            if handler.id not in handler_states:
+                handler_states[handler.id] = HandlerState.from_scratch(purpose=self.purpose)
         cls = type(self)
-        return cls(handler_states)
+        return cls(handler_states, purpose=self.purpose)
 
     def with_outcomes(
             self,
@@ -198,7 +229,7 @@ def with_outcomes(
             handler_id: (handler_state if handler_id not in outcomes else
                          handler_state.with_outcome(outcomes[handler_id]))
             for handler_id, handler_state in self.items()
-        })
+        }, purpose=self.purpose)
 
     def store(
             self,
@@ -245,13 +276,39 @@ def __getitem__(self, item: handlers_.HandlerId) -> HandlerState:
     @property
     def done(self) -> bool:
         # In particular, no handlers means that it is "done" even before doing.
-        return all(handler_state.finished for handler_state in self._states.values())
+        return all(
+            handler_state.finished for handler_state in self._states.values()
+            if self.purpose is None or handler_state.purpose is None
+               or handler_state.purpose == self.purpose
+        )
+
+    @property
+    def extras(self) -> Mapping[handlers_.Reason, StateCounters]:
+        return {
+            reason: StateCounters(
+                success=len([1 for handler_state in self._states.values()
+                            if handler_state.purpose == reason and handler_state.success]),
+                failure=len([1 for handler_state in self._states.values()
+                            if handler_state.purpose == reason and handler_state.failure]),
+                running=len([1 for handler_state in self._states.values()
+                            if handler_state.purpose == reason and not handler_state.finished]),
+            )
+            for reason in handlers_.HANDLER_REASONS
+            if self.purpose is not None and reason != self.purpose
+            if any(handler_state.purpose == reason for handler_state in self._states.values())
+        }
 
     @property
-    def counts(self) -> Tuple[int, int]:
-        return (
-            len([1 for handler_state in self._states.values() if handler_state.success]),
-            len([1 for handler_state in self._states.values() if handler_state.failure]),
+    def counts(self) -> StateCounters:
+        purposeful_states = [
+            handler_state for handler_state in self._states.values()
+            if self.purpose is None or handler_state.purpose is None
+               or handler_state.purpose == self.purpose
+        ]
+        return StateCounters(
+            success=len([1 for handler_state in purposeful_states if handler_state.success]),
+            failure=len([1 for handler_state in purposeful_states if handler_state.failure]),
+            running=len([1 for handler_state in purposeful_states if not handler_state.finished]),
         )
 
     @property
@@ -273,6 +330,8 @@ def delays(self) -> Collection[float]:
             max(0, (handler_state.delayed - now).total_seconds()) if handler_state.delayed else 0
             for handler_state in self._states.values()
             if not handler_state.finished
+            if self.purpose is None or handler_state.purpose is None
+               or handler_state.purpose == self.purpose
         ]
 
 
diff --git a/tests/handling/test_cause_logging.py b/tests/handling/test_cause_logging.py
index 91a617cb..c040b0b5 100644
--- a/tests/handling/test_cause_logging.py
+++ b/tests/handling/test_cause_logging.py
@@ -1,10 +1,13 @@
 import asyncio
+import datetime
 import logging
 
+import freezegun
 import pytest
 
 import kopf
 from kopf.reactor.processing import process_resource_event
+from kopf.storage.progress import StatusProgressStorage
 from kopf.structs.containers import ResourceMemories
 from kopf.structs.handlers import ALL_REASONS, HANDLER_REASONS, Reason
 
@@ -91,3 +94,46 @@ async def test_diffs_not_logged_if_absent(registry, settings, resource, handlers
     ], prohibited=[
         " diff: "
     ])
+
+
+
+# Timestamps: time zero (0), before (B), after (A), and time zero+1s (1).
+TS0 = datetime.datetime(2020, 12, 31, 23, 59, 59, 123456)
+TS1_ISO = '2021-01-01T00:00:00.123456'
+
+
+@pytest.mark.parametrize('cause_types', [
+    # All combinations except for same-to-same (it is not an "extra" then).
+    (a, b) for a in HANDLER_REASONS for b in HANDLER_REASONS if a != b
+])
+@freezegun.freeze_time(TS0)
+async def test_supersession_is_logged(
+        registry, settings, resource, handlers, cause_types, cause_mock, caplog, assert_logs):
+    caplog.set_level(logging.DEBUG)
+
+    settings.persistence.progress_storage = StatusProgressStorage()
+    body = {'status': {'kopf': {'progress': {
+        'create_fn': {'purpose': cause_types[0]},
+        'update_fn': {'purpose': cause_types[0]},
+        'resume_fn': {'purpose': cause_types[0]},
+        'delete_fn': {'purpose': cause_types[0]},
+    }}}}
+
+    cause_mock.reason = cause_types[1]
+    event_type = None if cause_types[1] == Reason.RESUME else 'irrelevant'
+
+    await process_resource_event(
+        lifecycle=kopf.lifecycles.all_at_once,
+        registry=registry,
+        settings=settings,
+        resource=resource,
+        memories=ResourceMemories(),
+        raw_event={'type': event_type, 'object': body},
+        replenished=asyncio.Event(),
+        event_queue=asyncio.Queue(),
+    )
+    assert_logs([
+        "(Creation|Updating|Resuming|Deletion) is superseded by (creation|updating|resuming|deletion): ",
+        "(Creation|Updating|Resuming|Deletion) is in progress: ",
+        "(Creation|Updating|Resuming|Deletion) is processed: ",
+    ])
diff --git a/tests/persistence/test_states.py b/tests/persistence/test_states.py
index 74e140b2..91d8524c 100644
--- a/tests/persistence/test_states.py
+++ b/tests/persistence/test_states.py
@@ -5,8 +5,9 @@
 import pytest
 
 from kopf.storage.progress import SmartProgressStorage, StatusProgressStorage
-from kopf.storage.states import HandlerOutcome, State, deliver_results
+from kopf.storage.states import HandlerOutcome, State, StateCounters, deliver_results
 from kopf.structs.bodies import Body
+from kopf.structs.handlers import HANDLER_REASONS, Reason
 from kopf.structs.patches import Patch
 
 # Timestamps: time zero (0), before (B), after (A), and time zero+1s (1).
@@ -33,22 +34,242 @@ def handler():
     return Mock(id='some-id', spec_set=['id'])
 
 
-@freezegun.freeze_time(TS0)
-def test_created_empty_from_scratch(storage, handler):
+#
+# State creation from scratch and from storage; basic properties checks:
+#
+
+
+def test_created_empty_from_scratch():
     state = State.from_scratch()
     assert len(state) == 0
+    assert state.purpose is None
+    assert state.done == True
+    assert state.delay is None
+    assert state.delays == []
+    assert state.counts == StateCounters(success=0, failure=0, running=0)
+    assert state.extras == {}
+
+
+@pytest.mark.parametrize('body', [
+    ({}),
+    ({'status': {}}),
+    ({'status': {'kopf': {}}}),
+    ({'status': {'kopf': {'progress': {}}}}),
+])
+def test_created_empty_from_empty_storage_with_handlers(storage, handler, body):
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 0
+    assert state.purpose is None
+    assert state.done == True
+    assert state.delay is None
+    assert state.delays == []
+    assert state.counts == StateCounters(success=0, failure=0, running=0)
+    assert state.extras == {}
+
+
+@pytest.mark.parametrize('body', [
+    ({'status': {'kopf': {'progress': {'some-id': {}}}}}),
+    ({'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
+    ({'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
+])
+def test_created_empty_from_filled_storage_without_handlers(storage, handler, body):
+    state = State.from_storage(body=Body(body), handlers=[], storage=storage)
+    assert len(state) == 0
+    assert state.purpose is None
+    assert state.done == True
+    assert state.delay is None
+    assert state.delays == []
+    assert state.counts == StateCounters(success=0, failure=0, running=0)
+    assert state.extras == {}
 
 
-@pytest.mark.parametrize('expected, body', [
-    (TS0_ISO, {}),
-    (TS0_ISO, {'status': {}}),
-    (TS0_ISO, {'status': {'kopf': {}}}),
-    (TS0_ISO, {'status': {'kopf': {'progress': {}}}}),
+#
+# Purpose propagation and re-purposing of the states (overall and per-handler):
+#
+
+
+def test_created_from_purposeless_storage(storage, handler):
+    body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 1
+    assert state.purpose is None
+    assert state['some-id'].purpose is None
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_created_from_purposeful_storage(storage, handler, reason):
+    body = {'status': {'kopf': {'progress': {'some-id': {'purpose': reason.value}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    assert len(state) == 1
+    assert state.purpose is None
+    assert state['some-id'].purpose is reason
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_enriched_with_handlers_keeps_the_original_purpose(handler, reason):
+    state = State.from_scratch()
+    state = state.with_purpose(reason)
+    state = state.with_handlers([handler])
+    assert state.purpose is reason
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_enriched_with_outcomes_keeps_the_original_purpose(reason):
+    state = State.from_scratch()
+    state = state.with_purpose(reason)
+    state = state.with_outcomes({})
+    assert state.purpose is reason
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_repurposed_before_handlers(handler, reason):
+    state = State.from_scratch()
+    state = state.with_purpose(reason).with_handlers([handler])
+    assert len(state) == 1
+    assert state.purpose is reason
+    assert state['some-id'].purpose is reason
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_repurposed_after_handlers(handler, reason):
+    state = State.from_scratch()
+    state = state.with_handlers([handler]).with_purpose(reason)
+    assert len(state) == 1
+    assert state.purpose is reason
+    assert state['some-id'].purpose is None
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_repurposed_with_handlers(handler, reason):
+    state = State.from_scratch()
+    state = state.with_handlers([handler]).with_purpose(reason, handlers=[handler])
+    assert len(state) == 1
+    assert state.purpose is reason
+    assert state['some-id'].purpose is reason
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_repurposed_not_affecting_the_existing_handlers_from_scratch(handler, reason):
+    state = State.from_scratch()
+    state = state.with_handlers([handler]).with_purpose(reason).with_handlers([handler])
+    assert len(state) == 1
+    assert state.purpose is reason
+    assert state['some-id'].purpose is None
+
+
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_repurposed_not_affecting_the_existing_handlers_from_storage(storage, handler, reason):
+    body = {'status': {'kopf': {'progress': {'some-id': {'purpose': None}}}}}
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_handlers([handler]).with_purpose(reason).with_handlers([handler])
+    assert len(state) == 1
+    assert state.purpose is reason
+    assert state['some-id'].purpose is None
+
+
+#
+# Counts & extras calculation with different combinations of purposes:
+#
+
+
+@pytest.mark.parametrize('expected_extras, body', [
+    # (success, failure, running)
+    (StateCounters(0, 0, 1), {'status': {'kopf': {'progress': {'some-id': {}}}}}),
+    (StateCounters(0, 1, 0), {'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
+    (StateCounters(1, 0, 0), {'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
 ])
+@pytest.mark.parametrize('stored_reason, processed_reason', [
+    # All combinations except for same-to-same (it is not an "extra" then).
+    (a, b) for a in HANDLER_REASONS for b in HANDLER_REASONS if a != b
+])
+def test_with_handlers_irrelevant_to_the_purpose(
+        storage, handler, body, expected_extras, stored_reason, processed_reason):
+    body['status']['kopf']['progress']['some-id']['purpose'] = stored_reason.value
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_purpose(processed_reason)
+    assert len(state) == 1
+    assert state.extras[stored_reason] == expected_extras
+    assert state.counts == StateCounters(success=0, failure=0, running=0)
+    assert state.done == True
+    assert state.delays == []
+
+
+@pytest.mark.parametrize('expected_counts, expected_done, expected_delays, body', [
+    # (success, failure)
+    (StateCounters(0, 0, 1), False, [0.0], {'status': {'kopf': {'progress': {'some-id': {}}}}}),
+    (StateCounters(0, 1, 0), True, [], {'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
+    (StateCounters(1, 0, 0), True, [], {'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
+])
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
+def test_with_handlers_relevant_to_the_purpose(
+        storage, handler, body, expected_counts, expected_done, expected_delays, reason):
+    body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
+    state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
+    state = state.with_purpose(reason)
+    assert len(state) == 1
+    assert state.extras == {}
+    assert state.counts == expected_counts
+    assert state.done == expected_done
+    assert state.delays == expected_delays
+
+
+@pytest.mark.parametrize('expected_counts, expected_done, expected_delays, body', [
+    (StateCounters(0, 0, 1), False, [1.0], {'status': {'kopf': {'progress': {'some-id': {}}}}}),
+    (StateCounters(0, 1, 0), True, [], {'status': {'kopf': {'progress': {'some-id': {'failure': True}}}}}),
+    (StateCounters(1, 0, 0), True, [], {'status': {'kopf': {'progress': {'some-id': {'success': True}}}}}),
+])
+@pytest.mark.parametrize('reason', HANDLER_REASONS)
 @freezegun.freeze_time(TS0)
-def test_created_empty_from_empty_storage(storage, handler, body, expected):
+def test_with_handlers_relevant_to_the_purpose_and_delayed(
+        storage, handler, body, expected_counts, expected_done, expected_delays, reason):
+    body['status']['kopf']['progress']['some-id']['delayed'] = TS1_ISO
+    body['status']['kopf']['progress']['some-id']['purpose'] = reason.value
     state = State.from_storage(body=Body(body), handlers=[handler], storage=storage)
-    assert len(state) == 0
+    state = state.with_purpose(reason)
+    assert len(state) == 1
+    assert state.extras == {}
+    assert state.counts == expected_counts
+    assert state.done == expected_done
+    assert state.delays == expected_delays
+
+
+@pytest.mark.parametrize('reason', [Reason.CREATE, Reason.UPDATE, Reason.RESUME])
+@freezegun.freeze_time(TS0)
+def test_issue_601_deletion_supersedes_other_processing(storage, reason):
+
+    body = {'status': {'kopf': {'progress': {
+        'fn1': {'purpose': reason.value, 'failure': True},
+        'fn2': {'purpose': reason.value, 'success': True},
+        'fn3': {'purpose': reason.value, 'delayed': TS1_ISO},
+    }}}}
+    create_handler1 = Mock(id='fn1', spec_set=['id'])
+    create_handler2 = Mock(id='fn2', spec_set=['id'])
+    create_handler3 = Mock(id='fn3', spec_set=['id'])
+    delete_handler9 = Mock(id='delete_fn', spec_set=['id'])
+    owned_handlers = [create_handler1, create_handler2, create_handler3, delete_handler9]
+    cause_handlers = [delete_handler9]
+
+    state = State.from_storage(body=Body(body), handlers=owned_handlers, storage=storage)
+    state = state.with_purpose(Reason.DELETE)
+    state = state.with_handlers(cause_handlers)
+
+    assert len(state) == 4
+    assert state.extras == {reason: StateCounters(success=1, failure=1, running=1)}
+    assert state.counts == StateCounters(success=0, failure=0, running=1)
+    assert state.done == False
+    assert state.delays == [0.0]
+
+    state = state.with_outcomes({'delete_fn': HandlerOutcome(final=True)})
+
+    assert state.extras == {reason: StateCounters(success=1, failure=1, running=1)}
+    assert state.counts == StateCounters(success=1, failure=0, running=0)
+    assert state.done == True
+    assert state.delays == []
+
+
+#
+# Handlers' time-based states: starting, running, retrying, finishing, etc.
+#
 
 
 @freezegun.freeze_time(TS0)
@@ -271,6 +492,11 @@ def test_set_retry_time(storage, handler, expected_retries, expected_delayed, bo
     assert patch['status']['kopf']['progress']['some-id']['delayed'] == expected_delayed
 
 
+#
+# Sub-handlers ids persistence for later purging of them.
+#
+
+
 def test_subrefs_added_to_empty_state(storage, handler):
     body = {}
     patch = Patch()
@@ -308,6 +534,11 @@ def test_subrefs_ignored_when_not_specified(storage, handler):
     assert patch['status']['kopf']['progress']['some-id']['subrefs'] is None
 
 
+#
+# Persisting outcomes: successes, failures, results, exceptions, etc.
+#
+
+
 @pytest.mark.parametrize('expected_retries, expected_stopped, body', [
     (1, TS0_ISO, {}),
     (6, TS0_ISO, {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}),
@@ -357,6 +588,11 @@ def test_store_result(handler, expected_patch, result):
     assert patch == expected_patch
 
 
+#
+# Purging the state in the storage.
+#
+
+
 def test_purge_progress_when_exists_in_body(storage, handler):
     body = {'status': {'kopf': {'progress': {'some-id': {'retries': 5}}}}}
     patch = Patch()