diff --git a/src/current_status/coordinator.py b/src/current_status/coordinator.py
new file mode 100644
index 0000000..9d0f841
--- /dev/null
+++ b/src/current_status/coordinator.py
@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import asyncio
+import copy
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+from uuid import UUID, uuid4
+
+
+class TargetNotFound(KeyError):
+    pass
+
+
+@dataclass
+class _TargetState:
+    incarnation: UUID
+    latest_started: int = 0
+
+
+@dataclass(frozen=True)
+class _Ticket:
+    target_id: str
+    incarnation: UUID
+    sequence: int
+
+
+class CurrentStatusCoordinator:
+    """Owns target incarnations and at most one completed status per target."""
+
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._targets: dict[str, _TargetState] = {}
+        self._statuses: dict[str, dict[str, Any]] = {}
+
+    async def register_target(self, target_id: str) -> None:
+        """Hook for the prior target-create path."""
+        async with self._lock:
+            if target_id in self._targets:
+                raise ValueError(f"target already exists: {target_id}")
+            self._targets[target_id] = _TargetState(incarnation=uuid4())
+            self._statuses.pop(target_id, None)
+
+    async def delete_target(self, target_id: str) -> bool:
+        """Hook for target deletion; status removal is in the same critical section."""
+        async with self._lock:
+            existed = self._targets.pop(target_id, None) is not None
+            self._statuses.pop(target_id, None)
+            return existed
+
+    async def _begin_check(self, target_id: str) -> _Ticket:
+        async with self._lock:
+            target = self._targets.get(target_id)
+            if target is None:
+                raise TargetNotFound(target_id)
+            target.latest_started += 1
+            return _Ticket(target_id, target.incarnation, target.latest_started)
+
+    async def _publish(self, ticket: _Ticket, result: Mapping[str, Any]) -> bool:
+        async with self._lock:
+            target = self._targets.get(ticket.target_id)
+            if target is None or target.incarnation != ticket.incarnation:
+                return False
+            if target.latest_started != ticket.sequence:
+                return False
+            self._statuses[ticket.target_id] = {
+                "payload": copy.deepcopy(dict(result)),
+                "checked_at": datetime.now(timezone.utc).isoformat(),
+                "check_sequence": ticket.sequence,
+            }
+            return True
+
+    async def run_check(
+        self,
+        target_id: str,
+        target: Mapping[str, Any],
+        checker: Callable[[Mapping[str, Any]], Awaitable[Mapping[str, Any]]],
+    ) -> bool:
+        """Run the prior checker and publish only if its ticket is still current."""
+        ticket = await self._begin_check(target_id)
+        result = await checker(target)
+        return await self._publish(ticket, result)
+
+    async def read(self, target_id: str) -> tuple[bool, dict[str, Any] | None]:
+        async with self._lock:
+            if target_id not in self._targets:
+                return False, None
+            status = self._statuses.get(target_id)
+            return True, copy.deepcopy(status)
+
+    async def retained_status_count(self) -> int:
+        async with self._lock:
+            return len(self._statuses)