feat(cloud-api): run lifecycle workers

apps/cloud-api/cloud_api/app.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from collections.abc import Callable
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
@@ -19,7 +20,9 @@ from cloud.internal_api.api import create_internal_router
 from cloud.plugins import PluginRegistry
 from cloud.pool import DevicePool
 from cloud.scheduler import TaskScheduler
+from cloud.schema import require_current_schema
 from cloud.sdk.api import create_cloud_router
+from core.models import utc_now
 
 
 DatabaseFactory = Callable[[CloudControlConfig], CloudDatabase]
@@ -85,13 +88,43 @@ def create_app(
         app.state.cloud_config = control_config
         app.state.database = database
         app.state.cloud_services = services
+        stop_workers = asyncio.Event()
+        worker_tasks: list[asyncio.Task[None]] = []
         try:
+            repository.health_check()
+            if control_config.environment == "production":
+                require_current_schema(control_config.database_url)
+            worker_tasks = [
+                asyncio.create_task(
+                    _run_scheduler_loop(
+                        services,
+                        stop_workers,
+                        interval_seconds=control_config.scheduler_interval_seconds,
+                    ),
+                    name="cloud-scheduler",
+                ),
+                asyncio.create_task(
+                    _run_lease_reaper_loop(
+                        services,
+                        stop_workers,
+                        interval_seconds=(control_config.lease_reaper_interval_seconds),
+                        max_attempts=control_config.max_task_attempts,
+                    ),
+                    name="cloud-lease-reaper",
+                ),
+            ]
+            app.state.worker_tasks = tuple(worker_tasks)
             yield
         finally:
+            stop_workers.set()
             try:
-                database.close()
+                if worker_tasks:
+                    await asyncio.gather(*worker_tasks)
             finally:
-                repository.unbind()
+                try:
+                    database.close()
+                finally:
+                    repository.unbind()
 
     app = FastAPI(title="Device Cloud API", lifespan=lifespan)
     app.include_router(
@@ -117,3 +150,39 @@ def _default_database_factory(config: CloudControlConfig) -> CloudDatabase:
         config.database_url,
         create_schema=config.environment != "production",
     )
+
+
+async def _run_scheduler_loop(
+    services: CloudApplicationServices,
+    stop: asyncio.Event,
+    *,
+    interval_seconds: float,
+) -> None:
+    while not stop.is_set():
+        services.scheduler.assign()
+        if await _wait_for_stop(stop, interval_seconds):
+            return
+
+
+async def _run_lease_reaper_loop(
+    services: CloudApplicationServices,
+    stop: asyncio.Event,
+    *,
+    interval_seconds: float,
+    max_attempts: int,
+) -> None:
+    while not stop.is_set():
+        services.repository.reap_expired_leases(
+            now=utc_now(),
+            max_attempts=max_attempts,
+        )
+        if await _wait_for_stop(stop, interval_seconds):
+            return
+
+
+async def _wait_for_stop(stop: asyncio.Event, interval_seconds: float) -> bool:
+    try:
+        await asyncio.wait_for(stop.wait(), timeout=interval_seconds)
+    except TimeoutError:
+        return False
+    return True

apps/cloud-api/tests/test_app.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import time
+
 import pytest
 from fastapi.testclient import TestClient
 
@@ -20,8 +22,20 @@ def test_create_app_returns_independent_cloud_application() -> None:
 def test_cloud_application_owns_database_lifecycle() -> None:
     events: list[str] = []
 
+    class FakeRepository:
+        def health_check(self) -> None:
+            events.append("healthy")
+
+        def list_queued_tasks(self) -> list[object]:
+            events.append("scheduled")
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            events.append("reaped")
+            return []
+
     class FakeDatabase:
-        repository = object()
+        repository = FakeRepository()
 
         def close(self) -> None:
             events.append("closed")
@@ -37,9 +51,9 @@ def test_cloud_application_owns_database_lifecycle() -> None:
         assert (
             app.state.cloud_services.pool.store is app.state.cloud_services.repository
         )
-        assert events == []
+        assert "healthy" in events
 
-    assert events == ["closed"]
+    assert events[-1] == "closed"
 
 
 def test_repository_proxy_is_available_only_during_lifespan() -> None:
@@ -47,6 +61,12 @@ def test_repository_proxy_is_available_only_during_lifespan() -> None:
         def health_check(self) -> None:
             return None
 
+        def list_queued_tasks(self) -> list[object]:
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            return []
+
     class FakeDatabase:
         repository = FakeRepository()
 
@@ -65,6 +85,47 @@ def test_repository_proxy_is_available_only_during_lifespan() -> None:
         app.state.cloud_services.repository.health_check()
 
 
+def test_lifespan_runs_scheduler_and_reaper_until_shutdown() -> None:
+    events: list[str] = []
+
+    class FakeRepository:
+        def health_check(self) -> None:
+            events.append("healthy")
+
+        def list_queued_tasks(self) -> list[object]:
+            events.append("scheduled")
+            return []
+
+        def reap_expired_leases(self, *, now, max_attempts: int) -> list[str]:
+            events.append(f"reaped:{max_attempts}")
+            return []
+
+    class FakeDatabase:
+        repository = FakeRepository()
+
+        def close(self) -> None:
+            events.append("closed")
+
+    app = create_app(
+        config=CloudControlConfig(
+            database_url="sqlite:///:memory:",
+            scheduler_interval_seconds=0.01,
+            lease_reaper_interval_seconds=0.01,
+            max_task_attempts=4,
+        ),
+        database_factory=lambda _config: FakeDatabase(),  # type: ignore[arg-type,return-value]
+    )
+
+    with TestClient(app):
+        time.sleep(0.04)
+        assert events.count("scheduled") >= 2
+        assert events.count("reaped:4") >= 2
+        assert all(not task.done() for task in app.state.worker_tasks)
+
+    assert all(task.done() for task in app.state.worker_tasks)
+    assert events[-1] == "closed"
+
+
 def test_production_app_rejects_missing_credentials() -> None:
     with pytest.raises(CloudConfigurationError, match="credential"):
         create_app(

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

@@ -44,7 +44,7 @@
 ## 6. Cloud Control Plane Composition
 
 - [x] 6.1 Compose repository, pool, scheduler, plugin registry, public router, internal router, and auth providers in the cloud app factory.
-- [ ] 6.2 Implement FastAPI lifespan startup/shutdown for configuration validation, database checks, scheduler loop, and lease-reaper loop.
+- [x] 6.2 Implement FastAPI lifespan startup/shutdown for configuration validation, database checks, scheduler loop, and lease-reaper loop.
 - [ ] 6.3 Ensure lifecycle iteration failures are logged and retried without terminating later iterations.
 - [ ] 6.4 Add `/health/live` and `/health/ready` with separate process, database/schema, and worker-state semantics.
 - [ ] 6.5 Add structured correlation-aware logging for requests and task lifecycle events with sensitive payload redaction.