"""Adapters exposing existing capability implementations through the thin lifecycle contract.""" from __future__ import annotations from typing import Any from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession from src.domain.capabilities.dependencies import ( CapabilityDependency, DependencyResolution, resolve_dependencies, ) from src.domain.capabilities.lifecycle import ( CapabilityIssue, CapabilityProvisioner, CapabilityState, CapabilityStatus, IssueSeverity, PlanAction, ProvisionPlan, ProvisionResult, ) from src.domain.capabilities.runtime import ( CapabilityRuntimeRequirements, RuntimePortRequirement, RuntimeConfigRequirement, RuntimeRouteRequirement, merge_runtime_requirements, ) from src.domain.feature_packs.models import ProjectCapabilityReconciliation from src.domain.forms.canonical import FORMS_CANONICAL_MODEL from src.domain.project_auth.projection_eligibility import ( DISABLED, ENABLED, SHADOW, project_state, ) # Catalog-pack primitives that imply a project-level dependency on another platform # capability's runtime wiring, not just a Cloud job primitive. Deliberately tiny and # literal (see requirements_spec.py's "no speculative primitive proliferation" bar) -- # add an entry only once a real pack demonstrates the need for the matching capability. _PRIMITIVE_CAPABILITY_HINTS: dict[str, str] = { "auth.roles": "auth", "form.public_create": "forms", "data.entity": "records", } class AuthV2Provisioner(CapabilityProvisioner): capability = "auth" version = 2 def __init__(self, db: AsyncSession, *, cloud_base_url: str, cloud_token: str): self.db = db self.cloud_base_url = cloud_base_url self.cloud_token = cloud_token async def declare(self, project: Any, desired_state: dict[str, Any]) -> dict[str, Any]: return {"capability": self.capability, "version": self.version, **desired_state} async def plan(self, project: Any) -> ProvisionPlan: state = await project_state(self.db, _project_id(project)) changes = () if state != DISABLED else ("authz_v2_backfill",) action = PlanAction.PROVISION if changes else PlanAction.NOOP return ProvisionPlan(self.capability, self.version, changes=changes, action=action) async def apply(self, project: Any, plan: ProvisionPlan) -> ProvisionResult: from src.domain.project_auth.provisioning_v2 import provision_authz_v2 outcome = await provision_authz_v2( self.db, _project_id(project), cloud_base_url=self.cloud_base_url, cloud_token=self.cloud_token, ) return ProvisionResult( self.capability, self.version, _auth_state(outcome.projection_state, outcome.status), blockers=(outcome.reason,) if outcome.reason else (), ) async def reconcile(self, project: Any) -> ProvisionResult: return await self.apply(project, await self.plan(project)) async def status(self, project: Any) -> CapabilityStatus: from src.domain.project_auth.repository import ( ProjectAuthConfigRepository, ProjectAuthKeyRepository, ) project_id = _project_id(project) config = await ProjectAuthConfigRepository(self.db).get_by_project(project_id) if config is None: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="auth_config_missing", issues=(CapabilityIssue( code="auth_config_missing", message="Core has no auth configuration for this project.", repairable=True, ),), ) if not config.enabled: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.DECLARED, blocked_reason="disabled", ) active_key = await ProjectAuthKeyRepository(self.db).get_active_for_project(project_id) if active_key is None: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="auth_public_key_missing", issues=(CapabilityIssue( code="auth_public_key_missing", message="Core has no active signing public key for this project.", repairable=True, ),), ) if config.cloud_synced_at is None: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="auth_config_projection_missing", details={"cloud_sync_error": config.cloud_sync_error}, issues=(CapabilityIssue( code="auth_config_projection_missing", message="Cloud has not acknowledged the required auth projection.", repairable=True, ),), ) if config.cloud_sync_error: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.DRIFTED, blocked_reason="auth_config_projection_missing", details={"cloud_sync_error": config.cloud_sync_error}, issues=(CapabilityIssue( code="auth_config_projection_missing", message="The last required auth projection attempt failed.", repairable=True, ),), ) state = await project_state(self.db, _project_id(project)) if state == DISABLED: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="authz_projection_missing", details={"projection_state": state}, issues=(CapabilityIssue( code="authz_projection_missing", message="Cloud authorization projection is not enabled for this project.", repairable=True, ),), ) return CapabilityStatus( capability=self.capability, version=self.version, state=_auth_state(state), drift=False, details={"projection_state": state}, ) async def runtime_requirements(self, project: Any) -> CapabilityRuntimeRequirements: return CapabilityRuntimeRequirements( capability_key=self.capability, capability_version=self.version, routes=( RuntimeRouteRequirement( path_prefix="/api/auth", purpose="auth_session_and_signin_routes", methods=("GET", "POST", "PUT"), ), ), ports=( RuntimePortRequirement( port_id="auth.magic_link_request", capability=self.capability, capability_version=self.version, method="POST", public_path="/api/auth/magic-link/request", auth="public", ), RuntimePortRequirement( port_id="auth.magic_link_consume", capability=self.capability, capability_version=self.version, method="GET", public_path="/api/auth/magic-link/consume", auth="public", ), RuntimePortRequirement( port_id="auth.me", capability=self.capability, capability_version=self.version, method="GET", public_path="/api/auth/me", auth="session", ), RuntimePortRequirement( port_id="auth.signout", capability=self.capability, capability_version=self.version, method="POST", public_path="/api/auth/signout", auth="session", ), RuntimePortRequirement( port_id="auth.permissions_check", capability=self.capability, capability_version=self.version, method="POST", public_path="/api/auth/permissions/check", auth="session", ), ), platform_services=("p2hs-auth",), config=( RuntimeConfigRequirement( key="P2HS_AUTH_PUBLIC_KEY", visibility="public", description="Project signing-key public half, for client-side session " "verification.", ), RuntimeConfigRequirement( key="P2HS_CLOUD_SERVICE_TOKEN", visibility="trusted", description="Server-side credential for the BFF to call Cloud auth on the " "project's behalf.", ), ), ) class FormsProvisioner(CapabilityProvisioner): """Adapter wrapping the existing Forms composition resolver (``src.domain.forms.composition_store``/``composition``). Canonical form definitions and their installed-Feature-Pack bindings already exist and are resolved deterministically from installed packs + the App Block plan; this provisioner reads that resolution as *observed* state rather than maintaining a second, parallel Forms ledger. There is still no separate "desired state" ledger for Forms (``declare``/``apply`` are no-ops) -- desired state today is implicit in which Feature Packs/App Blocks are installed, same as before this provisioner existed. See docs/CAPABILITY_RUNTIME_PROJECTION.md's "Deferred" section. """ capability = "forms" version = 1 def __init__(self, db: AsyncSession): self.db = db async def declare(self, project: Any, desired_state: dict[str, Any]) -> dict[str, Any]: return {"capability": self.capability, "version": self.version, **desired_state} async def plan(self, project: Any) -> ProvisionPlan: resolution = (await self._load_store(project)).resolution if not resolution.sources: return ProvisionPlan(self.capability, self.version, action=PlanAction.NOOP) if resolution.conflicts: return ProvisionPlan( self.capability, self.version, changes=("resolve_form_owner_conflicts",), action=PlanAction.BLOCKED, ) unresolved = _unresolved_form_keys(resolution) if unresolved: return ProvisionPlan( self.capability, self.version, changes=("fix_form_runtime_validation",), action=PlanAction.BLOCKED, ) return ProvisionPlan(self.capability, self.version, action=PlanAction.NOOP) async def apply(self, project: Any, plan: ProvisionPlan) -> ProvisionResult: return await self.reconcile(project) async def reconcile(self, project: Any) -> ProvisionResult: # Forms has no separate write/push step: composition resolution over # already-installed packs/blocks *is* the provisioning. Reconcile just # re-reads the same resolution status() does. report = await self.status(project) return ProvisionResult( self.capability, self.version, report.state, blockers=(report.blocked_reason,) if report.blocked_reason else (), details=report.details, ) async def status(self, project: Any) -> CapabilityStatus: resolution = (await self._load_store(project)).resolution issues = await self._pending_requirement_issues(project) if not resolution.sources: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.DECLARED, details={"contract_status": FORMS_CANONICAL_MODEL["status"], "forms": []}, issues=issues, ) if resolution.conflicts: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="form_owner_conflict", details={"conflicting_forms": sorted(resolution.conflicts)}, issues=issues, ) unresolved = _unresolved_form_keys(resolution) if unresolved: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.BLOCKED, blocked_reason="form_runtime_validation_failed", details={"unresolved_forms": sorted(unresolved)}, issues=issues, ) return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.DRIFTED if issues else CapabilityState.READY, details={"forms": sorted(resolution.runtimes)}, issues=issues, ) async def _pending_requirement_issues(self, project: Any) -> tuple[CapabilityIssue, ...]: """Surface a build prompt whose form-field intent could not be confidently extracted (see ``src.domain.projects.form_requirement_sync``) -- the prior resolved schema was deliberately left untouched rather than guessed, but that must not be silent: a follow-up prompt may be needed to actually apply the change the customer asked for.""" from src.domain.projects.discovery_repository import DiscoveryRepository project_obj = await DiscoveryRepository(self.db).get(_project_id(project)) marker = ((project_obj.discovery_json or {}) if project_obj else {}).get( "form_requirement_sync" ) if not isinstance(marker, dict) or not marker.get("ambiguous"): return () return ( CapabilityIssue( code="form_requirement_extraction_ambiguous", message="A recent build prompt seemed to ask for a form field change but " "could not be confidently resolved. The form schema was left unchanged; " "describe the field change again more specifically.", severity=IssueSeverity.WARNING, repairable=True, ), ) async def runtime_requirements(self, project: Any) -> CapabilityRuntimeRequirements: resolution = (await self._load_store(project)).resolution if not resolution.runtimes: return CapabilityRuntimeRequirements( capability_key=self.capability, capability_version=self.version ) has_file_fields = any( any(getattr(field, "type", None) == "file" for field in runtime.definition.fields) for runtime in resolution.runtimes.values() ) ports = [ RuntimePortRequirement( port_id="forms.submit", capability=self.capability, capability_version=self.version, method="POST", public_path="/api/forms/:form_key/submissions", auth="public", ) ] if has_file_fields: ports.append( RuntimePortRequirement( port_id="forms.file_upload", capability=self.capability, capability_version=self.version, method="POST", public_path="/api/forms/:form_key/files", auth="public", ) ) return CapabilityRuntimeRequirements( capability_key=self.capability, capability_version=self.version, routes=( RuntimeRouteRequirement( path_prefix="/api/forms", purpose="form_submission_routes", methods=("POST",), ), ), platform_services=("p2hs-forms",), config=tuple( RuntimeConfigRequirement( key=f"forms_definition:{form_key}", visibility="trusted", description=f"Canonical submission definition for form '{form_key}'.", ) for form_key in sorted(resolution.runtimes) ), ports=tuple(ports), ) async def _load_store(self, project: Any): from src.domain.forms.composition_store import load_composition_form_store return await load_composition_form_store(self.db, _project_id(project)) def _unresolved_form_keys(resolution) -> set[str]: """Form keys with a declared binding source but no resolved+validated runtime -- the composition resolver dropped them (unknown form_key or validation errors in build_resolved_runtime), distinct from an owner conflict.""" declared = {source.form_key for source in resolution.sources} - resolution.conflicts return declared - set(resolution.runtimes) class FeaturePackProvisioner(CapabilityProvisioner): """Aggregates runtime requirements of the capabilities a project's installed packs actually depend on (e.g. a pack requiring the ``auth.roles`` primitive pulls in Auth's runtime requirements too). Avoids any hosting knowledge itself -- it only re-exposes what ``AuthV2Provisioner``/``FormsProvisioner`` already declare. """ capability = "feature_packs" version = 1 def __init__( self, db: AsyncSession, *, db_factory=None, capability_providers: dict[str, CapabilityProvisioner] | None = None, ): self.db = db self.db_factory = db_factory # Keyed by capability key (e.g. "auth", "forms") -- supplied by the caller # that already constructed those provisioners, so this adapter never has to # know connection settings for capabilities it doesn't own. self._capability_providers = capability_providers or {} async def declare(self, project: Any, desired_state: dict[str, Any]) -> dict[str, Any]: return {"capability": self.capability, "version": self.version, **desired_state} async def plan(self, project: Any) -> ProvisionPlan: row = await self.db.get(ProjectCapabilityReconciliation, _project_id(project)) if row is None: return ProvisionPlan( self.capability, self.version, changes=("reconcile_feature_packs",), action=PlanAction.PROVISION, ) return ProvisionPlan( self.capability, self.version, desired_revision=row.attempt_count, action=PlanAction.RECONCILE, ) async def apply(self, project: Any, plan: ProvisionPlan) -> ProvisionResult: return await self.reconcile(project) async def reconcile(self, project: Any) -> ProvisionResult: if self.db_factory is None: return ProvisionResult( self.capability, self.version, CapabilityState.BLOCKED, blockers=("db_factory_required_for_reconciliation",), ) from src.workers.tasks.reconcile_project_capabilities import reconcile_project_capabilities result = await reconcile_project_capabilities( {"db_factory": self.db_factory}, str(_project_id(project)) ) state = ( CapabilityState.READY if not result.get("errors") and not result.get("drift") else CapabilityState.DRIFTED ) return ProvisionResult( self.capability, self.version, state, drift=bool(result.get("drift")), details=result, ) async def _installed_pack_contracts(self, project_id: UUID) -> list[Any]: from sqlalchemy import select from src.domain.feature_packs.models import FeaturePack, ProjectFeaturePack from src.domain.feature_packs.normalize import load_pack_contract installed = await self.db.execute( select(ProjectFeaturePack.pack_slug).where(ProjectFeaturePack.project_id == project_id) ) pack_slugs = {row[0] for row in installed.all()} if not pack_slugs: return [] packs = await self.db.execute(select(FeaturePack).where(FeaturePack.slug.in_(pack_slugs))) return [ load_pack_contract(pack_row.slug, pack_row.composition or {}) for pack_row in packs.scalars().all() ] async def _required_capability_keys(self, project_id: UUID) -> set[str]: """Capabilities implied by installed packs' *primitive* requires -- the pre-existing heuristic (``_PRIMITIVE_CAPABILITY_HINTS``), kept for packs that only declare the Cloud primitive, not the capability itself. Packs declaring ``{"kind": "capability", ...}`` directly are resolved formally instead, via ``_capability_dependency_resolution``.""" needed: set[str] = set() for contract in await self._installed_pack_contracts(project_id): for dep in (*contract.requires, *contract.optional_requires): if dep.kind == "primitive" and dep.id in _PRIMITIVE_CAPABILITY_HINTS: needed.add(_PRIMITIVE_CAPABILITY_HINTS[dep.id]) for dep in contract.required_capabilities(): needed.add(dep.key) return needed async def _capability_dependency_resolution(self, project_id: UUID) -> DependencyResolution: """Transitive/version-conflict resolution over every installed pack's formally declared ``kind="capability"`` requirements, treated as one collective requirer (``"feature_packs"``) -- see ``src.domain.capabilities.dependencies.resolve_dependencies``. Packs that only declare a primitive (not a capability) are invisible to this resolver; they're still covered by ``_required_capability_keys`` for aggregation, just without version-conflict detection.""" deps: list[CapabilityDependency] = [] for contract in await self._installed_pack_contracts(project_id): deps.extend(contract.required_capabilities()) return resolve_dependencies( roots={self.capability: self.version}, dependencies={self.capability: deps}, ) async def runtime_requirements(self, project: Any) -> CapabilityRuntimeRequirements: needed = await self._required_capability_keys(_project_id(project)) dependency_requirements = [ await self._capability_providers[key].runtime_requirements(project) for key in sorted(needed) if key in self._capability_providers ] empty = CapabilityRuntimeRequirements( capability_key=self.capability, capability_version=self.version ) if not dependency_requirements: return empty result = merge_runtime_requirements(dependency_requirements) if not result.ok or result.merged is None: # Fail closed to "no aggregated requirement" rather than guessing which # conflicting entry wins -- the capabilities involved still report their # own requirements individually to anything merging at a higher level. return empty merged = result.merged return CapabilityRuntimeRequirements( capability_key=self.capability, capability_version=self.version, contract_version=merged.contract_version, routes=merged.routes, platform_services=merged.platform_services, config=merged.config, health_checks=merged.health_checks, feature_flags=merged.feature_flags, ) async def status(self, project: Any) -> CapabilityStatus: project_id = _project_id(project) issues = await self._dependency_issues(project_id) row = await self.db.get(ProjectCapabilityReconciliation, project_id) if row is None: return CapabilityStatus( capability=self.capability, version=self.version, state=CapabilityState.DECLARED, issues=issues, ) state = { "applied": CapabilityState.READY, "upgrade_available": CapabilityState.DRIFTED, "error": CapabilityState.BLOCKED, "pending": CapabilityState.PLANNED, "drift": CapabilityState.DRIFTED, }.get(row.status, CapabilityState.DECLARED) return CapabilityStatus( capability=self.capability, version=self.version, state=state, desired_revision=row.attempt_count, applied_revision=row.attempt_count if state == CapabilityState.READY else None, drift=state == CapabilityState.DRIFTED, blocked_reason=row.last_error if state == CapabilityState.BLOCKED else None, details={"required_action": row.required_action}, issues=issues, ) async def _dependency_issues(self, project_id: UUID) -> tuple[CapabilityIssue, ...]: """Structured, non-fatal issues from capability-level dependency resolution -- reported alongside the reconciliation ledger's own ``state``, never overriding it. A real version conflict here means two installed packs declare incompatible capability requirements (e.g. one needs auth >= 2, another pins auth == 1); it is surfaced for visibility, not auto-blocked, since the reconciliation ledger is still the authoritative state source.""" resolution = await self._capability_dependency_resolution(project_id) issues: list[CapabilityIssue] = [] for conflict in resolution.version_conflicts: issues.append( CapabilityIssue( code="CAPABILITY_VERSION_CONFLICT", message=( f"Installed packs require incompatible versions of " f"capability '{conflict.key}'" ), severity=IssueSeverity.ERROR, repairable=False, ) ) for cycle in resolution.cycles: issues.append( CapabilityIssue( code="CAPABILITY_DEPENDENCY_CYCLE", message=f"Circular capability dependency: {' -> '.join(cycle.cycle)}", severity=IssueSeverity.ERROR, repairable=False, ) ) return tuple(issues) def _project_id(project: Any) -> UUID: value = project if isinstance(project, (UUID, str)) else project.id return value if isinstance(value, UUID) else UUID(str(value)) def _auth_state(projection_state: str, outcome_status: str | None = None) -> CapabilityState: if projection_state == ENABLED: return CapabilityState.ENABLED if projection_state == SHADOW: return CapabilityState.SHADOW if outcome_status == "blocked": return CapabilityState.BLOCKED if outcome_status == "retry": return CapabilityState.APPLYING return CapabilityState.DECLARED