"""Typed GenerationPlan.v1 contract and deterministic Discovery compiler. This module is deliberately free of database and provider side effects. It turns the already-approved Discovery result and App Block install plan into a single private artifact that later orchestration stages can consume. """ from __future__ import annotations import hashlib import json from collections.abc import Sequence from datetime import datetime from typing import Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field from src.domain.app_blocks.catalogs import ( canonical_feature_pack, default_known_packs, default_primitives, ) from src.domain.app_blocks.job_packs import known_job_pack_ids from src.domain.app_blocks.registry import AppBlockRegistry, load_default_registry from src.domain.app_blocks.schemas import AppBlockGap, AppBlockPlan from src.domain.codegen.application_experience import ( ApplicationExperience, compile_application_experience, empty_application_experience, ) from src.domain.forms.canonical import FormDefinition from src.domain.projects.discovery_catalog import CLOUD_CAPABILITIES from src.domain.projects.discovery_schemas import DiscoveryResult from src.domain.projects.form_requirements import FormRequirement from src.domain.starter_library.archetypes import RECIPE_STARTER_IDS, known_archetype_ids from src.domain.starter_library.catalog import known_contract_ids, known_proposed_ids PlanGapKind = Literal[ "composition_gap", "configuration_gap", "provider_gap", "semantic_gap", ] RequirementKind = Literal[ "native", "credential_configuration", "external_provider_required", "external_system_dependency", "unimplemented_external_semantics", ] class GenerationPlanMetadata(BaseModel): model_config = ConfigDict(extra="forbid") schema_version: Literal["1"] = "1" project_id: UUID | None = None source_discovery_revision: int | None = Field(default=None, ge=0) catalog_digests: dict[str, str] = Field(default_factory=dict) class GenerationPlanProvenance(BaseModel): model_config = ConfigDict(extra="forbid") schema_version: Literal["generation_plan.v1"] = "generation_plan.v1" compiler_version: str = "generation-plan-compiler.v1" app_block_registry_version: str = "1" feature_pack_catalog_version: str = "2" starter_catalog_version: str = "1" cloud_capability_catalog_version: str = "2" recipe_catalog_version: str = "1" source_discovery_revision: int | None = Field(default=None, ge=0) source_app_block_plan_hash: str | None = None compiled_at: datetime | None = None plan_hash: str | None = None class GenerationIntent(BaseModel): model_config = ConfigDict(extra="forbid") product_name: str = "" purpose: str = "" first_build_goal: str = "" original_intent: str = "" domain: str = "" requested_product_shape: str = "" summary: str = "" class GenerationActor(BaseModel): model_config = ConfigDict(extra="forbid") id: str name: str access_scope: str | None = None class GenerationWorkflow(BaseModel): model_config = ConfigDict(extra="forbid") id: str name: str description: str outcome: str | None = None steps: list[str] = Field(default_factory=list) class GenerationEntity(BaseModel): model_config = ConfigDict(extra="forbid") id: str name: str description: str ownership: str | None = None lifecycle: list[str] = Field(default_factory=list) class GenerationBusinessRule(BaseModel): model_config = ConfigDict(extra="forbid") id: str rule: str related_entities: list[str] = Field(default_factory=list) related_actors: list[str] = Field(default_factory=list) class GenerationComposition(BaseModel): model_config = ConfigDict(extra="forbid") starter_id: str | None = None archetype_id: str | None = None recipe_ids: list[str] = Field(default_factory=list) recipe_versions: dict[str, int] = Field(default_factory=dict) feature_pack_ids: list[str] = Field(default_factory=list) app_block_ids: list[str] = Field(default_factory=list) primitive_ids: list[str] = Field(default_factory=list) cloud_capability_ids: list[str] = Field(default_factory=list) # Resolved, renderer-neutral form schemas selected by composition. Keeping # these in the completed plan is what lets the Core-owned runtime projection # agree with generated UI on incremental codegen runs. form_definitions: dict[str, FormDefinition] = Field(default_factory=dict) class ProviderRequirement(BaseModel): model_config = ConfigDict(extra="forbid") id: str kind: RequirementKind required: bool = True providers: list[str] = Field(default_factory=list) source: str degraded_path: str | None = None class GenerationRuntime(BaseModel): model_config = ConfigDict(extra="forbid") starter_contract_ids: list[str] = Field(default_factory=list) runtime_dependencies: list[str] = Field(default_factory=list) authentication_required: bool = False tenancy_model: str | None = None forms: list[str] = Field(default_factory=list) notifications: list[str] = Field(default_factory=list) auth_contract: GenerationAuthContract | None = None class GenerationAuthAudience(BaseModel): """Exact auth operations admitted for one generated-app audience.""" model_config = ConfigDict(extra="forbid") audience: Literal["staff", "customer"] methods: list[str] = Field(default_factory=list) signup_allowed: bool = False password_setup_allowed: bool = False password_reset_allowed: bool = False login_route: str post_auth_route: str class GenerationAuthContract(BaseModel): model_config = ConfigDict(extra="forbid") contract_version: str = "auth.v1" audiences: list[GenerationAuthAudience] = Field(default_factory=list) def methods_for(self, audience: str) -> set[str]: for item in self.audiences: if item.audience == audience: return set(item.methods) return set() class GenerationConstraints(BaseModel): model_config = ConfigDict(extra="forbid") required_ids: list[str] = Field(default_factory=list) forbidden_capabilities: list[str] = Field(default_factory=list) forbidden_implementations: list[str] = Field(default_factory=list) security_constraints: list[str] = Field(default_factory=list) compatibility_constraints: list[str] = Field(default_factory=list) class GenerationValidation(BaseModel): model_config = ConfigDict(extra="forbid") required_contracts: list[str] = Field(default_factory=list) smoke_flows: list[str] = Field(default_factory=list) conformance_checks: list[str] = Field(default_factory=list) class ClassifiedGenerationGap(BaseModel): model_config = ConfigDict(extra="forbid") id: str kind: PlanGapKind severity: str message: str affects: list[str] = Field(default_factory=list) prohibited_codegen_actions: list[str] = Field(default_factory=list) degraded_path: str | None = None class GenerationGaps(BaseModel): model_config = ConfigDict(extra="forbid") unsupported_semantics: list[ClassifiedGenerationGap] = Field(default_factory=list) deferred_scope: list[str] = Field(default_factory=list) open_decisions: list[str] = Field(default_factory=list) def _canonical_auth_method(value: str) -> str: method = str(value).strip().lower() if method in {"passwordless", "magic-link", "magiclink", "email"}: return "magic_link" return method def _auth_contract(result: DiscoveryResult) -> GenerationAuthContract | None: spec = result.spec.authentication if not spec or not spec.required: return None raw_audiences = [str(value).strip().lower() for value in spec.audiences] audiences = [value for value in raw_audiences if value in {"staff", "customer"}] if not audiences: audiences = ["staff"] for value in spec.audience_methods: normalized = str(value).strip().lower() if normalized in {"staff", "customer"} and normalized not in audiences: audiences.append(normalized) global_methods = list(dict.fromkeys(_canonical_auth_method(value) for value in spec.methods)) per_audience = { str(audience).strip().lower(): list( dict.fromkeys(_canonical_auth_method(value) for value in methods) ) for audience, methods in spec.audience_methods.items() if str(audience).strip().lower() in {"staff", "customer"} } signup_audiences = { str(value).strip().lower() for value in spec.signup_audiences if str(value).strip().lower() in {"staff", "customer"} } contracts: list[GenerationAuthAudience] = [] for audience in audiences: methods = list(dict.fromkeys(per_audience.get(audience, global_methods))) password = "password" in methods contracts.append( GenerationAuthAudience( audience=audience, methods=methods, signup_allowed=audience in signup_audiences, password_setup_allowed=password, password_reset_allowed=password, login_route="/admin/login" if audience == "staff" else "/login", post_auth_route="/admin" if audience == "staff" else "/", ) ) return GenerationAuthContract(audiences=contracts) def _form_definitions_from_requirements( requirements: Sequence[FormRequirement], ) -> dict[str, FormDefinition]: """Project Core-resolved Discovery form requirements into the typed Forms contract GenerationPlan carries forward. ``requirements`` is the full, already-merged per-form field state (``DiscoverySpec.form_requirements`` -- see ``src.domain.projects.form_requirements``), never a per-run delta and never an ``AppBlock`` slot or agent-authored ``p2hs/feature-packs.json`` entry: fields are a runtime data-contract requirement, not presentation customization, and generated files never become authoritative. A requirement with no fields yet is omitted so the Forms runtime keeps using its historical canonical definition instead of a degenerate empty schema. """ definitions: dict[str, FormDefinition] = {} for requirement in requirements: if not requirement.fields: continue definitions[requirement.form_key] = FormDefinition( key=requirement.form_key, fields=list(requirement.fields) ) return definitions def _form_composition_from_definitions( definitions: dict[str, FormDefinition], ) -> tuple[set[str], set[str]]: """Return the canonical composition implied by resolved public forms. Form field requirements are persisted independently from App Block selection so a later prompt can extend a form on an already-created project. A resolved definition is nevertheless proof that the corresponding canonical form runtime is part of the product, and therefore must select its owning App Block and Feature Pack as well. Without this bridge a generated app can contain the generic ``/api/forms/{formKey}`` adapter while Core has never installed a concrete form binding. Unknown form keys remain project-local data and are deliberately not promoted into platform composition here. """ block_by_form_key = { "contact": ("forms.contact", "contact-capture"), "quote_request": ("forms.quote_request", "contact-capture"), } blocks: set[str] = set() packs: set[str] = set() for form_key in definitions: selected = block_by_form_key.get(form_key) if selected is None: continue block_id, pack_id = selected blocks.add(block_id) packs.add(pack_id) return blocks, packs class GenerationProductModel(BaseModel): model_config = ConfigDict(extra="forbid") actors: list[GenerationActor] = Field(default_factory=list) workflows: list[GenerationWorkflow] = Field(default_factory=list) entities: list[GenerationEntity] = Field(default_factory=list) business_rules: list[GenerationBusinessRule] = Field(default_factory=list) acceptance_outcomes: list[str] = Field(default_factory=list) class GenerationPlanV1(BaseModel): """Private, reproducible composition contract for one generation.""" model_config = ConfigDict(extra="forbid") metadata: GenerationPlanMetadata provenance: GenerationPlanProvenance intent: GenerationIntent composition: GenerationComposition product_model: GenerationProductModel experience: ApplicationExperience = Field(default_factory=empty_application_experience) runtime: GenerationRuntime provider_requirements: list[ProviderRequirement] = Field(default_factory=list) constraints: GenerationConstraints validation: GenerationValidation gaps: GenerationGaps def compile_generation_plan( discovery: DiscoveryResult | dict, app_block_plan: AppBlockPlan | dict, *, project_id: UUID | None = None, source_discovery_revision: int | None = None, catalog_digests: dict[str, str] | None = None, compiled_at: datetime | None = None, validate_catalog: bool = True, ) -> GenerationPlanV1: """Compile a stable plan without LLM, database, or provider access.""" result = ( discovery if isinstance(discovery, DiscoveryResult) else DiscoveryResult.model_validate(discovery) ) plan = ( app_block_plan if isinstance(app_block_plan, AppBlockPlan) else AppBlockPlan.model_validate(app_block_plan) ) registry = load_default_registry() requested_block_ids = { capability for capability in plan.requested_capabilities if registry.get(capability) is not None } form_definitions = _form_definitions_from_requirements(result.spec.form_requirements) form_blocks, form_packs = _form_composition_from_definitions(form_definitions) blocks = sorted({block.id for block in plan.blocks} | requested_block_ids | form_blocks) feature_packs = sorted( set(plan.required_feature_packs) | set(plan.feature_pack_ids) | form_packs ) primitives = sorted(set(plan.required_primitives)) cloud_capabilities = sorted( set(plan.requested_capabilities) - requested_block_ids ) starter_contracts = sorted(set(plan.starter_contracts)) provider_requirements = _provider_requirements(result, plan) classified_gaps = [_classify_gap(gap) for gap in plan.gaps] required_ids = sorted( set(blocks) | set(feature_packs) | set(primitives) | set(starter_contracts) | set(cloud_capabilities) ) security_constraints = [ "preserve workspace and project ownership boundaries", "do not expose provider credentials to browser code", ] if result.spec.authentication and result.spec.authentication.required: security_constraints.append("enforce the approved authentication contract") generation_plan = GenerationPlanV1( metadata=GenerationPlanMetadata( project_id=project_id, source_discovery_revision=source_discovery_revision, catalog_digests=dict(catalog_digests or {}), ), provenance=GenerationPlanProvenance( source_discovery_revision=source_discovery_revision, source_app_block_plan_hash=_model_hash(plan), compiled_at=compiled_at, ), intent=GenerationIntent( product_name=result.product_name or (result.spec.product.name or ""), purpose=result.spec.product.purpose, first_build_goal=result.spec.product.first_build_goal, original_intent=result.spec.original_intent, domain=result.spec.domain, requested_product_shape=result.spec.requested_product_shape, summary=result.summary, ), composition=GenerationComposition( starter_id=plan.starter_id, archetype_id=plan.product_archetype, recipe_ids=sorted( set(plan.collection_recipe_ids) | ({plan.matched_recipe_id} if plan.matched_recipe_id else set()) ), recipe_versions=( {plan.matched_recipe_id: plan.composition_recipe_version} if plan.matched_recipe_id and plan.composition_recipe_version else {} ), feature_pack_ids=feature_packs, app_block_ids=blocks, primitive_ids=primitives, cloud_capability_ids=cloud_capabilities, form_definitions=form_definitions, ), product_model=GenerationProductModel( actors=[ GenerationActor( id=actor.id, name=actor.name, access_scope=actor.access_scope, ) for actor in result.spec.actors ], workflows=[ GenerationWorkflow( id=workflow.id, name=workflow.name, description=workflow.description, outcome=workflow.outcome, steps=[step.name for step in workflow.steps], ) for workflow in result.spec.workflows ], entities=[ GenerationEntity( id=entity.id, name=entity.name, description=entity.description, ownership=entity.ownership, lifecycle=list(entity.lifecycle), ) for entity in result.spec.entities ], business_rules=[ GenerationBusinessRule( id=rule.id, rule=rule.rule, related_entities=list(rule.related_entities), related_actors=list(rule.related_actors), ) for rule in result.spec.business_rules ], acceptance_outcomes=list(result.spec.acceptance_outcomes), ), experience=compile_application_experience( domain=result.spec.domain, archetype_id=plan.product_archetype, purpose=result.spec.product.purpose, app_block_ids=blocks, feature_pack_ids=feature_packs, authentication_required=bool( result.spec.authentication and result.spec.authentication.required ), ), runtime=GenerationRuntime( starter_contract_ids=starter_contracts, runtime_dependencies=sorted(set(plan.runtime_dependencies)), authentication_required=bool( result.spec.authentication and result.spec.authentication.required ), tenancy_model=result.spec.tenancy.model if result.spec.tenancy else None, forms=list(result.spec.forms), notifications=[item.trigger for item in result.spec.notifications], auth_contract=_auth_contract(result), ), provider_requirements=provider_requirements, constraints=GenerationConstraints( required_ids=required_ids, forbidden_capabilities=sorted(set(plan.forbidden_capabilities)), forbidden_implementations=[ "invent_private_backend_for_unimplemented_capability", "expose_provider_credentials_to_browser", ], security_constraints=security_constraints, compatibility_constraints=[ "use admitted Starter and App Block contracts", "preserve Cloud capability boundaries", ], ), validation=GenerationValidation( required_contracts=starter_contracts, smoke_flows=[workflow.name for workflow in result.spec.workflows], conformance_checks=[ "selected composition contracts are present", "workspace and project authorization is preserved", "unsupported semantics are not fabricated", ], ), gaps=GenerationGaps( unsupported_semantics=classified_gaps, deferred_scope=list(result.spec.deferred_scope), open_decisions=[decision.id for decision in result.spec.open_decisions], ), ) if validate_catalog: validate_generation_plan_catalog(generation_plan) generation_plan.provenance.plan_hash = _plan_hash(generation_plan) return generation_plan class GenerationPlanCatalogError(ValueError): """Raised when a plan references a non-admitted platform identifier.""" def validate_generation_plan_catalog( plan: GenerationPlanV1, *, registry: AppBlockRegistry | None = None, known_packs: frozenset[str] | None = None, known_primitives: frozenset[str] | None = None, ) -> None: """Validate plan references against the canonical in-memory catalogs.""" registry = registry or load_default_registry() known_packs = ( known_packs if known_packs is not None else default_known_packs() | known_job_pack_ids() ) known_primitives = ( known_primitives if known_primitives is not None else default_primitives() ) errors: list[str] = [] for block_id in plan.composition.app_block_ids: block = registry.get(block_id) if block is None: errors.append(f"unknown_app_block:{block_id}") elif not block.implemented: errors.append(f"non_implemented_app_block:{block_id}") for pack_id in plan.composition.feature_pack_ids: if canonical_feature_pack(pack_id) not in known_packs: errors.append(f"unknown_feature_pack:{pack_id}") for primitive_id in plan.composition.primitive_ids: if primitive_id not in known_primitives: errors.append(f"unknown_primitive:{primitive_id}") for contract_id in plan.runtime.starter_contract_ids: if contract_id in known_proposed_ids(): errors.append(f"proposed_starter_contract:{contract_id}") elif contract_id not in known_contract_ids(): errors.append(f"unknown_starter_contract:{contract_id}") if plan.composition.archetype_id and plan.composition.archetype_id not in known_archetype_ids(): errors.append(f"unknown_archetype:{plan.composition.archetype_id}") for recipe_id in plan.composition.recipe_ids: if ( recipe_id not in RECIPE_STARTER_IDS and recipe_id not in plan.composition.recipe_versions ): errors.append(f"unknown_recipe:{recipe_id}") if plan.composition.starter_id and plan.composition.starter_id not in known_archetype_ids(): errors.append(f"unknown_starter:{plan.composition.starter_id}") known_capabilities = set(CLOUD_CAPABILITIES) known_capabilities.update( capability.id for block in registry.all() for capability in block.provides ) for capability_id in plan.composition.cloud_capability_ids: if capability_id not in known_capabilities: errors.append(f"unknown_cloud_capability:{capability_id}") provider_kinds: dict[str, set[RequirementKind]] = {} for requirement in plan.provider_requirements: provider_kinds.setdefault(requirement.id, set()).add(requirement.kind) for capability_id, kinds in provider_kinds.items(): if "native" in kinds and len(kinds) > 1: errors.append(f"provider_promoted_native:{capability_id}") if errors: raise GenerationPlanCatalogError("; ".join(sorted(set(errors)))) def _model_hash(model: BaseModel) -> str: payload = model.model_dump(mode="json") return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def _plan_hash(plan: GenerationPlanV1) -> str: payload = plan.model_dump(mode="json", exclude={"provenance": {"plan_hash"}}) return hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def _provider_requirements( discovery: DiscoveryResult, plan: AppBlockPlan ) -> list[ProviderRequirement]: requirements: dict[tuple[str, RequirementKind], ProviderRequirement] = {} for item in plan.configuration_requirements: kind: RequirementKind = "credential_configuration" key = (item.capability, kind) requirements[key] = ProviderRequirement( id=item.capability, kind=kind, providers=sorted(set(item.providers)), source="app_block_plan.configuration_requirements", degraded_path=item.behavior if item.behavior != "block_install" else None, ) for integration in discovery.spec.integrations: kind: RequirementKind = ( "external_provider_required" if integration.provider else "external_system_dependency" ) key = (integration.id, kind) requirements[key] = ProviderRequirement( id=integration.id, kind=kind, required=integration.required_for_first_build, providers=[integration.provider] if integration.provider else [], source="discovery.integrations", degraded_path="configuration_required" if not integration.required_for_first_build else None, ) for capability in plan.requested_capabilities: key = (capability, "native") requirements.setdefault( key, ProviderRequirement( id=capability, kind="native", source="app_block_plan.requested_capabilities", ), ) return [requirements[key] for key in sorted(requirements)] def _classify_gap(gap: AppBlockGap) -> ClassifiedGenerationGap: kind: PlanGapKind if gap.type == "configuration_required": kind = "configuration_gap" elif gap.type == "unsupported": kind = "semantic_gap" elif gap.type in {"feature_pack_missing", "primitive_missing"}: kind = "composition_gap" else: kind = "composition_gap" return ClassifiedGenerationGap( id=gap.id, kind=kind, severity=gap.severity, message=gap.message, affects=list(gap.affects), prohibited_codegen_actions=( [ "invent_backend_endpoint", "invent_cloud_capability", "represent_as_native_capability", ] if kind == "semantic_gap" else [] ), degraded_path="available" if gap.severity == "optional" else None, ) __all__ = [ "GenerationPlanCatalogError", "GenerationPlanV1", "compile_generation_plan", "validate_generation_plan_catalog", ]