From c981cfd81e05efc4a5dc2fb63dac0e75253f5e53 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Thu, 30 Jul 2026 22:07:07 +0800 Subject: [PATCH] feat: reconcile active metadata provider binding --- .github/workflows/ci.yml | 10 + ..._active_metadata_binding_reconciliation.py | 1753 +++++++++++++++++ .../metadata_fabric_binding_contract.py | 15 +- data_agent/platform_truth.py | 20 + ..._active_metadata_binding_reconciliation.py | 468 +++++ ...etadata_binding_reconciliation_postgres.py | 157 ++ data_agent/test_platform_truth.py | 6 + ...-active-metadata-binding-reconciliation.md | 78 + ...ata-binding-reconciliation-2026-07-30.json | 296 +++ docs/roadmap-ar0-platform-truth-2026-07-24.md | 5 +- docs/system-of-record-matrix-2026-07-24.md | 10 +- ...-active-metadata-binding-reconciliation.sh | 22 + 12 files changed, 2832 insertions(+), 8 deletions(-) create mode 100644 data_agent/metadata_fabric_active_metadata_binding_reconciliation.py create mode 100644 data_agent/test_metadata_fabric_active_metadata_binding_reconciliation.py create mode 100644 data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py create mode 100644 docs/architecture-decisions/adr-065-local-active-metadata-binding-reconciliation.md create mode 100644 docs/evidence/metadata-fabric-active-metadata-binding-reconciliation-2026-07-30.json create mode 100755 scripts/metadata-fabric-active-metadata-binding-reconciliation.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6b64059e..3ce8b4c8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -34,6 +34,7 @@ on: - feat/ar1-metadata-fabric-active-metadata-authorization - feat/ar1-metadata-fabric-active-metadata-scheduler-delivery - feat/ar1-metadata-fabric-active-metadata-projection-execution + - feat/ar1-metadata-fabric-active-metadata-binding-reconciliation env: PYTHON_VERSION: "3.13" @@ -185,6 +186,9 @@ jobs: - name: Validate metadata fabric Active Metadata projection execution evidence run: python -m data_agent.metadata_fabric_active_metadata_projection_execution validate + - name: Validate metadata fabric Active Metadata binding reconciliation evidence + run: python -m data_agent.metadata_fabric_active_metadata_binding_reconciliation validate + - name: Validate Active Metadata consumer deployment boundary run: python -m data_agent.active_metadata_consumer_deployment validate @@ -229,6 +233,11 @@ jobs: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test run: python -m pytest data_agent/test_active_metadata_authorization_postgres.py -q + - name: Verify Active Metadata binding reconciliation on PostgreSQL + env: + DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test + run: python -m pytest data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py -q + - name: Run required platform tests env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test @@ -267,6 +276,7 @@ jobs: data_agent/test_metadata_fabric_active_metadata_authorization.py \ data_agent/test_metadata_fabric_active_metadata_scheduler_delivery.py \ data_agent/test_metadata_fabric_active_metadata_projection_execution.py \ + data_agent/test_metadata_fabric_active_metadata_binding_reconciliation.py \ data_agent/test_metadata_fabric_lineage_delivery.py \ data_agent/test_metadata_fabric_provider_identity.py \ data_agent/test_metadata_fabric_gravitino_identity.py \ diff --git a/data_agent/metadata_fabric_active_metadata_binding_reconciliation.py b/data_agent/metadata_fabric_active_metadata_binding_reconciliation.py new file mode 100644 index 00000000..7da0d236 --- /dev/null +++ b/data_agent/metadata_fabric_active_metadata_binding_reconciliation.py @@ -0,0 +1,1753 @@ +"""Persist a scheduler-triggered Active Metadata provider binding. + +M3-19 consumes the checked M3-18 execution evidence and reconciles the exact +real Chongqing projection through DolphinScheduler. An exact retained +OpenMetadata projection is mandatory. The only permitted provider repair is +recreating the missing Gravitino projection, followed by a mutation-free +read-back. The same transaction chain then records the content-bound provider +evidence and an immutable Metadata Fabric binding through PlatformGateway. + +This remains a local Docker Desktop rehearsal. Provider and scheduler success +leave PlatformRun in ``reconciling`` and do not establish production identity, +terminal run evidence, or production ingestion. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import platform +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any +from uuid import UUID + +from pydantic import SecretStr +from sqlalchemy import create_engine, text +from sqlalchemy.exc import DBAPIError + +from . import metadata_fabric_active_metadata_authorization as authorization +from . import metadata_fabric_active_metadata_projection_execution as execution +from . import metadata_fabric_active_metadata_scheduler_delivery as delivery +from . import metadata_fabric_bridge as bridge +from . import metadata_fabric_ingestion as ingestion +from . import metadata_fabric_ingestion_replay as replay +from . import metadata_fabric_provider_metrics as provider_metrics +from .active_metadata_authorization import build_metadata_activation_authorization +from .dolphinscheduler_adapter import ( + DOLPHINSCHEDULER_API_PROFILE, + DOLPHINSCHEDULER_SERVER_VERSION, + DolphinSchedulerAdapter, + DolphinSchedulerClient, + DolphinSchedulerDefinitionBinding, + DolphinSchedulerProfile, + DolphinSchedulerWorkflowSpec, + build_dolphinscheduler_binding_artifact, + compile_dolphinscheduler_workflow, +) +from .dolphinscheduler_command_consumer import DolphinSchedulerCommandConsumer +from .metadata_fabric_binding_contract import ( + ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + MetadataFabricBindingRecord, + build_metadata_fabric_binding_record, + build_metadata_fabric_provider_evidence, + build_metadata_fabric_provider_evidence_artifact, +) +from .platform_authorization import ( + build_approval_artifact, + build_policy_decision_artifact, +) +from .platform_contracts import ( + ApprovalRecord, + Artifact, + PlatformDefinitionVersion, + PlatformRun, + PolicyDecision, + Resource, + ResourceVersion, + RunPolicyReferences, + RunStatus, + SubjectContext, + canonical_json_fingerprint, + platform_definition_fingerprint, +) +from .platform_gateway import ( + DefinitionRegistration, + GatewayNotFoundError, + PlatformGateway, +) +from .spatial_dataset_bundle import validate_shapefile_bundle_inventory + +CONTRACT_SCHEMA = "gda.active_metadata_binding_reconciliation_contract.v1" +EVIDENCE_SCHEMA = "gda.active_metadata_binding_reconciliation_evidence.v1" +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_SOURCE_EVIDENCE_PATH = execution.DEFAULT_EVIDENCE_PATH +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-active-metadata-binding-reconciliation-2026-07-30.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-active-metadata-binding-reconciliation.sh" +) +TENANT = authorization.TENANT +SOURCE_ID = authorization.SOURCE_ID +DEFINITION_ID = UUID("a9000000-0000-4000-8000-000000000002") +RUN_ID = UUID("a9000000-0000-4000-8000-000000000003") +TASK_CODE = 180000000000002 +WORKER = "worker:active-metadata-binding-reconciliation-1" +RUNNER = delivery.RUNNER +POLICY_EVALUATOR = delivery.POLICY_EVALUATOR +AUTHORIZER = delivery.AUTHORIZER +APPROVER = delivery.APPROVER +MIGRATIONS = tuple( + Path(__file__).resolve().parent / "migrations" / filename + for filename in ( + "092_platform_control_ledger.sql", + "093_app_user_tenant_context.sql", + "094_platform_control_gateway.sql", + "095_platform_command_outbox.sql", + "096_platform_success_verdict.sql", + "097_metadata_fabric_binding_ledger.sql", + "099_active_metadata_change_outbox.sql", + "100_active_metadata_activation_request.sql", + "101_active_metadata_authorization.sql", + ) +) +FALSE_CLAIMS = ( + "dataset_source_committed", + "dataset_absolute_path_committed", + "dataset_required_in_ci", + "deployment_applied", + "protected_workload_identity_verified", + "provider_minimum_privilege_verified", + "gravitino_authentication_verified", + "durable_catalog_verified", + "oidc_verified", + "tls_verified", + "live_openlineage_emission_verified", + "production_scheduler_submission_verified", + "production_ingestion_verified", + "production_ready", + "platform_run_succeeded", +) + + +class ActiveMetadataBindingReconciliationError(RuntimeError): + """The scheduler-triggered binding reconciliation failed closed.""" + + +@dataclass(frozen=True) +class BoundSource: + resource: Resource + version: ResourceVersion + binding: bridge.MetadataFabricBinding + + +@dataclass(frozen=True) +class ProjectionDefinitionBundle: + registration: DefinitionRegistration + definition: PlatformDefinitionVersion + workflow: DolphinSchedulerWorkflowSpec + + +@dataclass(frozen=True) +class ProjectionDispatchBundle: + source_resource: Resource + source_version: ResourceVersion + request: Any + registration: Any + definition_registration: DefinitionRegistration + dispatch_plan: Artifact + dispatch_policy_decision: Artifact + dispatch_approval: Artifact + run: PlatformRun + activation_authorization: Any + + +def _load_json_object(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise ActiveMetadataBindingReconciliationError( + f"{path.name} is not valid JSON" + ) from exc + if not isinstance(value, dict): + raise ActiveMetadataBindingReconciliationError( + f"{path.name} must contain an object" + ) + return value + + +def _file_record(path: Path) -> dict[str, str | None]: + resolved = path.resolve() + return { + "path": resolved.relative_to(REPO_ROOT).as_posix(), + "sha256": hashlib.sha256(resolved.read_bytes()).hexdigest() + if resolved.is_file() + else None, + } + + +def validate_source_evidence(source: dict[str, Any]) -> None: + errors = execution.validate_rehearsal_evidence(source) + if errors: + raise ActiveMetadataBindingReconciliationError( + "M3-18 projection execution evidence is invalid: " + ", ".join(errors) + ) + if source.get("binding_persisted_to_gda_control") is not False: + raise ActiveMetadataBindingReconciliationError( + "M3-18 source must not already claim binding persistence" + ) + first = source.get("first_apply") + replayed = source.get("replay") + if not isinstance(first, dict) or not isinstance(replayed, dict): + raise ActiveMetadataBindingReconciliationError( + "M3-18 provider observations are missing" + ) + if ( + first.get("status") != "created" + or first.get("mutation_count", 0) <= 0 + or replayed.get("status") != "no_op" + or replayed.get("mutation_count") != 0 + or first.get("binding_candidate_sha256") + != replayed.get("binding_candidate_sha256") + ): + raise ActiveMetadataBindingReconciliationError( + "M3-18 source does not prove create plus zero-mutation replay" + ) + + +def build_bound_source( + source: dict[str, Any], + profile: replay.LocalIngestionProfile, +) -> BoundSource: + validate_source_evidence(source) + content_sha256 = str(source["resource_version_content_sha256"]) + base = authorization.build_authorization_bundle(content_sha256) + first = source["first_apply"] + open_observation = first["openmetadata"] + gravitino_observation = first["gravitino"] + openmetadata_ref = bridge.OpenMetadataTableRef( + entity_id=UUID(open_observation["entity_id"]), + fully_qualified_name=open_observation["fully_qualified_name"], + entity_version=open_observation["entity_version"], + server_version=profile.providers.openmetadata.version, + ) + gravitino_ref = bridge.GravitinoTableRef( + metalake=profile.targets.gravitino.metalake, + catalog=profile.targets.gravitino.catalog, + schema_name=profile.targets.gravitino.schema_name, + table_name=profile.targets.gravitino.table, + provider_revision=gravitino_observation["provider_revision"], + server_version=profile.providers.gravitino.version, + ) + expected_identity = ( + base.source_resource.resource_urn, + str(base.registration.resource_version.resource_version_id), + content_sha256, + ) + for observation in (open_observation, gravitino_observation): + if tuple( + observation[key] + for key in ("resource_urn", "resource_version_id", "content_sha256") + ) != expected_identity: + raise ActiveMetadataBindingReconciliationError( + "M3-18 provider observation does not match the ResourceVersion" + ) + if openmetadata_ref.fully_qualified_name != profile.targets.openmetadata.table_fqn: + raise ActiveMetadataBindingReconciliationError( + "M3-18 OpenMetadata target does not match the M3-19 profile" + ) + if gravitino_ref.identity != gravitino_observation["identity"]: + raise ActiveMetadataBindingReconciliationError( + "M3-18 Gravitino target identity does not match" + ) + resource = base.source_resource.model_copy( + update={ + "governance_ref": bridge.openmetadata_governance_ref(openmetadata_ref), + "technical_refs": (bridge.gravitino_technical_ref(gravitino_ref),), + } + ) + version = base.registration.resource_version + binding = bridge.build_metadata_fabric_binding( + resource, + version, + openmetadata=openmetadata_ref, + gravitino=(gravitino_ref,), + ) + if binding.binding_sha256 != first["binding_candidate_sha256"]: + raise ActiveMetadataBindingReconciliationError( + "M3-18 binding candidate does not match exact provider refs" + ) + return BoundSource(resource=resource, version=version, binding=binding) + + +def build_projection_plan( + content_sha256: str, + profile: replay.LocalIngestionProfile, +) -> replay.LocalApplyPlan: + resource_urn = f"gda://{TENANT}/dataset/chongqing-cultural-districts" + common = { + "resource_urn": resource_urn, + "resource_version_id": str(SOURCE_ID), + "content_sha256": content_sha256, + } + projections = ( + ingestion._projection( + provider="openmetadata", + target_identity=profile.targets.openmetadata.table_fqn, + desired_state={ + **common, + "owner_refs": ["team:data-platform"], + "domain_refs": ["domain:natural-resources"], + "tag_refs": [ + "CulturalHeritage.CulturalDistrict", + "Sensitivity.Internal", + ], + }, + ), + ingestion._projection( + provider="gravitino", + target_identity=profile.targets.gravitino.identity, + desired_state={ + **common, + "provider_revision": f"shapefile-bundle-{content_sha256[:16]}", + }, + ), + ) + source_plan_sha256 = canonical_json_fingerprint( + { + "schema": "gda.active_metadata_binding_reconciliation_intent.v1", + "tenant_id": TENANT, + "resource_urn": resource_urn, + "resource_version_id": str(SOURCE_ID), + "content_sha256": content_sha256, + "source_execution_evidence_schema": execution.EVIDENCE_SCHEMA, + "targets": [item.target_identity for item in projections], + } + ) + values: dict[str, Any] = { + "source_plan_sha256": source_plan_sha256, + "tenant_id": TENANT, + "run_id": RUN_ID, + "definition_version_id": DEFINITION_ID, + "source_resource_version_id": SOURCE_ID, + "resource_urn": resource_urn, + "resource_version_id": SOURCE_ID, + "content_sha256": content_sha256, + "openmetadata_fqn": profile.targets.openmetadata.table_fqn, + "gravitino_identity": profile.targets.gravitino.identity, + "projections": projections, + } + stable = { + "schema": replay.APPLY_PLAN_SCHEMA, + **{ + key: ( + [item.model_dump(mode="json") for item in value] + if key == "projections" + else str(value) + if isinstance(value, UUID) + else value + ) + for key, value in values.items() + }, + "provider_apply_authorized": False, + "writes_to_gda_control": False, + "writes_to_legacy": False, + } + return replay.LocalApplyPlan( + **values, + apply_plan_sha256=canonical_json_fingerprint(stable), + ) + + +def _verify_retained_openmetadata( + plan: replay.LocalApplyPlan, + profile: replay.LocalIngestionProfile, + payload: dict[str, Any], + source_evidence: dict[str, Any], + *, + observed_at: datetime, +) -> bridge.OpenMetadataObservation: + source = source_evidence["first_apply"]["openmetadata"] + ref = bridge.OpenMetadataTableRef( + entity_id=UUID(str(payload["id"])), + fully_qualified_name=profile.targets.openmetadata.table_fqn, + entity_version=str(payload["version"]), + server_version=profile.providers.openmetadata.version, + ) + observation = bridge.parse_openmetadata_table_observation( + ref, + payload, + observed_at=observed_at, + ) + projection = next( + item for item in plan.projections if item.provider == "openmetadata" + ) + desired = projection.desired_state + exact_identity = ( + observation.resource_urn == plan.resource_urn + and observation.resource_version_id == plan.resource_version_id + and observation.content_sha256 == plan.content_sha256 + and sorted(observation.owner_refs) == desired["owner_refs"] + and sorted(observation.domain_refs) == desired["domain_refs"] + and sorted(observation.tag_refs) == desired["tag_refs"] + ) + source_matches = ( + str(observation.ref.entity_id) == source["entity_id"] + and observation.ref.fully_qualified_name == source["fully_qualified_name"] + and observation.ref.entity_version == source["entity_version"] + and observation.snapshot_sha256 == source["snapshot_sha256"] + ) + if not exact_identity or not source_matches: + raise ActiveMetadataBindingReconciliationError( + "retained OpenMetadata projection drifted from M3-18 evidence" + ) + return observation + + +def _reset_stale_empty_gravitino_memory_catalog( + gravitino: replay.GravitinoApplyClient, + target: replay.GravitinoTarget, +) -> bool: + if target.catalog_backend != "memory": + raise replay.MetadataFabricPartialProjectionError( + "stale catalog reset is limited to the memory backend" + ) + catalog_path = f"metalakes/{target.metalake}/catalogs/{target.catalog}" + catalog_payload = gravitino._request("GET", catalog_path, allow_not_found=True) + if catalog_payload is None: + return False + catalog = catalog_payload.get("catalog") + if not isinstance(catalog, dict): + raise replay.MetadataFabricPartialProjectionError( + "Gravitino catalog read-back is incomplete" + ) + properties = catalog.get("properties") + expected_properties = { + "catalog-backend": target.catalog_backend, + "uri": target.uri, + "warehouse": target.warehouse, + } + if ( + catalog.get("name") != target.catalog + or str(catalog.get("type", "")).upper() != target.catalog_type + or catalog.get("provider") != target.catalog_provider + or not isinstance(properties, dict) + or any( + properties.get(key) != value + for key, value in expected_properties.items() + ) + ): + raise replay.MetadataFabricPartialProjectionError( + "Gravitino memory catalog configuration drifted" + ) + schema_path = f"{catalog_path}/schemas/{target.schema_name}" + if gravitino._request("GET", schema_path, allow_not_found=True) is not None: + return False + schema_listing = gravitino._request("GET", f"{catalog_path}/schemas") + identifiers = None if schema_listing is None else schema_listing.get("identifiers") + if identifiers != []: + raise replay.MetadataFabricPartialProjectionError( + "Gravitino memory catalog is not visibly empty" + ) + gravitino._request("DELETE", catalog_path, params={"force": "true"}) + gravitino.mutations.append("gravitino.catalog.reset_stale_empty_memory") + return True + + +def apply_or_repair_once( + plan: replay.LocalApplyPlan, + profile: replay.LocalIngestionProfile, + apply_authorization: replay.ApplyAuthorizationBundle, + run: PlatformRun, + source_evidence: dict[str, Any], + *, + openmetadata: replay.OpenMetadataApplyClient, + gravitino: replay.GravitinoApplyClient, + at: datetime, +) -> replay.ApplyOutcome: + replay.validate_apply_authorization( + plan, + run, + apply_authorization, + at=at, + ) + openmetadata_before = openmetadata.get_table( + profile.targets.openmetadata.table_fqn + ) + gravitino_before = gravitino.get_table(profile.targets.gravitino) + if openmetadata_before is None: + raise replay.MetadataFabricPartialProjectionError( + "M3-19 requires the exact retained M3-18 OpenMetadata projection" + ) + retained = _verify_retained_openmetadata( + plan, + profile, + openmetadata_before, + source_evidence, + observed_at=at, + ) + if gravitino_before is not None: + outcome = replay.apply_once( + plan, + profile, + apply_authorization, + run, + openmetadata=openmetadata, + gravitino=gravitino, + at=at, + ) + if outcome.binding_candidate_sha256 != source_evidence["first_apply"][ + "binding_candidate_sha256" + ]: + raise ActiveMetadataBindingReconciliationError( + "retained provider binding drifted from M3-18 evidence" + ) + return outcome + + start_om = len(openmetadata.mutations) + start_gravitino = len(gravitino.mutations) + try: + _reset_stale_empty_gravitino_memory_catalog( + gravitino, + profile.targets.gravitino, + ) + gravitino_payload = gravitino.apply(plan, profile.targets.gravitino) + governance, technical, binding_sha256 = replay._verify_provider_state( + plan, + profile, + openmetadata_before, + gravitino_payload, + observed_at=at, + ) + mutations = ( + *openmetadata.mutations[start_om:], + *gravitino.mutations[start_gravitino:], + ) + if governance != retained: + raise ActiveMetadataBindingReconciliationError( + "OpenMetadata changed during technical projection repair" + ) + if not mutations or any( + not mutation.startswith("gravitino.") for mutation in mutations + ): + raise ActiveMetadataBindingReconciliationError( + "partial projection repair must mutate only Gravitino" + ) + if binding_sha256 != source_evidence["first_apply"][ + "binding_candidate_sha256" + ]: + raise ActiveMetadataBindingReconciliationError( + "repaired provider binding does not match M3-18 evidence" + ) + except Exception: + try: + gravitino.compensate() + except Exception as compensation_exc: + raise ActiveMetadataBindingReconciliationError( + "Gravitino repair failed and compensation was incomplete" + ) from compensation_exc + raise + return replay.ApplyOutcome( + status=replay.ApplyStatus.CREATED, + mutations=mutations, + openmetadata=governance, + gravitino=technical, + binding_candidate_sha256=binding_sha256, + ) + + +class BindingReconciliationExecutor(execution.ProjectionExecutor): + def __init__(self, *args: Any, source_evidence: dict[str, Any], **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.source_evidence = source_evidence + + def execute(self, payload: dict[str, Any]) -> dict[str, Any]: + self.request_count += 1 + try: + observed = execution.ProjectionExecutionRequest.model_validate(payload) + if observed != self.request: + raise ActiveMetadataBindingReconciliationError( + "binding callback request does not match the compiled workflow" + ) + if self.request_count != 1: + raise ActiveMetadataBindingReconciliationError( + "binding executor accepts exactly one scheduler callback" + ) + self.first = apply_or_repair_once( + self.plan, + self.profile, + self.apply_authorization, + self.run, + self.source_evidence, + openmetadata=self.openmetadata, + gravitino=self.gravitino, + at=datetime.now(UTC), + ) + self.replayed = replay.apply_once( + self.plan, + self.profile, + self.apply_authorization, + self.run, + openmetadata=self.openmetadata, + gravitino=self.gravitino, + at=datetime.now(UTC), + ) + if self.replayed.status != replay.ApplyStatus.NO_OP or self.replayed.mutations: + raise ActiveMetadataBindingReconciliationError( + "exact repaired provider replay performed duplicate mutations" + ) + if self.first.binding_candidate_sha256 != self.replayed.binding_candidate_sha256: + raise ActiveMetadataBindingReconciliationError( + "provider binding drifted across repair and replay" + ) + return { + "schema": "gda.active_metadata_binding_reconciliation_response.v1", + "status": "reconciled_and_replayed", + "request_sha256": self.request.request_sha256, + } + except Exception as exc: + self.error_type = type(exc).__name__ + raise + + +def build_scheduler_definition( + callback_url: str, + request: execution.ProjectionExecutionRequest, + *, + created_at: datetime, +) -> ProjectionDefinitionBundle: + definition_document = execution._workflow_document(callback_url, request) + scheduler = definition_document["dolphinscheduler"] + scheduler["name"] = "gda_active_metadata_binding_reconciliation_v1" + scheduler["description"] = "Read back and persist an authorized provider binding" + task = scheduler["task_definitions"][0] + task["code"] = TASK_CODE + task["name"] = "reconcile_active_metadata_binding" + task["description"] = "Verify retained provider projections before binding commit" + relation = scheduler["task_relations"][0] + relation["postTaskCode"] = TASK_CODE + definition_urn = f"gda://{TENANT}/definition/metadata-binding-reconciliation" + input_contract = { + "metadata_change": "gis.cultural_districts", + "execution_request_sha256": request.request_sha256, + "source_execution_evidence_schema": execution.EVIDENCE_SCHEMA, + } + output_contract = { + "provider_projection_readback": True, + "metadata_fabric_binding_commit": True, + "platform_run_terminal_success": False, + } + definition_sha256 = platform_definition_fingerprint( + orchestration_class="dataops", + capability_id="metadata_fabric.projection_plan", + portability_class="provider_native", + definition_document=definition_document, + input_contract=input_contract, + output_contract=output_contract, + ) + resource = Resource( + tenant_id=TENANT, + resource_urn=definition_urn, + resource_kind="definition", + authority_system="gda", + authority_locator="definition/metadata-binding-reconciliation", + owner_ref="team:metadata-platform", + ) + resource_version = ResourceVersion( + tenant_id=TENANT, + resource_urn=definition_urn, + resource_version_id=DEFINITION_ID, + version_key="dolphinscheduler-3.4.2-local-binding-reconciliation-v1", + content_sha256=definition_sha256, + authority_version_ref={ + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + "callback_transport": "docker_desktop_host_gateway_http", + }, + created_by="workload:metadata-definition-registrar", + created_at=created_at, + ) + definition = PlatformDefinitionVersion( + tenant_id=TENANT, + definition_urn=definition_urn, + definition_version_id=DEFINITION_ID, + orchestration_class="dataops", + capability_id="metadata_fabric.projection_plan", + portability_class="provider_native", + definition_document=definition_document, + input_contract=input_contract, + output_contract=output_contract, + definition_sha256=definition_sha256, + ) + return ProjectionDefinitionBundle( + registration=DefinitionRegistration( + resource=resource, + resource_version=resource_version, + definition=definition, + ), + definition=definition, + workflow=compile_dolphinscheduler_workflow(definition), + ) + + +def build_dispatch_bundle( + content_sha256: str, + bound_source: BoundSource, + definition_bundle: ProjectionDefinitionBundle, + scheduler_binding: DolphinSchedulerDefinitionBinding, + *, + authorized_at: datetime, +) -> ProjectionDispatchBundle: + base = authorization.build_authorization_bundle(content_sha256) + if base.registration.resource_version != bound_source.version: + raise ActiveMetadataBindingReconciliationError( + "bound source version does not match the activation registration" + ) + if scheduler_binding.definition_version_id != DEFINITION_ID: + raise ActiveMetadataBindingReconciliationError( + "DolphinScheduler binding does not match the reconciliation definition" + ) + if scheduler_binding.compiled_sha256 != definition_bundle.workflow.compiled_sha256: + raise ActiveMetadataBindingReconciliationError( + "DolphinScheduler binding does not match the compiled workflow" + ) + dispatch_plan = build_dolphinscheduler_binding_artifact( + scheduler_binding, + created_by=RUNNER, + created_at=authorized_at - timedelta(seconds=3), + ) + subject = SubjectContext( + tenant_id=TENANT, + subject_id=RUNNER.removeprefix("workload:"), + subject_type="workload", + roles=("metadata_projector",), + purpose="reconcile and persist authorized active metadata binding", + ) + decision = PolicyDecision( + tenant_id=TENANT, + run_id=RUN_ID, + subject_context=subject, + action="dolphinscheduler.dispatch", + definition_version_id=DEFINITION_ID, + resource_version_ids=(DEFINITION_ID, SOURCE_ID), + execution_plan_artifact_id=dispatch_plan.artifact_id, + effect="allow", + policy_version_ref=f"gda://{TENANT}/policy/metadata-dispatch-v1", + evaluator_subject=POLICY_EVALUATOR, + requires_approval=True, + decided_at=authorized_at - timedelta(seconds=3), + expires_at=authorized_at + timedelta(days=365), + ) + dispatch_policy = build_policy_decision_artifact(decision) + dispatch_approval = build_approval_artifact( + ApprovalRecord( + tenant_id=TENANT, + run_id=RUN_ID, + definition_version_id=DEFINITION_ID, + policy_decision_artifact_id=dispatch_policy.artifact_id, + policy_decision_sha256=dispatch_policy.content_sha256, + verdict="approved", + approver_subject=APPROVER, + reason="approved local scheduler-triggered binding reconciliation", + decided_at=authorized_at - timedelta(seconds=2), + expires_at=authorized_at + timedelta(days=180), + ) + ) + run = PlatformRun( + tenant_id=TENANT, + run_id=RUN_ID, + definition_version_id=DEFINITION_ID, + orchestration_class="dataops", + subject_context=subject, + input_bindings=( + { + "binding_name": "metadata_change", + "resource_version_id": SOURCE_ID, + "semantic_type": "gis.cultural_districts", + }, + ), + idempotency_key="metadata-binding:cultural-districts:reconcile:v1", + policy_refs=RunPolicyReferences( + policy_decision_artifact_id=dispatch_policy.artifact_id, + approval_artifact_id=dispatch_approval.artifact_id, + ), + submitted_at=authorized_at - timedelta(seconds=1), + ) + activation = build_metadata_activation_authorization( + base.request, + bound_source.version, + definition_bundle.definition, + run, + dispatch_plan, + dispatch_policy, + dispatch_approval, + authorized_by=AUTHORIZER, + authorized_at=authorized_at, + ) + return ProjectionDispatchBundle( + source_resource=bound_source.resource, + source_version=bound_source.version, + request=base.request, + registration=base.registration, + definition_registration=definition_bundle.registration, + dispatch_plan=dispatch_plan, + dispatch_policy_decision=dispatch_policy, + dispatch_approval=dispatch_approval, + run=run, + activation_authorization=activation, + ) + + +def _apply_migrations(engine: Any) -> None: + with engine.begin() as connection: + is_superuser = connection.exec_driver_sql( + "SELECT rolsuper FROM pg_roles WHERE rolname = current_user" + ).scalar_one() + if not is_superuser: + raise ActiveMetadataBindingReconciliationError( + "local reconciliation requires a fresh superuser database" + ) + connection.exec_driver_sql( + """ + CREATE TABLE IF NOT EXISTS agent_app_users ( + id SERIAL PRIMARY KEY, + username VARCHAR(100) UNIQUE NOT NULL + ) + """ + ) + for migration in MIGRATIONS: + connection.execute(text(migration.read_text(encoding="utf-8"))) + + +def _register_control_chain( + gateway: PlatformGateway, + bundle: ProjectionDispatchBundle, + apply_authorization: replay.ApplyAuthorizationBundle, +) -> None: + gateway.register_resource(bundle.source_resource) + gateway.register_resource_version_with_metadata_event(bundle.registration) + claimed = gateway.claim_metadata_changes( + TENANT, + WORKER, + consumer_subject=authorization.CONSUMER_SUBJECT, + ) + if len(claimed) != 1: + raise ActiveMetadataBindingReconciliationError( + "expected exactly one Active Metadata change" + ) + gateway.stage_metadata_activation_request( + TENANT, + claimed[0].event.event_id, + worker_id=WORKER, + request=bundle.request, + ) + gateway.register_definition(bundle.definition_registration) + for artifact in ( + bundle.dispatch_plan, + bundle.dispatch_policy_decision, + bundle.dispatch_approval, + apply_authorization.execution_plan_artifact, + apply_authorization.policy_decision_artifact, + apply_authorization.approval_artifact, + ): + gateway.record_artifact(artifact) + gateway.submit_run(bundle.run) + + +def _direct_binding_mutations_blocked( + gateway: PlatformGateway, + record: MetadataFabricBindingRecord, +) -> tuple[bool, bool]: + results: list[bool] = [] + with gateway._transaction(record.tenant_id) as connection: + for statement in ( + """ + UPDATE gda_control.metadata_fabric_binding + SET recorded_by = 'workload:tamper' + WHERE tenant_id = :tenant_id AND binding_id = :binding_id + """, + """ + DELETE FROM gda_control.metadata_fabric_binding + WHERE tenant_id = :tenant_id AND binding_id = :binding_id + """, + ): + blocked = False + try: + with connection.begin_nested(): + connection.execute( + text(statement), + { + "tenant_id": record.tenant_id, + "binding_id": record.binding_id, + }, + ) + except DBAPIError: + blocked = True + results.append(blocked) + return results[0], results[1] + + +def _attempt_summary(engine: Any) -> tuple[int, int, int, int, list[str]]: + with engine.connect() as connection: + row = connection.execute( + text( + """ + SELECT + count(*) AS total, + count(DISTINCT external_namespace || ':' || external_run_id) + AS correlations, + count(*) FILTER (WHERE observed_state = 'submitted') AS submitted, + count(*) FILTER (WHERE observed_state = 'success') AS succeeded, + array_agg(observed_state ORDER BY observed_at, observation_id) + AS states + FROM gda_control.framework_attempt_observation + WHERE tenant_id = :tenant_id AND run_id = :run_id + """ + ), + {"tenant_id": TENANT, "run_id": RUN_ID}, + ).one() + return row.total, row.correlations, row.submitted, row.succeeded, list(row.states) + + +def _binding_ledger_state( + engine: Any, + record: MetadataFabricBindingRecord, +) -> tuple[int, bool, bool]: + with engine.connect() as connection: + count = connection.execute( + text( + """ + SELECT count(*) + FROM gda_control.metadata_fabric_binding + WHERE tenant_id = :tenant_id AND resource_version_id = :version_id + """ + ), + { + "tenant_id": record.tenant_id, + "version_id": record.binding.resource_version_id, + }, + ).scalar_one() + append_only = connection.exec_driver_sql( + """ + SELECT NOT has_table_privilege( + 'gda_control_gateway', + 'gda_control.metadata_fabric_binding', 'UPDATE' + ) + AND NOT has_table_privilege( + 'gda_control_gateway', + 'gda_control.metadata_fabric_binding', 'DELETE' + ) + """ + ).scalar_one() + force_rls = connection.exec_driver_sql( + """ + SELECT relforcerowsecurity + FROM pg_class + WHERE oid = 'gda_control.metadata_fabric_binding'::regclass + """ + ).scalar_one() + return int(count), bool(append_only), bool(force_rls) + + +def run_scheduler_binding_reconciliation( + database_url: str, + scheduler_profile: DolphinSchedulerProfile, + source_evidence: dict[str, Any], + projection_profile: replay.LocalIngestionProfile, + runtime_identity: dict[str, Any], + principal: dict[str, Any], + gravitino_version: str, + openmetadata: replay.OpenMetadataApplyClient, + gravitino: replay.GravitinoApplyClient, + callback_server: execution.ProjectionExecutionServer, + *, + terminal_timeout_seconds: float = 600, +) -> dict[str, Any]: + validate_source_evidence(source_evidence) + dataset = source_evidence["dataset_bundle"] + if validate_shapefile_bundle_inventory(dataset): + raise ActiveMetadataBindingReconciliationError( + "real Chongqing dataset bundle inventory is invalid" + ) + if scheduler_profile.workload_subject != RUNNER: + raise ActiveMetadataBindingReconciliationError( + "scheduler workload does not match the authorized runner" + ) + if scheduler_profile.policy_evaluator_subject != POLICY_EVALUATOR: + raise ActiveMetadataBindingReconciliationError( + "scheduler evaluator does not match policy evidence" + ) + + started_at = datetime.now(UTC) + bound_source = build_bound_source(source_evidence, projection_profile) + plan = build_projection_plan(dataset["content_sha256"], projection_profile) + request = execution.build_execution_request(plan) + if callback_server.request != request: + raise ActiveMetadataBindingReconciliationError( + "callback server is not bound to the exact reconciliation request" + ) + definition_bundle = build_scheduler_definition( + callback_server.callback_url, + request, + created_at=started_at, + ) + engine = create_engine(database_url) + client = DolphinSchedulerClient(scheduler_profile) + try: + scheduler_binding = client.create_workflow(definition_bundle.workflow) + authorized_at = datetime.now(UTC) + bundle = build_dispatch_bundle( + dataset["content_sha256"], + bound_source, + definition_bundle, + scheduler_binding, + authorized_at=authorized_at, + ) + apply_authorization = execution.build_provider_apply_authorization( + plan, + bundle.run, + projection_profile, + ) + callback_server.executor = BindingReconciliationExecutor( + request, + projection_profile, + plan, + bundle.run, + apply_authorization, + source_evidence=source_evidence, + openmetadata=openmetadata, + gravitino=gravitino, + ) + _apply_migrations(engine) + gateway = PlatformGateway(engine) + _register_control_chain(gateway, bundle, apply_authorization) + first_auth = gateway.authorize_metadata_activation( + bundle.activation_authorization + ) + replay_auth = gateway.authorize_metadata_activation( + bundle.activation_authorization + ) + callback_server.start() + + adapter = DolphinSchedulerAdapter( + scheduler_profile, + gateway=gateway, + client=client, + clock=lambda: authorized_at, + ) + consumer_result = DolphinSchedulerCommandConsumer( + adapter, + gateway=gateway, + ).run_once(TENANT, worker_id=WORKER, limit=1, lease_seconds=600) + if consumer_result.completed != 1: + raise ActiveMetadataBindingReconciliationError( + "authorized binding reconciliation command was not completed" + ) + command = gateway.get_command( + TENANT, + bundle.activation_authorization.command_id, + ) + with engine.connect() as connection: + instance_id = int( + connection.execute( + text( + """ + SELECT external_run_id + FROM gda_control.framework_attempt_observation + WHERE tenant_id = :tenant_id + AND run_id = :run_id + AND observed_state = 'submitted' + """ + ), + {"tenant_id": TENANT, "run_id": RUN_ID}, + ).scalar_one() + ) + terminal = delivery._wait_for_terminal_instance( + client, + instance_id, + scheduler_binding.workflow_definition_code, + timeout_seconds=terminal_timeout_seconds, + ) + variables = client.get_instance_variables(instance_id) + expected_variables = { + str(item["prop"]): str(item["value"]) + for item in definition_bundle.workflow.global_params + } + expected_variables.update(DolphinSchedulerClient.start_params(bundle.run)) + matching_instances = client.find_instances(scheduler_binding, bundle.run) + reconciled = adapter.reconcile( + TENANT, + RUN_ID, + bundle.dispatch_plan.artifact_id, + actor_subject=RUNNER, + attempt_no=1, + ) + attempts = _attempt_summary(engine) + executor = callback_server.executor + if executor is None or executor.first is None or executor.replayed is None: + raise ActiveMetadataBindingReconciliationError( + "scheduler callback did not produce both provider read-backs: " + f"request_count={0 if executor is None else executor.request_count}, " + f"error_type={None if executor is None else executor.error_type}, " + f"first_present={False if executor is None else executor.first is not None}, " + "replay_present=" + f"{False if executor is None else executor.replayed is not None}" + ) + first_outcome = replay._outcome_evidence(executor.first) + replay_outcome = replay._outcome_evidence(executor.replayed) + live_binding = bridge.build_metadata_fabric_binding( + bound_source.resource, + bound_source.version, + openmetadata=executor.first.openmetadata.ref, + gravitino=(executor.first.gravitino.ref,), + ) + observed_at = max( + executor.first.openmetadata.observed_at, + executor.first.gravitino.observed_at, + ) + provider_evidence = build_metadata_fabric_provider_evidence( + binding=live_binding, + source_evidence_schema=ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + source_evidence_sha256=source_evidence["evidence_sha256"], + openmetadata_snapshot_sha256=(executor.first.openmetadata.snapshot_sha256), + gravitino_snapshot_sha256=executor.first.gravitino.snapshot_sha256, + first_apply_status=executor.first.status.value, + first_apply_mutation_count=len(executor.first.mutations), + observed_at=observed_at, + ) + provider_artifact = build_metadata_fabric_provider_evidence_artifact( + provider_evidence, + created_by=RUNNER, + ) + record = build_metadata_fabric_binding_record( + binding=live_binding, + execution_plan_artifact_id=( + apply_authorization.execution_plan_artifact.artifact_id + ), + policy_decision_artifact_id=( + apply_authorization.policy_decision_artifact.artifact_id + ), + approval_artifact_id=apply_authorization.approval_artifact.artifact_id, + provider_evidence_artifact_id=provider_artifact.artifact_id, + recorded_by=RUNNER, + recorded_at=datetime.now(UTC), + ) + gateway.record_artifact(provider_artifact) + first_commit = gateway.commit_metadata_fabric_binding(record) + replay_commit = gateway.commit_metadata_fabric_binding(record) + stored = gateway.get_metadata_fabric_binding(TENANT, SOURCE_ID) + cross_tenant_read_blocked = False + try: + gateway.get_metadata_fabric_binding("isolated-tenant", SOURCE_ID) + except GatewayNotFoundError: + cross_tenant_read_blocked = True + update_blocked, delete_blocked = _direct_binding_mutations_blocked( + gateway, + record, + ) + binding_count, append_only, force_rls = _binding_ledger_state(engine, record) + final_run = gateway.get_run(TENANT, RUN_ID) + + verified = ( + first_auth.created + and not replay_auth.created + and command.status.value == "done" + and consumer_result.claimed == consumer_result.completed == 1 + and terminal.state.upper() == "SUCCESS" + and variables == expected_variables + and len(matching_instances) == 1 + and matching_instances[0].instance_id == instance_id + and reconciled.provider_state == "SUCCESS" + and attempts[:4] == (2, 1, 1, 1) + and attempts[4] == ["submitted", "success"] + and final_run.status == RunStatus.RECONCILING + and executor.request_count == 1 + and first_outcome["status"] in {"created", "no_op"} + and ( + ( + first_outcome["status"] == "created" + and first_outcome["mutation_count"] > 0 + and all( + mutation.startswith("gravitino.") + for mutation in first_outcome["mutations"] + ) + ) + or ( + first_outcome["status"] == "no_op" + and first_outcome["mutation_count"] == 0 + ) + ) + and replay_outcome["status"] == "no_op" + and replay_outcome["mutation_count"] == 0 + and first_outcome["binding_candidate_sha256"] + == replay_outcome["binding_candidate_sha256"] + == live_binding.binding_sha256 + == bound_source.binding.binding_sha256 + and first_commit.created + and not replay_commit.created + and first_commit.value == replay_commit.value == stored == record + and binding_count == 1 + and cross_tenant_read_blocked + and update_blocked + and delete_blocked + and append_only + and force_rls + ) + contract = build_contract_report() + stable = { + "schema": EVIDENCE_SCHEMA, + "status": ( + "local_scheduler_binding_reconciliation_verified" + if verified + else "blocked" + ), + "contract_sha256": contract["contract_sha256"], + "source_execution_evidence_sha256": source_evidence["evidence_sha256"], + "dataset_bundle": dataset, + "dataset_source_committed": False, + "dataset_absolute_path_committed": False, + "dataset_required_in_ci": False, + "real_dataset_resource_version_bound": True, + "resource_version_id": str(SOURCE_ID), + "resource_version_content_sha256": bound_source.version.content_sha256, + "definition_version_id": str(DEFINITION_ID), + "definition_sha256": definition_bundle.definition.definition_sha256, + "compiled_workflow_sha256": definition_bundle.workflow.compiled_sha256, + "run_id": str(RUN_ID), + "dispatch_execution_plan_artifact_id": str(bundle.dispatch_plan.artifact_id), + "dispatch_authorization_id": str( + bundle.activation_authorization.authorization_id + ), + "dispatch_authorization_created": first_auth.created, + "exact_dispatch_authorization_replay_created": replay_auth.created, + "provider_apply_execution_plan_artifact_id": str( + apply_authorization.execution_plan_artifact.artifact_id + ), + "provider_apply_policy_decision_artifact_id": str( + apply_authorization.policy_decision_artifact.artifact_id + ), + "provider_apply_approval_artifact_id": str( + apply_authorization.approval_artifact.artifact_id + ), + "provider_apply_authorization_sha256": ( + apply_authorization.authorization_sha256 + ), + "provider_evidence_artifact_id": str(provider_artifact.artifact_id), + "provider_evidence_sha256": provider_evidence.evidence_sha256, + "provider_apply_authorized": True, + "execution_request_sha256": request.request_sha256, + "execution_callback_request_count": executor.request_count, + "execution_callback_exact_request_verified": executor.error_type is None, + "command_id": str(bundle.activation_authorization.command_id), + "command_status": command.status.value, + "command_claimed_count": consumer_result.claimed, + "command_completed_count": consumer_result.completed, + "scheduler_provider": { + "name": "apache-dolphinscheduler", + "server_version": scheduler_binding.server_version, + "api_profile": scheduler_binding.api_profile, + "image": delivery.IMAGE, + "image_id": delivery.IMAGE_ID, + "architecture": platform.machine(), + "project_code": scheduler_binding.project_code, + "workflow_definition_code": ( + scheduler_binding.workflow_definition_code + ), + "workflow_definition_version": ( + scheduler_binding.workflow_definition_version + ), + "workflow_instance_id": instance_id, + "terminal_state": terminal.state.upper(), + }, + "correlation_variables": variables, + "exact_correlation_variable_readback_verified": ( + variables == expected_variables + ), + "matching_provider_instance_count": len(matching_instances), + "attempt_observation_count": attempts[0], + "external_correlation_count": attempts[1], + "submitted_observation_count": attempts[2], + "success_observation_count": attempts[3], + "attempt_states": attempts[4], + "scheduler_success_readback_verified": ( + reconciled.provider_state == "SUCCESS" + ), + "platform_run_status": final_run.status.value, + "platform_run_succeeded": final_run.status == RunStatus.SUCCEEDED, + "provider_runtime": runtime_identity, + "provider_security": { + "openmetadata": { + "auth_mode": projection_profile.providers.openmetadata.auth_mode, + "authenticated_principal": principal, + "minimum_privilege_verified": False, + }, + "gravitino": { + "auth_mode": projection_profile.providers.gravitino.auth_mode, + "version": gravitino_version, + "authentication_verified": False, + }, + }, + "first_readback": first_outcome, + "replay_readback": replay_outcome, + "source_created_apply_verified": True, + "partial_projection_detected": ( + first_outcome["status"] == "created" + ), + "gravitino_partial_projection_repaired": ( + first_outcome["status"] == "created" + ), + "openmetadata_mutations_executed": False, + "provider_mutations_executed": bool(first_outcome["mutation_count"]), + "scheduler_triggered_provider_readback_verified": verified, + "binding_id": str(record.binding_id), + "binding_sha256": record.binding.binding_sha256, + "binding_record_sha256": record.record_sha256, + "openmetadata_entity_id": str(record.binding.openmetadata.entity_id), + "openmetadata_fqn": record.binding.openmetadata.fully_qualified_name, + "gravitino_identity": record.binding.gravitino[0].identity, + "gravitino_provider_revision": ( + record.binding.gravitino[0].provider_revision + ), + "first_binding_commit_created": first_commit.created, + "replay_binding_commit_created": replay_commit.created, + "binding_row_count": binding_count, + "stored_binding_matches": stored == record, + "binding_persisted_to_gda_control": verified, + "writes_to_gda_control": verified, + "append_only_privileges_verified": append_only, + "force_rls_verified": force_rls, + "cross_tenant_read_blocked": cross_tenant_read_blocked, + "direct_binding_update_blocked": update_blocked, + "direct_binding_delete_blocked": delete_blocked, + "callback_server_cleanup_verified": False, + "provider_port_forwards_cleanup_verified": False, + "standalone_container_cleanup_verified": False, + "temporary_database_cleanup_verified": False, + "provider_objects_retained_for_readback": True, + "writes_to_legacy": False, + "deployment_applied": False, + "protected_workload_identity_verified": False, + "provider_minimum_privilege_verified": False, + "gravitino_authentication_verified": False, + "durable_catalog_verified": False, + "oidc_verified": False, + "tls_verified": False, + "live_openlineage_emission_verified": False, + "production_scheduler_submission_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + "errors": [] if verified else ["local binding reconciliation failed"], + } + return stable + finally: + client.close() + engine.dispose() + + +def run_managed_rehearsal( + database_admin_url: str, + source_evidence: dict[str, Any], + admin_password: SecretStr, + openmetadata_username: str, + openmetadata_password: SecretStr, + *, + readiness_timeout_seconds: float = 180, + terminal_timeout_seconds: float = 600, +) -> dict[str, Any]: + validate_source_evidence(source_evidence) + authorized_at = datetime.now(UTC) + profile = execution.build_projection_profile(authorized_at) + plan = build_projection_plan( + source_evidence["dataset_bundle"]["content_sha256"], + profile, + ) + request = execution.build_execution_request(plan) + database = delivery.EphemeralPostgresDatabase(database_admin_url) + scheduler = delivery.EphemeralDolphinScheduler( + admin_password, + readiness_timeout=readiness_timeout_seconds, + ) + callback_server = execution.ProjectionExecutionServer(request) + om_forward = provider_metrics._PortForward( + kubectl="kubectl", + context=profile.cluster.context, + namespace=profile.cluster.namespace, + service=profile.providers.openmetadata.service, + target_port=profile.providers.openmetadata.service_port, + ) + gravitino_forward = provider_metrics._PortForward( + kubectl="kubectl", + context=profile.cluster.context, + namespace=profile.cluster.namespace, + service=profile.providers.gravitino.service, + target_port=profile.providers.gravitino.service_port, + ) + evidence: dict[str, Any] | None = None + provider_forwards_stopped = False + openmetadata: replay.OpenMetadataApplyClient | None = None + gravitino: replay.GravitinoApplyClient | None = None + try: + om_forward.start() + gravitino_forward.start() + openmetadata = replay.OpenMetadataApplyClient( + base_url=f"http://127.0.0.1:{om_forward.local_port}/api/v1", + username=openmetadata_username, + password=openmetadata_password, + ) + gravitino = replay.GravitinoApplyClient( + base_url=f"http://127.0.0.1:{gravitino_forward.local_port}/api" + ) + principal = openmetadata.authenticated_principal() + gravitino_version = gravitino.version() + runtime_identity = replay._provider_runtime_identity(profile) + with database: + with scheduler: + project_code, access_token = scheduler.provision_project() + scheduler_profile = DolphinSchedulerProfile( + base_url=scheduler.base_url, + access_token=access_token, + project_code=project_code, + workload_subject=RUNNER, + policy_evaluator_subject=POLICY_EVALUATOR, + tenant_code="default", + worker_group="default", + timezone_name="UTC", + request_timeout_seconds=300, + reconciliation_page_limit=5, + ) + if database.database_url is None: + raise ActiveMetadataBindingReconciliationError( + "temporary PostgreSQL database was not created" + ) + evidence = run_scheduler_binding_reconciliation( + database.database_url, + scheduler_profile, + source_evidence, + profile, + runtime_identity, + principal, + gravitino_version, + openmetadata, + gravitino, + callback_server, + terminal_timeout_seconds=terminal_timeout_seconds, + ) + finally: + callback_server.stop() + if openmetadata is not None: + openmetadata.close() + if gravitino is not None: + gravitino.close() + provider_forwards_stopped = om_forward.stop() and gravitino_forward.stop() + if evidence is None: + raise ActiveMetadataBindingReconciliationError( + "local scheduler binding rehearsal produced no evidence" + ) + evidence["callback_server_cleanup_verified"] = callback_server.cleanup_verified + evidence["provider_port_forwards_cleanup_verified"] = provider_forwards_stopped + evidence["standalone_container_cleanup_verified"] = scheduler.cleanup_verified + evidence["temporary_database_cleanup_verified"] = database.cleanup_verified + cleanup_verified = all( + ( + callback_server.cleanup_verified, + provider_forwards_stopped, + scheduler.cleanup_verified, + database.cleanup_verified, + ) + ) + if not cleanup_verified: + evidence["errors"].append("ephemeral binding runtime cleanup failed") + evidence["status"] = "blocked" + evidence["binding_persisted_to_gda_control"] = False + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + return {**stable, "evidence_sha256": canonical_json_fingerprint(stable)} + + +def build_contract_report( + *, + source_evidence_path: Path = DEFAULT_SOURCE_EVIDENCE_PATH, + wrapper_path: Path = DEFAULT_WRAPPER_PATH, +) -> dict[str, Any]: + errors: list[str] = [] + source_evidence_sha256: str | None = None + binding_sha256: str | None = None + try: + source = _load_json_object(source_evidence_path) + validate_source_evidence(source) + profile = execution.build_projection_profile( + datetime(2026, 7, 30, 12, 0, tzinfo=UTC) + ) + bound = build_bound_source(source, profile) + source_evidence_sha256 = source["evidence_sha256"] + binding_sha256 = bound.binding.binding_sha256 + except (KeyError, TypeError, ValueError, ActiveMetadataBindingReconciliationError) as exc: + errors.append(f"M3-19 source contract is invalid: {type(exc).__name__}") + try: + wrapper = wrapper_path.read_text(encoding="utf-8") + for marker in ( + "set -euo pipefail", + "metadata_fabric_active_metadata_binding_reconciliation", + '"$@"', + ): + if marker not in wrapper: + errors.append(f"M3-19 wrapper is missing marker: {marker}") + except OSError as exc: + errors.append(f"M3-19 wrapper is invalid: {type(exc).__name__}") + files = { + "implementation": _file_record(Path(__file__)), + "binding_contract": _file_record( + REPO_ROOT / "data_agent/metadata_fabric_binding_contract.py" + ), + "source_evidence": _file_record(source_evidence_path), + "wrapper": _file_record(wrapper_path), + **{path.name: _file_record(path) for path in MIGRATIONS}, + } + stable = { + "schema": CONTRACT_SCHEMA, + "source_execution_evidence_sha256": source_evidence_sha256, + "expected_binding_sha256": binding_sha256, + "provider_operation_mode": ( + "authorized_gravitino_partial_repair_then_no_op_replay_and_binding_commit" + ), + "platform_run_terminal_gate": "reconciling_until_success_evidence", + "files": files, + "errors": errors, + } + return { + **stable, + "status": "valid" if not errors else "invalid", + "contract_sha256": canonical_json_fingerprint(stable), + "binding_persisted_to_gda_control": False, + "provider_mutations_executed": False, + "production_ingestion_verified": False, + "production_ready": False, + } + + +def validate_rehearsal_evidence(evidence: dict[str, Any]) -> list[str]: + errors: list[str] = [] + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + if evidence.get("schema") != EVIDENCE_SCHEMA: + errors.append("binding reconciliation evidence schema does not match") + if evidence.get("evidence_sha256") != canonical_json_fingerprint(stable): + errors.append("binding reconciliation evidence SHA-256 does not match") + contract = build_contract_report() + if evidence.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("binding reconciliation contract fingerprint is stale") + try: + source = _load_json_object(DEFAULT_SOURCE_EVIDENCE_PATH) + validate_source_evidence(source) + except ActiveMetadataBindingReconciliationError: + source = {} + errors.append("M3-18 source execution evidence is unavailable") + if evidence.get("source_execution_evidence_sha256") != source.get( + "evidence_sha256" + ): + errors.append("M3-18 source execution fingerprint is stale") + dataset = evidence.get("dataset_bundle") + if not isinstance(dataset, dict): + errors.append("binding reconciliation dataset bundle is missing") + else: + errors.extend(validate_shapefile_bundle_inventory(dataset)) + if evidence.get("resource_version_content_sha256") != dataset.get( + "content_sha256" + ): + errors.append("real dataset fingerprint is not bound to ResourceVersion") + for claim in FALSE_CLAIMS: + if evidence.get(claim) is not False: + errors.append(f"local binding reconciliation may not claim {claim}") + for claim in ( + "real_dataset_resource_version_bound", + "provider_apply_authorized", + "execution_callback_exact_request_verified", + "exact_correlation_variable_readback_verified", + "scheduler_success_readback_verified", + "source_created_apply_verified", + "scheduler_triggered_provider_readback_verified", + "stored_binding_matches", + "binding_persisted_to_gda_control", + "writes_to_gda_control", + "append_only_privileges_verified", + "force_rls_verified", + "cross_tenant_read_blocked", + "direct_binding_update_blocked", + "direct_binding_delete_blocked", + "callback_server_cleanup_verified", + "provider_port_forwards_cleanup_verified", + "standalone_container_cleanup_verified", + "temporary_database_cleanup_verified", + "provider_objects_retained_for_readback", + ): + if evidence.get(claim) is not True: + errors.append(f"binding reconciliation did not verify {claim}") + if evidence.get("partial_projection_detected") is not True: + errors.append("M3-19 did not record the partial provider projection") + if evidence.get("gravitino_partial_projection_repaired") is not True: + errors.append("M3-19 did not repair the missing Gravitino projection") + if evidence.get("openmetadata_mutations_executed") is not False: + errors.append("M3-19 may not mutate retained OpenMetadata state") + if evidence.get("provider_mutations_executed") is not True: + errors.append("M3-19 did not record the authorized Gravitino repair") + if evidence.get("first_binding_commit_created") is not True: + errors.append("first binding commit did not create a row") + if evidence.get("replay_binding_commit_created") is not False: + errors.append("exact binding replay created a row") + if evidence.get("binding_row_count") != 1: + errors.append("binding reconciliation must persist exactly one row") + if evidence.get("dispatch_authorization_created") is not True: + errors.append("binding dispatch authorization was not created") + if evidence.get("exact_dispatch_authorization_replay_created") is not False: + errors.append("exact binding dispatch authorization replay created a row") + if evidence.get("execution_callback_request_count") != 1: + errors.append("binding executor must receive exactly one callback") + if evidence.get("command_status") != "done": + errors.append("authorized binding command must be done") + if evidence.get("command_claimed_count") != 1: + errors.append("binding execution must claim one command") + if evidence.get("command_completed_count") != 1: + errors.append("binding execution must complete one command") + scheduler = evidence.get("scheduler_provider") + expected_scheduler = { + "name": "apache-dolphinscheduler", + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "image": delivery.IMAGE, + "image_id": delivery.IMAGE_ID, + "terminal_state": "SUCCESS", + } + if not isinstance(scheduler, dict) or any( + scheduler.get(key) != value for key, value in expected_scheduler.items() + ): + errors.append("binding scheduler provider identity or state does not match") + if evidence.get("matching_provider_instance_count") != 1: + errors.append("binding reconciliation must read back one scheduler instance") + if evidence.get("attempt_observation_count") != 2: + errors.append("binding reconciliation must record two attempt observations") + if evidence.get("external_correlation_count") != 1: + errors.append("binding reconciliation must retain one external correlation") + if evidence.get("attempt_states") != ["submitted", "success"]: + errors.append("binding scheduler attempt states do not match") + if evidence.get("platform_run_status") != "reconciling": + errors.append("binding persistence must leave PlatformRun reconciling") + first = evidence.get("first_readback") + replayed = evidence.get("replay_readback") + if not isinstance(first, dict) or not isinstance(replayed, dict): + errors.append("provider read-back outcomes are missing") + else: + if ( + first.get("status") != "created" + or first.get("mutation_count", 0) <= 0 + or any( + not mutation.startswith("gravitino.") + for mutation in first.get("mutations", []) + ) + ): + errors.append("first provider reconciliation was not Gravitino-only repair") + if replayed.get("status") != "no_op" or replayed.get("mutation_count") != 0: + errors.append("provider repair replay was not mutation-free") + for key in ("binding_candidate_sha256", "openmetadata", "gravitino"): + if first.get(key) != replayed.get(key): + errors.append(f"provider read-back drifted across replay: {key}") + if evidence.get("binding_sha256") != first.get("binding_candidate_sha256"): + errors.append("stored binding does not match provider read-back") + source_first = source.get("first_apply") if isinstance(source, dict) else None + if isinstance(source_first, dict): + if evidence.get("openmetadata_entity_id") != source_first.get( + "openmetadata", {} + ).get("entity_id"): + errors.append("stored OpenMetadata entity does not match M3-18") + if evidence.get("binding_sha256") != source_first.get( + "binding_candidate_sha256" + ): + errors.append("stored binding does not match M3-18 candidate") + serialized = json.dumps(evidence, ensure_ascii=True, sort_keys=True) + for forbidden in ( + "/Users/", + "Downloads/", + ".tmp/", + "host.docker.internal", + '"token"', + '"password"', + '"session"', + ): + if forbidden in serialized: + errors.append("binding evidence contains sensitive local material") + break + return errors + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + validate = subparsers.add_parser("validate") + validate.add_argument( + "--source-evidence", + type=Path, + default=DEFAULT_SOURCE_EVIDENCE_PATH, + ) + validate.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + rehearse = subparsers.add_parser("rehearse") + rehearse.add_argument("--database-admin-url", required=True) + rehearse.add_argument( + "--admin-password-env", + default="GDA_DOLPHINSCHEDULER_ADMIN_PASSWORD", + ) + rehearse.add_argument( + "--source-evidence", + type=Path, + default=DEFAULT_SOURCE_EVIDENCE_PATH, + ) + rehearse.add_argument("--readiness-timeout-seconds", type=float, default=180) + rehearse.add_argument("--terminal-timeout-seconds", type=float, default=600) + rehearse.add_argument("--evidence-out", type=Path, required=True) + args = parser.parse_args(argv) + + if args.command == "validate": + report = build_contract_report(source_evidence_path=args.source_evidence) + try: + report["errors"].extend( + validate_rehearsal_evidence(_load_json_object(args.evidence)) + ) + except ActiveMetadataBindingReconciliationError as exc: + report["errors"].append( + f"binding reconciliation evidence is invalid: {type(exc).__name__}" + ) + report["status"] = "valid" if not report["errors"] else "invalid" + print(json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not report["errors"] else 1 + + source = _load_json_object(args.source_evidence) + provider_profile = replay.load_profile() + try: + username = os.environ[provider_profile.providers.openmetadata.username_env] + password = SecretStr( + os.environ[provider_profile.providers.openmetadata.password_env] + ) + except KeyError as exc: + raise ActiveMetadataBindingReconciliationError( + "OpenMetadata local bootstrap credential environment is missing" + ) from exc + evidence = run_managed_rehearsal( + args.database_admin_url, + source, + delivery._read_admin_password(args.admin_password_env), + username, + password, + readiness_timeout_seconds=args.readiness_timeout_seconds, + terminal_timeout_seconds=args.terminal_timeout_seconds, + ) + args.evidence_out.write_text( + json.dumps(evidence, ensure_ascii=True, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + print(json.dumps(evidence, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not evidence["errors"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/metadata_fabric_binding_contract.py b/data_agent/metadata_fabric_binding_contract.py index 63ca6734..4dda1392 100644 --- a/data_agent/metadata_fabric_binding_contract.py +++ b/data_agent/metadata_fabric_binding_contract.py @@ -37,6 +37,9 @@ "application/vnd.gda.metadata-fabric-provider-binding-evidence+json" ) SOURCE_EVIDENCE_SCHEMA = "gda.metadata_fabric_local_ingestion_evidence.v1" +ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA = ( + "gda.active_metadata_projection_execution_evidence.v1" +) NonEmptyText = Annotated[ str, @@ -163,6 +166,7 @@ def parse_metadata_fabric_execution_plan_artifact( def metadata_fabric_provider_evidence_fingerprint( *, binding: MetadataFabricBinding, + source_evidence_schema: str = SOURCE_EVIDENCE_SCHEMA, source_evidence_sha256: str, openmetadata_snapshot_sha256: str, gravitino_snapshot_sha256: str, @@ -175,7 +179,7 @@ def metadata_fabric_provider_evidence_fingerprint( return canonical_json_fingerprint( { "binding": binding.model_dump(mode="json", by_alias=True), - "source_evidence_schema": SOURCE_EVIDENCE_SCHEMA, + "source_evidence_schema": source_evidence_schema, "source_evidence_sha256": source_evidence_sha256, "openmetadata_snapshot_sha256": openmetadata_snapshot_sha256, "gravitino_snapshot_sha256": gravitino_snapshot_sha256, @@ -196,7 +200,8 @@ class MetadataFabricProviderEvidence(_FrozenModel): ] = Field(default=PROVIDER_EVIDENCE_SCHEMA, alias="schema") binding: MetadataFabricBinding source_evidence_schema: Literal[ - "gda.metadata_fabric_local_ingestion_evidence.v1" + "gda.metadata_fabric_local_ingestion_evidence.v1", + "gda.active_metadata_projection_execution_evidence.v1", ] = SOURCE_EVIDENCE_SCHEMA source_evidence_sha256: Sha256 openmetadata_snapshot_sha256: Sha256 @@ -225,6 +230,7 @@ def _content_bound(self) -> Self: raise ValueError("first apply status does not match its mutation count") expected = metadata_fabric_provider_evidence_fingerprint( binding=self.binding, + source_evidence_schema=self.source_evidence_schema, source_evidence_sha256=self.source_evidence_sha256, openmetadata_snapshot_sha256=self.openmetadata_snapshot_sha256, gravitino_snapshot_sha256=self.gravitino_snapshot_sha256, @@ -242,6 +248,10 @@ def _content_bound(self) -> Self: def build_metadata_fabric_provider_evidence( *, binding: MetadataFabricBinding, + source_evidence_schema: Literal[ + "gda.metadata_fabric_local_ingestion_evidence.v1", + "gda.active_metadata_projection_execution_evidence.v1", + ] = SOURCE_EVIDENCE_SCHEMA, source_evidence_sha256: str, openmetadata_snapshot_sha256: str, gravitino_snapshot_sha256: str, @@ -251,6 +261,7 @@ def build_metadata_fabric_provider_evidence( ) -> MetadataFabricProviderEvidence: values: dict[str, Any] = { "binding": binding, + "source_evidence_schema": source_evidence_schema, "source_evidence_sha256": source_evidence_sha256, "openmetadata_snapshot_sha256": openmetadata_snapshot_sha256, "gravitino_snapshot_sha256": gravitino_snapshot_sha256, diff --git a/data_agent/platform_truth.py b/data_agent/platform_truth.py index f0ae4355..0fcb55a6 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -938,6 +938,26 @@ def _config( ), "Protected identity, durable executor and production scheduler/provider path", ), + RuntimeSpec( + "metadata_active_metadata_binding_reconciliation_rehearsal", + "active_metadata_binding_reconciliation_rehearsal", + "governed", + "evidence_durable", + "temporary scheduler/provider/PostgreSQL session + committed local evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_active_metadata_binding_reconciliation.py", + "scripts/metadata-fabric-active-metadata-binding-reconciliation.sh", + ), + ( + ( + "data_agent/metadata_fabric_active_metadata_binding_reconciliation.py", + "def run_managed_rehearsal", + ), + ), + "Protected identity, durable catalog/executor and production binding ledger", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation.py b/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation.py new file mode 100644 index 00000000..e1db33c0 --- /dev/null +++ b/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation.py @@ -0,0 +1,468 @@ +import json +from copy import deepcopy +from datetime import UTC, datetime + +import pytest + +from data_agent import metadata_fabric_active_metadata_binding_reconciliation as binding +from data_agent import metadata_fabric_active_metadata_projection_execution as execution +from data_agent import metadata_fabric_ingestion_replay as replay +from data_agent.dolphinscheduler_adapter import DolphinSchedulerDefinitionBinding +from data_agent.metadata_fabric_binding_contract import ( + ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + build_metadata_fabric_provider_evidence, + build_metadata_fabric_provider_evidence_artifact, + parse_metadata_fabric_provider_evidence_artifact, +) + +AT = datetime(2026, 7, 30, 12, 0, tzinfo=UTC) +EXPECTED_EVIDENCE_SHA256 = ( + "e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33" +) + + +def _source(): + return json.loads(binding.DEFAULT_SOURCE_EVIDENCE_PATH.read_text(encoding="utf-8")) + + +def _profile(): + return execution.build_projection_profile(AT) + + +def _scheduler_binding(definition): + return DolphinSchedulerDefinitionBinding( + tenant_id=binding.TENANT, + definition_version_id=binding.DEFINITION_ID, + project_code=190000000000101, + workflow_definition_code=190000000000102, + workflow_definition_version=1, + compiled_sha256=definition.workflow.compiled_sha256, + ) + + +class FakeOpenMetadata: + def __init__(self, table=None): + self.table = deepcopy(table) + self.mutations = [] + + def get_table(self, _fqn): + return deepcopy(self.table) + + +class FakeGravitino: + def __init__(self, table=None): + self.table = deepcopy(table) + self.mutations = [] + self.apply_count = 0 + self.compensated = False + + def get_table(self, _target): + return deepcopy(self.table) + + def _request( + self, + _method, + _path, + *, + json_body=None, + params=None, + allow_not_found=False, + ): + del json_body, params, allow_not_found + return None + + def apply(self, plan, target): + self.apply_count += 1 + projection = next( + item for item in plan.projections if item.provider == "gravitino" + ) + self.table = { + "code": 0, + "table": { + "name": target.table, + "properties": { + "gda.resource_urn": plan.resource_urn, + "gda.resource_version_id": str(plan.resource_version_id), + "gda.content_sha256": plan.content_sha256, + "gda.provider_revision": projection.desired_state[ + "provider_revision" + ], + }, + }, + } + self.mutations.append("gravitino.table.create") + return deepcopy(self.table) + + def compensate(self): + self.table = None + self.compensated = True + return True + + +class FakeStaleMemoryCatalogGravitino(FakeGravitino): + def __init__(self, target, *, identifiers): + super().__init__() + self.target = target + self.identifiers = identifiers + self.requests = [] + + def _request( + self, + method, + path, + *, + json_body=None, + params=None, + allow_not_found=False, + ): + del json_body, allow_not_found + self.requests.append((method, path, params)) + catalog_path = ( + f"metalakes/{self.target.metalake}/catalogs/{self.target.catalog}" + ) + if method == "GET" and path == catalog_path: + return { + "code": 0, + "catalog": { + "name": self.target.catalog, + "type": self.target.catalog_type.lower(), + "provider": self.target.catalog_provider, + "properties": { + "catalog-backend": self.target.catalog_backend, + "uri": self.target.uri, + "warehouse": self.target.warehouse, + "in-use": "true", + }, + }, + } + if method == "GET" and path == f"{catalog_path}/schemas": + return {"code": 0, "identifiers": self.identifiers} + if method == "GET": + return None + return {"code": 0} + + +def _retained_openmetadata_payload(source): + observed = source["first_apply"]["openmetadata"] + return { + "id": observed["entity_id"], + "name": "cultural_districts", + "fullyQualifiedName": observed["fully_qualified_name"], + "version": observed["entity_version"], + "deleted": False, + "owners": [ + { + "type": "team", + "fullyQualifiedName": "data-platform", + } + ], + "domains": [ + { + "type": "domain", + "fullyQualifiedName": "natural-resources", + } + ], + "tags": [ + {"tagFQN": "CulturalHeritage.CulturalDistrict"}, + {"tagFQN": "Sensitivity.Internal"}, + ], + "extension": { + "gdaResourceUrn": observed["resource_urn"], + "gdaResourceVersionId": observed["resource_version_id"], + "gdaContentSha256": observed["content_sha256"], + }, + } + + +def _repair_inputs(): + source = _source() + profile = _profile() + payload = _retained_openmetadata_payload(source) + source["first_apply"]["openmetadata"]["snapshot_sha256"] = ( + binding.canonical_json_fingerprint(payload) + ) + bound = binding.build_bound_source(_source(), profile) + plan = binding.build_projection_plan(bound.version.content_sha256, profile) + request = execution.build_execution_request(plan) + definition = binding.build_scheduler_definition( + "http://host.docker.internal:43123/v1/execute-projection", + request, + created_at=AT, + ) + dispatch = binding.build_dispatch_bundle( + bound.version.content_sha256, + bound, + definition, + _scheduler_binding(definition), + authorized_at=AT, + ) + apply_authorization = execution.build_provider_apply_authorization( + plan, + dispatch.run, + profile, + ) + return source, profile, plan, dispatch.run, apply_authorization, payload + + +def test_m3_18_provider_refs_build_exact_real_data_binding(): + source = _source() + bound = binding.build_bound_source(source, _profile()) + + assert bound.version.resource_version_id == binding.SOURCE_ID + assert bound.version.content_sha256 == source["dataset_bundle"]["content_sha256"] + assert str(bound.binding.openmetadata.entity_id) == ( + source["first_apply"]["openmetadata"]["entity_id"] + ) + assert bound.binding.openmetadata.fully_qualified_name == ( + "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts" + ) + assert bound.binding.gravitino[0].identity == ( + "gda_chongqing_m3_18/iceberg/cultural_heritage/cultural_districts" + ) + assert bound.binding.binding_sha256 == ( + source["first_apply"]["binding_candidate_sha256"] + ) + assert bound.resource.governance_ref == ( + bound.binding.openmetadata.model_dump(mode="json") + ) + assert bound.resource.technical_refs == ( + bound.binding.gravitino[0].model_dump(mode="json"), + ) + + +def test_binding_reconciliation_uses_distinct_dispatch_and_apply_authorization(): + source = _source() + profile = _profile() + bound = binding.build_bound_source(source, profile) + plan = binding.build_projection_plan(bound.version.content_sha256, profile) + request = execution.build_execution_request(plan) + definition = binding.build_scheduler_definition( + "http://host.docker.internal:43123/v1/execute-projection", + request, + created_at=AT, + ) + dispatch = binding.build_dispatch_bundle( + bound.version.content_sha256, + bound, + definition, + _scheduler_binding(definition), + authorized_at=AT, + ) + apply_authorization = execution.build_provider_apply_authorization( + plan, + dispatch.run, + profile, + ) + apply_decision = replay.parse_policy_decision_artifact( + apply_authorization.policy_decision_artifact + ) + + assert plan.definition_version_id == binding.DEFINITION_ID + assert plan.run_id == binding.RUN_ID + assert definition.definition.capability_id == "metadata_fabric.projection_plan" + assert dispatch.source_resource == bound.resource + assert dispatch.run.status.value == "accepted" + assert dispatch.activation_authorization.provider_mutations_executed is False + assert apply_decision.action == replay.ACTION + assert apply_decision.execution_plan_artifact_id != ( + dispatch.dispatch_plan.artifact_id + ) + + +def test_provider_evidence_accepts_active_metadata_execution_source(): + source = _source() + bound = binding.build_bound_source(source, _profile()) + first = source["first_apply"] + provider_evidence = build_metadata_fabric_provider_evidence( + binding=bound.binding, + source_evidence_schema=ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + source_evidence_sha256=source["evidence_sha256"], + openmetadata_snapshot_sha256=first["openmetadata"]["snapshot_sha256"], + gravitino_snapshot_sha256=first["gravitino"]["snapshot_sha256"], + first_apply_status="no_op", + first_apply_mutation_count=0, + observed_at=AT, + ) + artifact = build_metadata_fabric_provider_evidence_artifact( + provider_evidence, + created_by=binding.RUNNER, + ) + + assert parse_metadata_fabric_provider_evidence_artifact(artifact) == ( + provider_evidence + ) + assert provider_evidence.source_evidence_schema == ( + ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA + ) + + +def test_source_evidence_tampering_is_rejected_before_binding_construction(): + tampered = deepcopy(_source()) + tampered["first_apply"]["openmetadata"]["entity_id"] = ( + "00000000-0000-4000-8000-000000000001" + ) + + with pytest.raises( + binding.ActiveMetadataBindingReconciliationError, + match="M3-18 projection execution evidence is invalid", + ): + binding.build_bound_source(tampered, _profile()) + + +def test_static_contract_is_source_bound_and_fail_closed(): + report = binding.build_contract_report() + + assert report["status"] == "valid" + assert report["errors"] == [] + assert report["source_execution_evidence_sha256"] == ( + _source()["evidence_sha256"] + ) + assert report["expected_binding_sha256"] == ( + _source()["first_apply"]["binding_candidate_sha256"] + ) + assert report["binding_persisted_to_gda_control"] is False + assert report["provider_mutations_executed"] is False + assert report["production_ready"] is False + + +def test_exact_openmetadata_and_missing_gravitino_repairs_then_replays_no_op(): + source, profile, plan, run, apply_authorization, payload = _repair_inputs() + openmetadata = FakeOpenMetadata(payload) + gravitino = FakeGravitino() + + repaired = binding.apply_or_repair_once( + plan, + profile, + apply_authorization, + run, + source, + openmetadata=openmetadata, + gravitino=gravitino, + at=AT, + ) + replayed = replay.apply_once( + plan, + profile, + apply_authorization, + run, + openmetadata=openmetadata, + gravitino=gravitino, + at=AT, + ) + + assert repaired.status == replay.ApplyStatus.CREATED + assert repaired.mutations == ("gravitino.table.create",) + assert replayed.status == replay.ApplyStatus.NO_OP + assert replayed.mutations == () + assert repaired.binding_candidate_sha256 == replayed.binding_candidate_sha256 + assert openmetadata.mutations == [] + assert gravitino.apply_count == 1 + + +def test_openmetadata_snapshot_drift_blocks_before_gravitino_repair(): + source, profile, plan, run, apply_authorization, payload = _repair_inputs() + payload["name"] = "drifted_name" + openmetadata = FakeOpenMetadata(payload) + gravitino = FakeGravitino() + gravitino.apply(plan, profile.targets.gravitino) + gravitino.mutations.clear() + gravitino.apply_count = 0 + + with pytest.raises( + binding.ActiveMetadataBindingReconciliationError, + match="OpenMetadata projection drifted", + ): + binding.apply_or_repair_once( + plan, + profile, + apply_authorization, + run, + source, + openmetadata=openmetadata, + gravitino=gravitino, + at=AT, + ) + + assert openmetadata.mutations == [] + assert gravitino.mutations == [] + assert gravitino.apply_count == 0 + + +def test_missing_openmetadata_and_retained_gravitino_blocks_before_mutation(): + source, profile, plan, run, apply_authorization, _payload = _repair_inputs() + gravitino = FakeGravitino(table={"code": 0, "table": {}}) + + with pytest.raises( + replay.MetadataFabricPartialProjectionError, + match="requires the exact retained M3-18 OpenMetadata projection", + ): + binding.apply_or_repair_once( + plan, + profile, + apply_authorization, + run, + source, + openmetadata=FakeOpenMetadata(), + gravitino=gravitino, + at=AT, + ) + + assert gravitino.mutations == [] + assert gravitino.apply_count == 0 + + +def test_exact_empty_stale_memory_catalog_is_reset_before_recreation(): + target = _profile().targets.gravitino + gravitino = FakeStaleMemoryCatalogGravitino(target, identifiers=[]) + + assert binding._reset_stale_empty_gravitino_memory_catalog( + gravitino, + target, + ) + + assert gravitino.mutations == ["gravitino.catalog.reset_stale_empty_memory"] + assert gravitino.requests[-1][0] == "DELETE" + assert gravitino.requests[-1][2] == {"force": "true"} + + +def test_nonempty_stale_memory_catalog_blocks_before_reset(): + target = _profile().targets.gravitino + gravitino = FakeStaleMemoryCatalogGravitino( + target, + identifiers=[{"namespace": [target.catalog], "name": "retained"}], + ) + + with pytest.raises( + replay.MetadataFabricPartialProjectionError, + match="not visibly empty", + ): + binding._reset_stale_empty_gravitino_memory_catalog(gravitino, target) + + assert gravitino.mutations == [] + assert all(method != "DELETE" for method, _path, _params in gravitino.requests) + + +def test_checked_reconciliation_evidence_is_current_and_fail_closed(): + evidence = json.loads(binding.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + + assert binding.validate_rehearsal_evidence(evidence) == [] + assert evidence["evidence_sha256"] == EXPECTED_EVIDENCE_SHA256 + assert evidence["binding_persisted_to_gda_control"] is True + assert evidence["first_readback"]["mutations"] == [ + "gravitino.catalog.reset_stale_empty_memory", + "gravitino.catalog.create", + "gravitino.schema.create", + "gravitino.table.create", + ] + assert evidence["replay_readback"]["mutation_count"] == 0 + assert evidence["platform_run_status"] == "reconciling" + assert evidence["durable_catalog_verified"] is False + assert evidence["production_ready"] is False + + tampered = deepcopy(evidence) + tampered["binding_persisted_to_gda_control"] = False + stable = {key: value for key, value in tampered.items() if key != "evidence_sha256"} + tampered["evidence_sha256"] = binding.canonical_json_fingerprint(stable) + assert "binding reconciliation did not verify binding_persisted_to_gda_control" in ( + binding.validate_rehearsal_evidence(tampered) + ) diff --git a/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py b/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py new file mode 100644 index 00000000..d5b031ff --- /dev/null +++ b/data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py @@ -0,0 +1,157 @@ +import json +import os +from datetime import UTC, datetime +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.engine import make_url + +from data_agent import metadata_fabric_active_metadata_binding_reconciliation as binding +from data_agent import metadata_fabric_active_metadata_projection_execution as execution +from data_agent.dolphinscheduler_adapter import DolphinSchedulerDefinitionBinding +from data_agent.metadata_fabric_binding_contract import ( + ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + build_metadata_fabric_binding_record, + build_metadata_fabric_provider_evidence, + build_metadata_fabric_provider_evidence_artifact, +) +from data_agent.platform_gateway import GatewayNotFoundError, PlatformGateway + +DATABASE_URL = os.environ.get("DATABASE_URL") +AT = datetime(2026, 7, 30, 12, 0, tzinfo=UTC) + + +def _temporary_database_url() -> tuple[object, str, str]: + admin_url = make_url(DATABASE_URL) + admin_engine = create_engine(admin_url, isolation_level="AUTOCOMMIT") + with admin_engine.connect() as connection: + if not connection.exec_driver_sql( + "SELECT rolsuper FROM pg_roles WHERE rolname = current_user" + ).scalar_one(): + admin_engine.dispose() + pytest.skip("binding reconciliation test requires a PostgreSQL superuser") + database_name = f"gda_binding_reconciliation_{uuid4().hex}" + connection.exec_driver_sql(f'CREATE DATABASE "{database_name}"') + database_url = admin_url.set(database=database_name).render_as_string( + hide_password=False + ) + return admin_engine, database_name, database_url + + +def _drop_temporary_database(admin_engine, database_name: str) -> None: + with admin_engine.connect() as connection: + connection.execute( + text( + """ + SELECT pg_terminate_backend(pid) + FROM pg_stat_activity + WHERE datname = :database_name + AND pid <> pg_backend_pid() + """ + ), + {"database_name": database_name}, + ) + connection.exec_driver_sql(f'DROP DATABASE "{database_name}"') + admin_engine.dispose() + + +@pytest.mark.skipif(not DATABASE_URL, reason="DATABASE_URL is not configured") +def test_real_m3_18_binding_commits_idempotently_to_fresh_postgres(): + admin_engine, database_name, database_url = _temporary_database_url() + engine = None + try: + source = json.loads( + binding.DEFAULT_SOURCE_EVIDENCE_PATH.read_text(encoding="utf-8") + ) + profile = execution.build_projection_profile(AT) + bound = binding.build_bound_source(source, profile) + plan = binding.build_projection_plan(bound.version.content_sha256, profile) + request = execution.build_execution_request(plan) + definition = binding.build_scheduler_definition( + "http://host.docker.internal:43123/v1/execute-projection", + request, + created_at=AT, + ) + scheduler_binding = DolphinSchedulerDefinitionBinding( + tenant_id=binding.TENANT, + definition_version_id=binding.DEFINITION_ID, + project_code=190000000000101, + workflow_definition_code=190000000000102, + workflow_definition_version=1, + compiled_sha256=definition.workflow.compiled_sha256, + ) + dispatch = binding.build_dispatch_bundle( + bound.version.content_sha256, + bound, + definition, + scheduler_binding, + authorized_at=AT, + ) + apply_authorization = execution.build_provider_apply_authorization( + plan, + dispatch.run, + profile, + ) + first = source["first_apply"] + provider_evidence = build_metadata_fabric_provider_evidence( + binding=bound.binding, + source_evidence_schema=ACTIVE_METADATA_PROJECTION_EVIDENCE_SCHEMA, + source_evidence_sha256=source["evidence_sha256"], + openmetadata_snapshot_sha256=first["openmetadata"]["snapshot_sha256"], + gravitino_snapshot_sha256=first["gravitino"]["snapshot_sha256"], + first_apply_status="created", + first_apply_mutation_count=4, + observed_at=AT, + ) + provider_artifact = build_metadata_fabric_provider_evidence_artifact( + provider_evidence, + created_by=binding.RUNNER, + ) + record = build_metadata_fabric_binding_record( + binding=bound.binding, + execution_plan_artifact_id=( + apply_authorization.execution_plan_artifact.artifact_id + ), + policy_decision_artifact_id=( + apply_authorization.policy_decision_artifact.artifact_id + ), + approval_artifact_id=apply_authorization.approval_artifact.artifact_id, + provider_evidence_artifact_id=provider_artifact.artifact_id, + recorded_by=binding.RUNNER, + recorded_at=AT, + ) + + engine = create_engine(database_url) + binding._apply_migrations(engine) + gateway = PlatformGateway(engine) + binding._register_control_chain(gateway, dispatch, apply_authorization) + gateway.record_artifact(provider_artifact) + + first_commit = gateway.commit_metadata_fabric_binding(record) + replay_commit = gateway.commit_metadata_fabric_binding(record) + stored = gateway.get_metadata_fabric_binding(binding.TENANT, binding.SOURCE_ID) + with pytest.raises(GatewayNotFoundError): + gateway.get_metadata_fabric_binding("isolated-tenant", binding.SOURCE_ID) + update_blocked, delete_blocked = binding._direct_binding_mutations_blocked( + gateway, + record, + ) + row_count, append_only, force_rls = binding._binding_ledger_state( + engine, + record, + ) + + assert first_commit.created is True + assert replay_commit.created is False + assert first_commit.value == replay_commit.value == stored == record + assert row_count == 1 + assert update_blocked and delete_blocked + assert append_only and force_rls + assert stored.binding.binding_sha256 == ( + source["first_apply"]["binding_candidate_sha256"] + ) + finally: + if engine is not None: + engine.dispose() + _drop_temporary_database(admin_engine, database_name) diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index 3cc292b7..ff84d6eb 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -274,6 +274,12 @@ def test_repository_source_access_and_runtime_baselines_match(): and item["production_role"] == "local_verification_only" for item in static_report["runtime"]["inventory"] ) + assert any( + item["runtime_id"] + == "metadata_active_metadata_binding_reconciliation_rehearsal" + and item["production_role"] == "local_verification_only" + for item in static_report["runtime"]["inventory"] + ) def test_runtime_report_detects_unregistered_background_mechanism(tmp_path): diff --git a/docs/architecture-decisions/adr-065-local-active-metadata-binding-reconciliation.md b/docs/architecture-decisions/adr-065-local-active-metadata-binding-reconciliation.md new file mode 100644 index 00000000..15f29f85 --- /dev/null +++ b/docs/architecture-decisions/adr-065-local-active-metadata-binding-reconciliation.md @@ -0,0 +1,78 @@ +# ADR-065: Local Active Metadata binding reconciliation + +**Status**: Accepted + +**Date**: 2026-07-30 + +**Decision owners**: Data Platform, Metadata Platform, Data Governance, Security, Platform Architecture + +**Related decisions**: [ADR-049](adr-049-tenant-scoped-metadata-fabric-binding-ledger.md) · [ADR-054](adr-054-local-gravitino-jdbc-catalog-restart-continuity.md) · [ADR-064](adr-064-local-scheduler-triggered-active-metadata-projection-execution.md) + +## Context + +M3-18 retained an exact OpenMetadata and Gravitino projection for the real Chongqing cultural-district ResourceVersion, but deliberately did not persist its provider binding in GDA Control. A later read-only probe found the OpenMetadata UUID, FQN, version, governance, content and snapshot unchanged while the Gravitino table was absent. + +The Gravitino target used an Iceberg `memory` catalog. After provider restart, the connector reported no visible schemas, while Gravitino's PostgreSQL entity index still retained the old schema row. A normal schema create therefore failed with a duplicate-key conflict. Treating this as a generic create, directly deleting provider database rows, or accepting the M3-18 evidence without a current read-back would all weaken the binding contract. + +## Decision + +### 1. M3-18 evidence and OpenMetadata state are immutable prerequisites + +M3-19 validates the checked M3-18 evidence and its dependency fingerprints before database or provider writes. It reconstructs the exact ResourceVersion, provider targets, binding candidate and independent `metadata_fabric.apply` authorization. The retained OpenMetadata UUID, FQN, version, GDA identity, owner, domain, tags and canonical snapshot must all match M3-18 before any repair. + +Missing or drifted OpenMetadata blocks. A Gravitino-present/OpenMetadata-missing state also blocks. OpenMetadata is never mutated by this reconciliation. + +### 2. Only an exact, visibly empty M3-18 memory catalog may be reset + +If the target table is absent, M3-19 reads the dedicated M3-18 catalog configuration. Name, type, provider, `memory` backend, URI and warehouse must match the authorized target. The target schema must be absent and the complete visible schema inventory must be empty. + +Only that state permits a provider-native forced catalog reset followed by recreation of the exact catalog, schema and table. Every recorded mutation must begin with `gravitino.`. A non-empty inventory, configuration drift, another backend, unexpected response or non-Gravitino mutation blocks before binding commit. Gravitino's PostgreSQL is never edited directly. + +### 3. Repair is accepted only after zero-mutation replay + +The repaired table must read back the M3-18 ResourceURN, ResourceVersion UUID, content SHA, provider revision and binding SHA. An immediate exact replay must be `no_op` with zero mutations and identical OpenMetadata/Gravitino observations. Only then is provider evidence materialized. + +### 4. PlatformGateway owns the immutable binding fact + +The same scheduler Run, dispatch authorization, provider-apply plan, PolicyDecision, Approval and provider evidence are registered on a fresh PostgreSQL database. `PlatformGateway.commit_metadata_fabric_binding` creates exactly one tenant-scoped row; exact replay returns `created=false`. FORCE RLS, cross-tenant invisibility, append-only grants and direct UPDATE/DELETE rejection remain mandatory. + +### 5. Provider and scheduler success do not bypass terminal evidence + +The PlatformRun remains `reconciling`. M3-19 does not produce the output Artifact, QualityResult, LineageEvent and RunSuccessEvidence required for `succeeded`. The callback, provider port-forwards, standalone scheduler container and temporary database are removed after the rehearsal. + +## Verification + +The local rehearsal recorded: + +- Chongqing content SHA `fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007`; +- exact retained OpenMetadata UUID `9d043410-02b5-487d-bb70-da5f3969a978` and snapshot SHA `a3ed5e2195c2f5847b5f5b59d78c8ba547c1f7170b3396cdd56b45f8559b0077`; +- four Gravitino-only mutations: stale empty memory-catalog reset, catalog create, schema create and table create; +- immediate replay `no_op/0 mutations` with binding SHA `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b`; +- first binding commit `created=true`, replay `created=false`, one ledger row, binding ID `ff669a52-1271-55f0-a1b6-ee6f57ddb1ea`; +- DolphinScheduler `SUCCESS`, two attempt observations, one external correlation and PlatformRun `reconciling`; +- callback, two port-forwards, standalone container and temporary database cleanup; +- contract SHA `012a7c86ba9fe53217e721ff7286b8f2a246b9394efd2999abbcd025e13ac7f5` and evidence SHA `e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33`. + +## Claim Boundary + +Allowed now: + +- the retained M3-18 OpenMetadata projection was verified before mutation; +- the exact absent Gravitino projection was repaired under a content-bound local authorization; +- repair replay was mutation-free and the resulting binding was persisted idempotently through PlatformGateway; +- the checked real-data binding is tenant-scoped, append-only and cross-tenant invisible in the recorded local PostgreSQL rehearsal. + +Fixed false now: + +- durable Gravitino catalog continuity, Gravitino authentication and provider-wide minimum privilege; +- protected workload identity, OIDC, TLS and a deployed durable executor; +- production scheduler/provider submission, production binding deployment and production ingestion; +- terminal PlatformRun success and `production_ready`. + +## Consequences + +**Positive**: M3-19 connects a real scheduler/provider execution to the authoritative GDA binding ledger without trusting stale evidence or mutating retained governance state. + +**Negative**: the repair exposes why a `memory` catalog cannot support a production durability claim. The local reset is intentionally limited to a dedicated, visibly empty catalog and is not a general recovery mechanism. + +**Revisit trigger**: replace this repair path with authenticated durable-catalog reconciliation when a protected executor, persistent provider identity, durable Gravitino backend, production PlatformGateway deployment and complete terminal evidence chain are available. diff --git a/docs/evidence/metadata-fabric-active-metadata-binding-reconciliation-2026-07-30.json b/docs/evidence/metadata-fabric-active-metadata-binding-reconciliation-2026-07-30.json new file mode 100644 index 00000000..b169ebbe --- /dev/null +++ b/docs/evidence/metadata-fabric-active-metadata-binding-reconciliation-2026-07-30.json @@ -0,0 +1,296 @@ +{ + "append_only_privileges_verified": true, + "attempt_observation_count": 2, + "attempt_states": [ + "submitted", + "success" + ], + "binding_id": "ff669a52-1271-55f0-a1b6-ee6f57ddb1ea", + "binding_persisted_to_gda_control": true, + "binding_record_sha256": "ef11bebe32f6b60d229698a1992108bb4eba5c0f64843150016a774bdb5544fa", + "binding_row_count": 1, + "binding_sha256": "7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b", + "callback_server_cleanup_verified": true, + "command_claimed_count": 1, + "command_completed_count": 1, + "command_id": "9c71b83d-2ac9-5a87-bec5-e40e30ab629e", + "command_status": "done", + "compiled_workflow_sha256": "ed435c3a4c5aa3d36cb8b25328224d864cf0205800f987606280b779fc77360b", + "contract_sha256": "012a7c86ba9fe53217e721ff7286b8f2a246b9394efd2999abbcd025e13ac7f5", + "correlation_variables": { + "gda_definition_sha256": "17302116f93c46a4e2c8c7ff82919dbbbca3d15bd8838381308b8933b419bb68", + "gda_definition_urn": "gda://metadata-authorization-local/definition/metadata-binding-reconciliation", + "gda_definition_version_id": "a9000000-0000-4000-8000-000000000002", + "gda_idempotency_key": "metadata-binding:cultural-districts:reconcile:v1", + "gda_run_id": "a9000000-0000-4000-8000-000000000003", + "gda_tenant_id": "metadata-authorization-local" + }, + "cross_tenant_read_blocked": true, + "dataset_absolute_path_committed": false, + "dataset_bundle": { + "components": [ + { + "component": ".cpg", + "sha256": "3ad3031f5503a4404af825262ee8232cc04d4ea6683d42c5dd0a2f2a27ac9824", + "size_bytes": 5 + }, + { + "component": ".dbf", + "sha256": "ee7c6c4c6957aea296b69d62118d416e5ee989aa77f7b98cf0fe580874ce5127", + "size_bytes": 44990 + }, + { + "component": ".prj", + "sha256": "b10dbe4d6d1de908d340f892c90b3d31a552630af3742bb515bfe1bd26124f2c", + "size_bytes": 176 + }, + { + "component": ".sbn", + "sha256": "7d0279465b18beec40308717e0ef0ea5701a586bc5c84e6a9aa309d5bc0ec99a", + "size_bytes": 308 + }, + { + "component": ".sbx", + "sha256": "019156149b2c7771ec0dd249c757dd7aa01a98075e246a77f77b080860c57333", + "size_bytes": 124 + }, + { + "component": ".shp", + "sha256": "6ac0d5c8c8db66fc0e2a74d8232b7779bd2454257df14efa2930e3dbc181aed0", + "size_bytes": 283640 + }, + { + "component": ".shp.xml", + "sha256": "8ef222ce1952552b366acf14a996e1c8cbdbe3eed0bb829dacfe7eafd068d948", + "size_bytes": 43100 + }, + { + "component": ".shx", + "sha256": "f3fbb6a7775ca833c066e3a3f2a332f99f979840045909ae187d08dac126a119", + "size_bytes": 260 + } + ], + "content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "format": "ESRI Shapefile", + "schema": "gda.spatial_dataset_bundle.v1", + "source_label": "chongqing-central-cultural-districts", + "spatial_inventory": { + "bounds": [ + 106.37987914500007, + 29.558008447000077, + 106.59532712300008, + 29.877271985000025 + ], + "crs": { + "authority": "EPSG", + "code": 4490, + "name": "China Geodetic Coordinate System 2000" + }, + "driver": "ESRI Shapefile", + "feature_count": 20, + "field_count": 33, + "geometry_type": "PolygonZ" + } + }, + "dataset_required_in_ci": false, + "dataset_source_committed": false, + "definition_sha256": "17302116f93c46a4e2c8c7ff82919dbbbca3d15bd8838381308b8933b419bb68", + "definition_version_id": "a9000000-0000-4000-8000-000000000002", + "deployment_applied": false, + "direct_binding_delete_blocked": true, + "direct_binding_update_blocked": true, + "dispatch_authorization_created": true, + "dispatch_authorization_id": "8e8f9737-3c07-5bd6-bc25-b61087fe3098", + "dispatch_execution_plan_artifact_id": "bcc07b86-9656-53c3-8a6b-5fe52c203612", + "durable_catalog_verified": false, + "errors": [], + "evidence_sha256": "e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33", + "exact_correlation_variable_readback_verified": true, + "exact_dispatch_authorization_replay_created": false, + "execution_callback_exact_request_verified": true, + "execution_callback_request_count": 1, + "execution_request_sha256": "027df912388bb4ab8e290100955ba42b83c4d570942523df38e4975b1424b1f8", + "external_correlation_count": 1, + "first_binding_commit_created": true, + "first_readback": { + "binding_candidate_sha256": "7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b", + "gravitino": { + "content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "identity": "gda_chongqing_m3_18/iceberg/cultural_heritage/cultural_districts", + "provider_revision": "shapefile-bundle-fd474fd65c8e4a71", + "resource_urn": "gda://metadata-authorization-local/dataset/chongqing-cultural-districts", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "snapshot_sha256": "c7998917cedb52f91c3ae0695223dbc4f505fd90bf9b78990991a60410c500b1" + }, + "mutation_count": 4, + "mutations": [ + "gravitino.catalog.reset_stale_empty_memory", + "gravitino.catalog.create", + "gravitino.schema.create", + "gravitino.table.create" + ], + "openmetadata": { + "content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "domain_refs": [ + "domain:natural-resources" + ], + "entity_id": "9d043410-02b5-487d-bb70-da5f3969a978", + "entity_version": "0.1", + "fully_qualified_name": "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts", + "owner_refs": [ + "team:data-platform" + ], + "resource_urn": "gda://metadata-authorization-local/dataset/chongqing-cultural-districts", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "snapshot_sha256": "a3ed5e2195c2f5847b5f5b59d78c8ba547c1f7170b3396cdd56b45f8559b0077", + "tag_refs": [ + "CulturalHeritage.CulturalDistrict", + "Sensitivity.Internal" + ] + }, + "status": "created" + }, + "force_rls_verified": true, + "gravitino_authentication_verified": false, + "gravitino_identity": "gda_chongqing_m3_18/iceberg/cultural_heritage/cultural_districts", + "gravitino_partial_projection_repaired": true, + "gravitino_provider_revision": "shapefile-bundle-fd474fd65c8e4a71", + "live_openlineage_emission_verified": false, + "matching_provider_instance_count": 1, + "oidc_verified": false, + "openmetadata_entity_id": "9d043410-02b5-487d-bb70-da5f3969a978", + "openmetadata_fqn": "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts", + "openmetadata_mutations_executed": false, + "partial_projection_detected": true, + "platform_run_status": "reconciling", + "platform_run_succeeded": false, + "production_ingestion_verified": false, + "production_ready": false, + "production_scheduler_submission_verified": false, + "protected_workload_identity_verified": false, + "provider_apply_approval_artifact_id": "a42bd367-bdb5-5d80-9319-86d460ad69f5", + "provider_apply_authorization_sha256": "8beae588e8cb831e5f25861ebb0536d4cff0ba7311dea17b6b0a94bd9bb7981b", + "provider_apply_authorized": true, + "provider_apply_execution_plan_artifact_id": "39f4c20d-81dd-5d59-bc81-12fa52904f4d", + "provider_apply_policy_decision_artifact_id": "29fc583e-1e42-54ff-93cd-8f5537697ee3", + "provider_evidence_artifact_id": "64a3fd6e-ab07-597c-a942-12f83971b4e1", + "provider_evidence_sha256": "6f1dfa0f9624b7b3877a4a85f0127138407bc198ac42f8fb26ccb99e1bfbe822", + "provider_minimum_privilege_verified": false, + "provider_mutations_executed": true, + "provider_objects_retained_for_readback": true, + "provider_port_forwards_cleanup_verified": true, + "provider_runtime": { + "context": "docker-desktop", + "namespace": { + "name": "gda-metadata-sandbox", + "uid": "812c9c5c-9ce2-409e-a6a8-a3f24785aa0c" + }, + "services": { + "gravitino": { + "name": "metadata-gravitino", + "type": "ClusterIP", + "uid": "b2fda004-0e59-40dd-b08e-45481593d179" + }, + "openmetadata": { + "name": "openmetadata", + "type": "ClusterIP", + "uid": "c903a96c-9aa9-420c-93d1-d2224df0cc85" + } + }, + "workloads": { + "gravitino": { + "kind": "StatefulSet", + "name": "metadata-gravitino", + "ready_replicas": 1, + "uid": "57a914fb-eca6-49f8-9c24-67c6d520a172" + }, + "openmetadata": { + "kind": "Deployment", + "name": "openmetadata", + "ready_replicas": 1, + "uid": "39bba7aa-b48e-4f5c-a9a8-c9cc9ca4ec1f" + } + } + }, + "provider_security": { + "gravitino": { + "auth_mode": "disabled", + "authentication_verified": false, + "version": "1.3.0" + }, + "openmetadata": { + "auth_mode": "local_basic_bootstrap", + "authenticated_principal": { + "id": "bfe46857-217e-43f5-a530-b0b33d73d624", + "is_admin": true, + "name": "admin" + }, + "minimum_privilege_verified": false + } + }, + "real_dataset_resource_version_bound": true, + "replay_binding_commit_created": false, + "replay_readback": { + "binding_candidate_sha256": "7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b", + "gravitino": { + "content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "identity": "gda_chongqing_m3_18/iceberg/cultural_heritage/cultural_districts", + "provider_revision": "shapefile-bundle-fd474fd65c8e4a71", + "resource_urn": "gda://metadata-authorization-local/dataset/chongqing-cultural-districts", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "snapshot_sha256": "c7998917cedb52f91c3ae0695223dbc4f505fd90bf9b78990991a60410c500b1" + }, + "mutation_count": 0, + "mutations": [], + "openmetadata": { + "content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "domain_refs": [ + "domain:natural-resources" + ], + "entity_id": "9d043410-02b5-487d-bb70-da5f3969a978", + "entity_version": "0.1", + "fully_qualified_name": "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts", + "owner_refs": [ + "team:data-platform" + ], + "resource_urn": "gda://metadata-authorization-local/dataset/chongqing-cultural-districts", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "snapshot_sha256": "a3ed5e2195c2f5847b5f5b59d78c8ba547c1f7170b3396cdd56b45f8559b0077", + "tag_refs": [ + "CulturalHeritage.CulturalDistrict", + "Sensitivity.Internal" + ] + }, + "status": "no_op" + }, + "resource_version_content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "run_id": "a9000000-0000-4000-8000-000000000003", + "scheduler_provider": { + "api_profile": "3.4", + "architecture": "arm64", + "image": "apache/dolphinscheduler-standalone-server:3.4.2", + "image_id": "sha256:485a1b37dd1c4088c8c8335f9fccbd229e5e703c32e21f318eb00cbb60b1af9d", + "name": "apache-dolphinscheduler", + "project_code": 180212973717376, + "server_version": "3.4.2", + "terminal_state": "SUCCESS", + "workflow_definition_code": 180212973810560, + "workflow_definition_version": 1, + "workflow_instance_id": 1 + }, + "scheduler_success_readback_verified": true, + "scheduler_triggered_provider_readback_verified": true, + "schema": "gda.active_metadata_binding_reconciliation_evidence.v1", + "source_created_apply_verified": true, + "source_execution_evidence_sha256": "397c0f1a29f53935c5508155470c4972cfc50260f0d0686fb48cb3f75519b17b", + "standalone_container_cleanup_verified": true, + "status": "local_scheduler_binding_reconciliation_verified", + "stored_binding_matches": true, + "submitted_observation_count": 1, + "success_observation_count": 1, + "temporary_database_cleanup_verified": true, + "tls_verified": false, + "writes_to_gda_control": true, + "writes_to_legacy": false +} diff --git a/docs/roadmap-ar0-platform-truth-2026-07-24.md b/docs/roadmap-ar0-platform-truth-2026-07-24.md index 419e240d..9d095a0b 100644 --- a/docs/roadmap-ar0-platform-truth-2026-07-24.md +++ b/docs/roadmap-ar0-platform-truth-2026-07-24.md @@ -199,7 +199,7 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 当前完成仅指本地合同、授权 evidence、outbox/callback 代码、数据库成功终局门、托管 worker 代码、默认关闭的部署模板及离线 activation/release preflight、candidate/registry/provenance/artifact-release/live observation evidence gate、合成 golden slice、定向测试、真实 PostgreSQL 16 事务边界和 canonical mainline 治理。`candidate_validated`、`registry_subject_bound`、本地合成 `provenance_verified`、`ready_for_activation`、`ready_for_staging_apply`、`verified_for_staging_apply` 和本地 live collection 都不等于真实镜像已 attested 或 staging 已部署；真实 IAM/OIDC 与 service token 生命周期、首次 GHCR publish/verify、真实 provenance artifact verify、registry-backed live staging revision、worker/callback 扩容运行、golden slice staging 运行链、受保护 release/live evidence provenance、独立 DolphinScheduler metadata PostgreSQL 和真实数据终局证据仍属于 4.7 后续切片。 -### 4.8 Metadata Fabric Bridge M1 + M2 + M3-18（本地 scheduler-triggered provider projection/read-back 已验证，生产验证待执行） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-19（本地 provider binding reconciliation 已验证，生产验证待执行） 第八块回到 AR-1 的 metadata control plane，以 [ADR-036](architecture-decisions/adr-036-read-only-metadata-fabric-bridge-contract.md) 固定 OpenMetadata + Gravitino + GDA Control Ledger 的首条 table slice： @@ -238,8 +238,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 33. [ADR-062](architecture-decisions/adr-062-atomic-active-metadata-authorization-and-dispatch.md) 新增 migration 101、内容绑定 `MetadataActivationAuthorization` 与专用 PlatformGateway 提升 API。`awaiting_authorization` request 只有在同租户真实 ResourceVersion/content hash、`metadata_fabric.projection_plan` DefinitionVersion、accepted workload Run/input、execution-plan Artifact、allow PolicyDecision、独立 approved Approval 与第四方 authorizer 完整匹配时，才能与一个 pending DolphinScheduler dispatch 同事务提交；普通 `request_dispatch` 绕过和无 command 的孤立授权均回滚。真实重庆中心城区历史文化街区 Shapefile 8 组件被规范化为不含路径的 inventory，20 个 `PolygonZ`、33 字段、EPSG:4490，其 bundle SHA `fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007` 精确成为 ResourceVersion content hash。PostgreSQL 演练最终只有 1 个 authorization、1 个 pending command，精确 replay 不新增，FORCE RLS、function-only INSERT、直接 UPDATE/DELETE 拒绝均通过。contract fingerprint 为 `cef78f91058a8529f4e86330790e714b52b73725a45ffe4dc9eded35bc8ccfa4`，evidence fingerprint 为 `6ae387240e3bcebaafe2ad7acc73f4e09d53df2e73b2ec63cd92edbc262d831e`；源数据/绝对路径不入 Git、CI 不依赖本机文件，scheduler submission、provider apply/mutation/ingestion 与 production readiness 仍为 `false`。 34. [ADR-063](architecture-decisions/adr-063-local-authorized-active-metadata-scheduler-delivery.md) 将同一重庆 ResourceVersion 指纹带入 provider-native DolphinScheduler `3.4.2` Shell DAG，先发布/release 无副作用 workflow，再把真实返回的 project/workflow code、version 与 compiled SHA 固化为 execution-plan binding。M3-16 授权原子创建的 command 被既有 `DolphinSchedulerCommandConsumer` 认领并真实提交；provider 回读精确包含 6 个 GDA definition/Run 关联变量，且只找到 1 个匹配实例。终态 `SUCCESS` 被记录为 `submitted/success` 两条 attempt observations 和 1 个 external correlation，PlatformRun 只到 `reconciling`、不进入 `succeeded`；authorization replay 不新增。官方 standalone 容器与临时 PostgreSQL 均清理。contract fingerprint 为 `dcf97c8fa002e9fe6b6bc3a7603ee2ebd5ddb053544801ce35143a095e648edb`，evidence fingerprint 为 `00d4ea062c40f8d97557eadc357a36c6d1ccd56e12a94a44694113681e5d55f4`。该结论只证明本地 scheduler control-plane delivery/read-back；受保护身份、常驻 controller、生产 scheduler metadata/HA、provider apply/mutation/ingestion 与 production readiness 仍为 `false`。 35. [ADR-064](architecture-decisions/adr-064-local-scheduler-triggered-active-metadata-projection-execution.md) 将 M3-17 的真实 dispatch 与 M3-2 provider client 串成单条本地执行链：官方 DolphinScheduler Shell task 经 Docker Desktop host gateway 向短生命周期 executor 发送 1 个内容绑定请求；executor 在内存中验证独立 `metadata_fabric.apply` PolicyDecision/Approval，首次向 OpenMetadata/Gravitino 创建 10 个 projection 层级对象并回读相同重庆 ResourceVersion，随后精确 replay 为 `no_op/0 mutations`。两次 read-back 的 OpenMetadata UUID、Gravitino identity 与 binding candidate 完全一致；scheduler 仍形成 `submitted/success` 两条 observation，PlatformRun 保持 `reconciling`。callback、两条 port-forward、standalone 容器和临时数据库均清理，provider projection 保留。contract fingerprint 为 `a6632ae0edd4d4f3389129a8c07411a8d101ae56fbfc26b03fb0aff6928bb7bd`，evidence fingerprint 为 `397c0f1a29f53935c5508155470c4972cfc50260f0d0686fb48cb3f75519b17b`。该结论不证明 protected identity、provider minimum privilege、Gravitino authentication/TLS、生产 scheduler/executor、持久 binding、production ingestion 或 `production_ready`。 +36. [ADR-065](architecture-decisions/adr-065-local-active-metadata-binding-reconciliation.md) 在提交 M3-18 binding 前先验证 retained OpenMetadata UUID/FQN/version/content/governance/snapshot 完全一致。Gravitino `memory` catalog 重启后出现 connector 空状态与 provider entity index 残留的分裂；M3-19 只在专用 catalog 配置精确且可见 schema inventory 为空时执行 provider-native reset，并以 4 个 `gravitino.*` mutations 重建 catalog/schema/table，OpenMetadata 零写入。即时 replay 为 `no_op/0 mutations` 且 binding SHA 仍为 `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b`。PlatformGateway 首次 binding commit `created=true`、重放 `created=false`、仅 1 行，FORCE RLS、跨租户隔离、append-only 和 direct UPDATE/DELETE 拒绝均通过；Run 保持 `reconciling`，所有临时资源清理。contract fingerprint 为 `012a7c86ba9fe53217e721ff7286b8f2a246b9394efd2999abbcd025e13ac7f5`，evidence fingerprint 为 `e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33`。`durable_catalog_verified=false`，该结论不证明生产 identity/catalog/executor/binding deployment、production ingestion 或 `production_ready`。 -此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file；M3-13 只证明单次本地 append 在 provider 200 响应丢失并映射为 commit-state-unknown 后，可以由即时 table readback 判定 committed 且不重提，不覆盖持久 controller、进程崩溃、并发写或任意 mutation；M3-14 只证明 ResourceVersion 注册与 Active Metadata 事件在本地 PostgreSQL 同事务创建，并验证租户/workload scoped claim/retry/complete；M3-15 只证明默认零副本 managed consumer 的代码/部署边界，以及本地 PostgreSQL 中 inert activation request 与 event completion 的原子性；M3-16 只证明本地真实数据 content fingerprint、证据绑定授权与 pending command 的 PostgreSQL 原子性；M3-17 只证明本地 standalone 中既有 consumer/adapter 的真实 submission、精确 correlation read-back 和 provider success observation；M3-18 只证明同一 Docker Desktop 主机上 scheduler 通过 ephemeral HTTP executor 触发 bootstrap-admin/unauthenticated providers 的一次创建和同进程零写 replay，未部署受保护 workload identity/authorization controller、持久 executor 或生产 scheduler/provider path。生产持久 binding、ResourceVersion 和 legacy authority 都未写入；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 +此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file；M3-13 只证明单次本地 append 在 provider 200 响应丢失并映射为 commit-state-unknown 后，可以由即时 table readback 判定 committed 且不重提，不覆盖持久 controller、进程崩溃、并发写或任意 mutation；M3-14 只证明 ResourceVersion 注册与 Active Metadata 事件在本地 PostgreSQL 同事务创建，并验证租户/workload scoped claim/retry/complete；M3-15 只证明默认零副本 managed consumer 的代码/部署边界，以及本地 PostgreSQL 中 inert activation request 与 event completion 的原子性；M3-16 只证明本地真实数据 content fingerprint、证据绑定授权与 pending command 的 PostgreSQL 原子性；M3-17 只证明本地 standalone 中既有 consumer/adapter 的真实 submission、精确 correlation read-back 和 provider success observation；M3-18 只证明同一 Docker Desktop 主机上 scheduler 通过 ephemeral HTTP executor 触发 bootstrap-admin/unauthenticated providers 的一次创建和同进程零写 replay；M3-19 只证明同一主机上 exact OpenMetadata + absent Gravitino 的受限修复、即时 no-op replay 和临时 PostgreSQL binding commit，且 `memory` catalog reset 明确不等于 durable catalog recovery。生产持久 binding deployment、ResourceVersion 和 legacy authority 都未切换；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index 105ec46e..373dfb23 100644 --- a/docs/system-of-record-matrix-2026-07-24.md +++ b/docs/system-of-record-matrix-2026-07-24.md @@ -2,9 +2,9 @@ 日期：2026-07-30 -阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity、M3-7 production identity readiness contract、M3-8 local Gravitino JDBC restart continuity、M3-9 local Spark/Iceberg REST interoperability、M3-10 local cross-node Spark/object-store interoperability、M3-11 production object-store readiness contract、M3-12 local Spark commit-failure recovery、M3-13 local uncertain-commit reconciliation、M3-14 local Active Metadata transactional outbox、M3-15 local durable activation request consumer、M3-16 local real-data authorization/dispatch promotion、M3-17 local real scheduler delivery/read-back 与 M3-18 local scheduler-triggered provider projection/read-back 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler/executor 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1 至 M3-19 local Active Metadata binding reconciliation 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler/executor 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-active-metadata-projection-execution` +适用分支：`feat/ar1-metadata-fabric-active-metadata-binding-reconciliation` ## 判定规则 @@ -20,7 +20,7 @@ | SQL schema 历史 | PostgreSQL `schema_migrations`，以完整 migration ID + checksum 为权威 | migration CLI 的 JSON 报告 | 保持现有 ledger；任何 drift fail closed | Data Platform | AR-0，已验证 | | 部署配置策略 | Compose/K8s/进程环境；`platform_truth.CONFIG_SPECS` 定义关键类型与策略；DolphinScheduler worker 与 Active Metadata consumer 均有默认零副本、外部 ConfigMap/Secret 驱动的 Kustomize 模板和静态 validator，前者另有 staging activation preflight | `.env` 仅补默认；脱敏 snapshot、Secret key attestation、未扩容 Deployment 和 `ready_for_activation` 都是观测/模板 | 版本化 DeploymentProfile + secret reference；部署环境始终优先；模板或 preflight 通过都不等于环境已启用 | Platform/SRE/Security | AR-0，部分实现；worker 模板/preflight 本地已验证 | | 环境发布与晋级 | 本地 candidate/registry/provenance/release/live 合同已绑定 publisher、verifier、OCI 和 manifest identity；canonical `main@0182406`、archive refs、三组 active ruleset 与 `staging-provenance` protected environment 已建立，但尚无成功 publisher/verifier 或 deployment | 旧 mainline、feature branch、CI artifact、JSON、离线 report 和合成 `verified_for_staging_apply` 都不能单独成为发布权威；publisher SHA、verifier SHA 与 branch lineage 必须分别验证 | 由受保护 environment 的 DeploymentRevision 绑定 OCI、provenance artifact、release manifest 与全部 live verdict | Platform/SRE/Security/Repository Owner | AR-1 mainline 治理已恢复 -> 首次 GHCR publish/verify -> 真实 staging | -| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；Metadata Fabric recovery/metrics/policy/catalog/interoperability/failure/uncertain-commit/outbox/consumer/authorization/scheduler-delivery/projection-execution rehearsal 均为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection evidence | PlatformRun ledger 唯一登记最终状态；activation request 只拥有待授权意图；M3-18 scheduler/provider `SUCCESS` 和 read-back 只形成 attempt/local projection evidence 并把 Run 留在 `reconciling`，不是平台成功终局权威；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 consumer、M3-16 authorization、M3-17 delivery 与 M3-18 projection execution 本地验证；常驻受保护 executor -> staging 控制链待接入 | +| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；Metadata Fabric recovery/metrics/policy/catalog/interoperability/failure/uncertain-commit/outbox/consumer/authorization/scheduler-delivery/projection-execution/binding-reconciliation rehearsal 均为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection/binding evidence | PlatformRun ledger 唯一登记最终状态；activation request 只拥有待授权意图；M3-19 scheduler/provider `SUCCESS`、repair/read-back 和 binding commit 仍把 Run 留在 `reconciling`，不是平台成功终局权威；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 consumer、M3-16 authorization、M3-17 delivery、M3-18 projection execution 与 M3-19 binding reconciliation 本地验证；常驻受保护 executor -> staging 控制链待接入 | | 原始文件/对象 | 当前 local uploads、S3/MinIO/OBS 均可能被直接写入，权威边界未统一 | 临时上传、下载缓存、预览文件 | Landing object 以 immutable URI + checksum + retention 为权威；本地 scratch 可删除 | Data Platform | AR-2 | | 湖仓表与 snapshot | Iceberg/STAC/S3A 有局部实现，尚无通用发布权威 | STAC item、GeoParquet export | Iceberg catalog snapshot 是分析表版本权威；对象是物理内容，STAC 是发现投影 | Data Platform | AR-2 | | 在线空间数据 | PostGIS 业务表是当前编辑/查询事实，部分临时表混入 | Martin MVT、API JSON、导出文件 | 已批准 DataProductVersion 物化到 PostGIS；不能由瓦片或临时表反向定义产品版本 | GIS/Data Platform | AR-2 -> AR-4 | @@ -31,7 +31,7 @@ | Definition | `gda_control.platform_definition_version` 已绑定 definition ResourceVersion、完整逻辑 hash 和原子 gateway registration；3.4.2 adapter 可编译、创建并上线 provider DAG；binding 已以 append-only `execution_plan` Artifact 持久化并可按 tenant + artifact UUID 读取，旧 workflow/template/YAML 仍在写入 | 编辑器状态、DolphinScheduler DAG/definition | 旧 workflow 必须规范化并完整 hash 后才可形成 PlatformDefinitionVersion；provider binding 作为 ExecutionPlanArtifact/evidence，不可反写 definition | DataOps | AR-1 binding persistence 代码已验证 -> staging 调用链待验收 | | Run 最终状态 | `gda_control.platform_run/event` 已实现受控 submit/read/CAS；通用 transition 已禁止 `succeeded`，专用数据库 finalizer 只接受精确 workload、DolphinScheduler success observation、内容匹配 output、独立 passed QualityResult/evidence 和 input-to-output lineage；adapter standalone API path 已验证，但端到端 staging 尚未完成，legacy 路径继续运行 | Redis progress、日志、DolphinScheduler state、attempt observation | 旧 run 到 PlatformRun 永久 prohibited；已有 PlatformRun correlation 时才可转为 observation；provider 终态只进入 `reconciling`，ledger 经证据门唯一裁决成功 | DataOps/AgentOps | AR-1 success authority 本地/PostgreSQL 已验证 -> staging/生产切换待验收 | | 调度与补数 | APScheduler、自进化 scheduler 和调用方定时逻辑并存；DolphinScheduler POC 只验证 manual start/list/variables/STOP | UI schedule 列表 | DolphinScheduler 管 DataOps schedule/complement；Temporal 只管需要 durable signal/compensation 的 Agent/GWM workflow | DataOps/AgentOps | AR-1 manual correlation 已验证；schedule/complement/failover 待验收 | -| 事件交付 | Standards outbox 已数据库耐久；`platform_command_outbox` 支持 DolphinScheduler dispatch/reconcile；M3-14 `metadata_change_outbox` 将新 ResourceVersion 与内容绑定事件同事务写入；M3-15 managed consumer 同事务创建 inert request；M3-16 将真实 ResourceVersion、Definition/Run/plan/PolicyDecision/Approval/authorizer 绑定后与一个 pending dispatch 同事务提交；M3-17 由既有 consumer 向本地真实 DolphinScheduler 提交并回读；M3-18 的单个任务触发独立授权的 provider apply/read-back 与零写 replay | command/metadata delivery status、消费者 claim、activation intent/request/authorization、FrameworkAttemptObservation、provider instance/correlation、provider apply/read-back、worker status JSON、WebSocket 消息 | command/event 与源事实同事务入 outbox，幂等 consumer 交付；Active Metadata consumer 不能授权或执行；dispatch 与 provider apply 分别授权；scheduler/provider `SUCCESS` 仍须经平台终局 evidence gate | Platform/Integrations/Metadata Platform | AR-1 command worker、M3-14/M3-15/M3-16/M3-17/M3-18 本地已验证 -> protected authorizer/worker/executor identity、生产 scheduler/provider 和 production scale-up 待执行 | +| 事件交付 | Standards outbox 已数据库耐久；`platform_command_outbox` 支持 DolphinScheduler dispatch/reconcile；M3-14 `metadata_change_outbox` 将新 ResourceVersion 与内容绑定事件同事务写入；M3-15 managed consumer 同事务创建 inert request；M3-16 将真实 ResourceVersion、Definition/Run/plan/PolicyDecision/Approval/authorizer 绑定后与一个 pending dispatch 同事务提交；M3-17 由既有 consumer 向本地真实 DolphinScheduler 提交并回读；M3-18 的单个任务触发独立授权的 provider apply/read-back 与零写 replay；M3-19 在 exact OpenMetadata 前提下修复缺失 Gravitino projection、零写 replay 后提交 immutable binding | command/metadata delivery status、消费者 claim、activation intent/request/authorization、FrameworkAttemptObservation、provider instance/correlation、provider apply/read-back/binding、worker status JSON、WebSocket 消息 | command/event 与源事实同事务入 outbox，幂等 consumer 交付；Active Metadata consumer 不能授权或执行；dispatch 与 provider apply 分别授权；scheduler/provider `SUCCESS` 和 binding persistence 仍须经平台终局 evidence gate | Platform/Integrations/Metadata Platform | AR-1 command worker、M3-14 至 M3-19 本地已验证 -> protected authorizer/worker/executor identity、生产 scheduler/provider 和 production scale-up 待执行 | | 质量结果 | `gda_control.quality_result` 已提供 tenant RLS、append-only gateway 写入，绑定 Run、output ResourceVersion、rule version、verdict、metrics、evidence Artifact 和独立 evaluator；standards、QC、MMFE 专项结果仍未迁移 | dashboard、OpenMetadata quality summary | GDA ledger 保存产品终局所需的不可变 verdict/evidence；OpenMetadata 与 UI 只作可重建发现投影；旧结果缺稳定版本和证据时不得升级为终局依据 | Governance/DataOps | AR-1 最小成功证据已验证 -> 真实规则/staging 待接入 | | 标准与语义定义 | `std_*`、semantic registry 和 YAML 共同存在，生命周期未统一 | prompt/context、搜索索引 | 版本化 Standard/SemanticDefinition 经审批后为权威；Agent context 只消费批准版本 | Governance | AR-1 -> AR-3 | | 身份与权限 | Chainlit user 可显式绑定 tenant；versioned API 从认证 principal 派生 SubjectContext；`gda_control_gateway` 是 non-login/non-bypass 最小权限角色；Run 可引用强类型 PolicyDecision/Approval Artifact；M3-5/M3-6 分别验证本地 provider scoped grant、越权拒绝和 credential rotation/revocation；M3-7 已冻结生产 OIDC/workload/tenant binding、TLS、持久 catalog 与 attestation contract，但 40 个外部输入仍 blocked；M3-8 证明同一 Gravitino Basic role 在本地 JDBC restart 后连续 | session/cache、前端菜单权限、本地 provider identity/JDBC restart evidence、pending profile 与合成 readiness report | IdP/workload identity 提供真实 service identity；PolicyDecision/Approval 继续绑定不可变资源与 execution plan；只有 fresh protected attestation 可派生双 provider production identity claims，profile、Basic/JWT evidence、restart continuity 或人工批准均不可替代 | Security | AR-1 local identities/persistence + production readiness contract 已验证 -> protected 双 provider IAM/attestation 待执行 | @@ -62,6 +62,7 @@ 17. M3-16 只证明本地重庆 Shapefile bundle content fingerprint 与 ResourceVersion 精确绑定，以及 authorization + pending dispatch 的 PostgreSQL 原子性；真实数据不自动形成 authority 或授权，源文件/绝对路径不入 Git、CI 不依赖本机路径。受保护 authorizer identity、常驻 promotion controller、真实 DolphinScheduler submission/read-back、provider apply/mutation/ingestion、告警/SLO 与 production readiness 仍未验证。 18. M3-17 只证明官方 DolphinScheduler `3.4.2` standalone 中既有 adapter/consumer 对精确授权 command 的本地真实 submission、6 个 GDA correlation variables 回读、单实例 `SUCCESS` 与 `submitted/success` attempt evidence；PlatformRun 留在 `reconciling`。本地 workflow/project/instance 是 scheduler control-plane 对象，但没有授权或执行 OpenMetadata、Gravitino、lakehouse、legacy 或源数据 mutation。protected workload identity、独立 scheduler metadata PostgreSQL/HA/backup、常驻 deployment、production submission、provider apply/ingestion、告警/SLO 与 production readiness 仍未验证。 19. M3-18 只证明同一 Docker Desktop 主机上的官方 DolphinScheduler task 经 ephemeral HTTP executor 触发一次独立授权的本地 OpenMetadata/Gravitino projection：首次 10 mutations，精确 replay 为 `no_op/0 mutations`，两次 provider read-back 与 binding candidate 一致，PlatformRun 仍为 `reconciling`。OpenMetadata 使用 bootstrap admin，Gravitino 无认证且为 memory catalog；callback 不是 protected workload identity 或生产服务。生产 scheduler/executor、双 provider minimum privilege/OIDC/TLS、持久 binding/catalog、告警/SLO、production ingestion 与 production readiness 仍未验证。 +20. M3-19 只允许在 retained OpenMetadata UUID/FQN/version/content/governance/snapshot 完全匹配时修复缺失 Gravitino target；专用 `memory` catalog 只有配置精确且可见 schema inventory 为空才可 provider-native reset。修复限于 4 个 `gravitino.*` mutations，OpenMetadata 零写入，即时 replay 为 `no_op/0 mutations`，binding 通过临时 PostgreSQL PlatformGateway 幂等追加且 Run 留在 `reconciling`。这不证明 durable catalog、protected identity、生产 executor/scheduler/provider、生产 binding deployment/ingestion 或 terminal success。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -103,6 +104,7 @@ - Metadata Fabric M3-16 已新增 migration 101、内容绑定 `MetadataActivationAuthorization`、PlatformGateway atomic authorize/dispatch API 与 activation dispatch 数据库 guard。重庆中心城区历史文化街区 Shapefile 8 组件被转换为 path-free inventory，20 个 `PolygonZ`、33 字段、EPSG:4490，bundle SHA `fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007` 精确成为 ResourceVersion content hash。真实 PostgreSQL 16 演练证明普通 dispatch 绕过被拒绝、无 command 的授权因 deferred FK 回滚，最终只有 1 个 authorization、1 个 pending command，精确 replay 不新增，FORCE RLS、function-only INSERT 与直接 UPDATE/DELETE 拒绝均通过。contract fingerprint 为 `cef78f91058a8529f4e86330790e714b52b73725a45ffe4dc9eded35bc8ccfa4`，evidence fingerprint 为 `6ae387240e3bcebaafe2ad7acc73f4e09d53df2e73b2ec63cd92edbc262d831e`。`deployment_applied=false`、`production_workload_identity_verified=false`、`provider_apply_authorized=false`、`provider_mutations_executed=false`、`production_scheduler_submission_verified=false`、`production_ingestion_verified=false`、`production_ready=false`。 - Metadata Fabric M3-17 已将相同重庆 ResourceVersion fingerprint 带入 provider-native DolphinScheduler binding，使用官方 standalone `3.4.2` 真实创建/release 无副作用 Shell workflow。M3-16 authorization 原子创建的 pending command 被既有 consumer 认领并完成；provider 回读 6 个受控 GDA definition/Run variables、1 个匹配实例和 `SUCCESS`，GDA Control 记录精确 `submitted/success` 两条 observations、1 个 external correlation，Run 保持 `reconciling` 而非 `succeeded`。authorization replay 不新增，临时容器与数据库均清理。contract fingerprint 为 `dcf97c8fa002e9fe6b6bc3a7603ee2ebd5ddb053544801ce35143a095e648edb`，evidence fingerprint 为 `00d4ea062c40f8d97557eadc357a36c6d1ccd56e12a94a44694113681e5d55f4`。`deployment_applied=false`、`production_workload_identity_verified=false`、`provider_apply_authorized=false`、`provider_mutations_executed=false`、`production_scheduler_submission_verified=false`、`production_ingestion_verified=false`、`production_ready=false`。 - Metadata Fabric M3-18 已由同一真实 DolphinScheduler `SUCCESS` 实例触发短生命周期 projection executor；独立 `metadata_fabric.apply` authorization 在 provider 调用前验证。重庆 ResourceVersion fingerprint 被写入 OpenMetadata `gda_chongqing_m3_18.cultural_heritage.published.cultural_districts` 与 Gravitino `gda_chongqing_m3_18/iceberg/cultural_heritage/cultural_districts`；首次 10 mutations，精确 replay 为 `no_op/0 mutations`，两次 read-back 的 OpenMetadata UUID `9d043410-02b5-487d-bb70-da5f3969a978`、Gravitino identity 与 binding SHA `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b` 一致。callback、两条 port-forward、standalone 容器和临时数据库已清理，Run 保持 `reconciling`。contract fingerprint 为 `a6632ae0edd4d4f3389129a8c07411a8d101ae56fbfc26b03fb0aff6928bb7bd`，evidence fingerprint 为 `397c0f1a29f53935c5508155470c4972cfc50260f0d0686fb48cb3f75519b17b`。`protected_workload_identity_verified=false`、`provider_minimum_privilege_verified=false`、`gravitino_authentication_verified=false`、`production_scheduler_submission_verified=false`、`production_ingestion_verified=false`、`production_ready=false`。 +- Metadata Fabric M3-19 已在 retained OpenMetadata snapshot 精确匹配 M3-18 后，识别 Gravitino `memory` connector 空状态与持久 entity index 残留。专用 catalog 配置精确且 visible schema inventory 为零后，scheduler callback 只执行 `gravitino.catalog.reset_stale_empty_memory`、catalog/schema/table create 四次 mutation；OpenMetadata 零写入，立即 replay 为 `no_op/0 mutations`，binding SHA 保持 `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b`。PlatformGateway 首次 commit `created=true`、重放 `created=false`、仅 1 行，FORCE RLS、跨租户隔离、append-only 和 direct UPDATE/DELETE 拒绝通过；Run 保持 `reconciling`，临时资源均清理。contract fingerprint 为 `012a7c86ba9fe53217e721ff7286b8f2a246b9394efd2999abbcd025e13ac7f5`，evidence fingerprint 为 `e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33`。`durable_catalog_verified=false`、`production_ingestion_verified=false`、`production_ready=false`。 ## 下一验收证据 diff --git a/scripts/metadata-fabric-active-metadata-binding-reconciliation.sh b/scripts/metadata-fabric-active-metadata-binding-reconciliation.sh new file mode 100755 index 00000000..5c145d81 --- /dev/null +++ b/scripts/metadata-fabric-active-metadata-binding-reconciliation.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +COMMON_GIT_DIR="$(git -C "$ROOT" rev-parse --path-format=absolute --git-common-dir 2>/dev/null || true)" +SHARED_ROOT="" +if [ -n "$COMMON_GIT_DIR" ]; then + SHARED_ROOT="$(cd "$COMMON_GIT_DIR/.." && pwd)" +fi + +if [ -n "${PYTHON:-}" ]; then + : +elif [ -x "$ROOT/.venv/bin/python" ]; then + PYTHON="$ROOT/.venv/bin/python" +elif [ -n "$SHARED_ROOT" ] && [ -x "$SHARED_ROOT/.venv/bin/python" ]; then + PYTHON="$SHARED_ROOT/.venv/bin/python" +else + PYTHON="python" +fi + +cd "$ROOT" +exec "$PYTHON" -m data_agent.metadata_fabric_active_metadata_binding_reconciliation "$@"