"""Feature Pack contract V2.1 — backend composition, not product UX. App Blocks bind to ``provides`` / ``events`` / ``provider_bindings``. This module does not import App Blocks (packs sit below that layer). Authoring rules: docs/FEATURE_PACKS_APP_BLOCKS_STARTER.md """ from __future__ import annotations from typing import TYPE_CHECKING, Any, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator from src.domain.calculation.ir import CalculationIR from src.domain.feature_packs.requirements_spec import SUPPORTED_PRIMITIVES if TYPE_CHECKING: from src.domain.capabilities.dependencies import CapabilityDependency # Catalog packs may declare Cloud job primitives that Level-1 compose() # must not synthesize. App Block resolution uses the union. CATALOG_PRIMITIVES = frozenset( { "booking.reserve", "booking.cancel", "booking.availability", "inventory.reserve", "inventory.release", "inventory.commit", "record.renew", "workforce.shift_assign", "workforce.shift_release", "payments.checkout.create", "payments.order.capture", "payments.refund.create", "payments.catalog.upsert", "billing.portal", "billing.tax_calculation", # Composition pattern (verified webhook -> bounded mapping -> upsert), not # a literal Cloud job endpoint. Catalog-pack-only on purpose: Level-1 # compose() must not try to synthesize webhook provisioning + identity # mapping from a bare `requires` token. Any pack may declare it via # entity_ingest_bindings; recipes reference it generically via # `requires`/`optional_requires` without naming a specific pack slug. "external.entity_ingest", } ) RESOLVABLE_PRIMITIVES = SUPPORTED_PRIMITIVES | CATALOG_PRIMITIVES PackDepKind = Literal["primitive", "feature_pack", "capability"] SchemaMode = Literal["extensible", "protected"] MissingConfigurationBehavior = Literal["degraded", "disabled", "block_install"] OwnershipMode = Literal[ "staff_managed", "owner_managed", "rls", "public_create", "mixed", ] FEATURE_PACK_CONTRACT_VERSION = 2 KNOWN_PROVIDERS: dict[str, str] = { "stripe": "payments.provider", "paypal": "payments.provider", "openai": "ai.chat", "salesforce": "crm.external_sync", "hubspot": "crm.external_sync", "resend": "email.send", "twilio": "messaging.sms.send", "google_calendar": "calendar.sync", "calendly": "calendar.sync", "accredible": "education.certificate", } class PackDependencyRef(BaseModel): model_config = ConfigDict(extra="forbid") kind: PackDepKind id: str min_version: str | None = None when: str | None = None class PackProviderBinding(BaseModel): model_config = ConfigDict(extra="forbid") capability: str providers: list[str] = Field(default_factory=list) configuration_required: bool = False missing_configuration_behavior: MissingConfigurationBehavior = "degraded" @model_validator(mode="before") @classmethod def accept_supported_providers(cls, data: Any) -> Any: if not isinstance(data, dict): return data payload = dict(data) supported = payload.pop("supported_providers", None) single = payload.pop("provider", None) if "providers" not in payload: if supported is not None: payload["providers"] = supported elif single: payload["providers"] = [single] return payload ProducerKind = Literal["record.created", "workflow.transition", "field.changed"] EventAliasKind = Literal["composition", "transport"] class PackEventProducer(BaseModel): """Declarative runtime fact that may emit this pack event. Absent = not emittable.""" model_config = ConfigDict(extra="forbid") kind: ProducerKind entity: str field: str | None = None from_state: str | None = None to_state: str | None = None @model_validator(mode="after") def shape_matches_kind(self) -> PackEventProducer: if self.kind == "record.created": if self.field or self.from_state or self.to_state: raise ValueError( "record.created producer cannot set field, from_state, or to_state" ) elif self.kind == "workflow.transition": if not self.field: raise ValueError("workflow.transition producer requires field") elif self.kind == "field.changed": if not self.field: raise ValueError("field.changed producer requires field") if self.from_state or self.to_state: raise ValueError( "field.changed producer cannot set from_state or to_state" ) return self def identity(self) -> tuple[str, str, str | None, str | None, str | None]: return (self.kind, self.entity, self.field, self.from_state, self.to_state) class PackEventAlias(BaseModel): """Subscriber/composition name for a canonical PackEvent.id. Never a peer identity.""" model_config = ConfigDict(extra="forbid") id: str kind: EventAliasKind # Phase 11R3 (docs/audits/NOTIFICATION_ROUTING_POLICY_11R3.md): bounded # recipient-routing vocabulary. Deliberately 5 strategies, not 6 -- # "explicit_event_recipient" was considered and rejected because no field # type on this platform can guarantee a value is a real app_user reference # (see the routing doc's Sec 1); every case that might have qualified is # actually `field_user_ref` (resolved, then validated against # cloud_app_users at runtime -- Cloud's job, not Core's, since Core has no # visibility into a specific project's live user rows). NotificationRouteStrategy = Literal[ "subject_owner", "field_user_ref", "assignment_assignee", "role_members", "none" ] class NotificationRoute(BaseModel): """Declares HOW a recipient is derived for an emittable event -- never WHO specifically (that is runtime data Core cannot see). Same channel and same authority as PackEventProducer: Core states the semantic rule as part of the pack's own event contract; Cloud resolves it at runtime against project-local data only (no new Core round-trip). App Blocks and Codegen never see or influence this -- it is not a slot, not a binding, not part of any Codegen-facing artifact.""" model_config = ConfigDict(extra="forbid") strategy: NotificationRouteStrategy field: str | None = None role_field: str | None = None exclude_actor: bool = False @model_validator(mode="after") def shape_matches_strategy(self) -> NotificationRoute: if self.strategy == "subject_owner": if self.field or self.role_field: raise ValueError("subject_owner route cannot set field or role_field") elif self.strategy == "field_user_ref": if not self.field: raise ValueError("field_user_ref route requires field") if self.role_field: raise ValueError("field_user_ref route cannot set role_field") elif self.strategy == "assignment_assignee": if self.field or self.role_field: raise ValueError("assignment_assignee route cannot set field or role_field") elif self.strategy == "role_members": if not self.role_field: raise ValueError("role_members route requires role_field") if self.field: raise ValueError("role_members route cannot set field") elif self.strategy == "none": if self.field or self.role_field or self.exclude_actor: raise ValueError("none route cannot set field, role_field, or exclude_actor") return self class PackEvent(BaseModel): model_config = ConfigDict(extra="forbid") id: str version: int = 1 description: str | None = None payload: dict[str, str] = Field( default_factory=lambda: {"record_id": "string", "entity_key": "string"} ) producer: PackEventProducer | None = None aliases: list[PackEventAlias] = Field(default_factory=list) recipient: NotificationRoute | None = None @property def emittable(self) -> bool: return self.producer is not None @model_validator(mode="after") def recipient_requires_emittable(self) -> PackEvent: if self.recipient is not None and self.producer is None: raise ValueError( f"event '{self.id}' declares a recipient route but is not emittable " "(no producer) -- nothing would ever trigger routing for it" ) return self class ProvidedCapability(BaseModel): model_config = ConfigDict(extra="forbid") id: str version: int = 1 @model_validator(mode="before") @classmethod def coerce_string(cls, data: Any) -> Any: if isinstance(data, str): return {"id": data, "version": 1} return data class PackEntityContract(BaseModel): """Entity-level policy only. Field flags live on schema.fields[].""" model_config = ConfigDict(extra="forbid") entity_key: str schema_mode: SchemaMode = "extensible" workflow_configurable: bool = False class PackSecurity(BaseModel): """Backend enforcement contract. App Blocks may project this into UI; they must not treat it as the security boundary.""" model_config = ConfigDict(extra="forbid") public_actions: list[str] = Field(default_factory=list) authenticated_actions: list[str] = Field(default_factory=list) required_roles: list[str] = Field(default_factory=list) ownership_mode: OwnershipMode | None = None rls_required: bool = False authenticated: bool = False PackCapabilityLifecycle = Literal[ "stable", "preview", "internal", "reserved", "compatibility", "deprecated" ] PackCapabilityReadiness = Literal["implemented", "partial", "unimplemented"] class PackPlannerMeta(BaseModel): model_config = ConfigDict(extra="forbid") selectable: bool = True preferred_for_new_projects: bool = True priority: int = 100 lifecycle: PackCapabilityLifecycle = "stable" readiness: PackCapabilityReadiness = "implemented" class PackWebhookBinding(BaseModel): model_config = ConfigDict(extra="forbid") provider: str events: list[str] = Field(default_factory=list) target: str forward_path: str protected: bool = True EntityIngestOperation = Literal["upsert"] class PackEntityIngestBinding(BaseModel): """Declares that a verified webhook on this pack may mutate one of the pack's own entities — the ``external.entity_ingest`` primitive (verified webhook -> bounded payload mapping -> idempotent entity mutation). Deliberately does NOT carry a field-mapping list: which payload fields may reach the entity is already governed by the existing per-field ``public_create`` flag (the same mechanism Forms' public submission uses). This binding only adds the two things that mechanism doesn't already express: which webhook feeds this entity, and how a repeat delivery of the same external object is recognized so it updates instead of duplicating. """ model_config = ConfigDict(extra="forbid") entity_key: str webhook_provider: str operation: EntityIngestOperation = "upsert" identity_fields: list[str] = Field(default_factory=lambda: ["source", "external_id"]) default_status_field: str | None = None default_status_value: str | None = None @model_validator(mode="after") def status_default_is_paired(self) -> PackEntityIngestBinding: if bool(self.default_status_field) != bool(self.default_status_value): raise ValueError( "default_status_field and default_status_value must be set together" ) return self class PackCompatibility(BaseModel): model_config = ConfigDict(extra="forbid") contract_version: Literal[2] = 2 implementation_version: str | None = None legacy_aggregate: bool = False implementation_notes: str | None = None class PackWorkflowContract(BaseModel): model_config = ConfigDict(extra="forbid") entity_key: str field: str configurable: bool = True protected: bool = True transitions: dict[str, list[str]] = Field(default_factory=dict) def validate_workflow_transitions( field_type: str | None, options: list[str] | None, transitions: dict[str, list[str]], ) -> dict[str, list[str]]: """The single canonical structural check a workflow transition graph must pass before it reaches schema.workflow, applied identically by both authoring paths: Level-1 synthesized packs (composer.py's derive_pack_composition) and hand- authored catalog_v2/migration-seeded packs (normalize.py's _apply_workflows_to_entities). Cloud independently re-validates from/to states against field.options at schema-push time (records.ts) as a runtime backstop -- this is the same rule, enforced earlier, so a malformed graph fails loudly in Core instead of silently at Cloud's schema-push boundary. Returns the transitions graph with duplicate targets removed (first-occurrence order preserved) -- duplicates carry no enforcement meaning (Cloud only checks `transitions[from].includes(to)`) but must not leak non-deterministically into the serialized schema pushed to Cloud. """ if field_type != "select": raise ValueError(f"workflow field must be a select field, got '{field_type}'") allowed = set(options or []) normalized: dict[str, list[str]] = {} for from_state, to_states in transitions.items(): if from_state not in allowed: raise ValueError( f"workflow transition source '{from_state}' is not one of the field's options" ) deduped: list[str] = [] seen: set[str] = set() for to_state in to_states: if to_state not in allowed: raise ValueError( f"workflow transition target '{to_state}' is not one of the field's options" ) if to_state not in seen: seen.add(to_state) deduped.append(to_state) normalized[from_state] = deduped return normalized class FeaturePackContract(BaseModel): """Normalized pack surface for install, App Block planning, and manifests.""" model_config = ConfigDict(extra="forbid") slug: str contract_version: Literal[2] = 2 provides: list[ProvidedCapability] = Field(default_factory=list) requires: list[PackDependencyRef] = Field(default_factory=list) optional_requires: list[PackDependencyRef] = Field(default_factory=list) provider_bindings: list[PackProviderBinding] = Field(default_factory=list) events: list[PackEvent] = Field(default_factory=list) entities: list[PackEntityContract] = Field(default_factory=list) calculators: list[CalculationIR] = Field(default_factory=list) workflows: list[PackWorkflowContract] = Field(default_factory=list) security: PackSecurity = Field(default_factory=PackSecurity) compatibility: PackCompatibility = Field(default_factory=PackCompatibility) planner: PackPlannerMeta = Field(default_factory=PackPlannerMeta) webhook_bindings: list[PackWebhookBinding] = Field(default_factory=list) entity_ingest_bindings: list[PackEntityIngestBinding] = Field(default_factory=list) def provide_ids(self) -> list[str]: return [item.id for item in self.provides] def required_pack_slugs(self) -> list[str]: return [dep.id for dep in self.requires if dep.kind == "feature_pack"] def required_primitive_ids(self) -> list[str]: return [dep.id for dep in self.requires if dep.kind == "primitive"] def required_capabilities(self) -> tuple[CapabilityDependency, ...]: """Capability-level dependencies (distinct from required_pack_slugs/ required_primitive_ids, which stay at the pack/Cloud-primitive level) -- e.g. a pack declaring `{"kind": "capability", "id": "auth", "min_version": "2"}` to require the platform Auth capability at version >= 2. Feeds directly into src.domain.capabilities.dependencies.resolve_dependencies. """ from src.domain.capabilities.dependencies import CapabilityDependency deps: list[CapabilityDependency] = [] for dep in self.requires: if dep.kind == "capability": deps.append( CapabilityDependency( key=dep.id, required=True, minimum_version=int(dep.min_version) if dep.min_version else None, ) ) for dep in self.optional_requires: if dep.kind == "capability": deps.append( CapabilityDependency( key=dep.id, required=False, minimum_version=int(dep.min_version) if dep.min_version else None, ) ) return tuple(deps) def configuration_providers(self) -> list[str]: keys: list[str] = [] seen: set[str] = set() for binding in self.provider_bindings: if not binding.configuration_required: continue for provider in binding.providers: if provider not in seen: seen.add(provider) keys.append(provider) return keys @model_validator(mode="after") def entity_ingest_bindings_reference_declared_entities_and_webhooks( self, ) -> FeaturePackContract: if not self.entity_ingest_bindings: return self entity_keys = {item.entity_key for item in self.entities} webhook_providers = {item.provider for item in self.webhook_bindings} for binding in self.entity_ingest_bindings: if binding.entity_key not in entity_keys: raise ValueError( f"entity_ingest_binding targets undeclared entity '{binding.entity_key}'" ) if binding.webhook_provider not in webhook_providers: raise ValueError( f"entity_ingest_binding webhook_provider '{binding.webhook_provider}' " "has no matching webhook_bindings entry on this pack" ) return self def to_composition_fields(self) -> dict[str, Any]: return { "schema_version": FEATURE_PACK_CONTRACT_VERSION, "provides": [item.model_dump() for item in self.provides], "requires": [dep.model_dump() for dep in self.requires], "optional_requires": [dep.model_dump() for dep in self.optional_requires], "provider_bindings": [ item.model_dump() for item in self.provider_bindings ], "events": [item.model_dump() for item in self.events], "entity_contracts": [item.model_dump() for item in self.entities], "calculators": [item.model_dump(by_alias=True) for item in self.calculators], "workflows": [item.model_dump() for item in self.workflows], "security": self.security.model_dump(), "compatibility": self.compatibility.model_dump(), "planner": self.planner.model_dump(), "webhook_bindings": [item.model_dump() for item in self.webhook_bindings], "entity_ingest_bindings": [ item.model_dump() for item in self.entity_ingest_bindings ], "connector_actions_role": "resolved_implementation", } def classify_legacy_token(token: str) -> PackDependencyRef | PackProviderBinding: """Split overloaded string ``requires`` into a typed dep or provider binding.""" value = token.strip() if value in RESOLVABLE_PRIMITIVES or ( "." in value and value not in KNOWN_PROVIDERS ): return PackDependencyRef(kind="primitive", id=value) if value in KNOWN_PROVIDERS: return PackProviderBinding( capability=KNOWN_PROVIDERS[value], providers=[value], configuration_required=True, missing_configuration_behavior="degraded", ) return PackDependencyRef(kind="feature_pack", id=value) def typed_requires_from_primitives(ids: list[str]) -> list[dict[str, str]]: return [{"kind": "primitive", "id": item} for item in ids]