"""Reconcile existing projects with current Core capability contracts. This is a policy/control-plane updater, not a source migration. It refreshes Cloud after a contract change and records whether generated code still needs a pack reinstall or regeneration. """ from __future__ import annotations from datetime import UTC, datetime from uuid import UUID import structlog from sqlalchemy import and_, select from src.domain.feature_packs.models import ( FeaturePack, ProjectCapabilityReconciliation, ProjectFeaturePack, ) from src.domain.feature_packs.reconciliation import ( build_contract_snapshot, contract_fingerprint, reconciliation_status, ) from src.domain.projects.models import Project logger = structlog.get_logger() _MAX_PER_RUN = 100 # ProjectFeaturePack states in which the pack's entities were actually provisioned in Cloud. # `blocked` means the capability itself could not be built and `uninstalled`/`supported`/ # `enabled` mean it was never (or is no longer) installed: none of those has an entity to # refresh, so asking Cloud about it can only 404 (see models.ProjectFeaturePack). _ENTITIES_PROVISIONED = frozenset( {"installed", "configured", "deployed", "verified", "configuration_required"} ) async def _refresh_entity_policies(cloud_client, project_id: str, project_packs) -> list[str]: """Re-apply the current access policy to every entity of every installed pack. Returns the keys of entities Cloud says do not exist (404). For a pack recorded as installed that is drift between Core's ledger and Cloud: reported as such (see ``drift_summary``), never raised, so it cannot fail the rest of the project's reconciliation or spam a stack trace every cycle. Any other Cloud error still raises.""" import httpx missing: list[str] = [] for installed, pack in project_packs: if installed.status not in _ENTITIES_PROVISIONED: continue for entity_def in (pack.composition or {}).get("entities") or []: if not isinstance(entity_def, dict): continue try: await cloud_client.reconcile_project_entity_policy(project_id, entity_def) except httpx.HTTPStatusError as exc: if exc.response.status_code != 404: raise key = entity_def.get("entity_key") or entity_def.get("key") or "unknown" missing.append(f"{installed.pack_slug}:{key}") return missing def drift_summary(missing: list[str]) -> tuple[str, str, str]: """(ledger status, required action, message) for entities missing in Cloud.""" return ( "drift", "reinstall_pack", "Installed pack entities are missing in Cloud: " + ", ".join(sorted(set(missing)))[:1800], ) def _project_feature_pack_contract_query(): """Return the pack projection consumed by ``build_contract_snapshot``. The reconciliation layer's contract uses the public key ``slug`` even though the ORM column is named ``pack_slug``. Keep that translation at the query boundary so mappings passed to the contract builder have the shape it expects. """ return select( ProjectFeaturePack.pack_slug.label("slug"), ProjectFeaturePack.pack_version.label("installed_version"), ProjectFeaturePack.status, FeaturePack.version.label("current_version"), ).join(FeaturePack, FeaturePack.slug == ProjectFeaturePack.pack_slug) async def reconcile_project_capabilities( ctx: dict, project_id: str | None = None, ) -> dict: from src.core.database import async_session_factory from src.integrations.cloud_control_client import get_cloud_client_for_workspace from src.workers.tasks.cloud_sync import sync_project_to_cloud db_factory = ctx.get("db_factory") or async_session_factory results = { "checked": 0, "skipped": 0, "applied": 0, "upgrade_available": 0, "drift": 0, "errors": 0, } async with db_factory() as db: stmt = select(Project).where( and_( Project.status.in_(["ready", "creating"]), Project.deleted_at.is_(None), ) ).order_by(Project.updated_at.asc()).limit(_MAX_PER_RUN) if project_id: stmt = stmt.where(Project.id == UUID(project_id)) projects = list((await db.execute(stmt)).scalars()) for project in projects: results["checked"] += 1 try: async with db_factory() as db: rows = list((await db.execute( _project_feature_pack_contract_query() .where(ProjectFeaturePack.project_id == project.id) )).mappings()) snapshot = build_contract_snapshot([dict(row) for row in rows]) fingerprint = contract_fingerprint(snapshot) status, required_action = reconciliation_status(snapshot) ledger = await db.get(ProjectCapabilityReconciliation, project.id) if ledger and ledger.desired_fingerprint == fingerprint and ledger.status in { "applied", "upgrade_available" }: results["skipped"] += 1 results[status] += 1 continue if ledger is None: ledger = ProjectCapabilityReconciliation( project_id=project.id, desired_fingerprint=fingerprint, desired_snapshot=snapshot, status="pending", attempt_count=0, ) db.add(ledger) else: ledger.desired_fingerprint = fingerprint ledger.desired_snapshot = snapshot ledger.status = "pending" ledger.required_action = None ledger.last_error = None ledger.attempt_count = (ledger.attempt_count or 0) + 1 await db.commit() sync_result = await sync_project_to_cloud(ctx, str(project.id)) if sync_result.get("status") == "skipped": raise RuntimeError(sync_result.get("reason") or "cloud_sync_skipped") # Refresh access policies for entities already installed by an older # pack version. The helper preserves all live fields and uses ETags. async with db_factory() as db: project_packs = list((await db.execute( select(ProjectFeaturePack, FeaturePack) .join(FeaturePack, FeaturePack.slug == ProjectFeaturePack.pack_slug) .where(ProjectFeaturePack.project_id == project.id) )).all()) cloud_client = await get_cloud_client_for_workspace(project.workspace_id, db) missing: list[str] = [] if cloud_client.api_url: missing = await _refresh_entity_policies( cloud_client, str(project.id), project_packs ) last_error = None if missing: status, required_action, last_error = drift_summary(missing) async with db_factory() as db: ledger = await db.get(ProjectCapabilityReconciliation, project.id) if ledger is None: raise RuntimeError("reconciliation_ledger_missing") ledger.status = status ledger.required_action = required_action ledger.applied_at = datetime.now(UTC) ledger.last_error = last_error await db.commit() results[status] += 1 logger.info( "project_capability_reconciled" if not missing else "project_capability_drift", project_id=str(project.id), status=status, required_action=required_action, missing_entities=missing or None, ) except Exception as exc: results["errors"] += 1 async with db_factory() as db: ledger = await db.get(ProjectCapabilityReconciliation, project.id) if ledger: ledger.status = "error" ledger.last_error = str(exc)[:2000] await db.commit() logger.error( "project_capability_reconciliation_failed", project_id=str(project.id), error=str(exc), exc_info=True, ) return results