"""Release a governed DataFlow as an immutable data production line.""" from __future__ import annotations from typing import Any from app.core.common.identifiers import ensure_governance_uid, new_governance_uid from app.core.data_rules.compiler import compile_rule_plan from app.core.data_rules.contracts import read_rule_spec, validate_dataflow_spec from app.core.data_rules.expressions import validate_rule_expressions from app.core.data_rules.production_line import resolve_production_line def _uid(value: Any, label: str) -> str: try: return ensure_governance_uid({"uid": str(value)}) except ValueError as exc: raise ValueError(f"{label} must be a valid UUIDv7") from exc def _source(value: Any) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError("source_text is required") normalized = value.strip() if len(normalized) > 20_000: raise ValueError("source_text exceeds 20000 characters") return normalized class ProductionLineReleaseService: def __init__( self, repository, *, schema_resolver=None, available_plan_backends=None, ): self.repository = repository self.schema_resolver = schema_resolver self.available_plan_backends = frozenset( available_plan_backends if available_plan_backends is not None else {"sql_pushdown", "polars_batch", "quality_check"} ) def release( self, *, dataflow_uid: str, dataflow_spec: dict[str, Any], source_text: str, created_by: str, ) -> dict[str, Any]: path_uid = _uid(dataflow_uid, "dataflow_uid") actor = _uid(created_by, "created_by") flow = validate_dataflow_spec(dataflow_spec) if flow["dataflow_uid"] != path_uid: raise ValueError("dataflow spec uid does not match path dataflow_uid") source = _source(source_text) if self.schema_resolver is None: raise ValueError("trusted schema resolver is not configured") input_snapshots = { ref: self.schema_resolver.resolve(ref) for ref in flow["input_schema_refs"] } output_snapshot = self.schema_resolver.resolve(flow["output_schema_ref"]) inputs = { ref: snapshot["schema_hash"] for ref, snapshot in sorted(input_snapshots.items()) } output = output_snapshot["schema_hash"] standards, rules = self.repository.load_published_assets(flow) rule_backend_support: dict[str, frozenset[str]] = {} def validate_published_rule(rule_version_id: str) -> None: if rule_version_id in rule_backend_support: return rule = rules.get(rule_version_id) if not isinstance(rule, dict): raise ValueError( f"published rule version {rule_version_id} was not found" ) rule_spec = read_rule_spec(rule.get("rule_spec")) rule_snapshot = self.schema_resolver.resolve( rule_spec["input_schema_ref"] ) snapshot_fields = { field["name"]: field["type"] for field in rule_snapshot["fields"] } supported = validate_rule_expressions( rule_spec, snapshot_fields ) if not supported: raise ValueError("rule has no supported execution backend") rule_backend_support[rule_version_id] = supported # Validate every loaded rule before allocating any release version. # The repository is expected to scope this mapping to published assets # available to the release, so no loaded expression is left unchecked. for loaded_rule_version_id in rules: validate_published_rule(loaded_rule_version_id) # Ensure referenced standards and their clauses resolve to loaded, # already-validated rule versions before beginning the release. for component in flow["components"]: if component["type"] == "standard.enforce": standard = standards.get(component["standard_version_id"]) if not isinstance(standard, dict): raise ValueError( "published standard version " f"{component['standard_version_id']} was not found" ) for clause in standard["clauses"]: validate_published_rule(str(clause["rule_version_id"])) else: validate_published_rule(component["rule_version_id"]) compiled: dict[str, dict[str, Any]] = {} for rule_version_id, supported_backends in sorted( rule_backend_support.items() ): plan = compile_rule_plan( rules[rule_version_id], supported_backends=supported_backends, ) if plan["backend"] not in self.available_plan_backends: raise ValueError( f"plan backend {plan['backend']} is not registered" ) compiled[rule_version_id] = plan version = self.repository.begin_dataflow_release( dataflow_spec=flow, source_text=source, input_schema_hashes=inputs, output_schema_hash=output, created_by=actor, ) version_id = _uid(version.get("id"), "dataflow_version_id") binding_ids: dict[str, str] = {} def add_binding( *, binding_key: str, component_id: str, component_kind: str, rule_version_id: str, stage: str, order_no: int, idempotency: dict[str, Any] | None, provenance: dict[str, str], ) -> None: rule = rules.get(rule_version_id) if not isinstance(rule, dict): raise ValueError( f"published rule version {rule_version_id} was not found" ) plan = compiled[rule_version_id] binding_id = new_governance_uid() binding_ids[binding_key] = binding_id self.repository.persist_component_plan( dataflow_version_id=version_id, component_binding_id=binding_id, component_id=component_id, component_kind=component_kind, rule_version_id=rule_version_id, stage=stage, order_no=order_no, idempotency=idempotency, provenance=provenance, plan=plan, schema_hashes={ "inputs": inputs, "output": output, }, ) for component in flow["components"]: if component["type"] == "standard.enforce": standard_id = component["standard_version_id"] standard = standards.get(standard_id) if not isinstance(standard, dict): raise ValueError( f"published standard version {standard_id} was not found" ) for clause_index, clause in enumerate(standard["clauses"]): clause_id = str(clause["clause_id"]) binding_key = f"{component['id']}:{clause_id}" add_binding( binding_key=binding_key, component_id=(f"{component['id']}__{clause_id}"[:100]), component_kind="quality.check", rule_version_id=str(clause["rule_version_id"]), stage=component["stage"], order_no=(component["order"] * 1000) + clause_index, idempotency=None, provenance={ "standard_version_id": standard_id, "clause_id": clause_id, }, ) continue add_binding( binding_key=component["id"], component_id=component["id"], component_kind=component["type"], rule_version_id=component["rule_version_id"], stage=component["stage"], order_no=component["order"] * 1000, idempotency=component.get("idempotency"), provenance={}, ) release_rules = { rule_id: { **rule, "execution_plan": { "backend": compiled[rule_id]["backend"], "plan_hash": compiled[rule_id]["plan_hash"], }, } for rule_id, rule in rules.items() if rule_id in compiled } package = resolve_production_line( flow, standards, release_rules, component_binding_ids=binding_ids, ) return self.repository.complete_dataflow_release( dataflow_version_id=version_id, package=package, ) class BoundSqlPlanService: """Compile and persist a physical plan only after deployment bindings exist. Compilation is deliberately persisted as ``compiled``. Test evidence and publication transitions remain separate lifecycle actions (Task 7), so this vertical slice never labels an unexecuted plan tested or published. """ def __init__(self, repository, compiler_registry): self.repository = repository self.compiler_registry = compiler_registry def compile_and_persist( self, *, component_binding_id: str, rule_version_id: str, input_schema_snapshot_id: str, output_schema_snapshot_id: str, input_binding_id: str, output_binding_id: str, ) -> dict[str, Any]: ids = { "component_binding_id": _uid( component_binding_id, "component_binding_id" ), "rule_version_id": _uid(rule_version_id, "rule_version_id"), "input_schema_snapshot_id": _uid( input_schema_snapshot_id, "input_schema_snapshot_id" ), "output_schema_snapshot_id": _uid( output_schema_snapshot_id, "output_schema_snapshot_id" ), "input_binding_id": _uid(input_binding_id, "input_binding_id"), "output_binding_id": _uid(output_binding_id, "output_binding_id"), } context = self.repository.load_bound_compile_context(**ids) if not isinstance(context, dict): raise ValueError("canonical bound compile context was not found") try: component = context["component_binding"] rule_version = context["rule_version"] input_schema = context["input_schema"] output_schema = context["output_schema"] input_binding = context["input_binding"] output_binding = context["output_binding"] backend = context["backend"] except KeyError as exc: raise ValueError( "canonical bound compile context is incomplete" ) from exc canonical_ids = { "component_binding_id": component.get("id"), "rule_version_id": rule_version.get("id"), "input_schema_snapshot_id": input_schema.get("id"), "output_schema_snapshot_id": output_schema.get("id"), "input_binding_id": input_binding.get("id"), "output_binding_id": output_binding.get("id"), } if canonical_ids != ids or component.get( "rule_version_id" ) != ids["rule_version_id"]: raise ValueError( "canonical bound compile context identifiers do not match" ) compiler = self.compiler_registry.select( rule_version.get("rule_spec"), input_binding, output_binding, ) compiled = compiler.compile( rule_version=rule_version, input_schema=input_schema, output_schema=output_schema, input_binding=input_binding, output_binding=output_binding, backend=backend, ) return self.repository.persist_bound_component_plan( component_binding_id=ids["component_binding_id"], rule_version_id=ids["rule_version_id"], input_binding_id=ids["input_binding_id"], output_binding_id=ids["output_binding_id"], compiled=compiled, status="compiled", )