feat(cloud-scheduler): reap expired leases

openspec/changes/cloud-control-plane-integration/tasks.md

@@ -22,7 +22,7 @@
 - [x] 3.3 Implement owning-host claim that atomically transitions one assigned attempt to dispatched under its active lease.
 - [x] 3.4 Implement lease renewal with host/task/attempt ownership validation and conflict responses for stale leases.
 - [x] 3.5 Implement idempotent terminal result recording and reservation release for active leases.
-- [ ] 3.6 Implement expired-lease requeue/failure behavior with bounded attempts and preserved attempt history.
+- [x] 3.6 Implement expired-lease requeue/failure behavior with bounded attempts and preserved attempt history.
 - [ ] 3.7 Add PostgreSQL concurrency tests proving one assignment/claim winner and SQLite tests documenting single-control-plane behavior.
 
 ## 4. Authentication And Authorization

packages/cloud-platform/cloud/sql_repository.py

@@ -451,6 +451,59 @@ class SQLAlchemyCloudRepository:
             attempt_row.result_json = result_json
             return "recorded"
 
+    def reap_expired_leases(
+        self,
+        *,
+        now: datetime,
+        max_attempts: int,
+    ) -> list[str]:
+        if max_attempts < 1:
+            raise ValueError("max_attempts must be at least 1")
+
+        with self._sessions.begin() as session:
+            statement = (
+                select(ScheduledTaskRow)
+                .where(
+                    ScheduledTaskRow.status.in_(("assigned", "dispatched")),
+                    ScheduledTaskRow.lease_expires_at.is_not(None),
+                    ScheduledTaskRow.lease_expires_at <= _iso(now),
+                )
+                .order_by(ScheduledTaskRow.lease_expires_at, ScheduledTaskRow.id)
+            )
+            if self.engine.dialect.name == "postgresql":
+                statement = statement.with_for_update(skip_locked=True)
+
+            tasks = session.scalars(statement).all()
+            reaped_task_ids: list[str] = []
+            for task in tasks:
+                attempt_row = session.get(
+                    TaskAttemptRow,
+                    (task.id, task.attempt_count),
+                    with_for_update=self.engine.dialect.name == "postgresql",
+                )
+                if attempt_row is None:
+                    continue
+
+                attempt_row.status = "expired"
+                attempt_row.completed_at = _iso(now)
+                attempt_row.failure_reason = "lease expired"
+                task.updated_at = _iso(now)
+                task.lease_id = None
+                task.lease_expires_at = None
+                task.result_json = None
+                if task.attempt_count < max_attempts:
+                    task.status = "queued"
+                    task.assigned_host_id = None
+                    task.assigned_device_id = None
+                    task.failure_reason = None
+                else:
+                    task.status = "failed"
+                    task.failure_reason = (
+                        f"lease expired after {task.attempt_count} attempts"
+                    )
+                reaped_task_ids.append(task.id)
+            return reaped_task_ids
+
     def list_task_attempts(self, task_id: str) -> list[Any]:
         with self._sessions() as session:
             rows = session.scalars(

tests/test_cloud_repository_contract.py

@@ -957,3 +957,191 @@ def test_stale_foreign_or_expired_result_is_rejected(
         assert task.terminal_result is None
     finally:
         database.close()
+
+
+@pytest.mark.parametrize("claimed", [False, True], ids=["assigned", "dispatched"])
+def test_expired_lease_requeues_with_auditable_history(
+    database_url: str,
+    claimed: bool,
+) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("requeue-host")
+    device_id = _unique_id("requeue-device")
+    task_id = _unique_id("requeue-task")
+    now = datetime(2026, 7, 12, 14, 0, tzinfo=UTC)
+    expired_at = now + timedelta(seconds=10)
+    reaped_at = expired_at + timedelta(seconds=1)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="retry after expiry",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="expired-attempt-1",
+            lease_expires_at=expired_at,
+            now=now,
+        )
+        if claimed:
+            database.repository.claim_assignment(host_id=host_id, now=now)
+
+        reaped_task_ids = database.repository.reap_expired_leases(
+            now=reaped_at,
+            max_attempts=2,
+        )
+        assert task_id in reaped_task_ids
+
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "queued"
+        assert task.attempt_count == 1
+        assert task.assigned_host_id is None
+        assert task.assigned_device_id is None
+        assert task.lease_id is None
+        assert task.lease_expires_at is None
+        assert task.failure_reason is None
+        attempts = database.repository.list_task_attempts(task_id)
+        assert len(attempts) == 1
+        assert attempts[0].status == "expired"
+        assert attempts[0].failure_reason == "lease expired"
+        assert attempts[0].completed_at == reaped_at
+        assert device_id not in database.repository.list_reserved_device_ids(
+            now=reaped_at
+        )
+
+        second_assignment = database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="active-attempt-2",
+            lease_expires_at=reaped_at + timedelta(minutes=1),
+            now=reaped_at,
+        )
+        assert second_assignment is not None
+        assert second_assignment.attempt == 2
+        assert [
+            attempt.status
+            for attempt in database.repository.list_task_attempts(task_id)
+        ] == ["expired", "assigned"]
+    finally:
+        database.close()
+
+
+def test_expired_lease_fails_at_attempt_limit(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("limit-host")
+    device_id = _unique_id("limit-device")
+    task_id = _unique_id("limit-task")
+    now = datetime(2026, 7, 12, 15, 0, tzinfo=UTC)
+    expired_at = now + timedelta(seconds=10)
+    reaped_at = expired_at + timedelta(seconds=1)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="fail after expiry",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="final-attempt",
+            lease_expires_at=expired_at,
+            now=now,
+        )
+
+        reaped_task_ids = database.repository.reap_expired_leases(
+            now=reaped_at,
+            max_attempts=1,
+        )
+        assert task_id in reaped_task_ids
+
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "failed"
+        assert task.failure_reason == "lease expired after 1 attempts"
+        assert task.lease_id is None
+        assert task.lease_expires_at is None
+        attempts = database.repository.list_task_attempts(task_id)
+        assert attempts[0].status == "expired"
+        assert device_id not in database.repository.list_reserved_device_ids(
+            now=reaped_at
+        )
+        assert (
+            database.repository.reap_expired_leases(
+                now=reaped_at,
+                max_attempts=1,
+            )
+            == []
+        )
+    finally:
+        database.close()
+
+
+def test_unexpired_lease_is_not_reaped(database_url: str) -> None:
+    database = CloudDatabase(database_url)
+    host_id = _unique_id("active-reaper-host")
+    device_id = _unique_id("active-reaper-device")
+    task_id = _unique_id("active-reaper-task")
+    now = datetime(2026, 7, 12, 16, 0, tzinfo=UTC)
+
+    try:
+        database.repository.upsert_host(host_id, address=None, last_seen_at=now)
+        database.repository.replace_host_devices(
+            host_id,
+            [_device(device_id, host_id)],
+        )
+        database.repository.enqueue_task(
+            ScheduledTask(
+                id=task_id,
+                goal="stay active",
+                workflow_definition_id=None,
+                constraints=TaskConstraints(),
+                created_at=now,
+            )
+        )
+        database.repository.assign_task(
+            task_id=task_id,
+            host_id=host_id,
+            device_id=device_id,
+            lease_id="active-reaper-lease",
+            lease_expires_at=now + timedelta(minutes=1),
+            now=now,
+        )
+
+        assert (
+            database.repository.reap_expired_leases(
+                now=now + timedelta(seconds=30),
+                max_attempts=2,
+            )
+            == []
+        )
+        task = database.repository.get_task(task_id)
+        assert task is not None
+        assert task.status == "assigned"
+        assert device_id in database.repository.list_reserved_device_ids(now=now)
+    finally:
+        database.close()