diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d083ca5c..39e11447 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,8 @@ on: - feat/ar1-metadata-fabric-spark-uncertain-commit-reconciliation - feat/ar1-metadata-fabric-active-metadata-outbox - feat/ar1-metadata-fabric-active-metadata-consumer + - feat/ar1-metadata-fabric-active-metadata-authorization + - feat/ar1-metadata-fabric-active-metadata-scheduler-delivery env: PYTHON_VERSION: "3.13" @@ -176,6 +178,9 @@ jobs: - name: Validate metadata fabric Active Metadata authorization evidence run: python -m data_agent.metadata_fabric_active_metadata_authorization validate + - name: Validate metadata fabric Active Metadata scheduler delivery evidence + run: python -m data_agent.metadata_fabric_active_metadata_scheduler_delivery validate + - name: Validate Active Metadata consumer deployment boundary run: python -m data_agent.active_metadata_consumer_deployment validate @@ -256,6 +261,7 @@ jobs: data_agent/test_metadata_fabric_active_metadata_outbox.py \ 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_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_scheduler_delivery.py b/data_agent/metadata_fabric_active_metadata_scheduler_delivery.py new file mode 100644 index 00000000..cf9a22c9 --- /dev/null +++ b/data_agent/metadata_fabric_active_metadata_scheduler_delivery.py @@ -0,0 +1,1139 @@ +"""Verify authorized Active Metadata delivery through real DolphinScheduler. + +This local-only rehearsal carries the checked Chongqing ResourceVersion content +fingerprint into a provider-native DolphinScheduler workflow, authorizes the +exact binding through the M3-16 ledger, lets the durable command consumer submit +it, and reads the successful provider instance back into attempt evidence. A +provider success leaves the PlatformRun in ``reconciling`` and never authorizes +or executes a governed metadata/data mutation. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import platform +import secrets +import socket +import subprocess +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any +from uuid import UUID + +import httpx +from pydantic import SecretStr +from sqlalchemy import create_engine, text +from sqlalchemy.engine import make_url + +from . import metadata_fabric_active_metadata_authorization as authorization +from .active_metadata_authorization import ( + MetadataActivationAuthorization, + build_metadata_activation_authorization, +) +from .dolphinscheduler_adapter import ( + DOLPHINSCHEDULER_ADAPTER_SCHEMA, + DOLPHINSCHEDULER_API_PROFILE, + DOLPHINSCHEDULER_SERVER_VERSION, + DolphinSchedulerAdapter, + DolphinSchedulerClient, + DolphinSchedulerDefinitionBinding, + DolphinSchedulerInstance, + 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_scheduler_delivery_contract.v1" +EVIDENCE_SCHEMA = "gda.active_metadata_scheduler_delivery_evidence.v1" +IMAGE = "apache/dolphinscheduler-standalone-server:3.4.2" +IMAGE_ID = "sha256:485a1b37dd1c4088c8c8335f9fccbd229e5e703c32e21f318eb00cbb60b1af9d" +CONTAINER_PORT = 12345 +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_DEPENDENCY_EVIDENCE_PATH = authorization.DEFAULT_EVIDENCE_PATH +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-active-metadata-scheduler-delivery-2026-07-30.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-active-metadata-scheduler-delivery.sh" +) +TENANT = authorization.TENANT +SOURCE_ID = authorization.SOURCE_ID +DEFINITION_ID = UUID("a7000000-0000-4000-8000-000000000002") +RUN_ID = UUID("a7000000-0000-4000-8000-000000000003") +TASK_CODE = 170000000000001 +WORKER = "worker:active-metadata-scheduler-delivery-1" +RUNNER = "workload:metadata-projection-runner" +POLICY_EVALUATOR = "workload:metadata-policy-evaluator" +AUTHORIZER = "workload:metadata-activation-authorizer" +APPROVER = "human:metadata-governance-approver" +TERMINAL_STATES = frozenset({"SUCCESS", "FAILURE", "STOP", "PAUSE"}) +FALSE_CLAIMS = ( + "dataset_source_committed", + "dataset_absolute_path_committed", + "dataset_required_in_ci", + "deployment_applied", + "production_workload_identity_verified", + "provider_apply_authorized", + "provider_mutations_executed", + "production_scheduler_submission_verified", + "production_ingestion_verified", + "production_ready", +) + + +class ActiveMetadataSchedulerDeliveryError(RuntimeError): + """The scheduler delivery contract or local rehearsal failed closed.""" + + +@dataclass(frozen=True) +class SchedulerDefinitionBundle: + registration: DefinitionRegistration + definition: PlatformDefinitionVersion + workflow: DolphinSchedulerWorkflowSpec + + +@dataclass(frozen=True) +class SchedulerDeliveryBundle: + source_resource: Resource + source_version: ResourceVersion + request: Any + registration: Any + definition_registration: DefinitionRegistration + execution_plan: Artifact + policy_decision: Artifact + approval: Artifact + run: PlatformRun + authorization: MetadataActivationAuthorization + + +def _workflow_document() -> dict[str, Any]: + task = { + "code": TASK_CODE, + "name": "project_active_metadata_noop", + "version": 1, + "description": "Validate delivery without mutating governed metadata", + "delayTime": 0, + "taskType": "SHELL", + "taskParams": { + "localParams": [], + "rawScript": "printf '%s\\n' 'gda active metadata delivery verified'", + "resourceList": [], + "dependence": {}, + "conditionResult": {"successNode": [], "failedNode": []}, + "waitStartTimeout": {}, + }, + "flag": "YES", + "taskPriority": "MEDIUM", + "workerGroup": "default", + "environmentCode": -1, + "failRetryTimes": 0, + "failRetryInterval": 1, + "timeoutFlag": "CLOSE", + "timeoutNotifyStrategy": "WARN", + "timeout": 0, + } + relation = { + "name": "", + "preTaskCode": 0, + "preTaskVersion": 0, + "postTaskCode": TASK_CODE, + "postTaskVersion": 1, + "conditionType": "NONE", + "conditionParams": {}, + } + return { + "dolphinscheduler": { + "name": "gda_active_metadata_projection_delivery_v1", + "description": "No-side-effect Active Metadata delivery verification", + "task_definitions": [task], + "task_relations": [relation], + "locations": [{"taskCode": TASK_CODE, "x": 160, "y": 100}], + "global_params": [], + "timeout_seconds": 0, + "execution_type": "PARALLEL", + } + } + + +def build_scheduler_definition(created_at: datetime) -> SchedulerDefinitionBundle: + definition_urn = f"gda://{TENANT}/definition/metadata-projection-delivery" + definition_document = _workflow_document() + input_contract = { + "metadata_change": "gis.cultural_districts", + "delivery_mode": "no_side_effect", + } + output_contract = {"governed_provider_mutation": False} + definition_sha256 = platform_definition_fingerprint( + orchestration_class="dataops", + capability_id="metadata_fabric.projection_plan", + portability_class="provider_native", + definition_document=definition_document, + input_contract=input_contract, + output_contract=output_contract, + ) + resource = Resource( + tenant_id=TENANT, + resource_urn=definition_urn, + resource_kind="definition", + authority_system="gda", + authority_locator="definition/metadata-projection-delivery", + 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-v1", + content_sha256=definition_sha256, + authority_version_ref={ + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + }, + 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 SchedulerDefinitionBundle( + registration=DefinitionRegistration( + resource=resource, + resource_version=resource_version, + definition=definition, + ), + definition=definition, + workflow=compile_dolphinscheduler_workflow(definition), + ) + + +def build_scheduler_delivery_bundle( + content_sha256: str, + definition_bundle: SchedulerDefinitionBundle, + binding: DolphinSchedulerDefinitionBinding, + *, + authorized_at: datetime, +) -> SchedulerDeliveryBundle: + base = authorization.build_authorization_bundle(content_sha256) + source_version = base.registration.resource_version + if binding.definition_version_id != definition_bundle.definition.definition_version_id: + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler binding does not match the delivery definition" + ) + if binding.compiled_sha256 != definition_bundle.workflow.compiled_sha256: + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler binding does not match the compiled workflow" + ) + execution_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="deliver 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=execution_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), + ) + policy_decision = build_policy_decision_artifact(decision) + approval = build_approval_artifact( + ApprovalRecord( + tenant_id=TENANT, + run_id=RUN_ID, + definition_version_id=DEFINITION_ID, + policy_decision_artifact_id=policy_decision.artifact_id, + policy_decision_sha256=policy_decision.content_sha256, + verdict="approved", + approver_subject=APPROVER, + reason="approved no-side-effect scheduler delivery rehearsal", + 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:scheduler-delivery:v1", + policy_refs=RunPolicyReferences( + policy_decision_artifact_id=policy_decision.artifact_id, + approval_artifact_id=approval.artifact_id, + ), + submitted_at=authorized_at - timedelta(seconds=1), + ) + activation_authorization = build_metadata_activation_authorization( + base.request, + source_version, + definition_bundle.definition, + run, + execution_plan, + policy_decision, + approval, + authorized_by=AUTHORIZER, + authorized_at=authorized_at, + ) + return SchedulerDeliveryBundle( + source_resource=base.source_resource, + source_version=source_version, + request=base.request, + registration=base.registration, + definition_registration=definition_bundle.registration, + execution_plan=execution_plan, + policy_decision=policy_decision, + approval=approval, + run=run, + authorization=activation_authorization, + ) + + +def _load_json_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ActiveMetadataSchedulerDeliveryError( + f"{path.name} must contain an object" + ) + return value + + +def _file_record(path: Path) -> dict[str, str | None]: + relative = path.resolve().relative_to(REPO_ROOT).as_posix() + if not path.is_file(): + return {"path": relative, "sha256": None} + return {"path": relative, "sha256": hashlib.sha256(path.read_bytes()).hexdigest()} + + +def build_contract_report() -> dict[str, Any]: + errors: list[str] = [] + paths = { + "rehearsal": Path(__file__).resolve(), + "adapter": REPO_ROOT / "data_agent/dolphinscheduler_adapter.py", + "consumer": REPO_ROOT / "data_agent/dolphinscheduler_command_consumer.py", + "authorization_migration": ( + REPO_ROOT / "data_agent/migrations/101_active_metadata_authorization.sql" + ), + "dependency_evidence": DEFAULT_DEPENDENCY_EVIDENCE_PATH, + "wrapper": DEFAULT_WRAPPER_PATH, + } + required = { + "rehearsal": ( + "def run_local_rehearsal(", + "DolphinSchedulerCommandConsumer", + "provider reached terminal state; platform verdict still pending", + "local_scheduler_submission_readback_verified", + ), + "adapter": ( + "class DolphinSchedulerAdapter", + "def reconcile(", + "RunStatus.RECONCILING", + ), + "consumer": ("class DolphinSchedulerCommandConsumer", "self.adapter.dispatch("), + "authorization_migration": ( + "authorize_metadata_activation", + "Active Metadata dispatch requires exact authorization", + ), + "wrapper": ( + "data_agent.metadata_fabric_active_metadata_scheduler_delivery", + '"$@"', + ), + } + files = {name: _file_record(path) for name, path in paths.items()} + for name, markers in required.items(): + path = paths[name] + try: + source = path.read_text(encoding="utf-8") + except OSError: + errors.append(f"{name} is missing") + continue + if any(marker not in source for marker in markers): + errors.append(f"{name} is missing scheduler delivery markers") + try: + dependency = _load_json_object(DEFAULT_DEPENDENCY_EVIDENCE_PATH) + dependency_errors = authorization.validate_rehearsal_evidence(dependency) + if dependency_errors: + errors.append("M3-16 authorization dependency is invalid") + except (OSError, ValueError, ActiveMetadataSchedulerDeliveryError): + dependency = {} + errors.append("M3-16 authorization dependency is unavailable") + stable = { + "schema": CONTRACT_SCHEMA, + "provider": "apache-dolphinscheduler", + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "image": IMAGE, + "image_id": IMAGE_ID, + "adapter_schema": DOLPHINSCHEDULER_ADAPTER_SCHEMA, + "delivery_boundary": "authorized_command_consumer_submission_and_readback", + "provider_success_platform_state": "reconciling", + "governed_mutation_mode": "no_side_effect", + "dependency_evidence_sha256": dependency.get("evidence_sha256"), + "files": files, + "errors": errors, + } + return { + **stable, + "status": "valid" if not errors else "invalid", + "contract_sha256": canonical_json_fingerprint(stable), + "production_scheduler_submission_verified": False, + "provider_mutations_executed": False, + "production_ready": False, + } + + +def _wait_for_terminal_instance( + client: DolphinSchedulerClient, + instance_id: int, + workflow_definition_code: int, + *, + timeout_seconds: float, +) -> DolphinSchedulerInstance: + deadline = time.monotonic() + timeout_seconds + last: DolphinSchedulerInstance | None = None + while time.monotonic() < deadline: + last = client.get_instance(instance_id, workflow_definition_code) + if last.state.upper() in TERMINAL_STATES: + return last + time.sleep(2) + state = last.state if last is not None else "unobserved" + raise ActiveMetadataSchedulerDeliveryError( + f"DolphinScheduler instance did not reach terminal state: {state}" + ) + + +def _register_control_chain( + gateway: PlatformGateway, bundle: SchedulerDeliveryBundle +) -> 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 ActiveMetadataSchedulerDeliveryError( + "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.execution_plan, + bundle.policy_decision, + bundle.approval, + ): + 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 run_local_rehearsal( + database_url: str, + profile: DolphinSchedulerProfile, + dependency_evidence: dict[str, Any], + *, + terminal_timeout_seconds: float = 600, +) -> dict[str, Any]: + dependency_errors = authorization.validate_rehearsal_evidence( + dependency_evidence + ) + if dependency_errors: + raise ActiveMetadataSchedulerDeliveryError( + "M3-16 authorization evidence is invalid" + ) + dataset_inventory = dependency_evidence["dataset_bundle"] + if validate_shapefile_bundle_inventory(dataset_inventory): + raise ActiveMetadataSchedulerDeliveryError( + "real dataset bundle inventory is invalid" + ) + if profile.workload_subject != RUNNER: + raise ActiveMetadataSchedulerDeliveryError( + "scheduler profile workload does not match the authorized runner" + ) + if profile.policy_evaluator_subject != POLICY_EVALUATOR: + raise ActiveMetadataSchedulerDeliveryError( + "scheduler profile evaluator does not match policy evidence" + ) + + started_at = datetime.now(UTC) + definition_bundle = build_scheduler_definition(started_at) + engine = create_engine(database_url) + client = DolphinSchedulerClient(profile) + try: + binding = client.create_workflow(definition_bundle.workflow) + authorized_at = datetime.now(UTC) + bundle = build_scheduler_delivery_bundle( + dataset_inventory["content_sha256"], + definition_bundle, + binding, + authorized_at=authorized_at, + ) + authorization._apply_migrations(engine) + gateway = PlatformGateway(engine) + _register_control_chain(gateway, bundle) + first = gateway.authorize_metadata_activation(bundle.authorization) + replay = gateway.authorize_metadata_activation(bundle.authorization) + + adapter = DolphinSchedulerAdapter( + 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 ActiveMetadataSchedulerDeliveryError( + "authorized scheduler command was not completed" + ) + command = gateway.get_command(TENANT, bundle.authorization.command_id) + submitted_observation = _attempt_summary(engine) + if submitted_observation[2] != 1: + raise ActiveMetadataSchedulerDeliveryError( + "scheduler submission observation is missing" + ) + 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 = _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.execution_plan.artifact_id, + actor_subject=RUNNER, + attempt_no=1, + ) + attempts = _attempt_summary(engine) + final_run = gateway.get_run(TENANT, RUN_ID) + + verified = ( + first.created + and not replay.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 bundle.source_version.content_sha256 + == dataset_inventory["content_sha256"] + == dependency_evidence["resource_version_content_sha256"] + ) + contract = build_contract_report() + stable = { + "schema": EVIDENCE_SCHEMA, + "status": ( + "local_real_data_authorized_scheduler_delivery_verified" + if verified + else "blocked" + ), + "contract_sha256": contract["contract_sha256"], + "dependency_evidence_sha256": dependency_evidence["evidence_sha256"], + "dataset_bundle": dataset_inventory, + "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, + "execution_plan_artifact_id": str(bundle.execution_plan.artifact_id), + "execution_plan_sha256": bundle.execution_plan.content_sha256, + "run_id": str(RUN_ID), + "authorization_id": str(bundle.authorization.authorization_id), + "authorization_sha256": bundle.authorization.authorization_sha256, + "authorization_created": first.created, + "exact_authorization_replay_created": replay.created, + "command_id": str(bundle.authorization.command_id), + "command_status": command.status.value, + "command_claimed_count": consumer_result.claimed, + "command_completed_count": consumer_result.completed, + "provider": { + "name": "apache-dolphinscheduler", + "server_version": binding.server_version, + "api_profile": binding.api_profile, + "image": IMAGE, + "image_id": 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], + "provider_success_readback_verified": ( + reconciled.provider_state == "SUCCESS" + ), + "platform_run_status": final_run.status.value, + "platform_run_succeeded": final_run.status == RunStatus.SUCCEEDED, + "provider_workflow_definition_created": True, + "provider_workflow_instance_executed": True, + "no_side_effect_workflow_verified": True, + "local_scheduler_submission_readback_verified": verified, + "standalone_container_cleanup_verified": False, + "temporary_database_cleanup_verified": False, + "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, + "errors": [] if verified else ["local scheduler delivery failed"], + } + return stable + finally: + client.close() + engine.dispose() + + +def _free_loopback_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as candidate: + candidate.bind(("127.0.0.1", 0)) + return int(candidate.getsockname()[1]) + + +def _run_command(args: list[str]) -> str: + completed = subprocess.run( + args, + check=False, + capture_output=True, + text=True, + timeout=120, + ) + if completed.returncode != 0: + raise ActiveMetadataSchedulerDeliveryError( + f"local runtime command failed: {Path(args[0]).name}" + ) + return completed.stdout.strip() + + +class EphemeralPostgresDatabase: + def __init__(self, admin_url: str): + self.admin_url = admin_url + self.database_name = f"gda_m3_17_{secrets.token_hex(6)}" + self.database_url: str | None = None + self.created = False + self.cleanup_verified = False + + def __enter__(self) -> EphemeralPostgresDatabase: + admin = create_engine(self.admin_url, isolation_level="AUTOCOMMIT") + try: + with admin.connect() as connection: + connection.exec_driver_sql(f'CREATE DATABASE "{self.database_name}"') + self.created = True + self.database_url = make_url(self.admin_url).set( + database=self.database_name + ).render_as_string( + hide_password=False + ) + return self + except BaseException: + admin.dispose() + self.__exit__() + raise + finally: + admin.dispose() + + def __exit__(self, *_args: object) -> None: + if not self.created: + return + admin = create_engine(self.admin_url, isolation_level="AUTOCOMMIT") + try: + with admin.connect() as connection: + connection.execute( + text( + """ + SELECT pg_terminate_backend(pid) + FROM pg_stat_activity + WHERE datname = :database_name + AND pid <> pg_backend_pid() + """ + ), + {"database_name": self.database_name}, + ) + connection.exec_driver_sql(f'DROP DATABASE "{self.database_name}"') + exists = connection.execute( + text("SELECT 1 FROM pg_database WHERE datname = :database_name"), + {"database_name": self.database_name}, + ).first() + self.cleanup_verified = exists is None + finally: + admin.dispose() + + +class EphemeralDolphinScheduler: + def __init__(self, admin_password: SecretStr, *, readiness_timeout: float): + self.admin_password = admin_password + self.readiness_timeout = readiness_timeout + self.container_name = f"gda-m3-17-{secrets.token_hex(5)}" + self.host_port = _free_loopback_port() + self.base_url = f"http://127.0.0.1:{self.host_port}/dolphinscheduler" + self.started = False + self.cleanup_verified = False + + def __enter__(self) -> EphemeralDolphinScheduler: + try: + image_id = _run_command( + ["docker", "image", "inspect", "--format", "{{.Id}}", IMAGE] + ) + if image_id != IMAGE_ID: + raise ActiveMetadataSchedulerDeliveryError( + "local DolphinScheduler image ID does not match ADR-023" + ) + self.started = True + _run_command( + [ + "docker", + "run", + "--detach", + "--name", + self.container_name, + "--publish", + f"127.0.0.1:{self.host_port}:{CONTAINER_PORT}", + IMAGE, + ] + ) + self._wait_ready() + return self + except BaseException: + self.__exit__() + raise + + def __exit__(self, *_args: object) -> None: + if not self.started: + return + completed = subprocess.run( + ["docker", "rm", "--force", self.container_name], + check=False, + capture_output=True, + text=True, + timeout=120, + ) + if completed.returncode == 0: + remaining = subprocess.run( + [ + "docker", + "container", + "inspect", + self.container_name, + ], + check=False, + capture_output=True, + text=True, + timeout=30, + ) + self.cleanup_verified = remaining.returncode != 0 + + def _wait_ready(self) -> None: + deadline = time.monotonic() + self.readiness_timeout + health_url = f"{self.base_url}/actuator/health" + while time.monotonic() < deadline: + try: + response = httpx.get(health_url, timeout=5) + if response.status_code == 200 and response.json().get("status") == "UP": + return + except (httpx.HTTPError, ValueError): + pass + time.sleep(2) + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler standalone did not become ready" + ) + + @staticmethod + def _api_data(response: httpx.Response) -> Any: + try: + value = response.json() + except ValueError as exc: + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler provisioning returned non-JSON" + ) from exc + if response.status_code >= 400 or not isinstance(value, dict): + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler provisioning request failed" + ) + if value.get("code") != 0: + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler provisioning API rejected the request" + ) + return value.get("data") + + def provision_project(self) -> tuple[int, SecretStr]: + with httpx.Client(timeout=30) as client: + login = client.post( + f"{self.base_url}/login", + data={ + "userName": "admin", + "userPassword": self.admin_password.get_secret_value(), + }, + ) + login_data = self._api_data(login) + if isinstance(login_data, dict) and login_data.get("sessionId"): + client.cookies.set("sessionId", str(login_data["sessionId"])) + project_name = f"gda_m3_17_{secrets.token_hex(5)}" + project_data = self._api_data( + client.post( + f"{self.base_url}/projects", + data={ + "projectName": project_name, + "description": "GDA M3-17 local delivery rehearsal", + }, + ) + ) + if not isinstance(project_data, dict): + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler project response is invalid" + ) + try: + project_code = int(project_data["code"]) + except (KeyError, TypeError, ValueError) as exc: + raise ActiveMetadataSchedulerDeliveryError( + "DolphinScheduler project code is invalid" + ) from exc + access_token = secrets.token_urlsafe(32) + expires_at = datetime.now(UTC) + timedelta(days=1) + self._api_data( + client.post( + f"{self.base_url}/access-tokens", + data={ + "userId": "1", + "expireTime": expires_at.strftime("%Y-%m-%d %H:%M:%S"), + "token": access_token, + }, + ) + ) + return project_code, SecretStr(access_token) + + +def run_managed_rehearsal( + database_admin_url: str, + dependency_evidence: dict[str, Any], + admin_password: SecretStr, + *, + readiness_timeout_seconds: float = 180, + terminal_timeout_seconds: float = 600, +) -> dict[str, Any]: + database = EphemeralPostgresDatabase(database_admin_url) + scheduler = EphemeralDolphinScheduler( + admin_password, + readiness_timeout=readiness_timeout_seconds, + ) + evidence: dict[str, Any] | None = None + with database: + with scheduler: + project_code, access_token = scheduler.provision_project() + 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 ActiveMetadataSchedulerDeliveryError( + "temporary PostgreSQL database was not created" + ) + evidence = run_local_rehearsal( + database.database_url, + profile, + dependency_evidence, + terminal_timeout_seconds=terminal_timeout_seconds, + ) + if evidence is None: + raise ActiveMetadataSchedulerDeliveryError( + "local scheduler delivery produced no evidence" + ) + evidence["standalone_container_cleanup_verified"] = scheduler.cleanup_verified + evidence["temporary_database_cleanup_verified"] = database.cleanup_verified + if not scheduler.cleanup_verified or not database.cleanup_verified: + evidence["errors"].append("ephemeral runtime cleanup failed") + evidence["status"] = "blocked" + evidence["local_scheduler_submission_readback_verified"] = False + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + return {**stable, "evidence_sha256": canonical_json_fingerprint(stable)} + + +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("scheduler delivery evidence schema does not match") + if evidence.get("evidence_sha256") != canonical_json_fingerprint(stable): + errors.append("scheduler delivery evidence SHA-256 does not match") + contract = build_contract_report() + if evidence.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("scheduler delivery contract fingerprint is stale") + try: + dependency = _load_json_object(DEFAULT_DEPENDENCY_EVIDENCE_PATH) + except (OSError, ValueError, ActiveMetadataSchedulerDeliveryError): + dependency = {} + errors.append("M3-16 dependency evidence is unavailable") + if evidence.get("dependency_evidence_sha256") != dependency.get( + "evidence_sha256" + ): + errors.append("scheduler delivery dependency fingerprint is stale") + dataset = evidence.get("dataset_bundle") + if not isinstance(dataset, dict): + errors.append("scheduler delivery 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 scheduler delivery may not claim {claim}") + for claim in ( + "real_dataset_resource_version_bound", + "authorization_created", + "exact_correlation_variable_readback_verified", + "provider_success_readback_verified", + "provider_workflow_definition_created", + "provider_workflow_instance_executed", + "no_side_effect_workflow_verified", + "local_scheduler_submission_readback_verified", + "standalone_container_cleanup_verified", + "temporary_database_cleanup_verified", + ): + if evidence.get(claim) is not True: + errors.append(f"scheduler delivery did not verify {claim}") + if evidence.get("exact_authorization_replay_created") is not False: + errors.append("authorization replay must not create a row") + if evidence.get("command_status") != "done": + errors.append("authorized dispatch command must be done") + if evidence.get("command_claimed_count") != 1: + errors.append("scheduler delivery must claim one command") + if evidence.get("command_completed_count") != 1: + errors.append("scheduler delivery must complete one command") + provider = evidence.get("provider") + if not isinstance(provider, dict): + errors.append("scheduler provider evidence is missing") + else: + expected = { + "name": "apache-dolphinscheduler", + "server_version": DOLPHINSCHEDULER_SERVER_VERSION, + "api_profile": DOLPHINSCHEDULER_API_PROFILE, + "image": IMAGE, + "image_id": IMAGE_ID, + "terminal_state": "SUCCESS", + } + if any(provider.get(key) != value for key, value in expected.items()): + errors.append("scheduler provider identity or state does not match") + if evidence.get("matching_provider_instance_count") != 1: + errors.append("scheduler delivery must read back one correlated instance") + if evidence.get("attempt_observation_count") != 2: + errors.append("scheduler delivery must record two attempt observations") + if evidence.get("external_correlation_count") != 1: + errors.append("scheduler delivery must retain one external correlation") + if evidence.get("submitted_observation_count") != 1: + errors.append("scheduler delivery must record submission evidence") + if evidence.get("success_observation_count") != 1: + errors.append("scheduler delivery must record provider success evidence") + if evidence.get("attempt_states") != ["submitted", "success"]: + errors.append("scheduler delivery 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") + serialized = json.dumps(evidence, ensure_ascii=True, sort_keys=True) + for forbidden in ("/Users/", "Downloads/", ".tmp/", '"token"', '"password"', '"session"'): + if forbidden in serialized: + errors.append("scheduler delivery evidence contains sensitive local material") + break + return errors + + +def _read_admin_password(environment_name: str) -> SecretStr: + value = os.environ.get(environment_name, "") + if not value: + raise ActiveMetadataSchedulerDeliveryError( + f"{environment_name} must provide the standalone admin password" + ) + return SecretStr(value) + + +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( + "--dependency-evidence", + type=Path, + default=DEFAULT_DEPENDENCY_EVIDENCE_PATH, + ) + rehearse.add_argument("--readiness-timeout-seconds", type=float, default=180) + rehearse.add_argument("--terminal-timeout-seconds", type=float, default=600) + rehearse.add_argument("--evidence-out", type=Path, required=True) + args = parser.parse_args(argv) + + if args.command == "validate": + report = build_contract_report() + try: + report["errors"].extend( + validate_rehearsal_evidence(_load_json_object(args.evidence)) + ) + except (OSError, ValueError, ActiveMetadataSchedulerDeliveryError) as exc: + report["errors"].append( + f"scheduler delivery 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 + + dependency = _load_json_object(args.dependency_evidence) + evidence = run_managed_rehearsal( + args.database_admin_url, + dependency, + _read_admin_password(args.admin_password_env), + 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 a33653d1..8a129eef 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -898,6 +898,26 @@ def _config( ), "Protected workload identity and real scheduler submission/read-back", ), + RuntimeSpec( + "metadata_active_metadata_scheduler_delivery_rehearsal", + "active_metadata_scheduler_delivery_rehearsal", + "governed", + "evidence_durable", + "temporary PostgreSQL/DolphinScheduler + committed local evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_active_metadata_scheduler_delivery.py", + "scripts/metadata-fabric-active-metadata-scheduler-delivery.sh", + ), + ( + ( + "data_agent/metadata_fabric_active_metadata_scheduler_delivery.py", + "def run_local_rehearsal", + ), + ), + "Protected identity, deployed controller and production scheduler/provider path", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_active_metadata_scheduler_delivery.py b/data_agent/test_metadata_fabric_active_metadata_scheduler_delivery.py new file mode 100644 index 00000000..3a271142 --- /dev/null +++ b/data_agent/test_metadata_fabric_active_metadata_scheduler_delivery.py @@ -0,0 +1,97 @@ +import json +from copy import deepcopy +from datetime import UTC, datetime, timedelta + +from data_agent import metadata_fabric_active_metadata_scheduler_delivery as delivery +from data_agent.dolphinscheduler_adapter import ( + DolphinSchedulerDefinitionBinding, + compile_dolphinscheduler_workflow, + parse_dolphinscheduler_binding_artifact, +) + + +def _binding(definition_bundle): + return DolphinSchedulerDefinitionBinding( + tenant_id=delivery.TENANT, + definition_version_id=delivery.DEFINITION_ID, + project_code=180000000000001, + workflow_definition_code=180000000000002, + workflow_definition_version=1, + compiled_sha256=definition_bundle.workflow.compiled_sha256, + ) + + +def test_provider_native_delivery_bundle_binds_real_resource_and_exact_workflow(): + created_at = datetime(2026, 7, 30, 9, 0, tzinfo=UTC) + definition_bundle = delivery.build_scheduler_definition(created_at) + content_sha256 = "f" * 64 + bundle = delivery.build_scheduler_delivery_bundle( + content_sha256, + definition_bundle, + _binding(definition_bundle), + authorized_at=created_at + timedelta(minutes=1), + ) + + assert definition_bundle.definition.portability_class.value == "provider_native" + assert compile_dolphinscheduler_workflow(definition_bundle.definition) == ( + definition_bundle.workflow + ) + assert bundle.source_version.content_sha256 == content_sha256 + assert bundle.authorization.content_sha256 == content_sha256 + assert bundle.authorization.execution_plan_artifact_id == ( + bundle.execution_plan.artifact_id + ) + assert parse_dolphinscheduler_binding_artifact(bundle.execution_plan) == ( + _binding(definition_bundle) + ) + assert bundle.run.subject_context.subject_id == "metadata-projection-runner" + assert bundle.authorization.provider_mutations_executed is False + + +def test_static_contract_requires_consumer_submission_readback_and_reconciling(): + report = delivery.build_contract_report() + + assert report["status"] == "valid" + assert report["errors"] == [] + assert report["provider_success_platform_state"] == "reconciling" + assert report["governed_mutation_mode"] == "no_side_effect" + assert report["production_scheduler_submission_verified"] is False + assert report["provider_mutations_executed"] is False + assert report["production_ready"] is False + assert all(not item["path"].startswith("/") for item in report["files"].values()) + + +def test_checked_scheduler_delivery_evidence_is_current_and_fail_closed(): + evidence = json.loads(delivery.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + + assert delivery.validate_rehearsal_evidence(evidence) == [] + assert evidence["provider"]["server_version"] == "3.4.2" + assert evidence["provider"]["terminal_state"] == "SUCCESS" + assert evidence["matching_provider_instance_count"] == 1 + assert evidence["attempt_states"] == ["submitted", "success"] + assert evidence["platform_run_status"] == "reconciling" + assert evidence["platform_run_succeeded"] is False + assert evidence["provider_mutations_executed"] is False + assert evidence["production_scheduler_submission_verified"] is False + assert evidence["production_ready"] is False + serialized = json.dumps(evidence) + assert "/Users/" not in serialized + assert "Downloads/" not in serialized + assert '"token"' not in serialized + assert '"password"' not in serialized + assert '"session"' not in serialized + + +def test_scheduler_delivery_evidence_rejects_tampering_and_overclaim(): + evidence = json.loads(delivery.DEFAULT_EVIDENCE_PATH.read_text(encoding="utf-8")) + tampered = deepcopy(evidence) + tampered["matching_provider_instance_count"] = 2 + tampered["platform_run_succeeded"] = True + tampered["production_ready"] = True + + errors = delivery.validate_rehearsal_evidence(tampered) + + assert "scheduler delivery evidence SHA-256 does not match" in errors + assert "scheduler delivery must read back one correlated instance" in errors + assert "local provider success may not claim platform success" in errors + assert "local scheduler delivery may not claim production_ready" in errors diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index f4e2c1ad..91ff5e86 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -262,6 +262,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_scheduler_delivery_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-063-local-authorized-active-metadata-scheduler-delivery.md b/docs/architecture-decisions/adr-063-local-authorized-active-metadata-scheduler-delivery.md new file mode 100644 index 00000000..91e48216 --- /dev/null +++ b/docs/architecture-decisions/adr-063-local-authorized-active-metadata-scheduler-delivery.md @@ -0,0 +1,130 @@ +# ADR-063: Local authorized Active Metadata scheduler delivery + +- Status: Accepted for local AR-1 verification +- Date: 2026-07-30 +- Owners: Metadata Platform / Data Platform / Security +- Related decisions: ADR-023, ADR-024, ADR-025, ADR-026, ADR-027, ADR-060, ADR-061, ADR-062 + +## Context + +ADR-062 ends with one content-bound authorization and one pending +DolphinScheduler dispatch in the same PostgreSQL transaction. That proves the +promotion boundary but not that the existing provider adapter and leased +command consumer can deliver this exact command to a real scheduler, recover +the correlation variables, and project provider state back into GDA Control. + +The next slice must close that local control-loop gap without turning a +standalone development server into production evidence. It must also preserve +the platform terminal-state rule: scheduler `SUCCESS` is attempt evidence, not +permission to mark a `PlatformRun` succeeded. + +## Options considered + +| Option | Benefit | Cost / risk | Decision | +|---|---|---|---| +| Add an HTTP mock around the adapter | Fast and deterministic | Repeats unit coverage and proves no provider compatibility | Rejected | +| Execute the metadata projection against OpenMetadata/Gravitino | Demonstrates a business mutation | Conflates scheduler delivery with provider authorization, rollback, and ingestion | Rejected | +| Submit a no-side-effect workflow to official local DolphinScheduler and read it back | Proves the exact adapter/consumer/provider boundary while keeping mutation scope closed | Requires ephemeral scheduler/database provisioning and bounded polling | Accepted | + +## Decision + +M3-17 uses the official ARM64 +`apache/dolphinscheduler-standalone-server:3.4.2` image with ID +`sha256:485a1b37dd1c4088c8c8335f9fccbd229e5e703c32e21f318eb00cbb60b1af9d`. +The rehearsal creates an ephemeral standalone container, project, API token, +workflow, and PostgreSQL database. Token and login session values remain only +in memory and are never written to evidence. + +The checked M3-16 evidence supplies the path-free Chongqing dataset inventory +and source ResourceVersion content fingerprint. M3-17 constructs a new +provider-native `PlatformDefinitionVersion`, compiles a single Shell task that +only emits a log line, publishes and releases the workflow, and converts the +returned project/code/version/compiled fingerprint into the immutable +DolphinScheduler binding Artifact. + +That exact binding is referenced by a new accepted Run, allow PolicyDecision, +independent ApprovalRecord, and independent +`MetadataActivationAuthorization`. Migration 101 and the existing gateway then +atomically append the authorization and pending command. Exact authorization +replay creates nothing. + +The existing `DolphinSchedulerCommandConsumer` claims the command and calls the +existing `DolphinSchedulerAdapter`; the rehearsal does not introduce another +queue or scheduler. Provider terminal-state polling is bounded by an explicit +timeout. The provider instance must return every controlled GDA definition and +Run correlation variable exactly, and a correlation scan must find exactly one +matching instance. + +The adapter records one `submitted` and one `success` attempt observation with +one external correlation. Reconciliation of provider `SUCCESS` moves the Run +to `reconciling`. Only the separate evidence-gated platform success finalizer +may later produce `succeeded`; M3-17 deliberately does not call it. + +## Real data boundary + +The same Chongqing central-city historical cultural district bundle remains the +acceptance input: 8 Shapefile components, 20 `PolygonZ` features, 33 fields, +EPSG:4490, and content fingerprint +`fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007`. +The source path and source files are not committed, and CI validates only the +checked inventory and evidence. + +Real data is required here because a synthetic hash would not prove continuity +from discovered spatial content through ResourceVersion, activation request, +authorization, Run, scheduler correlation, and attempt evidence. Reopening the +raw Shapefile is unnecessary for every downstream slice once that content +inventory is checked, hashed, and validated as a dependency. + +DEM, roads, CLCD, buildings, population, POI/AOI, commuting, search index, TAP, +and township boundaries remain valuable for later raster, temporal, scale, and +multi-source conformance. They should be introduced only when the next claim +needs their specific structure; loading the entire corpus into every control +plane rehearsal would add cost without increasing the claim strength. + +## Consequences and trade-offs + +- The local workflow definition and workflow instance are real scheduler + control-plane mutations. `provider_mutations_executed=false` refers narrowly + to governed metadata/data providers: no OpenMetadata, Gravitino, lakehouse, + legacy, or source dataset mutation was authorized or executed. +- Standalone uses its development identity and embedded runtime. It proves API + compatibility, not protected workload identity, tenant isolation, durable + scheduler metadata, HA, upgrade, backup/restore, or production capacity. +- Readiness and terminal state use bounded polling instead of fixed delays. + Host load protection can delay command consumption, so the explicit terminal + timeout is ten minutes. +- The container and temporary database are removed on success and exception; + cleanup is part of checked evidence. +- Workflow creation precedes binding-bound authorization. A production + definition publisher must be separated from the protected promotion + controller and must provide durable provider identity evidence. + +## Local verification + +The final rehearsal produced one authorization, one claimed/completed dispatch, +one successful provider instance, two observations (`submitted`, `success`), +one external correlation, and exact read-back of six controlled GDA variables. +The final PlatformRun status was `reconciling`, never `succeeded`. Both the +standalone container and temporary PostgreSQL database were absent after the +run. + +- Contract fingerprint: + `dcf97c8fa002e9fe6b6bc3a7603ee2ebd5ddb053544801ce35143a095e648edb` +- Evidence fingerprint: + `00d4ea062c40f8d97557eadc357a36c6d1ccd56e12a94a44694113681e5d55f4` +- M3-16 dependency fingerprint: + `6ae387240e3bcebaafe2ad7acc73f4e09d53df2e73b2ec63cd92edbc262d831e` + +`deployment_applied`, `production_workload_identity_verified`, +`provider_apply_authorized`, `provider_mutations_executed`, +`production_scheduler_submission_verified`, `production_ingestion_verified`, +and `production_ready` remain `false`. + +## Revisit triggers + +Revisit when a protected identity can publish and submit the same immutable +binding in staging, when DolphinScheduler metadata moves to an independently +backed-up PostgreSQL service, when worker lease/restart and callback failure are +injected against the deployed consumer, or when a separately authorized +idempotent metadata projection is ready to prove provider mutation and +read-back without weakening the platform success gate. diff --git a/docs/evidence/metadata-fabric-active-metadata-scheduler-delivery-2026-07-30.json b/docs/evidence/metadata-fabric-active-metadata-scheduler-delivery-2026-07-30.json new file mode 100644 index 00000000..20e417c5 --- /dev/null +++ b/docs/evidence/metadata-fabric-active-metadata-scheduler-delivery-2026-07-30.json @@ -0,0 +1,140 @@ +{ + "attempt_observation_count": 2, + "attempt_states": [ + "submitted", + "success" + ], + "authorization_created": true, + "authorization_id": "9d9e985b-b698-5aa3-b3c2-ce9f65c85310", + "authorization_sha256": "4fca1b84ee1fd9056054e91ef25123c80cae08b959ef4c39b276058a788acda8", + "command_claimed_count": 1, + "command_completed_count": 1, + "command_id": "6db53d29-a37e-5f4a-b222-5df5ab837e58", + "command_status": "done", + "compiled_workflow_sha256": "66ce899d55caaa668da20a7b189fb7b64992c2e04a38d74c60a39e549b8792bd", + "contract_sha256": "dcf97c8fa002e9fe6b6bc3a7603ee2ebd5ddb053544801ce35143a095e648edb", + "correlation_variables": { + "gda_definition_sha256": "96d77bb34e1cfbb802fa6b34782e39ed5b1a2256ee543c3ce81052833334a37c", + "gda_definition_urn": "gda://metadata-authorization-local/definition/metadata-projection-delivery", + "gda_definition_version_id": "a7000000-0000-4000-8000-000000000002", + "gda_idempotency_key": "metadata-projection:cultural-districts:scheduler-delivery:v1", + "gda_run_id": "a7000000-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": "96d77bb34e1cfbb802fa6b34782e39ed5b1a2256ee543c3ce81052833334a37c", + "definition_version_id": "a7000000-0000-4000-8000-000000000002", + "dependency_evidence_sha256": "6ae387240e3bcebaafe2ad7acc73f4e09d53df2e73b2ec63cd92edbc262d831e", + "deployment_applied": false, + "errors": [], + "evidence_sha256": "00d4ea062c40f8d97557eadc357a36c6d1ccd56e12a94a44694113681e5d55f4", + "exact_authorization_replay_created": false, + "exact_correlation_variable_readback_verified": true, + "execution_plan_artifact_id": "88a4dbdc-19b8-5155-b67b-7f337506e026", + "execution_plan_sha256": "8f111f84cac5b4146586107f289f1c45162941c928bd6a13b81b9ff9c6c0f65c", + "external_correlation_count": 1, + "local_scheduler_submission_readback_verified": true, + "matching_provider_instance_count": 1, + "no_side_effect_workflow_verified": true, + "platform_run_status": "reconciling", + "platform_run_succeeded": false, + "production_ingestion_verified": false, + "production_ready": false, + "production_scheduler_submission_verified": false, + "production_workload_identity_verified": false, + "provider": { + "api_profile": "3.4", + "architecture": "arm64", + "image": "apache/dolphinscheduler-standalone-server:3.4.2", + "image_id": "sha256:485a1b37dd1c4088c8c8335f9fccbd229e5e703c32e21f318eb00cbb60b1af9d", + "name": "apache-dolphinscheduler", + "project_code": 180200967129920, + "server_version": "3.4.2", + "terminal_state": "SUCCESS", + "workflow_definition_code": 180200967308096, + "workflow_definition_version": 1, + "workflow_instance_id": 1 + }, + "provider_apply_authorized": false, + "provider_mutations_executed": false, + "provider_success_readback_verified": true, + "provider_workflow_definition_created": true, + "provider_workflow_instance_executed": true, + "real_dataset_resource_version_bound": true, + "resource_version_content_sha256": "fd474fd65c8e4a71da241eb3fd07748ca3b972fbd2d3c32833376dbe71104007", + "resource_version_id": "a6000000-0000-4000-8000-000000000001", + "run_id": "a7000000-0000-4000-8000-000000000003", + "schema": "gda.active_metadata_scheduler_delivery_evidence.v1", + "standalone_container_cleanup_verified": true, + "status": "local_real_data_authorized_scheduler_delivery_verified", + "submitted_observation_count": 1, + "success_observation_count": 1, + "temporary_database_cleanup_verified": true +} diff --git a/docs/roadmap-ar0-platform-truth-2026-07-24.md b/docs/roadmap-ar0-platform-truth-2026-07-24.md index 90091163..4ce84238 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-16（本地真实数据授权/dispatch 原子提升已验证，生产验证待执行） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-17（本地真实数据授权与 scheduler delivery/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： @@ -236,8 +236,9 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 31. [ADR-060](architecture-decisions/adr-060-transactional-active-metadata-change-outbox.md) 已新增 migration 099、内容绑定 `MetadataChangeEvent`、deterministic activation intent 与 PlatformGateway 原子注册/claim/fail/complete API。真实 PostgreSQL 16 演练中，ResourceVersion 与事件同事务创建，精确 replay 与 processed replay 均不新增事件；错误 consumer/worker 被拒绝，一次 retry 和一次强制租约过期后由第三个 worker 完成，旧 ResourceVersion 的补事件尝试整笔回滚。最终只有 1 条权威事件、3 次 attempt，FORCE RLS、跨租户拒绝与 gateway 无直接 UPDATE/DELETE 均通过。因共享 contract/gateway 源码演进后已在 fresh database 重跑，当前 contract fingerprint 为 `c3d94228456aff7e9b134fa6bc746bbe6b7485950c16f8ec41ec34fc7a5ae567`，evidence fingerprint 为 `2b8a408e078cec44fde9a6d63e4b94f988dc820668c87ac0d2ef0a434d1a16a3`。该切片只生成 `metadata_fabric.projection_plan` 意图，不新增常驻 consumer、不提交 DolphinScheduler、不授权或执行 provider mutation，也不证明 production ingestion。 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`。 -此处 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 原子性，不包含受保护 workload identity、已部署 authorization controller、真实 DolphinScheduler submission/read-back 或 provider mutation。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，未部署受保护 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`。 ## 5. 重新评估条件 diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index 2fcb368a..e842e914 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 已验证，生产 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 已验证，生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-active-metadata-authorization` +适用分支：`feat/ar1-metadata-fabric-active-metadata-scheduler-delivery` ## 判定规则 @@ -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 rehearsal 均为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization evidence | PlatformRun ledger 唯一登记最终状态；activation request 只拥有待授权意图；M3-16 authorization 只授权本地 pending dispatch，不是 provider mutation 或生产 scheduler submission 权威；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 consumer 与 M3-16 authorization 本地验证；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 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 控制链待接入 | | 原始文件/对象 | 当前 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 同事务提交 | command/metadata delivery status、消费者 claim、activation intent/request/authorization、worker status JSON、WebSocket 消息 | command/event 与源事实同事务入 outbox，幂等 consumer 交付；Active Metadata consumer 不能授权或执行；activation capability 的普通 dispatch 被拒绝，只有 append-only authorization + deferred command FK + trigger 可创建命令 | Platform/Integrations/Metadata Platform | AR-1 command worker、M3-14/M3-15/M3-16 本地已验证 -> protected authorizer、真实 scheduler submission/read-back 与 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 提交并回读 | 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 待执行 | | 质量结果 | `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 待执行 | @@ -60,6 +60,7 @@ 15. M3-14 只证明本地 PostgreSQL 16 中 ResourceVersion 与 `resource_version.registered` 事件同事务创建，以及 tenant/workload scoped claim、retry、lease reclaim 和 complete；它没有部署常驻 consumer、没有提交 DolphinScheduler、没有取得 provider apply 授权、没有执行 provider mutation，也不构成 production Active Metadata readiness。 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 仍未验证。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -99,6 +100,7 @@ - Metadata Fabric M3-14 已新增内容绑定 `MetadataChangeEvent`、migration 099 transactional outbox 与 PlatformGateway 原子注册/claim/fail/complete。真实 PostgreSQL 16 演练验证首次注册 `created=true`、pending/processed 精确 replay 均 `created=false`、错误 consumer/worker 拒绝、retry、lease-expiry reclaim、第三次 attempt 完成、processed 不再认领、旧 ResourceVersion 补事件整笔回滚、跨租户不可见、FORCE RLS 和 gateway 无直接 UPDATE/DELETE；最终只有 1 条权威事件。共享 contract/gateway 源码演进后已在 fresh database 重跑，当前 contract fingerprint 为 `c3d94228456aff7e9b134fa6bc746bbe6b7485950c16f8ec41ec34fc7a5ae567`，evidence fingerprint 为 `2b8a408e078cec44fde9a6d63e4b94f988dc820668c87ac0d2ef0a434d1a16a3`。激活意图只路由到 `metadata_fabric.projection_plan`，`provider_apply_authorized=false`、`provider_mutations_executed=false`、`production_ingestion_verified=false`、`production_scheduler_submission_verified=false`、`production_ready=false`。 - 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`。 ## 下一验收证据 @@ -106,7 +108,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 read-back、重试/死信告警与生产 SLO 证据； +- Active Metadata consumer/authorizer 的 production scale-up、受保护 workload identity、生产 DolphinScheduler submission/read-back、幂等 projection execution、受治理 provider mutation/read-back、重试/死信告警与生产 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-scheduler-delivery.sh b/scripts/metadata-fabric-active-metadata-scheduler-delivery.sh new file mode 100755 index 00000000..b99f9712 --- /dev/null +++ b/scripts/metadata-fabric-active-metadata-scheduler-delivery.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_scheduler_delivery "$@"