diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md
index 28eab75..6a86dbd 100644
--- a/openspec/changes/cloud-control-plane-integration/tasks.md
+++ b/openspec/changes/cloud-control-plane-integration/tasks.md
@@ -8,7 +8,7 @@
 
 ## 2. Repository Contract And Migrations
 
-- [ ] 2.1 Define the cloud repository contract for hosts, device snapshots, plugins, tasks, attempts, leases, reservations, and transactional assignment operations.
+- [x] 2.1 Define the cloud repository contract for hosts, device snapshots, plugins, tasks, attempts, leases, reservations, and transactional assignment operations.
 - [ ] 2.2 Implement SQLAlchemy models and a repository adapter that preserves existing `CloudStore` observable behavior.
 - [ ] 2.3 Add PostgreSQL and SQLite database URL support with engine/session lifecycle owned by the cloud application.
 - [ ] 2.4 Add Alembic configuration and a baseline migration that preserves existing host, device, task, and plugin data while adding lease/attempt/result fields.
diff --git a/packages/cloud-platform/cloud/repository.py b/packages/cloud-platform/cloud/repository.py
new file mode 100644
index 0000000..f02b1f3
--- /dev/null
+++ b/packages/cloud-platform/cloud/repository.py
@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import TYPE_CHECKING, Literal, Protocol
+
+if TYPE_CHECKING:
+    from cloud.plugins import PluginManifest
+    from cloud.pool import HostRegistration, PooledDevice
+    from cloud.scheduler import ScheduledTask
+
+
+AttemptStatus = Literal["assigned", "dispatched", "done", "failed", "expired"]
+TerminalTaskStatus = Literal["done", "failed"]
+ResultRecordStatus = Literal["recorded", "already_recorded", "conflict"]
+
+
+@dataclass(frozen=True)
+class TaskAttemptRecord:
+    task_id: str
+    attempt: int
+    lease_id: str
+    host_id: str
+    device_id: str
+    status: AttemptStatus
+    lease_expires_at: datetime
+    created_at: datetime
+    completed_at: datetime | None = None
+    failure_reason: str | None = None
+
+
+@dataclass(frozen=True)
+class LeasedAssignment:
+    task_id: str
+    attempt: int
+    lease_id: str
+    lease_expires_at: datetime
+    host_id: str
+    device_id: str
+    goal: str | None
+    workflow_definition_id: str | None
+
+
+class CloudRepository(Protocol):
+    """Persistence port for cloud state and atomic scheduling operations."""
+
+    def upsert_host(
+        self,
+        host_id: str,
+        *,
+        address: str | None,
+        last_seen_at: datetime,
+    ) -> None: ...
+
+    def replace_host_devices(
+        self,
+        host_id: str,
+        devices: list[PooledDevice],
+    ) -> None: ...
+
+    def list_hosts(self) -> list[HostRegistration]: ...
+
+    def get_host(self, host_id: str) -> HostRegistration | None: ...
+
+    def list_devices(self) -> list[PooledDevice]: ...
+
+    def get_device(self, device_id: str) -> PooledDevice | None: ...
+
+    def enqueue_task(self, task: ScheduledTask) -> None: ...
+
+    def list_queued_tasks(self) -> list[ScheduledTask]: ...
+
+    def get_task(self, task_id: str) -> ScheduledTask | None: ...
+
+    def update_task(
+        self,
+        task_id: str,
+        *,
+        status: str | None = None,
+        assigned_device_id: str | None = None,
+        assigned_host_id: str | None = None,
+    ) -> None: ...
+
+    def count_queued_tasks(self) -> int: ...
+
+    def save_plugin(self, manifest: PluginManifest, *, wired: bool) -> None: ...
+
+    def list_plugins(self) -> list[tuple[PluginManifest, bool]]: ...
+
+    def get_plugin(self, name: str) -> tuple[PluginManifest, bool] | None: ...
+
+    def list_reserved_device_ids(self, *, now: datetime) -> set[str]: ...
+
+    def assign_task(
+        self,
+        *,
+        task_id: str,
+        host_id: str,
+        device_id: str,
+        lease_id: str,
+        lease_expires_at: datetime,
+        now: datetime,
+    ) -> LeasedAssignment | None: ...
+
+    def claim_assignment(
+        self,
+        *,
+        host_id: str,
+        now: datetime,
+    ) -> LeasedAssignment | None: ...
+
+    def renew_lease(
+        self,
+        *,
+        task_id: str,
+        attempt: int,
+        lease_id: str,
+        host_id: str,
+        lease_expires_at: datetime,
+        now: datetime,
+    ) -> bool: ...
+
+    def record_task_result(
+        self,
+        *,
+        task_id: str,
+        attempt: int,
+        lease_id: str,
+        host_id: str,
+        status: TerminalTaskStatus,
+        failure_reason: str | None,
+        completed_at: datetime,
+    ) -> ResultRecordStatus: ...
+
+    def reap_expired_leases(
+        self,
+        *,
+        now: datetime,
+        max_attempts: int,
+    ) -> list[str]: ...
+
+    def list_task_attempts(self, task_id: str) -> list[TaskAttemptRecord]: ...
+
+    def health_check(self) -> None: ...
+
+    def close(self) -> None: ...
diff --git a/tests/test_cloud_repository_contract.py b/tests/test_cloud_repository_contract.py
new file mode 100644
index 0000000..321e3cf
--- /dev/null
+++ b/tests/test_cloud_repository_contract.py
@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import get_protocol_members
+
+from cloud.repository import CloudRepository, LeasedAssignment, TaskAttemptRecord
+
+
+def test_cloud_repository_exposes_crud_and_atomic_lease_operations() -> None:
+    members = get_protocol_members(CloudRepository)
+
+    assert {
+        "upsert_host",
+        "replace_host_devices",
+        "list_hosts",
+        "list_devices",
+        "enqueue_task",
+        "get_task",
+        "save_plugin",
+        "assign_task",
+        "claim_assignment",
+        "renew_lease",
+        "record_task_result",
+        "reap_expired_leases",
+        "list_task_attempts",
+        "health_check",
+        "close",
+    } <= members
+
+
+def test_repository_transfer_records_are_immutable() -> None:
+    assert TaskAttemptRecord.__dataclass_params__.frozen is True
+    assert LeasedAssignment.__dataclass_params__.frozen is True