From aa11742cd5d11caf646afbf13dd24e363827dee2 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Thu, 30 Jul 2026 20:22:29 +0800 Subject: [PATCH] feat: execute active metadata projection through scheduler --- .github/workflows/ci.yml | 5 + ...ic_active_metadata_projection_execution.py | 1522 +++++++++++++++++ data_agent/platform_truth.py | 22 +- ...ic_active_metadata_projection_execution.py | 130 ++ data_agent/test_platform_truth.py | 6 + ...ed-active-metadata-projection-execution.md | 92 + ...adata-projection-execution-2026-07-30.json | 285 +++ docs/roadmap-ar0-platform-truth-2026-07-24.md | 5 +- docs/system-of-record-matrix-2026-07-24.md | 12 +- ...ic-active-metadata-projection-execution.sh | 22 + 10 files changed, 2093 insertions(+), 8 deletions(-) create mode 100644 data_agent/metadata_fabric_active_metadata_projection_execution.py create mode 100644 data_agent/test_metadata_fabric_active_metadata_projection_execution.py create mode 100644 docs/architecture-decisions/adr-064-local-scheduler-triggered-active-metadata-projection-execution.md create mode 100644 docs/evidence/metadata-fabric-active-metadata-projection-execution-2026-07-30.json create mode 100755 scripts/metadata-fabric-active-metadata-projection-execution.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 39e11447..6b64059e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -33,6 +33,7 @@ on: - feat/ar1-metadata-fabric-active-metadata-consumer - feat/ar1-metadata-fabric-active-metadata-authorization - feat/ar1-metadata-fabric-active-metadata-scheduler-delivery + - feat/ar1-metadata-fabric-active-metadata-projection-execution env: PYTHON_VERSION: "3.13" @@ -181,6 +182,9 @@ jobs: - name: Validate metadata fabric Active Metadata scheduler delivery evidence run: python -m data_agent.metadata_fabric_active_metadata_scheduler_delivery validate + - name: Validate metadata fabric Active Metadata projection execution evidence + run: python -m data_agent.metadata_fabric_active_metadata_projection_execution validate + - name: Validate Active Metadata consumer deployment boundary run: python -m data_agent.active_metadata_consumer_deployment validate @@ -262,6 +266,7 @@ jobs: data_agent/test_metadata_fabric_active_metadata_consumer.py \ 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_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_projection_execution.py b/data_agent/metadata_fabric_active_metadata_projection_execution.py new file mode 100644 index 00000000..e1ec1dc7 --- /dev/null +++ b/data_agent/metadata_fabric_active_metadata_projection_execution.py @@ -0,0 +1,1522 @@ +"""Execute an authorized Active Metadata projection through DolphinScheduler. + +This M3-18 local-only rehearsal extends the M3-17 delivery boundary. An +authorized provider-native DolphinScheduler task calls a short-lived local +projection executor. The executor validates an independent, content-bound +``metadata_fabric.apply`` authorization before reusing the M3-2 provider +clients to create and read back OpenMetadata and Gravitino projections. It +then applies the exact plan again and requires a zero-mutation replay. + +The callback is ephemeral HTTP on the Docker Desktop host gateway, +OpenMetadata still uses its local bootstrap administrator, and Gravitino is +still unauthenticated. Provider and scheduler success leave PlatformRun in +``reconciling``; no production readiness claim is made. +""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import shlex +import threading +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from typing import Any, Literal, Self +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator +from sqlalchemy import create_engine, text + +from . import metadata_fabric_active_metadata_authorization as authorization +from . import metadata_fabric_active_metadata_scheduler_delivery as delivery +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 ( + MetadataActivationAuthorization, + 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 .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, PlatformGateway +from .spatial_dataset_bundle import validate_shapefile_bundle_inventory + +CONTRACT_SCHEMA = "gda.active_metadata_projection_execution_contract.v1" +REQUEST_SCHEMA = "gda.active_metadata_projection_execution_request.v1" +EVIDENCE_SCHEMA = "gda.active_metadata_projection_execution_evidence.v1" +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_SCHEDULER_DEPENDENCY_PATH = delivery.DEFAULT_EVIDENCE_PATH +DEFAULT_INGESTION_DEPENDENCY_PATH = replay.DEFAULT_EVIDENCE_PATH +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT / "docs/evidence/metadata-fabric-active-metadata-projection-execution-2026-07-30.json" +) +DEFAULT_WRAPPER_PATH = REPO_ROOT / "scripts/metadata-fabric-active-metadata-projection-execution.sh" +TENANT = authorization.TENANT +SOURCE_ID = authorization.SOURCE_ID +DEFINITION_ID = UUID("a8000000-0000-4000-8000-000000000002") +RUN_ID = UUID("a8000000-0000-4000-8000-000000000003") +TASK_CODE = 180000000000001 +WORKER = "worker:active-metadata-projection-execution-1" +RUNNER = delivery.RUNNER +POLICY_EVALUATOR = delivery.POLICY_EVALUATOR +AUTHORIZER = delivery.AUTHORIZER +APPROVER = delivery.APPROVER +CALLBACK_PATH = "/v1/execute-projection" +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", + "oidc_verified", + "tls_verified", + "binding_persisted_to_gda_control", + "live_openlineage_emission_verified", + "production_scheduler_submission_verified", + "production_ingestion_verified", + "production_ready", +) + + +class ActiveMetadataProjectionExecutionError(RuntimeError): + """The local scheduler-to-provider projection rehearsal failed closed.""" + + +class ProjectionExecutionRequest(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + request_schema: Literal["gda.active_metadata_projection_execution_request.v1"] = Field( + default=REQUEST_SCHEMA, alias="schema" + ) + tenant_id: str + run_id: UUID + definition_version_id: UUID + source_resource_version_id: UUID + content_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + apply_plan_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + request_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + + @model_validator(mode="after") + def _fingerprint(self) -> Self: + stable = self.model_dump(mode="json", by_alias=True, exclude={"request_sha256"}) + if self.request_sha256 != canonical_json_fingerprint(stable): + raise ValueError("projection execution request fingerprint does not match") + return self + + +@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: MetadataActivationAuthorization + + +def _file_sha256(path: Path) -> str: + return replay.recovery._file_sha256(path) + + +def _file_record(path: Path) -> dict[str, str | None]: + relative = path.resolve().relative_to(REPO_ROOT).as_posix() + return { + "path": relative, + "sha256": _file_sha256(path) if path.is_file() else None, + } + + +def build_projection_profile(authorized_at: datetime) -> replay.LocalIngestionProfile: + if authorized_at.tzinfo is None or authorized_at.utcoffset() is None: + raise ActiveMetadataProjectionExecutionError( + "projection authorization time must include a timezone" + ) + base = replay.load_profile() + profile = { + "schema": replay.PROFILE_SCHEMA, + "environment": "local_docker_desktop", + "cluster": base.cluster.model_dump(mode="json"), + "providers": base.providers.model_dump(mode="json"), + "targets": { + "openmetadata": { + "service": "gda_chongqing_m3_18", + "service_type": "CustomDatabase", + "database": "cultural_heritage", + "schema": "published", + "table": "cultural_districts", + "owner_team": "data-platform", + "domain": "natural-resources", + "classification": "Sensitivity", + "classification_tag": "Internal", + "glossary": "CulturalHeritage", + "glossary_term": "CulturalDistrict", + }, + "gravitino": { + "metalake": "gda_chongqing_m3_18", + "catalog": "iceberg", + "schema": "cultural_heritage", + "table": "cultural_districts", + "catalog_type": "RELATIONAL", + "catalog_provider": "lakehouse-iceberg", + "catalog_backend": "memory", + "uri": "file:///tmp/gda-m3-local", + "warehouse": "file:///tmp/gda-m3-local", + }, + }, + "authorization": { + "action": replay.ACTION, + "policy_version_ref": (f"gda://{TENANT}/policy/active-metadata-provider-apply-v1"), + "evaluator_subject": POLICY_EVALUATOR, + "approver_subject": APPROVER, + "approval_reason": ("Approved bounded scheduler-triggered Metadata Fabric projection"), + "decided_at": authorized_at - timedelta(minutes=3), + "approval_decided_at": authorized_at - timedelta(minutes=2), + "authorized_at": authorized_at - timedelta(minutes=1), + "approval_expires_at": authorized_at + timedelta(days=180), + "expires_at": authorized_at + timedelta(days=365), + }, + "claims": base.claims.model_dump(mode="json"), + } + return replay.LocalIngestionProfile.model_validate(profile) + + +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_projection_intent.v1", + "tenant_id": TENANT, + "resource_urn": resource_urn, + "resource_version_id": str(SOURCE_ID), + "content_sha256": content_sha256, + "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 build_execution_request(plan: replay.LocalApplyPlan) -> ProjectionExecutionRequest: + stable = { + "schema": REQUEST_SCHEMA, + "tenant_id": plan.tenant_id, + "run_id": str(plan.run_id), + "definition_version_id": str(plan.definition_version_id), + "source_resource_version_id": str(plan.source_resource_version_id), + "content_sha256": plan.content_sha256, + "apply_plan_sha256": plan.apply_plan_sha256, + } + return ProjectionExecutionRequest( + **stable, + request_sha256=canonical_json_fingerprint(stable), + ) + + +def build_provider_apply_authorization( + plan: replay.LocalApplyPlan, + run: PlatformRun, + profile: replay.LocalIngestionProfile, +) -> replay.ApplyAuthorizationBundle: + """Authorize a source projection without fabricating a target ResourceVersion.""" + auth = profile.authorization + actor = f"{run.subject_context.subject_type.value}:{run.subject_context.subject_id}" + execution_plan = replay.build_execution_plan_artifact( + plan, + created_by=actor, + created_at=auth.decided_at, + ) + resource_versions = tuple( + sorted( + { + plan.definition_version_id, + plan.source_resource_version_id, + plan.resource_version_id, + }, + key=str, + ) + ) + decision = PolicyDecision( + tenant_id=plan.tenant_id, + run_id=plan.run_id, + subject_context=run.subject_context, + action=auth.action, + definition_version_id=plan.definition_version_id, + resource_version_ids=resource_versions, + execution_plan_artifact_id=execution_plan.artifact_id, + effect="allow", + policy_version_ref=auth.policy_version_ref, + evaluator_subject=auth.evaluator_subject, + requires_approval=True, + obligations=(), + decided_at=auth.decided_at, + expires_at=auth.expires_at, + ) + decision_artifact = build_policy_decision_artifact(decision) + approval_artifact = build_approval_artifact( + ApprovalRecord( + tenant_id=plan.tenant_id, + run_id=plan.run_id, + definition_version_id=plan.definition_version_id, + policy_decision_artifact_id=decision_artifact.artifact_id, + policy_decision_sha256=decision_artifact.content_sha256, + verdict="approved", + approver_subject=auth.approver_subject, + reason=auth.approval_reason, + decided_at=auth.approval_decided_at, + expires_at=auth.approval_expires_at, + ) + ) + values = { + "execution_plan_artifact": execution_plan, + "policy_decision_artifact": decision_artifact, + "approval_artifact": approval_artifact, + } + bundle = replay.ApplyAuthorizationBundle( + **values, + authorization_sha256=canonical_json_fingerprint( + {key: value.model_dump(mode="json") for key, value in values.items()} + ), + ) + replay.validate_apply_authorization( + plan, + run, + bundle, + at=auth.authorized_at, + ) + return bundle + + +def _workflow_document( + callback_url: str, + request: ProjectionExecutionRequest, +) -> dict[str, Any]: + request_json = json.dumps( + request.model_dump(mode="json", by_alias=True), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ) + raw_script = " ".join( + ( + "curl --fail --silent --show-error --max-time 300", + "--request POST --header 'Content-Type: application/json'", + f"--data-binary {shlex.quote(request_json)}", + shlex.quote(callback_url), + ) + ) + task = { + "code": TASK_CODE, + "name": "execute_active_metadata_projection", + "version": 1, + "description": "Execute one authorized local Metadata Fabric projection", + "delayTime": 0, + "taskType": "SHELL", + "taskParams": { + "localParams": [], + "rawScript": raw_script, + "resourceList": [], + "dependence": {}, + "conditionResult": {"successNode": [], "failedNode": []}, + "waitStartTimeout": {}, + }, + "flag": "YES", + "taskPriority": "MEDIUM", + "workerGroup": "default", + "environmentCode": -1, + "failRetryTimes": 0, + "failRetryInterval": 1, + "timeoutFlag": "OPEN", + "timeoutNotifyStrategy": "WARN", + "timeout": 360, + } + relation = { + "name": "", + "preTaskCode": 0, + "preTaskVersion": 0, + "postTaskCode": TASK_CODE, + "postTaskVersion": 1, + "conditionType": "NONE", + "conditionParams": {}, + } + return { + "dolphinscheduler": { + "name": "gda_active_metadata_projection_execution_v1", + "description": "Authorized local provider projection execution", + "task_definitions": [task], + "task_relations": [relation], + "locations": [{"taskCode": TASK_CODE, "x": 160, "y": 100}], + "global_params": [], + "timeout_seconds": 420, + "execution_type": "PARALLEL", + } + } + + +def build_scheduler_definition( + callback_url: str, + request: ProjectionExecutionRequest, + *, + created_at: datetime, +) -> ProjectionDefinitionBundle: + definition_urn = f"gda://{TENANT}/definition/metadata-projection-execution" + definition_document = _workflow_document(callback_url, request) + input_contract = { + "metadata_change": "gis.cultural_districts", + "execution_request_sha256": request.request_sha256, + } + output_contract = { + "openmetadata_projection_readback": True, + "gravitino_projection_readback": True, + "zero_mutation_replay": True, + } + 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-projection-execution", + 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-provider-apply-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, + definition_bundle: ProjectionDefinitionBundle, + binding: DolphinSchedulerDefinitionBinding, + *, + authorized_at: datetime, +) -> ProjectionDispatchBundle: + base = authorization.build_authorization_bundle(content_sha256) + if binding.definition_version_id != DEFINITION_ID: + raise ActiveMetadataProjectionExecutionError( + "DolphinScheduler binding does not match the projection definition" + ) + if binding.compiled_sha256 != definition_bundle.workflow.compiled_sha256: + raise ActiveMetadataProjectionExecutionError( + "DolphinScheduler binding does not match the compiled projection workflow" + ) + dispatch_plan = build_dolphinscheduler_binding_artifact( + 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="execute authorized active metadata projection", + ) + 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 bounded local scheduler-triggered provider projection", + 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-projection:cultural-districts:execution: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, + base.registration.resource_version, + definition_bundle.definition, + run, + dispatch_plan, + dispatch_policy, + dispatch_approval, + authorized_by=AUTHORIZER, + authorized_at=authorized_at, + ) + return ProjectionDispatchBundle( + source_resource=base.source_resource, + source_version=base.registration.resource_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, + ) + + +class ProjectionExecutor: + def __init__( + self, + request: ProjectionExecutionRequest, + profile: replay.LocalIngestionProfile, + plan: replay.LocalApplyPlan, + run: PlatformRun, + apply_authorization: replay.ApplyAuthorizationBundle, + *, + openmetadata: replay.OpenMetadataApplyClient, + gravitino: replay.GravitinoApplyClient, + ) -> None: + self.request = request + self.profile = profile + self.plan = plan + self.run = run + self.apply_authorization = apply_authorization + self.openmetadata = openmetadata + self.gravitino = gravitino + self.request_count = 0 + self.first: replay.ApplyOutcome | None = None + self.replayed: replay.ApplyOutcome | None = None + self.error_type: str | None = None + + def execute(self, payload: dict[str, Any]) -> dict[str, Any]: + self.request_count += 1 + try: + observed = ProjectionExecutionRequest.model_validate(payload) + if observed != self.request: + raise ActiveMetadataProjectionExecutionError( + "projection callback request does not match the compiled workflow" + ) + if self.request_count != 1: + raise ActiveMetadataProjectionExecutionError( + "projection executor accepts exactly one scheduler callback" + ) + now = datetime.now(UTC) + replay.validate_apply_authorization( + self.plan, + self.run, + self.apply_authorization, + at=now, + ) + self.first = replay.apply_once( + self.plan, + self.profile, + self.apply_authorization, + self.run, + openmetadata=self.openmetadata, + gravitino=self.gravitino, + at=now, + ) + 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.first.status != replay.ApplyStatus.CREATED or not self.first.mutations: + raise ActiveMetadataProjectionExecutionError( + "first scheduler-triggered provider apply did not create projection state" + ) + if self.replayed.status != replay.ApplyStatus.NO_OP or self.replayed.mutations: + raise ActiveMetadataProjectionExecutionError( + "exact provider replay performed duplicate mutations" + ) + if self.first.binding_candidate_sha256 != self.replayed.binding_candidate_sha256: + raise ActiveMetadataProjectionExecutionError( + "provider binding read-back drifted across exact replay" + ) + return { + "schema": "gda.active_metadata_projection_execution_response.v1", + "status": "applied_and_replayed", + "request_sha256": self.request.request_sha256, + } + except Exception as exc: + self.error_type = type(exc).__name__ + raise + + +class ProjectionExecutionServer: + def __init__(self, request: ProjectionExecutionRequest) -> None: + self.request = request + self.executor: ProjectionExecutor | None = None + self.started = False + self.cleanup_verified = False + owner = self + + class Handler(BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 + if self.path != CALLBACK_PATH: + self.send_error(404) + return + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError: + self.send_error(400) + return + if length <= 0 or length > 16384: + self.send_error(413) + return + try: + payload = json.loads(self.rfile.read(length)) + if not isinstance(payload, dict) or owner.executor is None: + raise ValueError("projection executor request is unavailable") + response = owner.executor.execute(payload) + except Exception: + self.send_error(500) + return + body = json.dumps( + response, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, _format: str, *args: object) -> None: + return + + self._server = HTTPServer(("0.0.0.0", 0), Handler) + self._thread = threading.Thread( + target=self._server.serve_forever, + name="gda-m3-18-projection-executor", + daemon=True, + ) + + @property + def callback_url(self) -> str: + return f"http://host.docker.internal:{self._server.server_port}{CALLBACK_PATH}" + + def start(self) -> None: + if self.executor is None: + raise ActiveMetadataProjectionExecutionError( + "projection executor must be attached before callback server starts" + ) + self._thread.start() + self.started = True + + def stop(self) -> bool: + if self.started: + self._server.shutdown() + self._thread.join(timeout=30) + self._server.server_close() + self.cleanup_verified = not self._thread.is_alive() + return self.cleanup_verified + + +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 ActiveMetadataProjectionExecutionError("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 _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 _validate_dependencies( + scheduler_evidence: dict[str, Any], + ingestion_evidence: dict[str, Any], +) -> None: + if delivery.validate_rehearsal_evidence(scheduler_evidence): + raise ActiveMetadataProjectionExecutionError("M3-17 scheduler delivery evidence is invalid") + ingestion_errors = replay.verify_evidence_integrity(ingestion_evidence) + expected_contract = replay.build_contract_report()["contract_fingerprint"] + observed_contract = ( + ingestion_evidence.get("observation", {}).get("contract", {}).get("contract_fingerprint") + ) + if ingestion_errors or observed_contract != expected_contract: + raise ActiveMetadataProjectionExecutionError("M3-2 provider ingestion evidence is invalid") + + +def build_contract_report() -> dict[str, Any]: + errors: list[str] = [] + paths = { + "projection_execution": Path(__file__).resolve(), + "scheduler_delivery": Path(delivery.__file__).resolve(), + "provider_apply": Path(replay.__file__).resolve(), + "wrapper": DEFAULT_WRAPPER_PATH, + } + required = { + "projection_execution": ( + "class ProjectionExecutor", + "class ProjectionExecutionServer", + "def run_scheduler_projection_rehearsal(", + "provider_mutations_executed", + "production_ready", + ), + "scheduler_delivery": ( + "class EphemeralDolphinScheduler", + "class EphemeralPostgresDatabase", + "def _wait_for_terminal_instance(", + ), + "provider_apply": ( + "class OpenMetadataApplyClient", + "class GravitinoApplyClient", + "def apply_once(", + ), + "wrapper": ( + "set -euo pipefail", + "metadata_fabric_active_metadata_projection_execution", + ), + } + files: dict[str, dict[str, str | None]] = {} + for name, path in paths.items(): + try: + text_value = path.read_text(encoding="utf-8") + for marker in required[name]: + if marker not in text_value: + errors.append(f"{name} is missing marker: {marker}") + except OSError as exc: + errors.append(f"{name} is unavailable: {type(exc).__name__}") + files[name] = _file_record(path) + stable = { + "schema": CONTRACT_SCHEMA, + "scheduler_provider": { + "name": "apache-dolphinscheduler", + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "image": delivery.IMAGE, + "image_id": delivery.IMAGE_ID, + }, + "projection_providers": { + "openmetadata": "1.13.1", + "gravitino": "1.3.0", + }, + "scheduler_execution_boundary": ( + "dolphinscheduler_shell_to_ephemeral_local_projection_executor" + ), + "provider_mutation_mode": "authorized_apply_then_zero_mutation_replay", + "provider_success_platform_state": "reconciling", + "local_static_contract_verified": not errors, + "files": files, + "errors": errors, + } + return { + **stable, + "status": "valid" if not errors else "invalid", + "contract_sha256": canonical_json_fingerprint(stable), + "provider_apply_authorized": False, + "provider_mutations_executed": False, + "production_scheduler_submission_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + } + + +def run_scheduler_projection_rehearsal( + database_url: str, + scheduler_profile: DolphinSchedulerProfile, + scheduler_dependency: dict[str, Any], + ingestion_dependency: 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: ProjectionExecutionServer, + *, + terminal_timeout_seconds: float = 600, +) -> dict[str, Any]: + _validate_dependencies(scheduler_dependency, ingestion_dependency) + dataset = scheduler_dependency["dataset_bundle"] + if validate_shapefile_bundle_inventory(dataset): + raise ActiveMetadataProjectionExecutionError( + "real Chongqing dataset bundle inventory is invalid" + ) + if scheduler_profile.workload_subject != RUNNER: + raise ActiveMetadataProjectionExecutionError( + "scheduler workload does not match the authorized runner" + ) + if scheduler_profile.policy_evaluator_subject != POLICY_EVALUATOR: + raise ActiveMetadataProjectionExecutionError( + "scheduler evaluator does not match policy evidence" + ) + + started_at = datetime.now(UTC) + plan = build_projection_plan(dataset["content_sha256"], projection_profile) + request = build_execution_request(plan) + if callback_server.request != request: + raise ActiveMetadataProjectionExecutionError( + "callback server is not bound to the exact projection 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: + binding = client.create_workflow(definition_bundle.workflow) + authorized_at = datetime.now(UTC) + bundle = build_dispatch_bundle( + dataset["content_sha256"], + definition_bundle, + binding, + authorized_at=authorized_at, + ) + apply_authorization = build_provider_apply_authorization( + plan, + bundle.run, + projection_profile, + ) + callback_server.executor = ProjectionExecutor( + request, + projection_profile, + plan, + bundle.run, + apply_authorization, + openmetadata=openmetadata, + gravitino=gravitino, + ) + authorization._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 ActiveMetadataProjectionExecutionError( + "authorized projection 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, + 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(binding, bundle.run) + reconciled = adapter.reconcile( + TENANT, + RUN_ID, + bundle.dispatch_plan.artifact_id, + actor_subject=RUNNER, + attempt_no=1, + ) + attempts = _attempt_summary(engine) + final_run = gateway.get_run(TENANT, RUN_ID) + executor = callback_server.executor + if executor is None or executor.first is None or executor.replayed is None: + raise ActiveMetadataProjectionExecutionError( + "scheduler callback did not produce both provider outcomes" + ) + first_outcome = replay._outcome_evidence(executor.first) + replay_outcome = replay._outcome_evidence(executor.replayed) + 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"] == "created" + 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"] + and bundle.source_version.content_sha256 + == dataset["content_sha256"] + == scheduler_dependency["resource_version_content_sha256"] + ) + contract = build_contract_report() + stable = { + "schema": EVIDENCE_SCHEMA, + "status": ( + "local_scheduler_provider_projection_execution_verified" if verified else "blocked" + ), + "contract_sha256": contract["contract_sha256"], + "scheduler_dependency_evidence_sha256": scheduler_dependency["evidence_sha256"], + "ingestion_dependency_evidence_sha256": ingestion_dependency["evidence_fingerprint"], + "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": bundle.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_sha256": (bundle.activation_authorization.authorization_sha256), + "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_authorization_sha256": (apply_authorization.authorization_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": binding.server_version, + "api_profile": binding.api_profile, + "image": delivery.IMAGE, + "image_id": delivery.IMAGE_ID, + "architecture": platform.machine(), + "project_code": binding.project_code, + "workflow_definition_code": binding.workflow_definition_code, + "workflow_definition_version": 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, + "projection_targets": { + "openmetadata": projection_profile.targets.openmetadata.table_fqn, + "gravitino": projection_profile.targets.gravitino.identity, + }, + "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_apply": first_outcome, + "replay": replay_outcome, + "provider_mutations_executed": verified, + "openmetadata_readback_verified": verified, + "gravitino_readback_verified": verified, + "deterministic_live_replay_verified": verified, + "local_scheduler_projection_execution_verified": verified, + "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, + "oidc_verified": False, + "tls_verified": False, + "binding_persisted_to_gda_control": False, + "live_openlineage_emission_verified": False, + "production_scheduler_submission_verified": False, + "production_ingestion_verified": False, + "production_ready": False, + "errors": [] if verified else ["local projection execution failed"], + } + return stable + finally: + client.close() + engine.dispose() + + +def run_managed_rehearsal( + database_admin_url: str, + scheduler_dependency: dict[str, Any], + ingestion_dependency: 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_dependencies(scheduler_dependency, ingestion_dependency) + authorized_at = datetime.now(UTC) + profile = build_projection_profile(authorized_at) + plan = build_projection_plan( + scheduler_dependency["dataset_bundle"]["content_sha256"], + profile, + ) + request = build_execution_request(plan) + database = delivery.EphemeralPostgresDatabase(database_admin_url) + scheduler = delivery.EphemeralDolphinScheduler( + admin_password, + readiness_timeout=readiness_timeout_seconds, + ) + callback_server = 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 ActiveMetadataProjectionExecutionError( + "temporary PostgreSQL database was not created" + ) + evidence = run_scheduler_projection_rehearsal( + database.database_url, + scheduler_profile, + scheduler_dependency, + ingestion_dependency, + 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 ActiveMetadataProjectionExecutionError( + "local scheduler projection 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 projection runtime cleanup failed") + evidence["status"] = "blocked" + evidence["local_scheduler_projection_execution_verified"] = False + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + return {**stable, "evidence_sha256": canonical_json_fingerprint(stable)} + + +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 ActiveMetadataProjectionExecutionError(f"{path.name} is not valid JSON") from exc + if not isinstance(value, dict): + raise ActiveMetadataProjectionExecutionError(f"{path.name} must contain an object") + return value + + +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("projection execution evidence schema does not match") + if evidence.get("evidence_sha256") != canonical_json_fingerprint(stable): + errors.append("projection execution evidence SHA-256 does not match") + contract = build_contract_report() + if evidence.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("projection execution contract fingerprint is stale") + try: + scheduler_dependency = _load_json_object(DEFAULT_SCHEDULER_DEPENDENCY_PATH) + ingestion_dependency = _load_json_object(DEFAULT_INGESTION_DEPENDENCY_PATH) + except ActiveMetadataProjectionExecutionError: + scheduler_dependency = {} + ingestion_dependency = {} + errors.append("projection execution dependency evidence is unavailable") + if evidence.get("scheduler_dependency_evidence_sha256") != ( + scheduler_dependency.get("evidence_sha256") + ): + errors.append("M3-17 scheduler dependency fingerprint is stale") + if evidence.get("ingestion_dependency_evidence_sha256") != ( + ingestion_dependency.get("evidence_fingerprint") + ): + errors.append("M3-2 ingestion dependency fingerprint is stale") + dataset = evidence.get("dataset_bundle") + if not isinstance(dataset, dict): + errors.append("projection execution 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 projection execution 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", + "openmetadata_readback_verified", + "gravitino_readback_verified", + "deterministic_live_replay_verified", + "provider_mutations_executed", + "local_scheduler_projection_execution_verified", + "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"projection execution did not verify {claim}") + if evidence.get("dispatch_authorization_created") is not True: + errors.append("projection dispatch authorization was not created") + if evidence.get("exact_dispatch_authorization_replay_created") is not False: + errors.append("exact projection dispatch authorization replay created a row") + if evidence.get("execution_callback_request_count") != 1: + errors.append("projection executor must receive exactly one callback") + if evidence.get("command_status") != "done": + errors.append("authorized projection command must be done") + if evidence.get("command_claimed_count") != 1: + errors.append("projection execution must claim one command") + if evidence.get("command_completed_count") != 1: + errors.append("projection 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("projection scheduler provider identity or state does not match") + if evidence.get("matching_provider_instance_count") != 1: + errors.append("projection execution must read back one scheduler instance") + if evidence.get("attempt_observation_count") != 2: + errors.append("projection execution must record two attempt observations") + if evidence.get("external_correlation_count") != 1: + errors.append("projection execution must retain one external correlation") + if evidence.get("attempt_states") != ["submitted", "success"]: + errors.append("projection scheduler attempt states do not match") + if evidence.get("platform_run_status") != "reconciling": + errors.append("provider success must leave PlatformRun reconciling") + if evidence.get("platform_run_succeeded") is not False: + errors.append("local provider success may not claim platform success") + first = evidence.get("first_apply") + replayed = evidence.get("replay") + if not isinstance(first, dict) or not isinstance(replayed, dict): + errors.append("provider apply outcomes are missing") + else: + if first.get("status") != "created" or first.get("mutation_count", 0) <= 0: + errors.append("first scheduler-triggered provider apply did not mutate") + if replayed.get("status") != "no_op" or replayed.get("mutation_count") != 0: + errors.append("exact scheduler-triggered 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}") + 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("projection execution 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("--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( + "--scheduler-dependency", + type=Path, + default=DEFAULT_SCHEDULER_DEPENDENCY_PATH, + ) + rehearse.add_argument( + "--ingestion-dependency", + type=Path, + default=DEFAULT_INGESTION_DEPENDENCY_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() + try: + report["errors"].extend(validate_rehearsal_evidence(_load_json_object(args.evidence))) + except ActiveMetadataProjectionExecutionError as exc: + report["errors"].append( + f"projection execution 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 + + scheduler_dependency = _load_json_object(args.scheduler_dependency) + ingestion_dependency = _load_json_object(args.ingestion_dependency) + 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 ActiveMetadataProjectionExecutionError( + "OpenMetadata local bootstrap credential environment is missing" + ) from exc + evidence = run_managed_rehearsal( + args.database_admin_url, + scheduler_dependency, + ingestion_dependency, + 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/platform_truth.py b/data_agent/platform_truth.py index 8a129eef..f0ae4355 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -918,6 +918,26 @@ def _config( ), "Protected identity, deployed controller and production scheduler/provider path", ), + RuntimeSpec( + "metadata_active_metadata_projection_execution_rehearsal", + "active_metadata_projection_execution_rehearsal", + "governed", + "evidence_durable", + "temporary scheduler/executor/provider session + committed local evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_active_metadata_projection_execution.py", + "scripts/metadata-fabric-active-metadata-projection-execution.sh", + ), + ( + ( + "data_agent/metadata_fabric_active_metadata_projection_execution.py", + "def run_scheduler_projection_rehearsal", + ), + ), + "Protected identity, durable executor and production scheduler/provider path", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", @@ -982,7 +1002,7 @@ def _config( "5ee717911c109b480328a050893296e37591bfca748e3ed1743b7e3def3d9048" ) RUNTIME_PRIMITIVE_BASELINE_FINGERPRINT = ( - "d6402d91e40ddb61591a7d258925d79e5eee964c3a9c0ace7de34acd10facbfd" + "499c531821c2138977724d8a9c12e0328cb45525944634c0cfba3763137652b6" ) _IGNORED_SOURCE_PARTS = frozenset( diff --git a/data_agent/test_metadata_fabric_active_metadata_projection_execution.py b/data_agent/test_metadata_fabric_active_metadata_projection_execution.py new file mode 100644 index 00000000..1591c91c --- /dev/null +++ b/data_agent/test_metadata_fabric_active_metadata_projection_execution.py @@ -0,0 +1,130 @@ +import json +from copy import deepcopy +from datetime import UTC, datetime + +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 + + +def _binding(definition_bundle): + return DolphinSchedulerDefinitionBinding( + tenant_id=execution.TENANT, + definition_version_id=execution.DEFINITION_ID, + project_code=190000000000001, + workflow_definition_code=190000000000002, + workflow_definition_version=1, + compiled_sha256=definition_bundle.workflow.compiled_sha256, + ) + + +def _request_and_plan(): + at = datetime(2026, 7, 30, 12, 0, tzinfo=UTC) + profile = execution.build_projection_profile(at) + plan = execution.build_projection_plan("f" * 64, profile) + return profile, plan, execution.build_execution_request(plan) + + +def test_real_data_projection_plan_and_callback_request_are_content_bound(): + profile, plan, request = _request_and_plan() + + assert plan.tenant_id == execution.TENANT + assert plan.resource_version_id == execution.SOURCE_ID + assert plan.content_sha256 == "f" * 64 + assert plan.openmetadata_fqn == ( + "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts" + ) + assert plan.gravitino_identity == ( + "gda_chongqing_m3_18.iceberg.cultural_heritage.cultural_districts" + ) + assert request.apply_plan_sha256 == plan.apply_plan_sha256 + assert request.content_sha256 == plan.content_sha256 + assert profile.authorization.action == replay.ACTION + + +def test_scheduler_definition_executes_exact_projection_request(): + _profile, _plan, request = _request_and_plan() + bundle = execution.build_scheduler_definition( + "http://host.docker.internal:43123/v1/execute-projection", + request, + created_at=datetime(2026, 7, 30, 12, 0, tzinfo=UTC), + ) + script = bundle.workflow.task_definitions[0]["taskParams"]["rawScript"] + + assert bundle.definition.portability_class.value == "provider_native" + assert "curl --fail" in script + assert request.request_sha256 in script + assert bundle.workflow.task_definitions[0]["name"] == ("execute_active_metadata_projection") + + +def test_dispatch_and_provider_apply_authorizations_are_independent(): + profile, plan, request = _request_and_plan() + definition = execution.build_scheduler_definition( + "http://host.docker.internal:43123/v1/execute-projection", + request, + created_at=datetime(2026, 7, 30, 12, 0, tzinfo=UTC), + ) + dispatch = execution.build_dispatch_bundle( + plan.content_sha256, + definition, + _binding(definition), + authorized_at=datetime(2026, 7, 30, 12, 0, tzinfo=UTC), + ) + apply_authorization = execution.build_provider_apply_authorization( + plan, + dispatch.run, + profile, + ) + apply_decision = replay.parse_policy_decision_artifact( + apply_authorization.policy_decision_artifact + ) + + 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) + assert dispatch.run.status.value == "accepted" + + +def test_static_contract_declares_reconciling_and_local_claim_boundary(): + report = execution.build_contract_report() + + assert report["status"] == "valid" + assert report["errors"] == [] + assert report["provider_mutation_mode"] == ("authorized_apply_then_zero_mutation_replay") + assert report["provider_success_platform_state"] == "reconciling" + assert report["provider_apply_authorized"] is False + assert report["provider_mutations_executed"] is False + assert report["production_ready"] is False + + +def test_checked_projection_execution_evidence_is_current_and_fail_closed(): + evidence = json.loads(execution.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + + assert execution.validate_rehearsal_evidence(evidence) == [] + assert evidence["scheduler_provider"]["terminal_state"] == "SUCCESS" + assert evidence["first_apply"]["status"] == "created" + assert evidence["first_apply"]["mutation_count"] > 0 + assert evidence["replay"]["status"] == "no_op" + assert evidence["replay"]["mutation_count"] == 0 + assert evidence["first_apply"]["openmetadata"] == (evidence["replay"]["openmetadata"]) + assert evidence["first_apply"]["gravitino"] == evidence["replay"]["gravitino"] + assert evidence["platform_run_status"] == "reconciling" + assert evidence["platform_run_succeeded"] is False + assert evidence["provider_mutations_executed"] is True + assert evidence["production_ingestion_verified"] is False + assert evidence["production_ready"] is False + + +def test_projection_execution_evidence_rejects_replay_drift_and_overclaim(): + evidence = json.loads(execution.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + tampered = deepcopy(evidence) + tampered["replay"]["mutation_count"] = 1 + tampered["platform_run_succeeded"] = True + tampered["production_ready"] = True + + errors = execution.validate_rehearsal_evidence(tampered) + + assert "projection execution evidence SHA-256 does not match" in errors + assert "exact scheduler-triggered replay was not mutation-free" in errors + assert "local provider success may not claim platform success" in errors + assert "local projection execution may not claim production_ready" in errors diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index 91ff5e86..3cc292b7 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -268,6 +268,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_projection_execution_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-064-local-scheduler-triggered-active-metadata-projection-execution.md b/docs/architecture-decisions/adr-064-local-scheduler-triggered-active-metadata-projection-execution.md new file mode 100644 index 00000000..e9db0c38 --- /dev/null +++ b/docs/architecture-decisions/adr-064-local-scheduler-triggered-active-metadata-projection-execution.md @@ -0,0 +1,92 @@ +# ADR-064: Local scheduler-triggered Active Metadata projection execution + +**Status**: Accepted + +**Date**: 2026-07-30 + +**Decision owners**: Data Platform, Metadata Platform, Data Governance, Security, Platform Architecture + +**Related decisions**: [ADR-048](adr-048-local-authorized-metadata-fabric-ingestion-replay.md) · [ADR-062](adr-062-atomic-active-metadata-authorization-and-dispatch.md) · [ADR-063](adr-063-local-authorized-active-metadata-scheduler-delivery.md) + +## Context + +M3-17 proved that an exact Active Metadata authorization can become a real DolphinScheduler `3.4.2` workflow instance and provider `SUCCESS`, but the workflow deliberately had no side effect. M3-2 separately proved authorized OpenMetadata/Gravitino projection and zero-mutation replay, but the provider clients were invoked directly by the rehearsal. The missing boundary was a scheduler task that actually triggers the authorized provider projection while preserving distinct dispatch, provider-apply and platform-success authorities. + +The retained real input is the Chongqing central cultural-district Shapefile bundle registered in M3-16. Its path-free inventory contains 20 `PolygonZ` features, 33 fields, EPSG:4490 and content SHA-256 `fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007`. M3-18 does not reopen or commit the source path because no new source-data claim is required. + +## Options Considered + +| Option | Benefit | Cost/risk | Decision | +|---|---|---|---| +| Treat M3-17 scheduler `SUCCESS` as provider execution | No new runtime | Confuses control-plane delivery with business mutation | Rejected | +| Copy provider mutation logic into the DolphinScheduler image | Self-contained task | Creates a second mutation engine and embeds provider credentials in the scheduler runtime | Rejected | +| Submit two independent workflow instances for create and replay | Strong scheduler-level replay signal | The second dispatch would require its own durable authorization and complicate this boundary | Rejected | +| Let one authorized task call an ephemeral executor that reuses M3-2 clients for apply and exact replay | Proves scheduler-to-provider execution without duplicating adapters | Local callback and bootstrap provider security are not production-grade | Adopted | + +## Decision + +### 1. Dispatch and provider apply remain independently authorized + +The existing M3-16/M3-17 chain authorizes `dolphinscheduler.dispatch` for the provider-native workflow binding. A second execution-plan Artifact, PolicyDecision and Approval authorize `metadata_fabric.apply` for the exact tenant, Run, DefinitionVersion, source ResourceVersion, content hash, natural provider targets and apply-plan fingerprint. + +The source dataset is itself the projected ResourceVersion; M3-18 does not fabricate a target ResourceVersion merely to satisfy authorization cardinality. The provider-apply decision therefore canonicalizes the exact unique ResourceVersion scope before M3-2 validation. + +### 2. DolphinScheduler triggers exactly one bounded callback + +The official standalone image runs one Shell task whose immutable body contains the projection request fingerprint and calls a short-lived HTTP executor through Docker Desktop's host gateway. The executor accepts exactly one request, rejects body or fingerprint drift, holds provider credentials and authorization only in memory, and returns success only after both provider phases pass. + +The callback is plaintext local HTTP and is not a protected workload identity, authenticated service endpoint, production network path or durable controller. Its URL and credentials are excluded from evidence, and the server is stopped after the rehearsal. + +### 3. Existing provider clients own mutation and read-back + +The executor reuses `OpenMetadataApplyClient`, `GravitinoApplyClient` and `apply_once` from ADR-048. The first apply must create provider state and read back the exact GDA ResourceURN, ResourceVersion, content hash, governance refs and technical revision. The immediate second apply must be `no_op` with zero mutations and the same OpenMetadata observation, Gravitino observation and binding-candidate fingerprint. + +Partial inventory, state drift, inactive authorization or provider failure remains fail closed under the M3-2 behavior. M3-18 does not add another provider mutation implementation. + +### 4. Scheduler and provider success are still not platform success + +DolphinScheduler `SUCCESS` produces the existing `submitted/success` FrameworkAttemptObservations and one external correlation. Provider apply/read-back forms additional local evidence, but the PlatformRun remains `reconciling`; only the existing platform success evidence gate may produce `succeeded`. + +### 5. Provider objects are retained, ephemeral control resources are removed + +The OpenMetadata and Gravitino projection objects remain available for read-back. The callback server, both Kubernetes port-forwards, DolphinScheduler container and temporary GDA Control PostgreSQL database must all be removed. No legacy authority, source data, production deployment or persistent GDA binding is written. + +## Verification + +The local rehearsal recorded: + +- one authorized DolphinScheduler instance with six exact GDA correlation variables and terminal `SUCCESS`; +- one callback request matching request SHA-256 `462d738064fb2352acee3b72b1b966bc5ce4f524904bc991b87c8c56d1f2f8ae`; +- first apply `created` with 10 mutations across OpenMetadata and Gravitino; +- replay `no_op` with zero mutations; +- OpenMetadata table UUID `9d043410-02b5-487d-bb70-da5f3969a978`; +- shared binding-candidate SHA-256 `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b`; +- PlatformRun `reconciling`, never `succeeded`; +- callback, two port-forwards, standalone container and temporary database cleanup; +- contract SHA-256 `a6632ae0edd4d4f3389129a8c07411a8d101ae56fbfc26b03fb0aff6928bb7bd`; +- evidence SHA-256 `397c0f1a29f53935c5508155470c4972cfc50260f0d0686fb48cb3f75519b17b`. + +## Claim Boundary + +Allowed now: + +- local scheduler-triggered provider projection execution is verified for the recorded Docker Desktop identities; +- the exact Chongqing ResourceVersion content hash reached both provider projections; +- independent local dispatch and provider-apply authorizations were validated; +- first apply, provider read-back and zero-mutation replay are correlated to one scheduler Run. + +Fixed false now: + +- protected workload identity, OIDC, TLS and provider-wide minimum privilege; +- Gravitino authentication and production durable catalog behavior; +- production scheduler submission, HA, backup and deployed controller; +- binding persistence, production OpenLineage delivery and production ingestion; +- platform Run success and `production_ready`. + +## Consequences + +**Positive**: M3-18 closes the local scheduler-to-provider execution gap without creating a second provider adapter or weakening platform success authority. + +**Negative**: the executor is an ephemeral host process, OpenMetadata uses bootstrap admin, Gravitino is unauthenticated and uses its local memory catalog, and the retained provider objects make this exact `created` rehearsal intentionally non-repeatable without a new target or explicit cleanup. + +**Revisit trigger**: replace this boundary only when a deployed executor with protected workload identity, minimum-privilege dual-provider credentials, authenticated durable Gravitino catalog, production scheduler metadata/HA, persistent binding and platform terminal evidence is available. diff --git a/docs/evidence/metadata-fabric-active-metadata-projection-execution-2026-07-30.json b/docs/evidence/metadata-fabric-active-metadata-projection-execution-2026-07-30.json new file mode 100644 index 00000000..444a3263 --- /dev/null +++ b/docs/evidence/metadata-fabric-active-metadata-projection-execution-2026-07-30.json @@ -0,0 +1,285 @@ +{ + "attempt_observation_count": 2, + "attempt_states": [ + "submitted", + "success" + ], + "binding_persisted_to_gda_control": false, + "callback_server_cleanup_verified": true, + "command_claimed_count": 1, + "command_completed_count": 1, + "command_id": "0f99a36f-b1f8-5a2b-8d19-178058075ae2", + "command_status": "done", + "compiled_workflow_sha256": "53fa7450ac7cb020d8e842d80d090cfb7c6ee0892b8a9038ccddcc27a74efe42", + "contract_sha256": "a6632ae0edd4d4f3389129a8c07411a8d101ae56fbfc26b03fb0aff6928bb7bd", + "correlation_variables": { + "gda_definition_sha256": "3b2318d538d33f97b20fd500b3ec36adb57b19f654ac21d2ab03f9fd16434ac5", + "gda_definition_urn": "gda://metadata-authorization-local/definition/metadata-projection-execution", + "gda_definition_version_id": "a8000000-0000-4000-8000-000000000002", + "gda_idempotency_key": "metadata-projection:cultural-districts:execution:v1", + "gda_run_id": "a8000000-0000-4000-8000-000000000003", + "gda_tenant_id": "metadata-authorization-local" + }, + "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": "3b2318d538d33f97b20fd500b3ec36adb57b19f654ac21d2ab03f9fd16434ac5", + "definition_version_id": "a8000000-0000-4000-8000-000000000002", + "deployment_applied": false, + "deterministic_live_replay_verified": true, + "dispatch_authorization_created": true, + "dispatch_authorization_id": "3d0a25e5-24ee-5304-a202-587065f1b67d", + "dispatch_authorization_sha256": "648cf42d2272c64c1720f2e5a203ef0219126e8b24dee47acad3dcee1f5fcdc0", + "dispatch_execution_plan_artifact_id": "ff320c6f-54ea-5906-9358-96676a962e6e", + "errors": [], + "evidence_sha256": "397c0f1a29f53935c5508155470c4972cfc50260f0d0686fb48cb3f75519b17b", + "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": "462d738064fb2352acee3b72b1b966bc5ce4f524904bc991b87c8c56d1f2f8ae", + "external_correlation_count": 1, + "first_apply": { + "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": "396497819b4187a0d71b5ea74d999775c9ec7201c0f1ae02f4194a11bb1f630c" + }, + "mutation_count": 10, + "mutations": [ + "openmetadata.glossary.create", + "openmetadata.glossary_term.create", + "openmetadata.database_service.create", + "openmetadata.database.create", + "openmetadata.database_schema.create", + "openmetadata.table.create", + "gravitino.metalake.create", + "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" + }, + "gravitino_authentication_verified": false, + "gravitino_readback_verified": true, + "ingestion_dependency_evidence_sha256": "3d5fb07267680520d2f03bf27f354787b7253210eb93ab85aae83d5f5a714dbe", + "live_openlineage_emission_verified": false, + "local_scheduler_projection_execution_verified": true, + "matching_provider_instance_count": 1, + "oidc_verified": false, + "openmetadata_readback_verified": true, + "platform_run_status": "reconciling", + "platform_run_succeeded": false, + "production_ingestion_verified": false, + "production_ready": false, + "production_scheduler_submission_verified": false, + "projection_targets": { + "gravitino": "gda_chongqing_m3_18.iceberg.cultural_heritage.cultural_districts", + "openmetadata": "gda_chongqing_m3_18.cultural_heritage.published.cultural_districts" + }, + "protected_workload_identity_verified": false, + "provider_apply_authorization_sha256": "571ec4b872a678fad13210f3317ec15a22f779c732c330aa9064dcec6c9f18d2", + "provider_apply_authorized": true, + "provider_apply_execution_plan_artifact_id": "f678c10a-e878-53c8-9ae6-2c5719d76796", + "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_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": "396497819b4187a0d71b5ea74d999775c9ec7201c0f1ae02f4194a11bb1f630c" + }, + "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": "a8000000-0000-4000-8000-000000000003", + "scheduler_dependency_evidence_sha256": "00d4ea062c40f8d97557eadc357a36c6d1ccd56e12a94a44694113681e5d55f4", + "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": 180204965431360, + "server_version": "3.4.2", + "terminal_state": "SUCCESS", + "workflow_definition_code": 180204965513280, + "workflow_definition_version": 1, + "workflow_instance_id": 1 + }, + "scheduler_success_readback_verified": true, + "schema": "gda.active_metadata_projection_execution_evidence.v1", + "standalone_container_cleanup_verified": true, + "status": "local_scheduler_provider_projection_execution_verified", + "submitted_observation_count": 1, + "success_observation_count": 1, + "temporary_database_cleanup_verified": true, + "tls_verified": false, + "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 4ce84238..419e240d 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-17（本地真实数据授权与 scheduler delivery/read-back 已验证，生产验证待执行） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-18（本地 scheduler-triggered provider projection/read-back 已验证，生产验证待执行） 第八块回到 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： @@ -237,8 +237,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 32. [ADR-061](architecture-decisions/adr-061-durable-inert-active-metadata-activation-request.md) 拒绝把缺少 Definition、Run、execution plan、PolicyDecision 与 Approval 的 metadata event 直接转换为 DolphinScheduler command。migration 100 新增 tenant-scoped `MetadataActivationRequest`；managed consumer 只持 PostgreSQL 权限，并在同一事务写入 `awaiting_authorization` request 与完成 event。真实 PostgreSQL 演练得到 2 个 processed events、2 个精确 durable requests 和 0 个 `platform_command_outbox` rows；旧 no-request complete 被阻断，精确 request replay 不新增行，跨租户、FORCE RLS、直接 UPDATE/DELETE 拒绝均通过。base Deployment 为 0 replicas、无 provider/scheduler Secret、禁用 Kubernetes token mount。contract fingerprint 为 `2256daa97c1f3a2e71f4d7026592171daea802b371f5616fbbd72a63939ee6b5`，evidence fingerprint 为 `029aaf7de476115dcf6385ca4a0e05bb84492ebea8ddecf6c3edf36edd76dbef`；deployment 未 apply，production workload identity、scheduler submission、provider mutation/ingestion 与 readiness 全部仍为 `false`。 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`。 -此处 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，未部署受保护 workload identity/authorization controller，未执行受治理 provider mutation，也不证明生产 scheduler。M3-2 ingestion 仍使用 bootstrap admin，生产持久 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，未部署受保护 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`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index e842e914..105ec46e 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 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler 和生产切换仍 `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-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` -适用分支：`feat/ar1-metadata-fabric-active-metadata-scheduler-delivery` +适用分支：`feat/ar1-metadata-fabric-active-metadata-projection-execution` ## 判定规则 @@ -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 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 evidence | PlatformRun ledger 唯一登记最终状态；activation request 只拥有待授权意图；M3-17 provider `SUCCESS` 只形成 attempt evidence 并把 Run 留在 `reconciling`，不是 provider mutation 或平台成功终局权威；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 consumer、M3-16 authorization 与 M3-17 scheduler delivery 本地验证；metadata runner 仅本地验证 -> 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 控制链待接入 | | 原始文件/对象 | 当前 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 提交并回读 | command/metadata delivery status、消费者 claim、activation intent/request/authorization、FrameworkAttemptObservation、provider instance/correlation、worker status JSON、WebSocket 消息 | command/event 与源事实同事务入 outbox，幂等 consumer 交付；Active Metadata consumer 不能授权或执行；activation capability 的普通 dispatch 被拒绝；scheduler `SUCCESS` 仍须经平台终局 evidence gate | Platform/Integrations/Metadata Platform | AR-1 command worker、M3-14/M3-15/M3-16/M3-17 本地已验证 -> protected authorizer/worker identity、生产 scheduler 和 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 | 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 待执行 | | 质量结果 | `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 待执行 | @@ -61,6 +61,7 @@ 16. M3-15 只证明 managed consumer 代码、默认 0 replicas/无 provider 或 scheduler credential 的部署边界，以及本地 PostgreSQL 中 durable inert request 与 event completion 的同事务原子性；`awaiting_authorization` request 不是 Definition、Run、execution plan、PolicyDecision、Approval 或 PlatformCommand，不能授权调度或 provider mutation。受保护 workload identity、实际 scale-up、scheduler submission、provider read-back、告警/SLO 与 production readiness 仍未验证。 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 仍未验证。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -101,6 +102,7 @@ - Metadata Fabric M3-15 已新增 migration 100、内容绑定 `MetadataActivationRequest`、PlatformGateway atomic stage/read API、managed consumer/worker、默认 0 replicas Kustomize manifest 与静态部署 validator。真实 PostgreSQL 16 演练得到 2 个 processed events、2 个 `awaiting_authorization` requests、0 个 platform commands；no-request legacy complete 被阻断，request 精确 replay 为 `created=false`，consumer staging、跨租户拒绝、FORCE RLS、gateway 无直接 UPDATE/DELETE 均通过。contract fingerprint 为 `2256daa97c1f3a2e71f4d7026592171daea802b371f5616fbbd72a63939ee6b5`，evidence fingerprint 为 `029aaf7de476115dcf6385ca4a0e05bb84492ebea8ddecf6c3edf36edd76dbef`。consumer 没有 provider/scheduler credential 或 Kubernetes token；`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-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`。 ## 下一验收证据 @@ -108,7 +110,7 @@ - 真实 provenance artifact verify、受保护 overlay 的 `verified_for_staging_apply` release report，以及 staging/production 的 schema、config/runtime snapshot、registry/live DeploymentRevision 绑定、release/live artifact attestation 和环境 compare 报告； - staging 的 migration role、应用 login membership、连接池 role/tenant 复位、双租户 API 和 success finalization 运行产物； - DolphinScheduler adapter 的真实 IAM/OIDC、service token provisioning/轮换、provider 最小权限、binding artifact staging 接入、managed outbox worker/provider callback 实际扩容部署、唯一 worker ID、status/lease 故障恢复和无双写证据； -- Active Metadata consumer/authorizer 的 production scale-up、受保护 workload identity、生产 DolphinScheduler submission/read-back、幂等 projection execution、受治理 provider mutation/read-back、重试/死信告警与生产 SLO 证据； +- Active Metadata consumer/authorizer/executor 的 production scale-up、受保护 workload identity、生产 DolphinScheduler submission/read-back、双 provider 最小权限/OIDC/TLS、持久幂等 projection execution、重试/死信告警与生产 SLO 证据； - 首条真实图斑链对 golden slice 的 output hash、独立质量结果/evidence、血缘、发布 revision 和 rollback 演练； - OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、已批准 provider profile 与受保护对象存储 attestation、生产对象存储、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back、受保护环境 commit failure injection、provider uncertain outcome reconcile、cancel/reconcile/lineage 和完整 Spark/Flink conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity、M3-7 pending profile/合成 attestation、M3-8 本地 JDBC restart continuity、M3-9 本地同节点 Spark interoperability、M3-10 本地同主机跨节点 MinIO interoperability、M3-11 pending object-store profile/合成 attestation、M3-12 local pre-forward commit-failure recovery 与 M3-13 local uncertain-commit readback/no-resubmit 均不计入生产退出门； - DolphinScheduler/Temporal sandbox 的独立数据库、备份恢复、身份、版本和升级责任证明；DolphinScheduler standalone/H2 不计入此退出门。 diff --git a/scripts/metadata-fabric-active-metadata-projection-execution.sh b/scripts/metadata-fabric-active-metadata-projection-execution.sh new file mode 100755 index 00000000..05b0d10e --- /dev/null +++ b/scripts/metadata-fabric-active-metadata-projection-execution.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_projection_execution "$@"